diff --git a/SW.Serverless.Contract/Protos/adapter.proto b/SW.Serverless.Contract/Protos/adapter.proto index 2655bb8..5f0d41e 100644 --- a/SW.Serverless.Contract/Protos/adapter.proto +++ b/SW.Serverless.Contract/Protos/adapter.proto @@ -47,6 +47,15 @@ message Invoke { // request. Traxis calls a command and then GetLogs; both must see the SAME session or the // audit trail comes back empty. Absent means "this call alone". string session_id = 4; + + // Configuration for THIS call, on top of the startup values the process was given. + // + // An exclusive resident instance is shared: in Bitween one database connection serves every + // subscription bound to it, and each of those has its own settings — which statement to run, + // which operation it is. Startup values cannot carry that, because they belong to the process + // and the process belongs to all of them. This is the same reason session_id exists: a shared + // instance has to be told whose call this is. + map properties = 5; } message Ping {} diff --git a/SW.Serverless.Samples.Ticker/Handler.cs b/SW.Serverless.Samples.Ticker/Handler.cs index d949a30..569921b 100644 --- a/SW.Serverless.Samples.Ticker/Handler.cs +++ b/SW.Serverless.Samples.Ticker/Handler.cs @@ -156,6 +156,30 @@ public async Task ClearCursor() return new { cleared = true }; } + /// + /// Per-call configuration, which is what a SHARED instance needs: one process serving many + /// callers cannot take per-caller settings from its startup values, because those belong to + /// the process. + /// + public Task ReadValue(string name) => Task.FromResult(new + { + invocation = context.InvocationValues.TryGetValue(name, out var v) ? v : null, + startup = context.StartupValueOf(name), + resolved = context.ValueOf(name) + }); + + /// + /// Reads the same value either side of an await, so a test can prove two concurrent callers + /// do not see each other's — the failure an ordinary field would have. + /// + public async Task ReadValueSlowly(string name) + { + var before = context.ValueOf(name); + await Task.Delay(300); + var after = context.ValueOf(name); + return new { before, after }; + } + /// 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 052c7cb..67bbd94 100644 --- a/SW.Serverless.Sdk/Resident/IAdapterContext.cs +++ b/SW.Serverless.Sdk/Resident/IAdapterContext.cs @@ -28,6 +28,24 @@ public interface IAdapterContext IReadOnlyDictionary AdapterValues { get; } string StartupValueOf(string name); + /// + /// Configuration for the command CURRENTLY running on this async flow, sent by the host + /// with the invocation. Empty outside a command, and empty when the caller sent none. + /// + /// It exists because an exclusive instance is shared. One database connection serves every + /// subscription pointed at it, and each of those has its own settings — which statement to + /// run, which operation it is. Startup values cannot carry that: they belong to the process, + /// and the process belongs to all of them. + /// + IReadOnlyDictionary InvocationValues { get; } + + /// + /// first, then . This is the one + /// to reach for: a per-call setting overrides the process default, and an adapter serving a + /// single caller behaves exactly as it did before any of this existed. + /// + string ValueOf(string name); + /// Cancelled when the host asks the adapter to shut down. CancellationToken Stopping { get; } diff --git a/SW.Serverless.Sdk/Resident/ResidentRunner.cs b/SW.Serverless.Sdk/Resident/ResidentRunner.cs index dc525d1..9c2116b 100644 --- a/SW.Serverless.Sdk/Resident/ResidentRunner.cs +++ b/SW.Serverless.Sdk/Resident/ResidentRunner.cs @@ -62,6 +62,11 @@ public sealed class ResidentRunner : IAdapterContext IReadOnlyDictionary startupValues = new Dictionary(); IReadOnlyDictionary adapterValues = new Dictionary(); + static readonly IReadOnlyDictionary Empty = new Dictionary(); + + // Per async flow, so concurrent commands on one shared instance do not read each other's. + static readonly AsyncLocal> invocationValues = new(); + ResidentRunner(Type handlerType, Func handlerFactory) { this.handlerType = handlerType; @@ -304,6 +309,14 @@ async Task OnInvokeAsync(long id, Invoke invoke) if (!commands.TryGetValue(invoke.Command, out var method)) throw new MissingMethodException(handlerType.FullName, invoke.Command); + // Set on THIS async flow, before the handler runs, so concurrent invocations on a + // shared instance each see their own caller's configuration. AsyncLocal rather than + // a field for exactly that reason: several commands are in flight at once by + // design, and a field would have the last one in overwrite the rest. + invocationValues.Value = invoke.Properties.Count == 0 + ? Empty + : new Dictionary(invoke.Properties); + // The host's session id when it grouped this call with others, otherwise the // call stands alone. using var session = AdapterSession.Begin( @@ -488,6 +501,16 @@ async Task PumpOutboundAsync(IClientStreamWriter stream) public string StartupValueOf(string name) => startupValues.TryGetValue(name, out var v) ? v : null; + public IReadOnlyDictionary InvocationValues => invocationValues.Value ?? Empty; + + public string ValueOf(string name) + { + var perCall = invocationValues.Value; + if (perCall != null && perCall.TryGetValue(name, out var value)) return value; + + return StartupValueOf(name); + } + public async Task PublishAsync( ReadOnlyMemory payload, string dedupeKey, string endpoint = null, IDictionary headers = null, string contentType = null, diff --git a/SW.Serverless.UnitTests/ResidentAdapterTests.cs b/SW.Serverless.UnitTests/ResidentAdapterTests.cs index a793c44..beac278 100644 --- a/SW.Serverless.UnitTests/ResidentAdapterTests.cs +++ b/SW.Serverless.UnitTests/ResidentAdapterTests.cs @@ -295,6 +295,84 @@ await WaitFor(() => adapters.Describe() // ------------------------------------------------------------------ helpers + // ------------------------------------------------------------------ per-invocation properties + + /// + /// Configuration that travels with the CALL rather than with the process. An exclusive + /// instance is shared by every caller pointed at it, so anything that varies per caller — + /// which statement to run, which tenant this is — cannot live in the startup values. + /// + [TestMethod] + public async Task Per_invocation_properties_reach_the_command() + { + var instance = await StartTicker("percall"); + + var answer = await instance.InvokeAsync>( + "ReadValue", "Tag", + properties: new Dictionary { ["Tag"] = "invoice-run" }); + + Assert.AreEqual("invoice-run", answer["invocation"]); + Assert.IsNull(answer["startup"], "the process was never told about Tag"); + Assert.AreEqual("invoice-run", answer["resolved"]); + + await adapters.StopAsync(TickerId, "percall", drain: false); + } + + /// + /// A per-call value wins over the process default, and an adapter called WITHOUT properties + /// still sees the startup value — which is what keeps every existing adapter working. + /// + [TestMethod] + public async Task A_per_invocation_value_overrides_the_startup_value() + { + var instance = await StartTicker("override", intervalSeconds: 30); + + var overridden = await instance.InvokeAsync>( + "ReadValue", "IntervalSeconds", + properties: new Dictionary { ["IntervalSeconds"] = "5" }); + + Assert.AreEqual("5", overridden["resolved"]); + Assert.AreEqual("30", overridden["startup"], "the startup value is still there underneath"); + + var plain = await instance.InvokeAsync>( + "ReadValue", "IntervalSeconds"); + + Assert.AreEqual("30", plain["resolved"], "no properties means the process default, as before"); + Assert.IsNull(plain["invocation"]); + + await adapters.StopAsync(TickerId, "override", drain: false); + } + + /// + /// The reason this is an AsyncLocal and not a field. Several commands run at once on one + /// instance by design — that is what multiplexing is for — and a field would have the last + /// caller in overwrite everyone else's configuration mid-flight. + /// + [TestMethod] + public async Task Concurrent_invocations_do_not_see_each_others_properties() + { + var instance = await StartTicker("concurrent"); + + var first = instance.InvokeAsync>( + "ReadValueSlowly", "Tag", + properties: new Dictionary { ["Tag"] = "alpha" }); + + var second = instance.InvokeAsync>( + "ReadValueSlowly", "Tag", + properties: new Dictionary { ["Tag"] = "beta" }); + + await Task.WhenAll(first, second); + + // Both sides of the await, because the value has to survive the continuation as well as + // arrive on the way in. + Assert.AreEqual("alpha", first.Result["before"]); + Assert.AreEqual("alpha", first.Result["after"]); + Assert.AreEqual("beta", second.Result["before"]); + Assert.AreEqual("beta", second.Result["after"]); + + await adapters.StopAsync(TickerId, "concurrent", drain: false); + } + // ------------------------------------------------------------------ host-held state /// diff --git a/SW.Serverless/Resident/AdapterPool.cs b/SW.Serverless/Resident/AdapterPool.cs index 42eb8ef..c9fc42e 100644 --- a/SW.Serverless/Resident/AdapterPool.cs +++ b/SW.Serverless/Resident/AdapterPool.cs @@ -195,8 +195,10 @@ public Lease(AdapterPool pool, ResidentAdapterInstance instance) public string SessionId { get; } public Task InvokeAsync(string command, object input = null, - int timeoutSeconds = 0, CancellationToken cancellationToken = default) => - Instance.InvokeAsync(command, input, timeoutSeconds, cancellationToken, SessionId); + int timeoutSeconds = 0, CancellationToken cancellationToken = default, + IDictionary properties = null) => + Instance.InvokeAsync(command, input, timeoutSeconds, cancellationToken, + SessionId, properties); public ValueTask DisposeAsync() => new(pool.ReturnAsync(Instance, SessionId)); } diff --git a/SW.Serverless/Resident/IResidentAdapterHost.cs b/SW.Serverless/Resident/IResidentAdapterHost.cs index 720129e..1063dd7 100644 --- a/SW.Serverless/Resident/IResidentAdapterHost.cs +++ b/SW.Serverless/Resident/IResidentAdapterHost.cs @@ -17,7 +17,8 @@ public interface IAdapterLease : IAsyncDisposable /// disposal then clears through IResettable. /// Task InvokeAsync(string command, object input = null, - int timeoutSeconds = 0, CancellationToken cancellationToken = default); + int timeoutSeconds = 0, CancellationToken cancellationToken = default, + IDictionary properties = null); } /// diff --git a/SW.Serverless/Resident/ResidentAdapterInstance.cs b/SW.Serverless/Resident/ResidentAdapterInstance.cs index cca24fc..1112b2d 100644 --- a/SW.Serverless/Resident/ResidentAdapterInstance.cs +++ b/SW.Serverless/Resident/ResidentAdapterInstance.cs @@ -336,12 +336,18 @@ void WriteLog(LogEntry entry) void Send(HostFrame frame) => outbound.Writer.TryWrite(frame); + /// + /// is configuration for THIS call, on top of the startup + /// values the process holds. An exclusive instance is shared by every caller pointed at it, + /// so anything that varies per caller has to travel with the call rather than with the + /// process. + /// public async Task InvokeAsync(string command, object input = null, int timeoutSeconds = 0, CancellationToken cancellationToken = default, - string sessionId = null) + string sessionId = null, IDictionary properties = null) { var bytes = await InvokeAsync(command, Serialize(input), timeoutSeconds, - cancellationToken, sessionId); + cancellationToken, sessionId, properties); if (bytes == null || bytes.Length == 0) return default; if (typeof(TResult) == typeof(byte[])) return (TResult)(object)bytes; var text = System.Text.Encoding.UTF8.GetString(bytes); @@ -351,7 +357,7 @@ public async Task InvokeAsync(string command, object input = n public async Task InvokeAsync(string command, byte[] payload = null, int timeoutSeconds = 0, CancellationToken cancellationToken = default, - string sessionId = null) + string sessionId = null, IDictionary properties = null) { if (State != InstanceState.Ready) throw new InvalidOperationException($"Adapter {AdapterId} is {State}, not Ready."); @@ -380,17 +386,23 @@ public async Task InvokeAsync(string command, byte[] payload = null, } }, null, timeout, Timeout.InfiniteTimeSpan); + var invoke = new Invoke + { + Command = command, + Payload = payload == null ? ByteString.Empty : ByteString.CopyFrom(payload), + TimeoutSeconds = (int)timeout.TotalSeconds, + SessionId = sessionId ?? "" + }; + + if (properties != null) + foreach (var kv in properties) + invoke.Properties[kv.Key] = kv.Value ?? ""; + Send(new HostFrame { Id = id, Traceparent = Activity.Current?.Id ?? "", - Invoke = new Invoke - { - Command = command, - Payload = payload == null ? ByteString.Empty : ByteString.CopyFrom(payload), - TimeoutSeconds = (int)timeout.TotalSeconds, - SessionId = sessionId ?? "" - } + Invoke = invoke }); using (cancellationToken.Register(() => diff --git a/docs/README.md b/docs/README.md index bb488d9..51b626e 100644 --- a/docs/README.md +++ b/docs/README.md @@ -44,7 +44,8 @@ You get, without asking for it: `AdapterValues:` so it can never shadow them. * **`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. + `SetStateAsync`), which is where a polling receiver keeps its cursor, and for per-call + configuration on a shared instance (`ValueOf` / `InvocationValues`). * **`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 f0f0667..0994b7b 100644 --- a/docs/resident-adapters-design.md +++ b/docs/resident-adapters-design.md @@ -1609,6 +1609,39 @@ Fixed by keying on `AdapterSpec.PoolKey` when set, and otherwise on the adapter the startup values, so identical configuration shares warm processes and differing configuration cannot. Hashed rather than concatenated because the key reaches logs and diagnostics. +### 14.11 A shared instance needs per-call configuration + +§14.5 gave the exclusive resident one instance per key, and §4 assumed that instance had one +caller. It does not. In Bitween a relational data source is one process holding one connection +pool, and *every* subscription bound to that data source runs through it — each with its own +settings: which statement to run, which operation it is, which tenant this is. + +Startup values cannot carry any of that. They are handed over once, in `Ready`, and they belong to +the process — which belongs to all of those callers at once. So the invocation grows a +`map properties`, alongside the `session_id` that exists for the same underlying +reason: a shared instance has to be told whose call this is. + +On the adapter side that surfaces as two members on `IAdapterContext`: + +```csharp +IReadOnlyDictionary InvocationValues { get; } // this call's, empty outside one +string ValueOf(string name); // invocation first, then startup +``` + +`ValueOf` is the one to reach for. A per-call setting overrides the process default, and an adapter +whose callers send no properties behaves exactly as it did before any of this existed — which is +what keeps the existing fleet working. + +**It is an `AsyncLocal`, and that is the whole subtlety.** Several commands run on one instance at +the same time; multiplexing is the point of the stream. A field would have the last caller in +overwrite everyone else's configuration mid-flight, and the failure would be intermittent, +load-dependent and near-impossible to reproduce — one subscription silently running another's +statement. It is set on the invoking flow before the handler is called, so it survives the +handler's own awaits and cannot leak sideways. + +Without this the shared-instance shape is only usable by callers that all want identical behaviour, +which for a database connection is nobody. + --- ## 15. What "gRPC over UDS / named pipe" actually means