Host-held adapter state, and fix pooled adapters sharing one configuration

SW.Serverless.Contract/Protos/adapter.proto

@@ -26,6 +26,7 @@ message HostFrame {
         Reset reset = 7;
         Shutdown shutdown = 8;
         EventAck event_ack = 9;
+        StateResult state_result = 10;
     }
 }
 
@@ -60,6 +61,13 @@ message Shutdown {
     bool drain = 2;           // true: stop fetching, finish in-flight, then exit
 }
 
+// Answers a StateRequest. Correlated by frame id, exactly as EventAck answers an Event.
+message StateResult {
+    bool found = 1;           // false for a get with nothing stored under that name
+    string value = 2;
+    Error error = 3;
+}
+
 message EventAck {
     bool accepted = 1;
     string reference = 2;     // the host's id for what it persisted, e.g. an Xchange id
@@ -78,6 +86,7 @@ message AdapterFrame {
         LogEntry log = 6;
         Metric metric = 7;
         Pong pong = 8;
+        StateRequest state = 9;
     }
 }
 
@@ -139,6 +148,25 @@ message Event {
     string endpoint = 5;      // which queue / topic / folder this came from
 }
 
+// Small, durable, host-held key/value state belonging to this adapter instance.
+//
+// It exists because an adapter must not be the system of record for its own progress: it is
+// restarted by the supervisor, it may run on a different node next time, and in the pooled shape
+// it is not even the same process twice. A polling receiver's cursor is the motivating case —
+// the equivalent of Airbyte's `state` argument. Deliberately small: this is a bookmark, not a
+// data store.
+message StateRequest {
+    Op op = 1;
+    string name = 2;
+    string value = 3;         // set only; ignored otherwise
+
+    enum Op {
+        GET = 0;
+        SET = 1;
+        DELETE = 2;
+    }
+}
+
 message LogEntry {
     int32 level = 1;          // Trace 0 .. Critical 5, matching ILogger
     string message = 2;

SW.Serverless.Samples.Ticker/Handler.cs

@@ -136,6 +136,26 @@ public Task<object> SetInterval(int seconds)
         public Task<object> GetCounters() =>
             Task.FromResult<object>(new { produced, accepted, rejected, lastMessageOn });
 
+        /// <summary>
+        /// Host-held state, the way a polling receiver keeps its cursor: written through the host
+        /// so that it survives a restart and is still there when the next instance comes up —
+        /// possibly on another node, possibly as a different process entirely.
+        /// </summary>
+        public async Task<object> SaveCursor(string value)
+        {
+            await context.SetStateAsync("cursor", value);
+            return new { saved = value };
+        }
+
+        public async Task<object> ReadCursor() =>
+            new { cursor = await context.GetStateAsync("cursor") };
+
+        public async Task<object> ClearCursor()
+        {
+            await context.SetStateAsync("cursor", null);
+            return new { cleared = true };
+        }
+
         /// <summary>Demonstrates that a command failure comes back as a typed error, not a hang.</summary>
         public Task Explode() => throw new InvalidOperationException("Deliberate failure from the ticker sample.");
 

SW.Serverless.Sdk/Resident/IAdapterContext.cs

@@ -44,6 +44,27 @@ Task<PublishResult> PublishAsync(
             string contentType = null,
             CancellationToken cancellationToken = default);
 
+        /// <summary>
+        /// Reads a small piece of durable state the HOST holds on this adapter's behalf. Null when
+        /// nothing is stored under that name.
+        ///
+        /// An adapter must not be the system of record for its own progress: the supervisor
+        /// restarts it, the next instance may be on another node, and a pooled one is not even the
+        /// same process twice. A polling receiver's cursor is the case this exists for — the same
+        /// role Airbyte's `state` argument plays. Keep it to a bookmark; it is not a data store,
+        /// and the host is entitled to refuse a large value.
+        /// </summary>
+        Task<string> GetStateAsync(string name, CancellationToken cancellationToken = default);
+
+        /// <summary>
+        /// Writes that state, durably, before returning. A null value deletes it.
+        ///
+        /// Call it at the point the progress is real — after the rows it describes have been
+        /// accepted by the host — because anything written earlier is a promise the next instance
+        /// will believe.
+        /// </summary>
+        Task SetStateAsync(string name, string value, CancellationToken cancellationToken = default);
+
         void Log(AdapterLogLevel level, string message, Exception exception = null,
             IDictionary<string, string> properties = null);
 

SW.Serverless.Sdk/Resident/ResidentRunner.cs

@@ -51,6 +51,7 @@ public sealed class ResidentRunner : IAdapterContext
             });
 
         readonly ConcurrentDictionary<long, TaskCompletionSource<EventAck>> pendingEvents = new();
+        readonly ConcurrentDictionary<long, TaskCompletionSource<StateResult>> pendingState = new();
         readonly CancellationTokenSource stopping = new();
 
         Handshake handshake;
@@ -249,6 +250,11 @@ async Task ReadLoopAsync(IAsyncStreamReader<HostFrame> stream)
                             tcs.TrySetResult(frame.EventAck);
                         break;
 
+                    case HostFrame.BodyOneofCase.StateResult:
+                        if (pendingState.TryRemove(frame.Id, out var stateTcs))
+                            stateTcs.TrySetResult(frame.StateResult);
+                        break;
+
                     case HostFrame.BodyOneofCase.SetLogLevel:
                         MinimumLogLevel = (AdapterLogLevel)frame.SetLogLevel.Level;
                         break;
@@ -533,6 +539,66 @@ public async Task<PublishResult> PublishAsync(
             }
         }
 
+        public async Task<string> GetStateAsync(string name, CancellationToken cancellationToken = default)
+        {
+            var result = await StateAsync(
+                new StateRequest { Op = StateRequest.Types.Op.Get, Name = Named(name) }, cancellationToken);
+
+            return result.Found ? result.Value : null;
+        }
+
+        public async Task SetStateAsync(string name, string value, CancellationToken cancellationToken = default)
+        {
+            var request = value == null
+                ? new StateRequest { Op = StateRequest.Types.Op.Delete, Name = Named(name) }
+                : new StateRequest { Op = StateRequest.Types.Op.Set, Name = Named(name), Value = value };
+
+            await StateAsync(request, cancellationToken);
+        }
+
+        static string Named(string name) =>
+            string.IsNullOrWhiteSpace(name)
+                ? throw new ArgumentException("A state name is required.", nameof(name))
+                : name;
+
+        /// <summary>
+        /// One state round trip. Deliberately NOT bounded by the in-flight window that guards
+        /// PublishAsync: that window exists to stop an adapter flooding the host with messages it
+        /// must persist, and a receiver saving its cursor after a batch would then be queued behind
+        /// the very events whose progress it is recording.
+        /// </summary>
+        async Task<StateResult> StateAsync(StateRequest request, CancellationToken cancellationToken)
+        {
+            var id = Interlocked.Increment(ref nextId);
+            var tcs = new TaskCompletionSource<StateResult>(TaskCreationOptions.RunContinuationsAsynchronously);
+            pendingState[id] = tcs;
+
+            try
+            {
+                Send(new AdapterFrame { Id = id, State = request });
+
+                using (cancellationToken.Register(() => tcs.TrySetCanceled()))
+                using (stopping.Token.Register(() => tcs.TrySetCanceled()))
+                {
+                    var result = await tcs.Task;
+
+                    // Surfaced rather than swallowed: a cursor that silently failed to save is a
+                    // batch that will be replayed, and the adapter is the only thing in a position
+                    // to stop rather than carry on.
+                    if (result.Error != null && !string.IsNullOrEmpty(result.Error.Message))
+                        throw new InvalidOperationException(
+                            $"The host could not {request.Op.ToString().ToLowerInvariant()} state "
+                            + $"'{request.Name}': {result.Error.Message}");
+
+                    return result;
+                }
+            }
+            finally
+            {
+                pendingState.TryRemove(id, out _);
+            }
+        }
+
         public void Log(AdapterLogLevel level, string message, Exception exception = null,
             IDictionary<string, string> properties = null)
         {

SW.Serverless.UnitTests/ResidentAdapterTests.cs

@@ -295,6 +295,95 @@ await WaitFor(() => adapters.Describe()
 
         // ------------------------------------------------------------------ helpers
 
+        // ------------------------------------------------------------------ host-held state
+
+        /// <summary>
+        /// The property a polling receiver's cursor depends on: state written through the host
+        /// outlives the adapter process, so a restart resumes where it left off instead of
+        /// replaying from the beginning.
+        /// </summary>
+        [TestMethod]
+        public async Task State_survives_a_restart_of_the_adapter()
+        {
+            var instance = await StartTicker("state");
+
+            await instance.InvokeAsync<object>("SaveCursor", "2026-09-08T10:00:00Z");
+
+            var before = await instance.InvokeAsync<Dictionary<string, string>>("ReadCursor");
+            Assert.AreEqual("2026-09-08T10:00:00Z", before["cursor"]);
+
+            // Same instance key, a brand new process — which is exactly what the supervisor does
+            // after a crash.
+            var restarted = await adapters.RestartAsync(TickerId, "state", drain: false);
+
+            var after = await restarted.InvokeAsync<Dictionary<string, string>>("ReadCursor");
+            Assert.AreEqual("2026-09-08T10:00:00Z", after["cursor"],
+                "the cursor is the host's, so a new process must still see it");
+
+            await restarted.InvokeAsync<object>("ClearCursor");
+            var cleared = await restarted.InvokeAsync<Dictionary<string, string>>("ReadCursor");
+            Assert.IsNull(cleared["cursor"]);
+
+            await adapters.StopAsync(TickerId, "state", drain: false);
+        }
+
+        /// <summary>
+        /// State belongs to the INSTANCE, not to the adapter. Two data sources served by one
+        /// adapter are two connections, and one's cursor read by the other would skip rows.
+        /// </summary>
+        [TestMethod]
+        public async Task State_is_scoped_to_the_instance()
+        {
+            var first = await StartTicker("state-a");
+            var second = await StartTicker("state-b");
+
+            await first.InvokeAsync<object>("SaveCursor", "a");
+
+            var read = await second.InvokeAsync<Dictionary<string, string>>("ReadCursor");
+            Assert.IsNull(read["cursor"], "one instance must not see another instance's state");
+
+            await adapters.StopAsync(TickerId, "state-a", drain: false);
+            await adapters.StopAsync(TickerId, "state-b", drain: false);
+        }
+
+        // ------------------------------------------------------------------ pool keying
+
+        /// <summary>
+        /// Two data sources on one adapter id must not share warm processes: the pool captures the
+        /// spec of whichever renter created it, credentials included, so sharing a key meant the
+        /// second data source silently ran against the first one's system.
+        /// </summary>
+        [TestMethod]
+        public void Pool_key_separates_specs_that_differ_in_configuration()
+        {
+            var one = new AdapterSpec
+            {
+                AdapterId = "db.oracle",
+                StartupValues = { ["Host"] = "one.example", ["Password"] = "s1" }
+            };
+            var two = new AdapterSpec
+            {
+                AdapterId = "db.oracle",
+                StartupValues = { ["Host"] = "two.example", ["Password"] = "s2" }
+            };
+            var alsoOne = new AdapterSpec
+            {
+                AdapterId = "db.oracle",
+                // Same pairs, written in the other order: the key is canonical, so these share.
+                StartupValues = { ["Password"] = "s1", ["Host"] = "one.example" }
+            };
+
+            Assert.AreNotEqual(ResidentAdapterHost.PoolKeyOf(one), ResidentAdapterHost.PoolKeyOf(two));
+            Assert.AreEqual(ResidentAdapterHost.PoolKeyOf(one), ResidentAdapterHost.PoolKeyOf(alsoOne));
+
+            // An explicit key wins, and nothing secret is readable in either form.
+            var keyed = new AdapterSpec { AdapterId = "db.oracle", PoolKey = "datasource-7" };
+            Assert.AreEqual("db.oracle:datasource-7", ResidentAdapterHost.PoolKeyOf(keyed));
+            StringAssert.Contains(ResidentAdapterHost.PoolKeyOf(one), "db.oracle:");
+            Assert.IsFalse(ResidentAdapterHost.PoolKeyOf(one).Contains("s1"),
+                "the pool key ends up in logs, so it must not carry credentials");
+        }
+
         static Task<ResidentAdapterInstance> StartTicker(string key, int intervalSeconds = 30) =>
             adapters.StartExclusiveAsync(new AdapterSpec
             {

SW.Serverless/Extensions/IServiceCollectionExtensions.cs

@@ -38,6 +38,11 @@ public static IServiceCollection AddResidentAdapters<TSink>(this IServiceCollect
 
             services.AddSingleton(options);
             services.TryAddSingleton<IAdapterEventSink, TSink>();
+
+            // Replaceable, and a real deployment must replace it: the in-memory store is per
+            // process, so a cursor saved on one node is invisible to the next one to run the
+            // adapter. TryAdd, so a host that registered its own keeps it.
+            services.TryAddSingleton<IAdapterStateStore, InMemoryAdapterStateStore>();
             services.TryAddSingleton<AdapterInstaller>();
             services.TryAddSingleton<IResidentAdapterLocator, DefaultResidentAdapterLocator>();
             services.AddSingleton<ResidentAdapterHost>();
@@ -46,5 +51,19 @@ public static IServiceCollection AddResidentAdapters<TSink>(this IServiceCollect
 
             return services;
         }
+
+        /// <summary>
+        /// As <see cref="AddResidentAdapters{TSink}"/>, with the host's own durable state store —
+        /// what a polling receiver's cursor is written to. Use this one anywhere state has to
+        /// outlive the process or be visible to another node.
+        /// </summary>
+        public static IServiceCollection AddResidentAdapters<TSink, TStateStore>(
+            this IServiceCollection services, Action<ResidentOptions> configure = null)
+            where TSink : class, IAdapterEventSink
+            where TStateStore : class, IAdapterStateStore
+        {
+            services.TryAddSingleton<IAdapterStateStore, TStateStore>();
+            return services.AddResidentAdapters<TSink>(configure);
+        }
     }
 }

SW.Serverless/Resident/AdapterSpec.cs

@@ -22,6 +22,16 @@ public class AdapterSpec
         /// <summary>Executable to launch. Defaults to `dotnet`; set for self-contained or non-.NET adapters.</summary>
         public string Executable { get; set; }
 
+        /// <summary>
+        /// Which warm pool a POOLED rental belongs to. Left null the host derives one from the
+        /// adapter id and a hash of <see cref="StartupValues"/>, so that two configurations of the
+        /// same adapter never share processes. Set it when the caller has a better name for the
+        /// grouping — a data source id, say — than the settings happen to hash to.
+        ///
+        /// Ignored for an exclusive instance, which is keyed by <see cref="InstanceKey"/>.
+        /// </summary>
+        public string PoolKey { get; set; }
+
         /// <summary>Configuration and credentials. Sent over the stream, never on argv.</summary>
         public IDictionary<string, string> StartupValues { get; set; } = new Dictionary<string, string>();