From d2e75ae5d4c452c24e9cf47a5b7200c2a48460b9 Mon Sep 17 00:00:00 2001 From: Muhannad Al-Khatib Date: Tue, 8 Sep 2026 16:28:36 +0300 Subject: [PATCH] feat: host-held adapter state, and fix pooled adapters sharing one config MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two changes the resident runtime needs before a database adapter can exist, both useful on their own. Host-held state. An adapter must not be the system of record for its own progress: the supervisor restarts it, the next instance may come up on another 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. IAdapterContext gains GetStateAsync / SetStateAsync, answered over a new StateRequest / StateResult frame pair the same way PublishAsync is answered by EventAck. Host side is IAdapterStateStore, the counterpart of IAdapterEventSink; InMemoryAdapterStateStore keeps samples and tests working untouched, and a real deployment registers its own through AddResidentAdapters(). State is keyed by instance, is not bounded by the in-flight window, and raises rather than swallows failures — see design doc 14.9 for why each of those matters. Pool keying. RentAsync keyed its pools on spec.AdapterId, and GetOrAdd captures the spec of whichever caller created the pool first, including its startup values — where the connection string and credentials live. One adapter serving two data sources therefore handed the second one a process connected as the first, silently. Masked so far because bus providers run as exclusive instances and never rent; anything going through ResidentAdapterRuntime was exposed. Fixed by keying on AdapterSpec.PoolKey when set, otherwise on the adapter id plus a hash of the startup values. Co-Authored-By: Claude Opus 5 --- SW.Serverless.Contract/Protos/adapter.proto | 28 ++++++ SW.Serverless.Samples.Ticker/Handler.cs | 20 +++++ SW.Serverless.Sdk/Resident/IAdapterContext.cs | 21 +++++ SW.Serverless.Sdk/Resident/ResidentRunner.cs | 66 ++++++++++++++ .../ResidentAdapterTests.cs | 89 +++++++++++++++++++ .../IServiceCollectionExtensions.cs | 19 ++++ SW.Serverless/Resident/AdapterSpec.cs | 10 +++ SW.Serverless/Resident/IAdapterStateStore.cs | 67 ++++++++++++++ SW.Serverless/Resident/ResidentAdapterHost.cs | 37 +++++++- .../Resident/ResidentAdapterInstance.cs | 54 ++++++++++- docs/README.md | 4 +- docs/resident-adapters-design.md | 52 +++++++++++ 12 files changed, 462 insertions(+), 5 deletions(-) create mode 100644 SW.Serverless/Resident/IAdapterStateStore.cs 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