diff --git a/SW.Serverless.Contract/Protos/adapter.proto b/SW.Serverless.Contract/Protos/adapter.proto index bf83704..2655bb8 100644 --- a/SW.Serverless.Contract/Protos/adapter.proto +++ b/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; diff --git a/SW.Serverless.Samples.Ticker/Handler.cs b/SW.Serverless.Samples.Ticker/Handler.cs index a6c80bd..d949a30 100644 --- a/SW.Serverless.Samples.Ticker/Handler.cs +++ b/SW.Serverless.Samples.Ticker/Handler.cs @@ -136,6 +136,26 @@ public Task SetInterval(int seconds) public Task GetCounters() => Task.FromResult(new { produced, accepted, rejected, lastMessageOn }); + /// + /// 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. + /// + public async Task SaveCursor(string value) + { + await context.SetStateAsync("cursor", value); + return new { saved = value }; + } + + public async Task ReadCursor() => + new { cursor = await context.GetStateAsync("cursor") }; + + public async Task ClearCursor() + { + await context.SetStateAsync("cursor", null); + return new { cleared = true }; + } + /// Demonstrates that a command failure comes back as a typed error, not a hang. public Task Explode() => throw new InvalidOperationException("Deliberate failure from the ticker sample."); diff --git a/SW.Serverless.Sdk/Resident/IAdapterContext.cs b/SW.Serverless.Sdk/Resident/IAdapterContext.cs index 8f662a1..052c7cb 100644 --- a/SW.Serverless.Sdk/Resident/IAdapterContext.cs +++ b/SW.Serverless.Sdk/Resident/IAdapterContext.cs @@ -44,6 +44,27 @@ Task PublishAsync( string contentType = null, CancellationToken cancellationToken = default); + /// + /// 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. + /// + Task GetStateAsync(string name, CancellationToken cancellationToken = default); + + /// + /// 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. + /// + Task SetStateAsync(string name, string value, CancellationToken cancellationToken = default); + void Log(AdapterLogLevel level, string message, Exception exception = null, IDictionary properties = null); diff --git a/SW.Serverless.Sdk/Resident/ResidentRunner.cs b/SW.Serverless.Sdk/Resident/ResidentRunner.cs index 5f0a305..dc525d1 100644 --- a/SW.Serverless.Sdk/Resident/ResidentRunner.cs +++ b/SW.Serverless.Sdk/Resident/ResidentRunner.cs @@ -51,6 +51,7 @@ public sealed class ResidentRunner : IAdapterContext }); readonly ConcurrentDictionary> pendingEvents = new(); + readonly ConcurrentDictionary> pendingState = new(); readonly CancellationTokenSource stopping = new(); Handshake handshake; @@ -249,6 +250,11 @@ async Task ReadLoopAsync(IAsyncStreamReader 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 PublishAsync( } } + public async Task 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; + + /// + /// 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. + /// + async Task StateAsync(StateRequest request, CancellationToken cancellationToken) + { + var id = Interlocked.Increment(ref nextId); + var tcs = new TaskCompletionSource(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 properties = null) { diff --git a/SW.Serverless.UnitTests/ResidentAdapterTests.cs b/SW.Serverless.UnitTests/ResidentAdapterTests.cs index ba8e07f..a793c44 100644 --- a/SW.Serverless.UnitTests/ResidentAdapterTests.cs +++ b/SW.Serverless.UnitTests/ResidentAdapterTests.cs @@ -295,6 +295,95 @@ await WaitFor(() => adapters.Describe() // ------------------------------------------------------------------ helpers + // ------------------------------------------------------------------ host-held state + + /// + /// 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. + /// + [TestMethod] + public async Task State_survives_a_restart_of_the_adapter() + { + var instance = await StartTicker("state"); + + await instance.InvokeAsync("SaveCursor", "2026-09-08T10:00:00Z"); + + var before = await instance.InvokeAsync>("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>("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("ClearCursor"); + var cleared = await restarted.InvokeAsync>("ReadCursor"); + Assert.IsNull(cleared["cursor"]); + + await adapters.StopAsync(TickerId, "state", drain: false); + } + + /// + /// 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. + /// + [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("SaveCursor", "a"); + + var read = await second.InvokeAsync>("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 + + /// + /// 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. + /// + [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 StartTicker(string key, int intervalSeconds = 30) => adapters.StartExclusiveAsync(new AdapterSpec { diff --git a/SW.Serverless/Extensions/IServiceCollectionExtensions.cs b/SW.Serverless/Extensions/IServiceCollectionExtensions.cs index 5f5af8e..7924dd7 100644 --- a/SW.Serverless/Extensions/IServiceCollectionExtensions.cs +++ b/SW.Serverless/Extensions/IServiceCollectionExtensions.cs @@ -38,6 +38,11 @@ public static IServiceCollection AddResidentAdapters(this IServiceCollect services.AddSingleton(options); services.TryAddSingleton(); + + // 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(); services.TryAddSingleton(); services.TryAddSingleton(); services.AddSingleton(); @@ -46,5 +51,19 @@ public static IServiceCollection AddResidentAdapters(this IServiceCollect return services; } + + /// + /// As , 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. + /// + public static IServiceCollection AddResidentAdapters( + this IServiceCollection services, Action configure = null) + where TSink : class, IAdapterEventSink + where TStateStore : class, IAdapterStateStore + { + services.TryAddSingleton(); + return services.AddResidentAdapters(configure); + } } } diff --git a/SW.Serverless/Resident/AdapterSpec.cs b/SW.Serverless/Resident/AdapterSpec.cs index 25a3973..8c1e5ca 100644 --- a/SW.Serverless/Resident/AdapterSpec.cs +++ b/SW.Serverless/Resident/AdapterSpec.cs @@ -22,6 +22,16 @@ public class AdapterSpec /// Executable to launch. Defaults to `dotnet`; set for self-contained or non-.NET adapters. public string Executable { get; set; } + /// + /// Which warm pool a POOLED rental belongs to. Left null the host derives one from the + /// adapter id and a hash of , 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 . + /// + public string PoolKey { get; set; } + /// Configuration and credentials. Sent over the stream, never on argv. public IDictionary StartupValues { get; set; } = new Dictionary(); diff --git a/SW.Serverless/Resident/IAdapterStateStore.cs b/SW.Serverless/Resident/IAdapterStateStore.cs new file mode 100644 index 0000000..e187639 --- /dev/null +++ b/SW.Serverless/Resident/IAdapterStateStore.cs @@ -0,0 +1,67 @@ +using System.Collections.Concurrent; +using System.Threading; +using System.Threading.Tasks; + +namespace SW.Serverless.Resident +{ + /// + /// Which piece of state is being addressed. Scoped to the instance rather than to the adapter, + /// because two instances of the same adapter are two different connections: one polling + /// receiver's cursor must never be read by another. + /// + public class AdapterStateKey + { + public string AdapterId { get; set; } + + /// DataSourceId for an exclusive instance, the pool slot id for a pooled one. + public string InstanceKey { get; set; } + + /// Chosen by the adapter. Namespace it yourself if one instance keeps several. + public string Name { get; set; } + + public override string ToString() => $"{AdapterId}/{InstanceKey}/{Name}"; + } + + /// + /// Implemented by the HOST APPLICATION, and the counterpart of : + /// where the sink is how an adapter hands work in, this is how it remembers where it got to. + /// + /// An adapter cannot hold its own progress. The supervisor restarts it, the next instance may + /// come up on a different node, and a pooled one is not the same process twice — so a cursor + /// kept in a field is a cursor that resets to the beginning at the least convenient moment. + /// Bitween backs this with a table; a sample host can use . + /// + /// Values are small — a bookmark, an offset, a timestamp. A host is entitled to refuse a large + /// one, and should say so in the returned error rather than storing it. + /// + public interface IAdapterStateStore + { + /// Null when nothing is stored under that key. + Task GetAsync(AdapterStateKey key, CancellationToken cancellationToken); + + /// Durable before it returns. A null value deletes the entry. + Task SetAsync(AdapterStateKey key, string value, CancellationToken cancellationToken); + } + + /// + /// The default, and only honest for a single-process host: state lives as long as the host does + /// and is not shared between nodes. Registered so that samples and tests work out of the box — + /// a real deployment replaces it, which is what the type parameter on + /// AddResidentAdapters is for. + /// + public class InMemoryAdapterStateStore : IAdapterStateStore + { + readonly ConcurrentDictionary entries = new(); + + public Task GetAsync(AdapterStateKey key, CancellationToken cancellationToken) => + Task.FromResult(entries.TryGetValue(key.ToString(), out var value) ? value : null); + + public Task SetAsync(AdapterStateKey key, string value, CancellationToken cancellationToken) + { + if (value == null) entries.TryRemove(key.ToString(), out _); + else entries[key.ToString()] = value; + + return Task.CompletedTask; + } + } +} diff --git a/SW.Serverless/Resident/ResidentAdapterHost.cs b/SW.Serverless/Resident/ResidentAdapterHost.cs index 0e47dda..16da951 100644 --- a/SW.Serverless/Resident/ResidentAdapterHost.cs +++ b/SW.Serverless/Resident/ResidentAdapterHost.cs @@ -11,6 +11,8 @@ using System.IO; using System.Linq; using System.Runtime.InteropServices; +using System.Security.Cryptography; +using System.Text; using System.Threading; using System.Threading.Tasks; @@ -20,6 +22,7 @@ public class ResidentAdapterHost : IResidentAdapterHost, IHostedService, IAsyncD { readonly ResidentOptions options; readonly IAdapterEventSink sink; + readonly IAdapterStateStore stateStore; readonly ILoggerFactory loggerFactory; readonly ILogger logger; readonly ResidentAdapterRegistry registry = new(); @@ -41,10 +44,11 @@ public class ResidentAdapterHost : IResidentAdapterHost, IHostedService, IAsyncD readonly IResidentAdapterLocator locator; public ResidentAdapterHost(ResidentOptions options, IAdapterEventSink sink, - IResidentAdapterLocator locator, ILoggerFactory loggerFactory) + IAdapterStateStore stateStore, IResidentAdapterLocator locator, ILoggerFactory loggerFactory) { this.options = options; this.sink = sink; + this.stateStore = stateStore; this.locator = locator; this.loggerFactory = loggerFactory; logger = loggerFactory.CreateLogger(); @@ -182,7 +186,7 @@ async Task SpawnAsync(Supervised supervised, CancellationToken cancellationToken spec.AdapterId, spec.InstanceKey ?? "default", Guid.NewGuid().ToString("N"), - options, sink, loggerFactory) + options, sink, stateStore, loggerFactory) { StartupValues = new Dictionary( spec.StartupValues ?? new Dictionary()), @@ -385,11 +389,38 @@ async Task StopSupervisedAsync(Supervised supervised, bool drain) public Task RentAsync(AdapterSpec spec, CancellationToken cancellationToken = default) { - var pool = pools.GetOrAdd(spec.AdapterId, + // Keyed by the SPEC, not by the adapter id. GetOrAdd captures the spec of whichever + // caller created the pool first — including its startup values, which is where the + // connection string and the credentials live. Keying on the id alone therefore handed + // the second data source a process connected as the first one: every later renter of + // that adapter silently ran against the wrong system. + var pool = pools.GetOrAdd(PoolKeyOf(spec), _ => new AdapterPool(spec, this, options, loggerFactory.CreateLogger())); return pool.RentAsync(cancellationToken); } + /// + /// Which pool a spec belongs in. An explicit wins; otherwise + /// it is the adapter id plus a hash of the startup values, so identical configuration shares + /// warm processes and differing configuration cannot. + /// + /// Hashed rather than concatenated because the values are credentials, and this string ends + /// up in logs and in pool diagnostics. + /// + public static string PoolKeyOf(AdapterSpec spec) + { + if (!string.IsNullOrWhiteSpace(spec.PoolKey)) return $"{spec.AdapterId}:{spec.PoolKey}"; + if (spec.StartupValues == null || spec.StartupValues.Count == 0) return spec.AdapterId; + + var canonical = new StringBuilder(); + foreach (var kv in spec.StartupValues.OrderBy(kv => kv.Key, StringComparer.Ordinal)) + canonical.Append(kv.Key).Append('\u001f').Append(kv.Value).Append('\u001e'); + + using var sha = SHA256.Create(); + var hash = sha.ComputeHash(Encoding.UTF8.GetBytes(canonical.ToString())); + return $"{spec.AdapterId}:{Convert.ToHexString(hash, 0, 8).ToLowerInvariant()}"; + } + internal async Task SpawnPooledAsync(AdapterSpec spec, string slot, CancellationToken ct) { diff --git a/SW.Serverless/Resident/ResidentAdapterInstance.cs b/SW.Serverless/Resident/ResidentAdapterInstance.cs index 371c62d..cca24fc 100644 --- a/SW.Serverless/Resident/ResidentAdapterInstance.cs +++ b/SW.Serverless/Resident/ResidentAdapterInstance.cs @@ -26,6 +26,7 @@ public sealed class ResidentAdapterInstance : IAsyncDisposable readonly ILogger logger; readonly ILogger adapterLogger; readonly IAdapterEventSink sink; + readonly IAdapterStateStore stateStore; // The correlation fix: every outstanding call is keyed, so a late reply can never // resolve an unrelated one the way the single v1 field did (design doc 14.3). @@ -45,13 +46,15 @@ public sealed class ResidentAdapterInstance : IAsyncDisposable CancellationTokenSource linkedCts; internal ResidentAdapterInstance(string adapterId, string instanceKey, string token, - ResidentOptions options, IAdapterEventSink sink, ILoggerFactory loggerFactory) + ResidentOptions options, IAdapterEventSink sink, IAdapterStateStore stateStore, + ILoggerFactory loggerFactory) { AdapterId = adapterId; InstanceKey = instanceKey; Token = token; this.options = options; this.sink = sink; + this.stateStore = stateStore; inbound = new SemaphoreSlim(Math.Max(1, options.MaxInFlight)); logger = loggerFactory.CreateLogger(); adapterLogger = loggerFactory.CreateLogger($"serverless.adapters.{adapterId}".ToLowerInvariant()); @@ -226,6 +229,13 @@ async Task OnFrameAsync(AdapterFrame frame, CancellationToken ct) }, ct); break; + case AdapterFrame.BodyOneofCase.State: + // Awaited inline rather than fanned out: state calls are small, ordered per + // adapter by construction, and an adapter that writes a cursor twice in a row + // means the second to be the one that lands. + await HandleStateAsync(frame, ct); + break; + case AdapterFrame.BodyOneofCase.Log: WriteLog(frame.Log); break; @@ -266,6 +276,48 @@ async Task HandleEventAsync(AdapterFrame frame, CancellationToken ct) Send(new HostFrame { Id = frame.Id, EventAck = ack }); } + async Task HandleStateAsync(AdapterFrame frame, CancellationToken ct) + { + var request = frame.State; + var key = new AdapterStateKey + { + AdapterId = AdapterId, + InstanceKey = InstanceKey, + Name = request.Name + }; + + var result = new StateResult(); + try + { + switch (request.Op) + { + case StateRequest.Types.Op.Get: + var value = await stateStore.GetAsync(key, ct); + result.Found = value != null; + result.Value = value ?? ""; + break; + + case StateRequest.Types.Op.Set: + await stateStore.SetAsync(key, request.Value ?? "", ct); + result.Found = true; + break; + + case StateRequest.Types.Op.Delete: + await stateStore.SetAsync(key, null, ct); + break; + } + } + catch (Exception ex) + { + // Reported back rather than logged and dropped. An adapter that believes it saved + // its cursor and did not will skip whatever it read next time. + logger.LogError(ex, "State store threw for {Key} ({Op}).", key, request.Op); + result.Error = new Error { Type = ex.GetType().Name, Message = ex.Message }; + } + + Send(new HostFrame { Id = frame.Id, StateResult = result }); + } + void WriteLog(LogEntry entry) { var level = (LogLevel)Math.Clamp(entry.Level, 0, 5); diff --git a/docs/README.md b/docs/README.md index 40adc73..bb488d9 100644 --- a/docs/README.md +++ b/docs/README.md @@ -42,7 +42,9 @@ You get, without asking for it: libraries they use — log reaches the host under `serverless.adapters.{id}`. * **`IConfiguration`** built from startup values, with cloud metadata namespaced under `AdapterValues:` so it can never shadow them. -* **`IAdapterContext`** for pushing events and metrics, injectable anywhere. +* **`IAdapterContext`** for pushing events and metrics, injectable anywhere — and for reading and + writing small durable state the host holds on the adapter's behalf (`GetStateAsync` / + `SetStateAsync`), which is where a polling receiver keeps its cursor. * **`AdapterSession.Id`** — ambient per-invocation identity, and the boundary a pooled adapter needs so state cannot leak between checkouts. diff --git a/docs/resident-adapters-design.md b/docs/resident-adapters-design.md index 20b24a1..f0f0667 100644 --- a/docs/resident-adapters-design.md +++ b/docs/resident-adapters-design.md @@ -1557,6 +1557,58 @@ capability behind metadata flags and new APIs, so Gateway can take a NuGet bump behavioural change and adopt features on its own schedule. A breaking protocol change would require a coordinated upgrade across ~190 binaries and is effectively off the table. +### 14.9 Host-held adapter state + +An adapter must not be the system of record for its own progress, and until now nothing in the +contract let it avoid being one. The supervisor restarts it, the next instance may come up on a +different node, and a pooled one is not the same process twice — so a polling receiver that keeps +its cursor in a field replays from the beginning at the least convenient moment. + +So `IAdapterContext` gains two calls, sitting beside `PublishAsync` and answered the same way — an +adapter-initiated frame, correlated by id, awaited before the adapter carries on: + +```csharp +Task GetStateAsync(string name, CancellationToken ct = default); +Task SetStateAsync(string name, string value, CancellationToken ct = default); // null deletes +``` + +On the wire that is `StateRequest` (get / set / delete) answered by `StateResult`. On the host side +it is `IAdapterStateStore`, the counterpart of `IAdapterEventSink`: where the sink is how an +adapter hands work in, this is how it remembers where it got to. `InMemoryAdapterStateStore` is the +default so samples and tests work untouched; a real deployment registers its own through +`AddResidentAdapters()` and backs it with a table. + +Three properties are deliberate: + +* **Keyed by instance, not by adapter.** Two instances of one adapter are two connections. One + reading the other's cursor would skip rows that were never processed. +* **Not bounded by the in-flight window.** That window exists to stop an adapter flooding the host + with events it must persist. A receiver saving its cursor *after* a batch would otherwise queue + behind the very events whose progress it is recording. +* **Failures are raised, not swallowed.** A cursor that silently failed to save is a batch that + will be replayed, and the adapter is the only thing positioned to stop rather than carry on. + +It is a bookmark, not a data store, and a host is entitled to refuse a large value. + +This is the same role Airbyte's `state` argument plays for its connectors, and it is what makes a +polling database receiver possible at all — see Bitween's `docs/provider-plan-databases.md`. + +### 14.10 A pool keyed by adapter id is a configuration leak + +`RentAsync` keyed its pools on `spec.AdapterId`, and `GetOrAdd` captures the spec of whichever +caller created the pool first — **including its startup values, which is where the connection +string and the credentials live**. One adapter serving two data sources therefore handed the second +one a process connected as the first, with no error anywhere: every later renter silently ran +against the wrong system. + +Masked until now because bus providers run as *exclusive* instances keyed by data source and never +go through the pool. Anything that rents — Bitween's Xchange pipeline does, through +`ResidentAdapterRuntime` — was exposed, and a pooled database adapter would be exposed by design. + +Fixed by keying on `AdapterSpec.PoolKey` when set, and otherwise on the adapter id plus a hash of +the startup values, so identical configuration shares warm processes and differing configuration +cannot. Hashed rather than concatenated because the key reaches logs and diagnostics. + --- ## 15. What "gRPC over UDS / named pipe" actually means