Server 3.2.0: connection context, targeted events, one ordered outbound queue per connection

CHANGELOG.md

@@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 > **Note**: Since 2.3.1, package versions diverge per package family. Each section below lists the package versions it produced (verified against csproj `<Version>` values).
 
+## [Server 3.2.0] - 2026-09-07
+
+### Added
+
+- **Connection context for a service.** `ConnectionContext.Current` (an `AsyncLocal` the server sets around every request and around the authorization handshake) tells a service method which connection it is serving: the connection id, the server (`ServerId`, `ServerName`, `Transport`) and the principal the connection authorized with. The principal comes from a token validator that also implements the new `IConnectionAuthenticator` (`TryAuthenticate(token, out ClaimsPrincipal?)`); any other validator is used exactly as before and the principal stays null. `IConnectionContextAccessor` / `ConnectionContextAccessor` for injection, `ConnectionContext.BeginScope(...)` for unit tests of services. `ConnectionInfo` gains `Principal`, `AuthorizedAtUtc`, `PendingCallbacks`, `PendingCallbackBytes`.
+- **Targeted events.** `CallbackScope.Target(connectionId)` / `Target(connectionIds)` / `TargetCaller()`: an event raised inside the scope goes to the named connection(s) only. The server reads the target when the callback reaches it (the raise chain is synchronous, so the `AsyncLocal` is visible there; it also flows into tasks started inside the scope). The scope reports what happened per server and connection (`Report` now, `Completion` once every accepted send finished: `Sent`, `UnknownConnection`, `NotAuthorized`, `QueueFull`, `SendFailed`, `SendTimedOut`, `Refused`). An event raised outside a scope still goes to every authorized connection.
+- **Callback delivery options** (`WithCallbackDelivery(...)`, `WithTargetedCallbacksOnly()`, `CallbackDeliveryOptions` on the builder options and on the new `WitServer` constructor overload): a per-connection bound on queued callbacks (`MaxPendingCallbacks`, `MaxPendingCallbackBytes`; 0 = unbounded) with an overflow policy — `Log` (default: queue anyway, warn, keep the connection), `CloseConnection` (refuse, close the connection; a callback send timeout closes it too), `DropNewest` (for lossy events) — and a `TargetedOnly` mode that refuses an untargeted raise. `WitServer.GetPendingCallbacks(connectionId)` and `AuthorizedConnectionCount` for diagnostics.
+
+### Changed
+
+- **One ordered outbound queue per connection.** Responses, handshake replies and callbacks of a connection are written by one writer task in the order they were enqueued (`ConnectionOutbox`), instead of a fire-and-forget send task per callback per connection waiting on the send lock. Wire order is now a stated guarantee: a callback raised inside a service method before it returns precedes that method's response; a connection whose transport is stuck holds only its own queue. With the default options every frame is still delivered, in the same order, with the same `Timeout` warning on a slow callback write — no behaviour change for an existing server. `ConnectionInfo.SendLock` stays and is taken by the writer around each write.
+- Nothing changes on the wire, in the core, in any client package or in the DI package (its floor stays `Server >= 3.1.1`; a consumer takes 3.2.0 with an explicit pin). 46 new tests: `ConnectionContextTests`, `TargetedCallbackTests` (Pipes, WebSocket, TCP), `ConnectionOutboxTests` (ordering, isolation, bounds, policies, timeouts, on a stub transport).
+
 ## [OutWit.Communication 3.1.2] - 2026-08-30
 
 ### Fixed

Communication/OutWit.Communication.LoadTests/Client/LoadNode.cs

@@ -0,0 +1,132 @@
+using OutWit.Communication.Client;
+using OutWit.Communication.Client.Tcp.Utils;
+using OutWit.Communication.Client.WebSocket.Utils;
+using OutWit.Communication.LoadTests.Contracts;
+
+namespace OutWit.Communication.LoadTests.Client
+{
+    /// <summary>
+    /// A simulated node: one client connection, one proxy, an ack for every task it receives.
+    /// </summary>
+    public sealed class LoadNode : IAsyncDisposable
+    {
+        #region Fields
+
+        private WitClient? m_client;
+
+        private ILoadService? m_service;
+
+        private long m_received;
+
+        #endregion
+
+        #region Constructors
+
+        public LoadNode(int index, string endpoint, bool webSocket, string token, TimeSpan timeout)
+        {
+            Index = index;
+            Endpoint = endpoint;
+            WebSocket = webSocket;
+            Token = token;
+            Timeout = timeout;
+        }
+
+        #endregion
+
+        #region Functions
+
+        public async Task<bool> ConnectAsync(CancellationToken cancellation)
+        {
+            m_client = WitClientBuilder.Build(options =>
+            {
+                if (WebSocket)
+                {
+                    options.WithWebSocket(Endpoint);
+                }
+                else
+                {
+                    var parts = Endpoint.Split(':');
+                    options.WithTcp(parts[0], int.Parse(parts[1]));
+                }
+
+                options.WithJson();
+                options.WithEncryption();
+                options.WithAccessToken(Token);
+                options.WithTimeout(Timeout);
+            });
+
+            if (!await m_client.ConnectAsync(Timeout, cancellation).ConfigureAwait(false))
+                return false;
+
+            m_service = m_client.GetService<ILoadService>();
+            m_service.TaskReceived += OnTaskReceived;
+            ConnectionId = m_service.Attach(Index);
+            return true;
+        }
+
+        private void OnTaskReceived(LoadTask task)
+        {
+            Interlocked.Increment(ref m_received);
+
+            // A broadcast reaches every node; only the addressee answers.
+            if (task.NodeIndex != Index)
+                return;
+
+            _ = AckAsync(task);
+        }
+
+        private async Task AckAsync(LoadTask task)
+        {
+            try
+            {
+                if (m_service != null)
+                    await m_service.AckAsync(task.Id, Index).ConfigureAwait(false);
+            }
+            catch
+            {
+                // A node that lost its connection cannot ack; the harness counts the gap.
+            }
+        }
+
+        #endregion
+
+        #region IAsyncDisposable
+
+        public async ValueTask DisposeAsync()
+        {
+            if (m_client == null)
+                return;
+
+            try
+            {
+                await m_client.Disconnect().ConfigureAwait(false);
+            }
+            catch
+            {
+                // Tearing down; nothing to report.
+            }
+
+            m_client.Dispose();
+        }
+
+        #endregion
+
+        #region Properties
+
+        public int Index { get; }
+
+        public string Endpoint { get; }
+
+        public bool WebSocket { get; }
+
+        public string Token { get; }
+
+        public TimeSpan Timeout { get; }
+
+        public Guid ConnectionId { get; private set; }
+
+        public long Received => Interlocked.Read(ref m_received);
+
+        #endregion
+    }
+}

Communication/OutWit.Communication.LoadTests/Contracts/ILoadService.cs

@@ -0,0 +1,24 @@
+namespace OutWit.Communication.LoadTests.Contracts
+{
+    /// <summary>
+    /// The contract between the load harness's server and its simulated nodes: a node attaches,
+    /// receives tasks by event, acknowledges each one.
+    /// </summary>
+    public interface ILoadService
+    {
+        /// <summary>A task pushed to one node (targeted) or to all (broadcast).</summary>
+        event Action<LoadTask> TaskReceived;
+
+        /// <summary>
+        /// Binds the calling connection to a node index; returns the connection id the server sees.
+        /// </summary>
+        Guid Attach(int nodeIndex);
+
+        /// <summary>
+        /// Acknowledges a task; the server measures the time from dispatch to this call. Async so
+        /// that a node's callback handler does not hold a thread while the answer travels (a node
+        /// that blocks per callback starves its own process long before the server is the limit).
+        /// </summary>
+        Task AckAsync(long taskId, int nodeIndex);
+    }
+}

Communication/OutWit.Communication.LoadTests/Contracts/LoadTask.cs

@@ -0,0 +1,16 @@
+namespace OutWit.Communication.LoadTests.Contracts
+{
+    /// <summary>
+    /// One unit of work pushed to a node.
+    /// </summary>
+    public sealed class LoadTask
+    {
+        public long Id { get; set; }
+
+        public int NodeIndex { get; set; }
+
+        public long DispatchedAtTicks { get; set; }
+
+        public byte[] Payload { get; set; } = Array.Empty<byte>();
+    }
+}

Communication/OutWit.Communication.LoadTests/LoadOptions.cs

@@ -0,0 +1,126 @@
+using OutWit.Communication.Server.Callbacks;
+
+namespace OutWit.Communication.LoadTests
+{
+    /// <summary>
+    /// What one run measures. Parsed from <c>--name value</c> arguments; see <see cref="Usage"/>.
+    /// </summary>
+    public sealed class LoadOptions
+    {
+        #region Constants
+
+        public const string Usage = """
+            OutWit.Communication.LoadTests -- the 3.2 shared-server model under load
+
+              --mode shared|per-client|broadcast   shared: one server, targeted callbacks (3.2)
+                                                   per-client: one server per node (the 1.7.x baseline)
+                                                   broadcast: one server, untargeted raise to everyone (what sharing
+                                                   a server meant before 3.2; every node receives every task)
+              --nodes N          simulated nodes (default 100)
+              --rate R           tasks per second, total (default 500)
+              --duration S       seconds of steady dispatch (default 20)
+              --payload BYTES    task payload size (default 256)
+              --transport websocket|tcp (default websocket)
+              --slow K           make node K's socket slow (server-side write delay), -1 = none (default -1)
+              --slow-delay MS    the delay per write of the slow node (default 50)
+              --stall            instead of a delay, the slow node's socket never drains
+              --max-pending N    CallbackDeliveryOptions.MaxPendingCallbacks (default 0 = unbounded)
+              --policy log|close|drop   overflow policy (default log)
+              --timeout MS       server callback send timeout (default 5000)
+              --json PATH        also write the report as JSON
+            """;
+
+        #endregion
+
+        #region Functions
+
+        public static LoadOptions Parse(string[] args)
+        {
+            var options = new LoadOptions();
+
+            for (var i = 0; i < args.Length; i++)
+            {
+                var name = args[i];
+                string Next() => i + 1 < args.Length ? args[++i] : throw new ArgumentException($"{name} needs a value");
+
+                switch (name)
+                {
+                    case "--mode": options.Mode = Next(); break;
+                    case "--nodes": options.Nodes = int.Parse(Next()); break;
+                    case "--rate": options.Rate = int.Parse(Next()); break;
+                    case "--duration": options.DurationSeconds = int.Parse(Next()); break;
+                    case "--payload": options.PayloadBytes = int.Parse(Next()); break;
+                    case "--transport": options.Transport = Next(); break;
+                    case "--slow": options.SlowNode = int.Parse(Next()); break;
+                    case "--slow-delay": options.SlowDelayMs = int.Parse(Next()); break;
+                    case "--stall": options.Stall = true; break;
+                    case "--max-pending": options.MaxPending = int.Parse(Next()); break;
+                    case "--policy": options.Policy = Next(); break;
+                    case "--timeout": options.TimeoutMs = int.Parse(Next()); break;
+                    case "--json": options.JsonPath = Next(); break;
+                    case "--help": case "-h": options.Help = true; break;
+                    default: throw new ArgumentException($"Unknown argument {name}\n{Usage}");
+                }
+            }
+
+            return options;
+        }
+
+        public CallbackDeliveryOptions ToDelivery()
+        {
+            return new CallbackDeliveryOptions
+            {
+                MaxPendingCallbacks = MaxPending,
+                OverflowPolicy = Policy switch
+                {
+                    "close" => CallbackOverflowPolicy.CloseConnection,
+                    "drop" => CallbackOverflowPolicy.DropNewest,
+                    _ => CallbackOverflowPolicy.Log
+                },
+                Mode = Mode == "shared" ? CallbackDeliveryMode.TargetedOnly : CallbackDeliveryMode.BroadcastAllowed
+            };
+        }
+
+        public override string ToString()
+        {
+            var slow = SlowNode < 0 ? "none" : Stall ? $"node {SlowNode} stalled" : $"node {SlowNode} +{SlowDelayMs} ms/write";
+            return $"mode={Mode} nodes={Nodes} rate={Rate}/s duration={DurationSeconds}s payload={PayloadBytes}B transport={Transport} slow={slow} maxPending={MaxPending} policy={Policy} timeout={TimeoutMs}ms";
+        }
+
+        #endregion
+
+        #region Properties
+
+        public string Mode { get; set; } = "shared";
+
+        public int Nodes { get; set; } = 100;
+
+        public int Rate { get; set; } = 500;
+
+        public int DurationSeconds { get; set; } = 20;
+
+        public int PayloadBytes { get; set; } = 256;
+
+        public string Transport { get; set; } = "websocket";
+
+        public int SlowNode { get; set; } = -1;
+
+        public int SlowDelayMs { get; set; } = 50;
+
+        public bool Stall { get; set; }
+
+        public int MaxPending { get; set; }
+
+        public string Policy { get; set; } = "log";
+
+        public int TimeoutMs { get; set; } = 5000;
+
+        public string? JsonPath { get; set; }
+
+        public bool Help { get; set; }
+
+        public bool WebSocket => Transport != "tcp";
+
+        #endregion
+    }
+}