diff --git a/CHANGELOG.md b/CHANGELOG.md index 497ba37..d717758 100644 --- a/CHANGELOG.md +++ b/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 `` 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 diff --git a/Communication/OutWit.Communication.LoadTests/Client/LoadNode.cs b/Communication/OutWit.Communication.LoadTests/Client/LoadNode.cs new file mode 100644 index 0000000..404f6cc --- /dev/null +++ b/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 +{ + /// + /// A simulated node: one client connection, one proxy, an ack for every task it receives. + /// + 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 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(); + 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 + } +} diff --git a/Communication/OutWit.Communication.LoadTests/Contracts/ILoadService.cs b/Communication/OutWit.Communication.LoadTests/Contracts/ILoadService.cs new file mode 100644 index 0000000..eac07bb --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/Contracts/ILoadService.cs @@ -0,0 +1,24 @@ +namespace OutWit.Communication.LoadTests.Contracts +{ + /// + /// The contract between the load harness's server and its simulated nodes: a node attaches, + /// receives tasks by event, acknowledges each one. + /// + public interface ILoadService + { + /// A task pushed to one node (targeted) or to all (broadcast). + event Action TaskReceived; + + /// + /// Binds the calling connection to a node index; returns the connection id the server sees. + /// + Guid Attach(int nodeIndex); + + /// + /// 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). + /// + Task AckAsync(long taskId, int nodeIndex); + } +} diff --git a/Communication/OutWit.Communication.LoadTests/Contracts/LoadTask.cs b/Communication/OutWit.Communication.LoadTests/Contracts/LoadTask.cs new file mode 100644 index 0000000..9364e48 --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/Contracts/LoadTask.cs @@ -0,0 +1,16 @@ +namespace OutWit.Communication.LoadTests.Contracts +{ + /// + /// One unit of work pushed to a node. + /// + 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(); + } +} diff --git a/Communication/OutWit.Communication.LoadTests/LoadOptions.cs b/Communication/OutWit.Communication.LoadTests/LoadOptions.cs new file mode 100644 index 0000000..bb9f433 --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/LoadOptions.cs @@ -0,0 +1,126 @@ +using OutWit.Communication.Server.Callbacks; + +namespace OutWit.Communication.LoadTests +{ + /// + /// What one run measures. Parsed from --name value arguments; see . + /// + 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 + } +} diff --git a/Communication/OutWit.Communication.LoadTests/LoadReport.cs b/Communication/OutWit.Communication.LoadTests/LoadReport.cs new file mode 100644 index 0000000..a49db11 --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/LoadReport.cs @@ -0,0 +1,88 @@ +using System.Text.Json; +using OutWit.Communication.LoadTests.Metrics; + +namespace OutWit.Communication.LoadTests +{ + /// + /// The numbers of one run, printed as a table and optionally written as JSON. + /// + public sealed class LoadReport + { + #region Properties + + public string Options { get; set; } = string.Empty; + + public int Nodes { get; set; } + + public double ConnectSeconds { get; set; } + + public int Dispatched { get; set; } + + public int Acked { get; set; } + + public double AchievedRatePerSecond { get; set; } + + public LatencySummary AllNodes { get; set; } = new(0, 0, 0, 0, 0, 0); + + public LatencySummary? NeighboursOfSlow { get; set; } + + public LatencySummary? SlowNode { get; set; } + + public Dictionary Outcomes { get; set; } = new(); + + public long SlowNodePendingMax { get; set; } + + public long AnyNodePendingMax { get; set; } + + public bool SlowNodeClosed { get; set; } + + public double SlowNodeClosedAfterSeconds { get; set; } + + public double ServerCpuSeconds { get; set; } + + public long WorkingSetMb { get; set; } + + public long BroadcastFramesPerNode { get; set; } + + #endregion + + #region Functions + + public void Print(TextWriter output) + { + output.WriteLine(); + output.WriteLine($"| run | {Options} |"); + output.WriteLine("|---|---|"); + output.WriteLine($"| nodes connected | {Nodes} in {ConnectSeconds:F2} s |"); + output.WriteLine($"| dispatched / acked | {Dispatched} / {Acked} ({AchievedRatePerSecond:F0} tasks/s achieved) |"); + output.WriteLine($"| dispatch -> ack, all nodes | {AllNodes} |"); + + if (NeighboursOfSlow != null) + output.WriteLine($"| dispatch -> ack, neighbours of the slow node | {NeighboursOfSlow} |"); + + if (SlowNode != null) + output.WriteLine($"| dispatch -> ack, the slow node | {SlowNode} |"); + + if (Outcomes.Count > 0) + output.WriteLine($"| delivery outcomes | {string.Join(", ", Outcomes.OrderByDescending(pair => pair.Value).Select(pair => $"{pair.Key} {pair.Value}"))} |"); + + if (SlowNodePendingMax > 0 || SlowNodeClosed) + output.WriteLine($"| slow node queue max / closed | {SlowNodePendingMax} / {(SlowNodeClosed ? $"yes after {SlowNodeClosedAfterSeconds:F1} s" : "no")} |"); + + output.WriteLine($"| queued callbacks, max over the other nodes | {AnyNodePendingMax} |"); + + if (BroadcastFramesPerNode > 0) + output.WriteLine($"| callback frames received per node | {BroadcastFramesPerNode} |"); + + output.WriteLine($"| process CPU / working set | {ServerCpuSeconds:F1} s / {WorkingSetMb} MB |"); + output.WriteLine(); + } + + public void WriteJson(string path) + { + File.WriteAllText(path, JsonSerializer.Serialize(this, new JsonSerializerOptions { WriteIndented = true })); + } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.LoadTests/LoadRunner.cs b/Communication/OutWit.Communication.LoadTests/LoadRunner.cs new file mode 100644 index 0000000..35c8a1d --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/LoadRunner.cs @@ -0,0 +1,315 @@ +using System.Diagnostics; +using OutWit.Communication.LoadTests.Client; +using OutWit.Communication.LoadTests.Metrics; +using OutWit.Communication.LoadTests.Server; +using OutWit.Communication.Server.Callbacks; + +namespace OutWit.Communication.LoadTests +{ + /// + /// One run: start the server(s), connect the nodes, dispatch tasks at the requested rate for + /// the requested time, collect the report. + /// + public sealed class LoadRunner + { + #region Constants + + private const string TOKEN = "load"; + + private static readonly TimeSpan CONNECT_TIMEOUT = TimeSpan.FromSeconds(30); + + private static readonly TimeSpan DRAIN = TimeSpan.FromSeconds(5); + + #endregion + + #region Fields + + private readonly LoadOptions m_options; + + private readonly List m_hosts = new(); + + private readonly List m_nodes = new(); + + private readonly Dictionary m_outcomes = new(); + + private long m_slowPendingMax; + + private long m_anyPendingMax; + + #endregion + + #region Constructors + + public LoadRunner(LoadOptions options) + { + m_options = options; + } + + #endregion + + #region Functions + + public async Task RunAsync(TextWriter log) + { + var report = new LoadReport { Options = m_options.ToString() }; + var process = Process.GetCurrentProcess(); + var cpuBefore = process.TotalProcessorTime; + + try + { + StartHosts(); + report.ConnectSeconds = await ConnectNodesAsync(log).ConfigureAwait(false); + report.Nodes = m_nodes.Count; + + log.WriteLine($"dispatching {m_options.Rate} tasks/s for {m_options.DurationSeconds} s ..."); + var (dispatched, seconds) = await DispatchAsync(log).ConfigureAwait(false); + + await DrainAsync().ConfigureAwait(false); + + report.Dispatched = dispatched; + report.AchievedRatePerSecond = dispatched / seconds; + Collect(report); + } + finally + { + // Teardown is not part of the measurement: a client whose close handshake never + // completes must not hold the run. + foreach (var node in m_nodes) + { + try + { + await node.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(3)).ConfigureAwait(false); + } + catch (TimeoutException) + { + log.WriteLine($" node {node.Index} did not disconnect within 3 s; abandoned"); + } + } + + foreach (var host in m_hosts) + { + try + { + await Task.Run(host.Dispose).WaitAsync(TimeSpan.FromSeconds(3)).ConfigureAwait(false); + } + catch (TimeoutException) + { + log.WriteLine(" a server did not dispose within 3 s; abandoned"); + } + } + } + + process.Refresh(); + report.ServerCpuSeconds = (process.TotalProcessorTime - cpuBefore).TotalSeconds; + report.WorkingSetMb = process.WorkingSet64 / (1024 * 1024); + return report; + } + + private void StartHosts() + { + var delivery = m_options.ToDelivery(); + var timeout = TimeSpan.FromMilliseconds(m_options.TimeoutMs); + + Func? ThrottleFor(int slowArrivalIndex) + { + if (m_options.SlowNode < 0) + return null; + + return arrival => arrival == slowArrivalIndex + ? (TimeSpan.FromMilliseconds(m_options.SlowDelayMs), m_options.Stall) + : null; + } + + if (m_options.Mode == "per-client") + { + var watch = Stopwatch.StartNew(); + for (var i = 0; i < m_options.Nodes; i++) + m_hosts.Add(LoadHost.Start(m_options.WebSocket, 4, TOKEN, timeout, delivery, i == m_options.SlowNode ? ThrottleFor(0) : null)); + + Console.WriteLine($"{m_hosts.Count} per-client servers started in {watch.Elapsed.TotalSeconds:F1} s"); + return; + } + + m_hosts.Add(LoadHost.Start(m_options.WebSocket, m_options.Nodes + 8, TOKEN, timeout, delivery, ThrottleFor(m_options.SlowNode))); + } + + private async Task ConnectNodesAsync(TextWriter log) + { + var watch = Stopwatch.StartNew(); + + // Sequential on purpose: the throttle is chosen by arrival order, so node K must be + // the K-th connection of the shared server. + for (var i = 0; i < m_options.Nodes; i++) + { + var host = m_options.Mode == "per-client" ? m_hosts[i] : m_hosts[0]; + var node = new LoadNode(i, host.ClientEndpoint, m_options.WebSocket, TOKEN, CONNECT_TIMEOUT); + + using var connectCancellation = new CancellationTokenSource(CONNECT_TIMEOUT); + if (!await node.ConnectAsync(connectCancellation.Token).ConfigureAwait(false)) + throw new InvalidOperationException($"node {i} could not connect"); + + m_nodes.Add(node); + + if ((i + 1) % 10 == 0) + log.WriteLine($" {i + 1} nodes connected ({watch.Elapsed.TotalSeconds:F1} s)"); + } + + watch.Stop(); + log.WriteLine($"{m_nodes.Count} nodes connected in {watch.Elapsed.TotalSeconds:F2} s"); + return watch.Elapsed.TotalSeconds; + } + + private async Task<(int Dispatched, double Seconds)> DispatchAsync(TextWriter log) + { + var payload = new byte[m_options.PayloadBytes]; + Random.Shared.NextBytes(payload); + + var interval = TimeSpan.FromSeconds(1.0 / m_options.Rate); + var deadline = Stopwatch.StartNew(); + var dispatched = 0; + var inFlight = new List(); + var nextAt = TimeSpan.Zero; + var sampler = SampleSlowPendingAsync(deadline); + + while (deadline.Elapsed < TimeSpan.FromSeconds(m_options.DurationSeconds)) + { + var wait = nextAt - deadline.Elapsed; + if (wait > TimeSpan.Zero) + await Task.Delay(wait).ConfigureAwait(false); + + nextAt += interval; + var node = dispatched % m_options.Nodes; + dispatched++; + + inFlight.Add(DispatchOneAsync(node, payload)); + + if (inFlight.Count >= 1024) + { + inFlight.RemoveAll(task => task.IsCompleted); + if (inFlight.Count >= 4096) + await Task.WhenAny(inFlight).ConfigureAwait(false); + } + + if (dispatched % (m_options.Rate * 5) == 0) + log.WriteLine($" {deadline.Elapsed.TotalSeconds:F0} s: {dispatched} dispatched, {TotalAcked()} acked"); + } + + var seconds = deadline.Elapsed.TotalSeconds; + await Task.WhenAll(inFlight).ConfigureAwait(false); + await sampler.ConfigureAwait(false); + return (dispatched, seconds); + } + + private async Task DispatchOneAsync(int node, byte[] payload) + { + switch (m_options.Mode) + { + case "per-client": + m_hosts[node].Service.DispatchToOnlyConnection(node, payload); + break; + + case "broadcast": + m_hosts[0].Service.DispatchBroadcast(node, payload); + break; + + default: + var status = await m_hosts[0].Service.DispatchAsync(node, payload).ConfigureAwait(false); + lock (m_outcomes) + m_outcomes[status.ToString()] = m_outcomes.GetValueOrDefault(status.ToString()) + 1; + break; + } + } + + private async Task SampleSlowPendingAsync(Stopwatch clock) + { + while (clock.Elapsed < TimeSpan.FromSeconds(m_options.DurationSeconds)) + { + for (var i = 0; i < m_nodes.Count; i++) + { + var host = m_options.Mode == "per-client" ? m_hosts[i] : m_hosts[0]; + var pending = host.Server.GetPendingCallbacks(m_nodes[i].ConnectionId); + + if (i == m_options.SlowNode && pending > m_slowPendingMax) + m_slowPendingMax = pending; + + if (i != m_options.SlowNode && pending > m_anyPendingMax) + m_anyPendingMax = pending; + } + + await Task.Delay(100).ConfigureAwait(false); + } + } + + private async Task DrainAsync() + { + var watch = Stopwatch.StartNew(); + while (watch.Elapsed < DRAIN && m_hosts.Any(host => host.Service.InFlight > 0)) + await Task.Delay(50).ConfigureAwait(false); + } + + private int TotalAcked() + { + return m_hosts.Sum(host => host.Service.Latency.Count); + } + + private void Collect(LoadReport report) + { + var all = new LatencyHistogram(); + var neighbours = new LatencyHistogram(); + var slow = new LatencyHistogram(); + + foreach (var host in m_hosts) + { + foreach (var (node, histogram) in host.Service.PerNode) + { + var summary = histogram.Summarize(); + _ = summary; + + foreach (var sample in Samples(histogram)) + { + all.Add(sample); + if (m_options.SlowNode >= 0) + (node == m_options.SlowNode ? slow : neighbours).Add(sample); + } + } + } + + report.Acked = all.Count; + report.AllNodes = all.Summarize(); + report.AnyNodePendingMax = m_anyPendingMax; + + if (m_options.SlowNode >= 0) + { + report.NeighboursOfSlow = neighbours.Summarize(); + report.SlowNode = slow.Summarize(); + report.SlowNodePendingMax = m_slowPendingMax; + + var throttled = m_hosts.Select(host => host.Throttling).FirstOrDefault(factory => factory != null)?.Throttled.Values.FirstOrDefault(); + if (throttled?.DisconnectedAt != null) + { + report.SlowNodeClosed = true; + report.SlowNodeClosedAfterSeconds = (throttled.DisconnectedAt.Value - RunStartedAt).TotalSeconds; + } + } + + lock (m_outcomes) + report.Outcomes = new Dictionary(m_outcomes); + + if (m_options.Mode == "broadcast") + report.BroadcastFramesPerNode = (long)m_nodes.Average(node => node.Received); + } + + private static IEnumerable Samples(LatencyHistogram histogram) + { + return histogram.Drain(); + } + + #endregion + + #region Properties + + private DateTime RunStartedAt { get; } = DateTime.UtcNow; + + #endregion + } +} diff --git a/Communication/OutWit.Communication.LoadTests/Metrics/LatencyHistogram.cs b/Communication/OutWit.Communication.LoadTests/Metrics/LatencyHistogram.cs new file mode 100644 index 0000000..9f83850 --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/Metrics/LatencyHistogram.cs @@ -0,0 +1,93 @@ +namespace OutWit.Communication.LoadTests.Metrics +{ + /// + /// Collects latencies in microseconds and answers percentiles. Thread-safe for adds. + /// + public sealed class LatencyHistogram + { + #region Fields + + private readonly object m_gate = new(); + + private readonly List m_samples = new(); + + #endregion + + #region Functions + + public void Add(double microseconds) + { + lock (m_gate) + m_samples.Add(microseconds); + } + + public LatencySummary Summarize() + { + double[] sorted; + lock (m_gate) + sorted = m_samples.OrderBy(sample => sample).ToArray(); + + if (sorted.Length == 0) + return new LatencySummary(0, 0, 0, 0, 0, 0); + + return new LatencySummary( + sorted.Length, + Percentile(sorted, 0.50), + Percentile(sorted, 0.95), + Percentile(sorted, 0.99), + sorted[^1], + sorted.Average()); + } + + /// + /// A copy of the samples, for merging into another histogram. + /// + public double[] Drain() + { + lock (m_gate) + return m_samples.ToArray(); + } + + private static double Percentile(double[] sorted, double p) + { + var index = (int)Math.Ceiling(p * sorted.Length) - 1; + return sorted[Math.Clamp(index, 0, sorted.Length - 1)]; + } + + #endregion + + #region Properties + + public int Count + { + get + { + lock (m_gate) + return m_samples.Count; + } + } + + #endregion + } + + /// + /// Percentiles in milliseconds. + /// + public sealed record LatencySummary(int Count, double P50Us, double P95Us, double P99Us, double MaxUs, double MeanUs) + { + public double P50Ms => P50Us / 1000.0; + + public double P95Ms => P95Us / 1000.0; + + public double P99Ms => P99Us / 1000.0; + + public double MaxMs => MaxUs / 1000.0; + + public double MeanMs => MeanUs / 1000.0; + + public override string ToString() + { + return $"n={Count} p50={P50Ms:F2} ms p95={P95Ms:F2} ms p99={P99Ms:F2} ms max={MaxMs:F1} ms"; + } + } +} diff --git a/Communication/OutWit.Communication.LoadTests/OutWit.Communication.LoadTests.csproj b/Communication/OutWit.Communication.LoadTests/OutWit.Communication.LoadTests.csproj new file mode 100644 index 0000000..5048c7d --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/OutWit.Communication.LoadTests.csproj @@ -0,0 +1,24 @@ + + + + Exe + net10.0 + enable + enable + false + true + true + Load harness for the 3.2 shared-server model: one WitServer serving N clients with targeted callbacks against N per-client servers; slow-neighbour isolation; connection storms. Not part of the CI gate; run it by hand and keep the numbers with the release notes. + + + + + + + + + + + + + diff --git a/Communication/OutWit.Communication.LoadTests/Program.cs b/Communication/OutWit.Communication.LoadTests/Program.cs new file mode 100644 index 0000000..97aec03 --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/Program.cs @@ -0,0 +1,25 @@ +using OutWit.Communication.LoadTests; + +var options = LoadOptions.Parse(args); +if (options.Help) +{ + Console.WriteLine(LoadOptions.Usage); + return 0; +} + +// One process hosts the server(s), the N clients and the measurement: give the pool the +// threads it will need at once instead of letting it inject them one or two per second. +ThreadPool.SetMinThreads(512, 512); + +Console.WriteLine(options); + +var report = await new LoadRunner(options).RunAsync(Console.Out); +report.Print(Console.Out); + +if (options.JsonPath != null) +{ + report.WriteJson(options.JsonPath); + Console.WriteLine($"report written to {options.JsonPath}"); +} + +return 0; diff --git a/Communication/OutWit.Communication.LoadTests/Server/LoadHost.cs b/Communication/OutWit.Communication.LoadTests/Server/LoadHost.cs new file mode 100644 index 0000000..163cfeb --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/Server/LoadHost.cs @@ -0,0 +1,99 @@ +using System.Net; +using System.Net.Sockets; +using OutWit.Communication.LoadTests.Contracts; +using OutWit.Communication.LoadTests.Transports; +using OutWit.Communication.Server; +using OutWit.Communication.Server.Callbacks; +using OutWit.Communication.Server.Tcp.Utils; +using OutWit.Communication.Server.WebSocket.Utils; + +namespace OutWit.Communication.LoadTests.Server +{ + /// + /// One server with one : the shared server of the 3.2 model, or one + /// of the N per-client servers of the baseline. Built on a real WebSocket or TCP transport, + /// optionally behind a . + /// + public sealed class LoadHost : IDisposable + { + #region Constructors + + private LoadHost(WitServer server, LoadService service, string clientEndpoint, ThrottlingTransportServerFactory? throttling) + { + Server = server; + Service = service; + ClientEndpoint = clientEndpoint; + Throttling = throttling; + } + + #endregion + + #region Functions + + public static LoadHost Start(bool webSocket, int maxClients, string token, TimeSpan timeout, + CallbackDeliveryOptions delivery, Func? throttleFor) + { + var port = FreePort(); + var service = new LoadService(); + ThrottlingTransportServerFactory? throttling = null; + + var server = WitServerBuilder.Build(options => + { + var transport = webSocket + ? options.WithWebSocket($"http://127.0.0.1:{port}/", maxClients, 10 * 1024 * 1024).TransportFactory! + : options.WithTcp(port, maxClients).TransportFactory!; + + if (throttleFor != null) + { + throttling = new ThrottlingTransportServerFactory(transport, throttleFor); + options.WithTransport(throttling); + } + + options.WithJson(); + options.WithEncryption(); + options.WithAccessToken(token); + options.WithTimeout(timeout); + options.WithService(service); + options.WithCallbackDelivery(delivery); + }); + + server.StartWaitingForConnection(); + + var endpoint = webSocket ? $"ws://127.0.0.1:{port}/" : $"127.0.0.1:{port}"; + return new LoadHost(server, service, endpoint, throttling); + } + + private static int FreePort() + { + var probe = new TcpListener(IPAddress.Loopback, 0); + probe.Start(); + var port = ((IPEndPoint)probe.LocalEndpoint).Port; + probe.Stop(); + return port; + } + + #endregion + + #region IDisposable + + public void Dispose() + { + Server.StopWaitingForConnection(); + Server.Dispose(); + } + + #endregion + + #region Properties + + public WitServer Server { get; } + + public LoadService Service { get; } + + public string ClientEndpoint { get; } + + public ThrottlingTransportServerFactory? Throttling { get; } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.LoadTests/Server/LoadService.cs b/Communication/OutWit.Communication.LoadTests/Server/LoadService.cs new file mode 100644 index 0000000..1772ed4 --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/Server/LoadService.cs @@ -0,0 +1,139 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using OutWit.Communication.LoadTests.Contracts; +using OutWit.Communication.LoadTests.Metrics; +using OutWit.Communication.Server.Callbacks; +using OutWit.Communication.Server.Connections; + +namespace OutWit.Communication.LoadTests.Server +{ + /// + /// The service the harness hosts: nodes attach (the connection id comes from + /// ), tasks are pushed through the event inside a + /// (or without one, for the broadcast comparison), and each + /// acknowledgement closes the dispatch-to-ack latency of its task. + /// + public sealed class LoadService : ILoadService + { + #region Events + + public event Action TaskReceived = delegate { }; + + #endregion + + #region Constants + + private static readonly TimeSpan COMPLETION_WAIT = TimeSpan.FromSeconds(10); + + #endregion + + #region Fields + + private readonly ConcurrentDictionary m_inFlight = new(); + + private long m_nextTaskId; + + #endregion + + #region ILoadService + + public Guid Attach(int nodeIndex) + { + var context = ConnectionContext.Current ?? throw new InvalidOperationException("Attach outside a request"); + Connections[nodeIndex] = context.ConnectionId; + return context.ConnectionId; + } + + public Task AckAsync(long taskId, int nodeIndex) + { + if (!m_inFlight.TryRemove(taskId, out var dispatchedAt)) + return Task.CompletedTask; + + var elapsedUs = (Stopwatch.GetTimestamp() - dispatchedAt) * 1_000_000.0 / Stopwatch.Frequency; + Latency.Add(elapsedUs); + PerNode.GetOrAdd(nodeIndex, _ => new LatencyHistogram()).Add(elapsedUs); + return Task.CompletedTask; + } + + #endregion + + #region Functions + + /// + /// Pushes one task to one node through a targeted callback and returns what the server did. + /// + public async Task DispatchAsync(int nodeIndex, byte[] payload) + { + if (!Connections.TryGetValue(nodeIndex, out var connectionId)) + return CallbackDeliveryStatus.UnknownConnection; + + var task = NewTask(nodeIndex, payload); + + using var scope = CallbackScope.Target(connectionId); + TaskReceived(task); + + // Under the Log policy a stalled socket's write never completes, by design; the + // harness gives up waiting after a while and reports the callback as still queued. + CallbackDeliveryReport report; + try + { + report = await scope.Completion.WaitAsync(COMPLETION_WAIT).ConfigureAwait(false); + } + catch (TimeoutException) + { + report = scope.Report; + } + + var status = report.Outcomes.Count == 0 ? CallbackDeliveryStatus.Refused : report.Outcomes[0].Status; + + if (status != CallbackDeliveryStatus.Sent) + m_inFlight.TryRemove(task.Id, out _); + + return status; + } + + /// + /// Pushes one task to every connection (the pre-3.2 shape) and lets one node ack it. + /// + public void DispatchBroadcast(int nodeIndex, byte[] payload) + { + TaskReceived(NewTask(nodeIndex, payload)); + } + + /// + /// Pushes a task without a scope: what a per-client server does (its only connection is the node). + /// + public void DispatchToOnlyConnection(int nodeIndex, byte[] payload) + { + TaskReceived(NewTask(nodeIndex, payload)); + } + + private LoadTask NewTask(int nodeIndex, byte[] payload) + { + var task = new LoadTask + { + Id = Interlocked.Increment(ref m_nextTaskId), + NodeIndex = nodeIndex, + DispatchedAtTicks = Stopwatch.GetTimestamp(), + Payload = payload + }; + + m_inFlight[task.Id] = task.DispatchedAtTicks; + return task; + } + + #endregion + + #region Properties + + public ConcurrentDictionary Connections { get; } = new(); + + public LatencyHistogram Latency { get; } = new(); + + public ConcurrentDictionary PerNode { get; } = new(); + + public int InFlight => m_inFlight.Count; + + #endregion + } +} diff --git a/Communication/OutWit.Communication.LoadTests/Transports/ThrottlingTransportServer.cs b/Communication/OutWit.Communication.LoadTests/Transports/ThrottlingTransportServer.cs new file mode 100644 index 0000000..817fd1e --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/Transports/ThrottlingTransportServer.cs @@ -0,0 +1,138 @@ +using OutWit.Communication.Interfaces; + +namespace OutWit.Communication.LoadTests.Transports +{ + /// + /// Wraps a real server transport and makes its writes slow: after the handshake replies, each + /// write waits before it goes to the wire, or forever while + /// . What the server sees is exactly a client whose socket does not drain. + /// Inbound frames are forwarded from the inner transport only once the server has subscribed, + /// so the inner transport's first-frame buffer flushes to a listener that exists. + /// + public sealed class ThrottlingTransportServer : ITransportServer + { + #region Constants + + /// + /// The initialization and authorization replies and the response to the node's Attach go + /// through untouched; the throttle starts with the first task callback. + /// + private const int HANDSHAKE_WRITES = 3; + + #endregion + + #region Events + + private TransportDataEventHandler? m_callback; + + public event TransportDataEventHandler Callback + { + add + { + var first = m_callback == null; + m_callback += value; + + if (first) + m_inner.Callback += OnInnerData; + } + remove => m_callback -= value; + } + + public event TransportEventHandler Disconnected = delegate { }; + + #endregion + + #region Fields + + private readonly ITransportServer m_inner; + + private readonly TaskCompletionSource m_stall = new(TaskCreationOptions.RunContinuationsAsynchronously); + + private int m_writes; + + #endregion + + #region Constructors + + public ThrottlingTransportServer(ITransportServer inner, TimeSpan delay, bool stalled) + { + m_inner = inner; + Delay = delay; + Stalled = stalled; + + m_inner.Disconnected += _ => + { + DisconnectedAt ??= DateTime.UtcNow; + Disconnected(Id); + }; + } + + #endregion + + #region Functions + + /// + /// Ends a stall; the writes waiting on it go through. + /// + public void Unstall() + { + Stalled = false; + m_stall.TrySetResult(true); + } + + private void OnInnerData(Guid sender, byte[] data) + { + m_callback?.Invoke(Id, data); + } + + #endregion + + #region ITransportServer + + public Task InitializeConnectionAsync(CancellationToken token) + { + return m_inner.InitializeConnectionAsync(token); + } + + public async Task SendBytesAsync(byte[] data) + { + var write = Interlocked.Increment(ref m_writes); + + if (write > HANDSHAKE_WRITES) + { + if (Stalled) + await m_stall.Task.ConfigureAwait(false); + else if (Delay > TimeSpan.Zero) + await Task.Delay(Delay).ConfigureAwait(false); + } + + await m_inner.SendBytesAsync(data).ConfigureAwait(false); + } + + public void Dispose() + { + DisconnectedAt ??= DateTime.UtcNow; + m_stall.TrySetResult(true); + m_inner.Dispose(); + } + + #endregion + + #region Properties + + public Guid Id => m_inner.Id; + + public bool CanReinitialize => m_inner.CanReinitialize; + + public TimeSpan Delay { get; } + + public bool Stalled { get; private set; } + + /// + /// When the connection went away (closed by the server's policy or by the client), or null. + /// + public DateTime? DisconnectedAt { get; private set; } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.LoadTests/Transports/ThrottlingTransportServerFactory.cs b/Communication/OutWit.Communication.LoadTests/Transports/ThrottlingTransportServerFactory.cs new file mode 100644 index 0000000..7ecf973 --- /dev/null +++ b/Communication/OutWit.Communication.LoadTests/Transports/ThrottlingTransportServerFactory.cs @@ -0,0 +1,92 @@ +using Microsoft.Extensions.Logging; +using OutWit.Communication.Interfaces; + +namespace OutWit.Communication.LoadTests.Transports +{ + /// + /// Wraps a real transport factory and hands the server a + /// for the connections the harness wants slow (chosen by arrival order), a pass-through for + /// the rest. The server's code path is the real one; only the socket "drains" slowly. + /// + public sealed class ThrottlingTransportServerFactory : ITransportServerFactory + { + #region Events + + public event TransportFactoryEventHandler NewClientConnected = delegate { }; + + #endregion + + #region Fields + + private readonly ITransportServerFactory m_inner; + + private readonly Func m_throttleFor; + + private int m_arrivals; + + #endregion + + #region Constructors + + /// The real factory. + /// Given the arrival index of a connection, the throttle to apply, or null for none. + public ThrottlingTransportServerFactory(ITransportServerFactory inner, Func throttleFor) + { + m_inner = inner; + m_throttleFor = throttleFor; + m_inner.NewClientConnected += OnNewClientConnected; + } + + #endregion + + #region Event Handlers + + private void OnNewClientConnected(ITransportServer transport) + { + var index = Interlocked.Increment(ref m_arrivals) - 1; + var throttle = m_throttleFor(index); + + if (throttle == null) + { + NewClientConnected(transport); + return; + } + + var throttled = new ThrottlingTransportServer(transport, throttle.Value.Delay, throttle.Value.Stalled); + Throttled[throttled.Id] = throttled; + NewClientConnected(throttled); + } + + #endregion + + #region ITransportServerFactory + + public void StartWaitingForConnection(ILogger? logger) + { + m_inner.StartWaitingForConnection(logger); + } + + public void StopWaitingForConnection() + { + m_inner.StopWaitingForConnection(); + } + + public void Dispose() + { + m_inner.Dispose(); + } + + #endregion + + #region Properties + + public IServerOptions Options => m_inner.Options; + + /// + /// The throttled transports by connection id. + /// + public System.Collections.Concurrent.ConcurrentDictionary Throttled { get; } = new(); + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Server/Authorization/IConnectionAuthenticator.cs b/Communication/OutWit.Communication.Server/Authorization/IConnectionAuthenticator.cs new file mode 100644 index 0000000..a60cf0b --- /dev/null +++ b/Communication/OutWit.Communication.Server/Authorization/IConnectionAuthenticator.cs @@ -0,0 +1,24 @@ +using System.Security.Claims; + +namespace OutWit.Communication.Server.Authorization +{ + /// + /// An optional extension of : a validator that also + /// implements this interface establishes a when a connection + /// authorizes, and the server keeps it on the connection for + /// . A validator that does not implement it + /// is used exactly as before, and the principal stays null. + /// + public interface IConnectionAuthenticator + { + /// + /// Validates the authorization token of a connecting client and, when it is valid, returns the + /// principal it identifies. + /// + /// The token from the authorization request. + /// The principal when the token is valid; null otherwise, or when the + /// validator cannot name one. + /// True when the token is valid and the connection may be authorized. + bool TryAuthenticate(string token, out ClaimsPrincipal? principal); + } +} diff --git a/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryMode.cs b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryMode.cs new file mode 100644 index 0000000..3946907 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryMode.cs @@ -0,0 +1,21 @@ +namespace OutWit.Communication.Server.Callbacks +{ + /// + /// Whether a server delivers an untargeted event raise to every authorized connection. + /// + public enum CallbackDeliveryMode + { + /// + /// The default and the pre-3.2 behaviour: an event raised outside a + /// goes to every authorized connection of the server. + /// + BroadcastAllowed, + + /// + /// An event must name its target: a raise outside a is refused + /// (logged, reported as , sent to nobody). For a + /// server whose clients must never see each other's events. + /// + TargetedOnly + } +} diff --git a/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryOptions.cs b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryOptions.cs new file mode 100644 index 0000000..9404ce2 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryOptions.cs @@ -0,0 +1,55 @@ +namespace OutWit.Communication.Server.Callbacks +{ + /// + /// How a server delivers callbacks: whether an untargeted raise is allowed, how many callbacks + /// may wait in one connection's outbound queue, and what happens beyond that. The defaults + /// reproduce the pre-3.2 behaviour exactly. + /// + public sealed class CallbackDeliveryOptions + { + #region Functions + + /// + /// Checks the option values. + /// + /// A bound is negative. + public void Validate() + { + if (MaxPendingCallbacks < 0) + throw new System.ArgumentOutOfRangeException(nameof(MaxPendingCallbacks), "The bound cannot be negative; 0 means unbounded"); + + if (MaxPendingCallbackBytes < 0) + throw new System.ArgumentOutOfRangeException(nameof(MaxPendingCallbackBytes), "The bound cannot be negative; 0 means unbounded"); + } + + #endregion + + #region Properties + + /// + /// Whether an event raised outside a goes to every authorized + /// connection (the default) or is refused. + /// + public CallbackDeliveryMode Mode { get; set; } = CallbackDeliveryMode.BroadcastAllowed; + + /// + /// The most callbacks that may wait in one connection's outbound queue; 0 (the default) is + /// unbounded. Responses are not counted. + /// + public int MaxPendingCallbacks { get; set; } + + /// + /// The most callback payload bytes that may wait in one connection's outbound queue; 0 (the + /// default) is unbounded. Responses are not counted. + /// + public long MaxPendingCallbackBytes { get; set; } + + /// + /// What happens when a bound is reached, and whether a callback send timeout closes the + /// connection. by default. + /// + public CallbackOverflowPolicy OverflowPolicy { get; set; } = CallbackOverflowPolicy.Log; + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryOutcome.cs b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryOutcome.cs new file mode 100644 index 0000000..03f8a65 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryOutcome.cs @@ -0,0 +1,49 @@ +using System; + +namespace OutWit.Communication.Server.Callbacks +{ + /// + /// One (server, connection) line of a . + /// + public sealed class CallbackDeliveryOutcome + { + #region Constructors + + public CallbackDeliveryOutcome(Guid serverId, Guid connectionId, CallbackDeliveryStatus status) + { + ServerId = serverId; + ConnectionId = connectionId; + Status = status; + } + + #endregion + + #region Functions + + public override string ToString() + { + return $"{Status} (connection {ConnectionId} on server {ServerId})"; + } + + #endregion + + #region Properties + + /// + /// The server that handled the raise (a service registered in several servers is raised in each). + /// + public Guid ServerId { get; } + + /// + /// The targeted connection; for a raise. + /// + public Guid ConnectionId { get; } + + /// + /// The status at the time the report was taken. + /// + public CallbackDeliveryStatus Status { get; } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryReport.cs b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryReport.cs new file mode 100644 index 0000000..1b9757b --- /dev/null +++ b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryReport.cs @@ -0,0 +1,57 @@ +using System.Collections.Generic; +using System.Linq; + +namespace OutWit.Communication.Server.Callbacks +{ + /// + /// What became of the events raised inside a : how many were raised + /// and, per server and target connection, what happened to each. A snapshot; read + /// for the current state or await + /// for the final one. + /// + public sealed class CallbackDeliveryReport + { + #region Constructors + + public CallbackDeliveryReport(int raised, IReadOnlyList outcomes) + { + Raised = raised; + Outcomes = outcomes; + } + + #endregion + + #region Functions + + public override string ToString() + { + return $"{Raised} raised: {string.Join(", ", Outcomes.Select(outcome => outcome.ToString()))}"; + } + + #endregion + + #region Properties + + /// + /// How many event raises the scope saw (each server that hosts the service counts the raise once). + /// + public int Raised { get; } + + /// + /// One line per (server, target connection); a refused broadcast is one line with an empty connection id. + /// + public IReadOnlyList Outcomes { get; } + + /// + /// True when at least one callback reached a transport. + /// + public bool AnySent => Outcomes.Any(outcome => outcome.Status == CallbackDeliveryStatus.Sent); + + /// + /// True when at least one callback was accepted (queued or sent) somewhere. + /// + public bool AnyAccepted => Outcomes.Any(outcome => outcome.Status is CallbackDeliveryStatus.Queued or CallbackDeliveryStatus.Sent); + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryStatus.cs b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryStatus.cs new file mode 100644 index 0000000..1dd3327 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Callbacks/CallbackDeliveryStatus.cs @@ -0,0 +1,32 @@ +namespace OutWit.Communication.Server.Callbacks +{ + /// + /// What happened to one callback on one connection. + /// + public enum CallbackDeliveryStatus + { + /// Accepted into the connection's outbound queue; not on the wire yet. + Queued, + + /// Written to the transport. + Sent, + + /// No connection with that id on this server. + UnknownConnection, + + /// The connection exists but has not finished the handshake. + NotAuthorized, + + /// The connection's callback queue is at its bound and the policy refused the callback. + QueueFull, + + /// The transport failed while writing the callback (the connection is gone). + SendFailed, + + /// The write did not finish within the server's timeout. + SendTimedOut, + + /// The server runs in and the raise named no target. + Refused + } +} diff --git a/Communication/OutWit.Communication.Server/Callbacks/CallbackOverflowPolicy.cs b/Communication/OutWit.Communication.Server/Callbacks/CallbackOverflowPolicy.cs new file mode 100644 index 0000000..be3be57 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Callbacks/CallbackOverflowPolicy.cs @@ -0,0 +1,31 @@ +namespace OutWit.Communication.Server.Callbacks +{ + /// + /// What the server does when a connection's callback queue reaches its bound + /// ( or + /// ), and what a callback send + /// timeout means. Responses are never subject to this policy. + /// + public enum CallbackOverflowPolicy + { + /// + /// The default and the pre-3.2 behaviour: the callback is queued anyway, the overflow and a + /// send timeout are logged as warnings, the connection stays open. + /// + Log, + + /// + /// The callback is refused () and the connection + /// is closed; a send timeout closes it too. For a server whose client is expected to keep up + /// and to reconnect when it cannot: closing is what tells it, and what keeps the other + /// connections of the server unaffected. + /// + CloseConnection, + + /// + /// The newest callback is dropped (), the + /// connection stays open. Only for events the service declares lossy. + /// + DropNewest + } +} diff --git a/Communication/OutWit.Communication.Server/Callbacks/CallbackScope.cs b/Communication/OutWit.Communication.Server/Callbacks/CallbackScope.cs new file mode 100644 index 0000000..7b7f6f2 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Callbacks/CallbackScope.cs @@ -0,0 +1,219 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using OutWit.Communication.Server.Connections; + +namespace OutWit.Communication.Server.Callbacks +{ + /// + /// Names the connections an event raised inside the scope is delivered to. A service opens the + /// scope, raises its ordinary C# event, and the server that hosts the service reads the target at + /// the moment the callback reaches it (the raise chain is synchronous, so an + /// set here is visible there; it also flows into tasks started inside + /// the scope). Without a scope the server delivers to every authorized connection, as it always + /// did. The scope collects what happened: for the state now, + /// for the state once every accepted send finished. + /// + /// + /// + /// using var scope = CallbackScope.Target(connectionId); + /// TaskReceived(delivery); // the contract's event + /// var report = await scope.Completion; // Sent, or UnknownConnection / QueueFull / ... + /// + /// + public sealed class CallbackScope : IDisposable + { + #region Fields + + private static readonly AsyncLocal CURRENT = new(); + + private readonly CallbackScope? m_previous; + + private readonly object m_gate = new(); + + private readonly List m_outcomes = new(); + + private int m_raised; + + private bool m_disposed; + + #endregion + + #region Constructors + + private CallbackScope(CallbackTarget recipients) + { + Recipients = recipients; + m_previous = CURRENT.Value; + CURRENT.Value = this; + } + + #endregion + + #region Functions + + /// + /// Opens a scope whose events go to one connection. + /// + /// The connection (see ). + /// The scope; dispose it when the raise is done. + public static CallbackScope Target(Guid connectionId) + { + return new CallbackScope(CallbackTarget.Connection(connectionId)); + } + + /// + /// Opens a scope whose events go to a set of connections. + /// + /// The connections; an empty set targets nobody. + /// The scope; dispose it when the raise is done. + /// is null. + public static CallbackScope Target(IReadOnlyCollection connectionIds) + { + return new CallbackScope(CallbackTarget.Connections(connectionIds)); + } + + /// + /// Opens a scope whose events go to the connection whose request is being processed: + /// "reply to the caller" without knowing its id. + /// + /// The scope; dispose it when the raise is done. + /// No request is being processed on this async flow. + public static CallbackScope TargetCaller() + { + var context = ConnectionContext.Current + ?? throw new InvalidOperationException("No request is being processed on this async flow; there is no caller to target"); + + return Target(context.ConnectionId); + } + + /// + /// Counts one event raise seen by a server. + /// + internal void RecordRaise() + { + Interlocked.Increment(ref m_raised); + } + + /// + /// Records what a server did with the callback for one target connection. + /// + /// The server. + /// The target; for a refused broadcast. + /// The status at enqueue time. + /// The send's final status, when the callback was queued. + internal void Record(Guid serverId, Guid connectionId, CallbackDeliveryStatus status, Task? completion) + { + lock (m_gate) + m_outcomes.Add(new PendingOutcome(serverId, connectionId, status, completion)); + } + + public override string ToString() + { + return $"callback scope: {Recipients}"; + } + + private CallbackDeliveryReport TakeReport() + { + PendingOutcome[] pending; + lock (m_gate) + pending = m_outcomes.ToArray(); + + var outcomes = pending + .Select(outcome => new CallbackDeliveryOutcome(outcome.ServerId, outcome.ConnectionId, outcome.StatusNow)) + .ToArray(); + + return new CallbackDeliveryReport(Volatile.Read(ref m_raised), outcomes); + } + + private async Task WaitForCompletionAsync() + { + Task[] sends; + lock (m_gate) + sends = m_outcomes.Where(outcome => outcome.Completion != null).Select(outcome => (Task)outcome.Completion!).ToArray(); + + if (sends.Length > 0) + await Task.WhenAll(sends).ConfigureAwait(false); + + return TakeReport(); + } + + #endregion + + #region IDisposable + + /// + /// Closes the scope: events raised afterwards on this flow are no longer targeted by it. + /// The report stays readable. + /// + public void Dispose() + { + if (m_disposed) + return; + + m_disposed = true; + CURRENT.Value = m_previous; + } + + #endregion + + #region Properties + + /// + /// The target of the scope open on this async flow, or + /// when none is. + /// + public static CallbackTarget Current => CURRENT.Value?.Recipients ?? CallbackTarget.Broadcast; + + /// + /// The scope open on this async flow, or null. + /// + internal static CallbackScope? CurrentScope => CURRENT.Value; + + /// + /// Who the events raised inside this scope go to. + /// + public CallbackTarget Recipients { get; } + + /// + /// What has happened so far: a callback still in a queue reads as . + /// + public CallbackDeliveryReport Report => TakeReport(); + + /// + /// The report once every callback accepted so far has been written or has failed. Await it + /// after the raise, inside or after the using. + /// + public Task Completion => WaitForCompletionAsync(); + + #endregion + + #region Nested Types + + private sealed class PendingOutcome + { + public PendingOutcome(Guid serverId, Guid connectionId, CallbackDeliveryStatus status, Task? completion) + { + ServerId = serverId; + ConnectionId = connectionId; + Status = status; + Completion = completion; + } + + public Guid ServerId { get; } + + public Guid ConnectionId { get; } + + public CallbackDeliveryStatus Status { get; } + + public Task? Completion { get; } + + public CallbackDeliveryStatus StatusNow => + Completion is { IsCompletedSuccessfully: true } ? Completion.Result : Status; + } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Server/Callbacks/CallbackTarget.cs b/Communication/OutWit.Communication.Server/Callbacks/CallbackTarget.cs new file mode 100644 index 0000000..91d08ae --- /dev/null +++ b/Communication/OutWit.Communication.Server/Callbacks/CallbackTarget.cs @@ -0,0 +1,100 @@ +using System; +using System.Collections.Generic; + +namespace OutWit.Communication.Server.Callbacks +{ + /// + /// Who receives a callback: every authorized connection of the server (a broadcast), one + /// connection, or a set of connections. + /// + public readonly struct CallbackTarget + { + #region Constants + + private static readonly Guid[] NO_CONNECTIONS = Array.Empty(); + + #endregion + + #region Constructors + + private CallbackTarget(IReadOnlyCollection? connectionIds) + { + ConnectionIds = connectionIds ?? NO_CONNECTIONS; + IsBroadcast = connectionIds == null; + } + + #endregion + + #region Functions + + /// + /// A callback for one connection. + /// + /// The connection (see ). + /// The target. + public static CallbackTarget Connection(Guid connectionId) + { + return new CallbackTarget(new[] { connectionId }); + } + + /// + /// A callback for a set of connections. An empty set targets nobody (and is not a broadcast). + /// + /// The connections. + /// The target. + /// is null. + public static CallbackTarget Connections(IReadOnlyCollection connectionIds) + { + if (connectionIds == null) + throw new ArgumentNullException(nameof(connectionIds)); + + return new CallbackTarget(connectionIds); + } + + /// + /// Whether is among the targets (always true for a broadcast). + /// + /// The connection. + /// True when the connection should receive the callback. + public bool Includes(Guid connectionId) + { + if (IsBroadcast) + return true; + + foreach (var id in ConnectionIds) + { + if (id == connectionId) + return true; + } + + return false; + } + + public override string ToString() + { + return IsBroadcast ? "broadcast" : $"{ConnectionIds.Count} connection(s)"; + } + + #endregion + + #region Properties + + /// + /// Every authorized connection of the server; what an event raised outside a + /// targets. + /// + public static CallbackTarget Broadcast => new(null); + + /// + /// True for . + /// + public bool IsBroadcast { get; } + + /// + /// The targeted connections; empty for a broadcast. + /// + public IReadOnlyCollection ConnectionIds { get; } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Server/Connections/ConnectionContext.cs b/Communication/OutWit.Communication.Server/Connections/ConnectionContext.cs new file mode 100644 index 0000000..79c32f6 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Connections/ConnectionContext.cs @@ -0,0 +1,128 @@ +using System; +using System.Security.Claims; +using System.Threading; + +namespace OutWit.Communication.Server.Connections +{ + /// + /// What a service can know about the connection a request arrived on: the connection id, the + /// server that accepted it, its transport, and the principal established when that connection + /// authorized. An immutable snapshot, taken per request; a service never sees the mutable + /// . Read it through (an + /// the server sets around every request and around the authorization + /// handshake) or through an . + /// + public sealed class ConnectionContext + { + #region Fields + + private static readonly AsyncLocal CURRENT = new(); + + #endregion + + #region Constructors + + public ConnectionContext(Guid connectionId, Guid serverId, string? serverName, string transport, + ClaimsPrincipal? principal, DateTimeOffset authorizedAtUtc) + { + ConnectionId = connectionId; + ServerId = serverId; + ServerName = serverName; + Transport = transport; + Principal = principal; + AuthorizedAtUtc = authorizedAtUtc; + } + + #endregion + + #region Functions + + /// + /// Makes the current one until the returned scope is disposed, + /// then restores what was current before. The server uses it around each request; a unit + /// test of a service uses it to simulate a connection. + /// + /// The context to expose; null clears the current one. + /// A scope that restores the previous context on dispose. + public static IDisposable BeginScope(ConnectionContext? context) + { + var previous = CURRENT.Value; + CURRENT.Value = context; + return new Scope(previous); + } + + public override string ToString() + { + return $"connection {ConnectionId} on server {ServerName ?? ServerId.ToString()} ({Transport})"; + } + + #endregion + + #region Properties + + /// + /// The context of the request being processed on this async flow, or null outside a + /// request (a timer, a background task, a raise from another thread). + /// + public static ConnectionContext? Current => CURRENT.Value; + + /// + /// The connection the request arrived on; unique per accepted transport for its lifetime. + /// + public Guid ConnectionId { get; } + + /// + /// The of the server that accepted the connection. + /// + public Guid ServerId { get; } + + /// + /// The of that server, when it has one. + /// + public string? ServerName { get; } + + /// + /// The transport name the server listens on (IServerOptions.Transport). + /// + public string Transport { get; } + + /// + /// The principal established at authorization when the token validator implements + /// ; null otherwise, and null during + /// the authorization handshake itself. + /// + public ClaimsPrincipal? Principal { get; } + + /// + /// When the connection authorized; during the handshake. + /// + public DateTimeOffset AuthorizedAtUtc { get; } + + #endregion + + #region Nested Types + + private sealed class Scope : IDisposable + { + private readonly ConnectionContext? m_previous; + + private bool m_disposed; + + public Scope(ConnectionContext? previous) + { + m_previous = previous; + } + + public void Dispose() + { + if (m_disposed) + return; + + m_disposed = true; + CURRENT.Value = m_previous; + } + } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Server/Connections/ConnectionContextAccessor.cs b/Communication/OutWit.Communication.Server/Connections/ConnectionContextAccessor.cs new file mode 100644 index 0000000..20c5a77 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Connections/ConnectionContextAccessor.cs @@ -0,0 +1,16 @@ +namespace OutWit.Communication.Server.Connections +{ + /// + /// The default : reads . + /// Register it as a singleton; it holds no state of its own. + /// + public sealed class ConnectionContextAccessor : IConnectionContextAccessor + { + #region IConnectionContextAccessor + + /// + public ConnectionContext? Current => ConnectionContext.Current; + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Server/Connections/ConnectionInfo.cs b/Communication/OutWit.Communication.Server/Connections/ConnectionInfo.cs index 48fbe02..2d7ceb8 100644 --- a/Communication/OutWit.Communication.Server/Connections/ConnectionInfo.cs +++ b/Communication/OutWit.Communication.Server/Connections/ConnectionInfo.cs @@ -1,5 +1,6 @@ using System; using System.Collections.Generic; +using System.Security.Claims; using System.Threading; using System.Threading.Channels; using OutWit.Communication.Interfaces; @@ -8,9 +9,11 @@ namespace OutWit.Communication.Server.Connections { /// /// One client's connection on the server: its transport, its per-connection - /// encryptor, its handshake state, and the two things that keep the server - /// correct under load — a single inbound queue processed in order, and a send - /// lock so responses and callbacks never interleave on the one transport. + /// encryptor, its handshake state, the principal it authorized with, and the + /// two things that keep the server correct under load — a single inbound queue + /// processed in order, and a single outbound queue () + /// so responses and callbacks leave in order and never interleave on the one + /// transport. /// public class ConnectionInfo : IDisposable { @@ -70,12 +73,20 @@ public void Reinitialize() return; State = ConnectionState.Connected; + Principal = null; + AuthorizedAtUtc = default; } /// - /// Stops accepting more inbound frames and lets the processing loop drain - /// and exit. Idempotent. + /// Gives the connection its outbound queue. Called once by the server right + /// after the connection is created; the queue needs the server's serializer + /// and options, which the connection does not know. /// + internal void AttachOutbox(ConnectionOutbox outbox) + { + Outbox = outbox; + } + /// /// Returns the cached response for an invocation already executed on /// this connection, if it is still within the bounded window. @@ -118,6 +129,10 @@ public void CacheResponse(Guid invocationId, byte[] response) } } + /// + /// Stops accepting more inbound frames and lets the processing loop drain + /// and exit. Idempotent. + /// public void CompleteInbound() { Inbound.Writer.TryComplete(); @@ -135,6 +150,7 @@ public void Dispose() m_disposed = true; Inbound.Writer.TryComplete(); + Outbox?.Dispose(); SendLock.Dispose(); m_encryptor?.Dispose(); } @@ -175,10 +191,39 @@ public IEncryptorServer Encryptor public Guid Id => Transport.Id; + /// + /// The principal established at authorization when the server's token + /// validator implements ; + /// null otherwise, and null again after a re-initialization. + /// + public ClaimsPrincipal? Principal { get; internal set; } + + /// + /// When the connection authorized; default until it has. + /// + public DateTimeOffset AuthorizedAtUtc { get; internal set; } + + /// + /// Callbacks waiting in the connection's outbound queue. + /// + public long PendingCallbacks => Outbox?.PendingCallbacks ?? 0; + + /// + /// Payload bytes of the callbacks waiting in the connection's outbound queue. + /// + public long PendingCallbackBytes => Outbox?.PendingCallbackBytes ?? 0; + + /// + /// Taken by the outbound writer around each transport write. The writer is + /// the only sender since 3.2; the lock stays for anyone who wrote to the + /// transport directly. + /// public SemaphoreSlim SendLock { get; } public Channel Inbound { get; } + internal ConnectionOutbox? Outbox { get; private set; } + #endregion } diff --git a/Communication/OutWit.Communication.Server/Connections/ConnectionOutbox.cs b/Communication/OutWit.Communication.Server/Connections/ConnectionOutbox.cs new file mode 100644 index 0000000..1b1a7b6 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Connections/ConnectionOutbox.cs @@ -0,0 +1,291 @@ +using System; +using System.Threading; +using System.Threading.Channels; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using OutWit.Communication.Messages; +using OutWit.Communication.Server.Callbacks; + +namespace OutWit.Communication.Server.Connections +{ + /// + /// The one ordered outbound queue of a connection, drained by one writer task. Responses, + /// handshake replies and callbacks all go through it, so the frames of a connection leave in + /// the order they were enqueued and the AEAD counter advances in wire order. Callbacks are + /// counted against the server's bounds; responses never + /// are and are never refused. A frame's completion task reports what became of it. + /// + internal sealed class ConnectionOutbox : IDisposable + { + #region Fields + + private readonly Channel m_frames = Channel.CreateUnbounded(new UnboundedChannelOptions + { + SingleReader = true, + SingleWriter = false + }); + + private readonly Guid m_connectionId; + + private readonly Func m_write; + + private readonly CallbackDeliveryOptions m_options; + + private readonly TimeSpan? m_callbackTimeout; + + private readonly ILogger? m_logger; + + private readonly Action m_close; + + private readonly Task m_writer; + + private long m_pendingCallbacks; + + private long m_pendingCallbackBytes; + + private int m_overflowLogged; + + private bool m_disposed; + + #endregion + + #region Constructors + + /// The connection, for the log. + /// Encrypts, serializes and writes one message to the transport. + /// The server's callback delivery options. + /// How long one callback write may take before it is reported as timed out; null or zero for no limit. + /// The server's logger. + /// Closes the connection with a reason; used by . + public ConnectionOutbox(Guid connectionId, Func write, CallbackDeliveryOptions options, + TimeSpan? callbackTimeout, ILogger? logger, Action close) + { + m_connectionId = connectionId; + m_write = write; + m_options = options; + m_callbackTimeout = callbackTimeout; + m_logger = logger; + m_close = close; + + m_writer = Task.Run(WriteAllAsync); + } + + #endregion + + #region Functions + + /// + /// Queues a response or a handshake reply. Never refused. + /// + /// The message. + /// Completes when the frame was written (or failed; the status says which). + public Task EnqueueResponse(WitMessage message) + { + var frame = new OutboundFrame(message, false, 0); + + if (!m_frames.Writer.TryWrite(frame)) + frame.Completion.TrySetResult(CallbackDeliveryStatus.SendFailed); + + return frame.Completion.Task; + } + + /// + /// Queues a callback, subject to the bounds and the overflow policy. + /// + /// The callback message. + /// The send's final status, when the callback was accepted. + /// Why it was not, otherwise. + /// True when the callback is in the queue. + public bool TryEnqueueCallback(WitMessage message, out Task completion, out CallbackDeliveryStatus refusal) + { + var bytes = message.Data?.Length ?? 0; + + if (IsOverBound(bytes)) + { + switch (m_options.OverflowPolicy) + { + case CallbackOverflowPolicy.CloseConnection: + m_logger?.LogWarning("Callback queue of client {ClientId} is full ({Pending} pending, {PendingBytes} bytes); closing the connection", + m_connectionId, Volatile.Read(ref m_pendingCallbacks), Volatile.Read(ref m_pendingCallbackBytes)); + completion = Task.FromResult(CallbackDeliveryStatus.QueueFull); + refusal = CallbackDeliveryStatus.QueueFull; + m_close("callback queue full"); + return false; + + case CallbackOverflowPolicy.DropNewest: + LogOverflowOnce("dropping the newest callback"); + completion = Task.FromResult(CallbackDeliveryStatus.QueueFull); + refusal = CallbackDeliveryStatus.QueueFull; + return false; + + default: + LogOverflowOnce("queuing anyway"); + break; + } + } + + var frame = new OutboundFrame(message, true, bytes); + Interlocked.Increment(ref m_pendingCallbacks); + Interlocked.Add(ref m_pendingCallbackBytes, bytes); + + if (!m_frames.Writer.TryWrite(frame)) + { + Interlocked.Decrement(ref m_pendingCallbacks); + Interlocked.Add(ref m_pendingCallbackBytes, -bytes); + completion = Task.FromResult(CallbackDeliveryStatus.SendFailed); + refusal = CallbackDeliveryStatus.SendFailed; + return false; + } + + completion = frame.Completion.Task; + refusal = CallbackDeliveryStatus.Queued; + return true; + } + + private bool IsOverBound(int bytes) + { + if (m_options.MaxPendingCallbacks > 0 && Volatile.Read(ref m_pendingCallbacks) >= m_options.MaxPendingCallbacks) + return true; + + if (m_options.MaxPendingCallbackBytes > 0 && Volatile.Read(ref m_pendingCallbackBytes) + bytes > m_options.MaxPendingCallbackBytes) + return true; + + return false; + } + + private void LogOverflowOnce(string action) + { + if (Interlocked.Exchange(ref m_overflowLogged, 1) != 0) + return; + + m_logger?.LogWarning("Callback queue of client {ClientId} passed its bound ({Pending} pending, {PendingBytes} bytes); {Action}", + m_connectionId, Volatile.Read(ref m_pendingCallbacks), Volatile.Read(ref m_pendingCallbackBytes), action); + } + + private async Task WriteAllAsync() + { + try + { + while (await m_frames.Reader.WaitToReadAsync().ConfigureAwait(false)) + { + while (m_frames.Reader.TryRead(out var frame)) + await WriteFrameAsync(frame).ConfigureAwait(false); + } + } + catch (Exception e) + { + m_logger?.LogError(e, "Outbound writer of client {ClientId} failed", m_connectionId); + } + } + + private async Task WriteFrameAsync(OutboundFrame frame) + { + if (frame.IsCallback) + { + Interlocked.Decrement(ref m_pendingCallbacks); + Interlocked.Add(ref m_pendingCallbackBytes, -frame.Bytes); + Volatile.Write(ref m_overflowLogged, 0); + } + + if (m_disposed) + { + frame.Completion.TrySetResult(CallbackDeliveryStatus.SendFailed); + return; + } + + var status = CallbackDeliveryStatus.Sent; + try + { + var write = m_write(frame.Message); + + if (frame.IsCallback && m_callbackTimeout is { } timeout && timeout > TimeSpan.Zero) + { + try + { + await write.WaitAsync(timeout).ConfigureAwait(false); + } + catch (TimeoutException) + { + status = CallbackDeliveryStatus.SendTimedOut; + m_logger?.LogWarning("Callback to client {ClientId} timed out", m_connectionId); + + if (m_options.OverflowPolicy == CallbackOverflowPolicy.CloseConnection) + m_close("callback send timed out"); + + // The write still owns the transport: the next frame must not start + // before it ends, or the frames would interleave on the wire. + await write.ConfigureAwait(false); + } + } + else + { + await write.ConfigureAwait(false); + } + } + catch (Exception e) + { + status = CallbackDeliveryStatus.SendFailed; + m_logger?.LogError(e, "Failed to send message to client {ClientId}", m_connectionId); + } + + frame.Completion.TrySetResult(status); + } + + #endregion + + #region IDisposable + + public void Dispose() + { + if (m_disposed) + return; + + m_disposed = true; + m_frames.Writer.TryComplete(); + } + + #endregion + + #region Properties + + /// + /// Callbacks waiting in the queue. + /// + public long PendingCallbacks => Volatile.Read(ref m_pendingCallbacks); + + /// + /// Payload bytes of the callbacks waiting in the queue. + /// + public long PendingCallbackBytes => Volatile.Read(ref m_pendingCallbackBytes); + + /// + /// The writer task, for tests that wait for the queue to drain. + /// + internal Task Writer => m_writer; + + #endregion + + #region Nested Types + + private sealed class OutboundFrame + { + public OutboundFrame(WitMessage message, bool isCallback, int bytes) + { + Message = message; + IsCallback = isCallback; + Bytes = bytes; + Completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + + public WitMessage Message { get; } + + public bool IsCallback { get; } + + public int Bytes { get; } + + public TaskCompletionSource Completion { get; } + } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Server/Connections/IConnectionContextAccessor.cs b/Communication/OutWit.Communication.Server/Connections/IConnectionContextAccessor.cs new file mode 100644 index 0000000..ac35c35 --- /dev/null +++ b/Communication/OutWit.Communication.Server/Connections/IConnectionContextAccessor.cs @@ -0,0 +1,15 @@ +namespace OutWit.Communication.Server.Connections +{ + /// + /// Gives a service the of the request it is handling by + /// injection rather than through the static , so a + /// unit test can hand the service a fake. + /// + public interface IConnectionContextAccessor + { + /// + /// The context of the request being processed on this async flow, or null outside a request. + /// + ConnectionContext? Current { get; } + } +} diff --git a/Communication/OutWit.Communication.Server/OutWit.Communication.Server.csproj b/Communication/OutWit.Communication.Server/OutWit.Communication.Server.csproj index 6556d9e..8a3b77b 100644 --- a/Communication/OutWit.Communication.Server/OutWit.Communication.Server.csproj +++ b/Communication/OutWit.Communication.Server/OutWit.Communication.Server.csproj @@ -2,7 +2,7 @@ net10.0;net9.0;net8.0;net7.0;net6.0 - 3.1.1 + 3.2.0 Base server library for the WitRPC framework, providing core functionality to host services and handle incoming RPC connections over various transports. OutWit;Communication;WitRPC;Server diff --git a/Communication/OutWit.Communication.Server/README.md b/Communication/OutWit.Communication.Server/README.md index 0ec2602..f7d4eac 100644 --- a/Communication/OutWit.Communication.Server/README.md +++ b/Communication/OutWit.Communication.Server/README.md @@ -97,7 +97,7 @@ When clients connect and invoke methods: - When a client calls a service method, the server receives the request, deserializes it to a `WitRequest` object, and invokes the corresponding method on your service object. The return value (or any exception) is captured and sent back as a response. -- If your service raises an event (e.g., calls an event delegate to notify of some change), the server framework will automatically forward that event to all connected clients that have subscribed to it. This allows for real-time push notifications from server to clients. +- If your service raises an event (e.g., calls an event delegate to notify of some change), the server framework will automatically forward that event to all connected clients that have subscribed to it. This allows for real-time push notifications from server to clients. Since 3.2 a service can also address an event to one client or a set of clients (see below). To stop the server when your application is shutting down: @@ -108,6 +108,46 @@ server.Dispose(); This will stop listening for new connections, close all existing client connections, and release resources like ports or pipe handles. +### Connection context and targeted events (3.2) + +A service method can find out which connection it is serving, and an event can be sent to one connection instead of everyone. Nothing changes on the wire and nothing changes for a service that does not use it: an event raised the ordinary way still reaches every connected client. + +**Who is calling.** `ConnectionContext.Current` is set for the duration of every request (an `AsyncLocal`, so it follows the method through its `await`s and into tasks it starts) and is `null` outside one: + +```csharp +public Guid WhoAmI() => ConnectionContext.Current!.ConnectionId; +``` + +It carries the connection id, the server (`ServerId`, `ServerName`, `Transport`) and, when your token validator also implements `IConnectionAuthenticator`, the `ClaimsPrincipal` established when the connection authorized. Prefer `IConnectionContextAccessor` (register `ConnectionContextAccessor` as a singleton) when the service takes it by injection, and `ConnectionContext.BeginScope(...)` to simulate a connection in a unit test. + +**Events for one client.** Open a `CallbackScope` around the raise; the server that hosts the service reads the target when the callback reaches it: + +```csharp +public void Notify(Guid connectionId, string message) +{ + using var scope = CallbackScope.Target(connectionId); // or Target(ids), or TargetCaller() + Notified(message); // the contract's ordinary event + // scope.Report says what happened now; await scope.Completion for the final word: + // Sent, UnknownConnection, NotAuthorized, QueueFull, SendFailed, SendTimedOut +} +``` + +`TargetCaller()` is "reply to whoever is calling" without knowing the id. A service registered in several servers is raised in each; the report lists the outcome per server, and the one that owns the connection delivers. + +**Delivery options.** Every connection has one ordered outbound queue: responses, handshake replies and events leave in the order they were queued, and an event raised inside a method before it returns is written before that method's response. The queue can be bounded per connection: + +```csharp +options.WithCallbackDelivery(delivery => +{ + delivery.MaxPendingCallbacks = 256; // 0 = unbounded (default) + delivery.MaxPendingCallbackBytes = 64 * 1024 * 1024; // 0 = unbounded (default) + delivery.OverflowPolicy = CallbackOverflowPolicy.CloseConnection; // Log (default) | CloseConnection | DropNewest +}); +options.WithTargetedCallbacksOnly(); // refuse an event raised outside a CallbackScope +``` + +Responses are never counted against the bound and never dropped. With the default `Log` policy an overflow and a send timeout are logged and the connection stays open, exactly as before 3.2; `CloseConnection` closes a client that cannot keep up (and one whose send times out), which is what keeps every other connection of the server unaffected; `DropNewest` is for events the service declares lossy. `WithTargetedCallbacksOnly()` is for a server whose clients must never see each other's events: a raise without a target is refused and logged. + ### Further Documentation Refer to the [WitRPC documentation](https://witrpc.io/) for more on server configuration, advanced options (like custom authentication via `WithAccessTokenValidator` or service discovery), and best practices for hosting WitRPC services. diff --git a/Communication/OutWit.Communication.Server/WitServer.cs b/Communication/OutWit.Communication.Server/WitServer.cs index da6bc13..b23213d 100644 --- a/Communication/OutWit.Communication.Server/WitServer.cs +++ b/Communication/OutWit.Communication.Server/WitServer.cs @@ -10,6 +10,8 @@ using OutWit.Communication.Model; using OutWit.Communication.Requests; using OutWit.Communication.Responses; +using OutWit.Communication.Server.Authorization; +using OutWit.Communication.Server.Callbacks; using OutWit.Communication.Server.Connections; using OutWit.Communication.Utils; @@ -23,6 +25,8 @@ public class WitServer : IDisposable private readonly SemaphoreSlim m_processingLimit; + private readonly CallbackDeliveryOptions m_callbackDelivery; + private bool m_isDisposed; #endregion @@ -34,7 +38,22 @@ public WitServer(ITransportServerFactory transportFactory, IEncryptorServerFacto IRequestProcessor requestProcessor, IDiscoveryServer? discoveryServer, ILogger? logger, TimeSpan? timeout, string? name, string? description, int maxConcurrentRequests = int.MaxValue, TimeSpan? handshakeTimeout = null) + : this(transportFactory, encryptorFactory, tokenValidator, parametersSerializer, messageSerializer, requestProcessor, + discoveryServer, logger, timeout, name, description, maxConcurrentRequests, handshakeTimeout, null) { + } + + /// How callbacks are delivered (targeting mode, per-connection + /// bounds, overflow policy); null for the defaults, which reproduce the pre-3.2 behaviour. + public WitServer(ITransportServerFactory transportFactory, IEncryptorServerFactory encryptorFactory, + IAccessTokenValidator tokenValidator, IMessageSerializer parametersSerializer, IMessageSerializer messageSerializer, + IRequestProcessor requestProcessor, IDiscoveryServer? discoveryServer, + ILogger? logger, TimeSpan? timeout, string? name, string? description, int maxConcurrentRequests, + TimeSpan? handshakeTimeout, CallbackDeliveryOptions? callbackDelivery) + { + m_callbackDelivery = callbackDelivery ?? new CallbackDeliveryOptions(); + m_callbackDelivery.Validate(); + TransportFactory = transportFactory; EncryptorFactory = encryptorFactory; ParametersSerializer = parametersSerializer; @@ -208,9 +227,23 @@ private WitMessage ProcessAuthorization(ConnectionInfo connection, WitMessage me try { - bool authorized = TokenValidator.IsAuthorizationTokenValid(request.Token); + // A validator that can name the principal does so once, here; the + // connection keeps it for ConnectionContext.Principal. Any other + // validator is used exactly as before. + bool authorized; + System.Security.Claims.ClaimsPrincipal? principal = null; + + if (TokenValidator is IConnectionAuthenticator authenticator) + authorized = authenticator.TryAuthenticate(request.Token, out principal); + else + authorized = TokenValidator.IsAuthorizationTokenValid(request.Token); + if (authorized) + { + connection.Principal = principal; + connection.AuthorizedAtUtc = DateTimeOffset.UtcNow; connection.State = ConnectionState.Authorized; + } var response = new WitResponseAuthorization { @@ -233,8 +266,22 @@ private WitMessage ProcessAuthorization(ConnectionInfo connection, WitMessage me #region Processing + /// + /// The snapshot a service sees as while it + /// handles a request of . + /// + private ConnectionContext CreateContext(ConnectionInfo connection) + { + return new ConnectionContext(connection.Id, Id, Name, TransportFactory.Options.Transport, + connection.Principal, connection.AuthorizedAtUtc); + } + private async Task ProcessMessage(ConnectionInfo connection, WitMessage message) { + // Visible to the token validator, the processor and the service method + // (and whatever they await or start); gone when this method completes. + using var contextScope = ConnectionContext.BeginScope(CreateContext(connection)); + var request = message.Data.GetRequest(MessageSerializer); if (request != null && request.InvocationId != Guid.Empty && @@ -313,18 +360,36 @@ private async Task Decrypt(ConnectionInfo connection, WitMessage mes #region Send + /// + /// Queues a response or a handshake reply on the connection's outbound queue and + /// waits until the writer has put it on the wire (or failed to). Order on the wire + /// is the order of enqueueing; callbacks share the same queue. + /// private async Task SendMessageAsync(ConnectionInfo connection, WitMessage message) + { + var outbox = connection.Outbox; + if (outbox == null) + { + await WriteMessageAsync(connection, message).ConfigureAwait(false); + return; + } + + await outbox.EnqueueResponse(message).ConfigureAwait(false); + } + + /// + /// The one place a frame is encrypted, serialized and written: called by the + /// connection's outbound writer, one frame at a time, so the AEAD counter advances + /// in wire order. A failure is the writer's to report; the exception propagates. + /// + private async Task WriteMessageAsync(ConnectionInfo connection, WitMessage message) { await connection.SendLock.WaitAsync().ConfigureAwait(false); try { - var encryptedMessage = await Encrypt(connection, message); + var encryptedMessage = await Encrypt(connection, message).ConfigureAwait(false); var data = MessageSerializer.Serialize(encryptedMessage); - await connection.Transport.SendBytesAsync(data); - } - catch (Exception e) - { - Logger?.LogError(e, "Failed to send message to client {ClientId}", connection.Id); + await connection.Transport.SendBytesAsync(data).ConfigureAwait(false); } finally { @@ -359,6 +424,12 @@ private void CloseConnection(ConnectionInfo connection) connection.Transport.Dispose(); } + private void CloseConnection(ConnectionInfo connection, string reason) + { + Logger?.LogWarning("Closing the connection of client {ClientId}: {Reason}", connection.Id, reason); + CloseConnection(connection); + } + #endregion #region Connection Loop @@ -438,7 +509,11 @@ private async Task ProcessFrameAsync(ConnectionInfo connection, byte[] dat return false; } - await SendMessageAsync(connection, ProcessAuthorization(connection, decrypted)); + WitMessage authorizationReply; + using (ConnectionContext.BeginScope(CreateContext(connection))) + authorizationReply = ProcessAuthorization(connection, decrypted); + + await SendMessageAsync(connection, authorizationReply); if (!connection.IsAuthorized) { @@ -457,7 +532,6 @@ private async Task ProcessFrameAsync(ConnectionInfo connection, byte[] dat return false; } - var tag = connection.Id.ToString().Substring(0, 4); var responseMessage = await ProcessMessage(connection, decrypted); await SendMessageAsync(connection, responseMessage); return true; @@ -472,11 +546,30 @@ private async Task ProcessFrameAsync(ConnectionInfo connection, byte[] dat #region Callbacks + /// + /// Delivers an event the service raised. The raise reaches this method on the + /// raising thread, so the the service opened (if any) + /// is still current: its target decides which connections get the callback, + /// and it collects what happened to each. Without a scope the callback goes to + /// every authorized connection, unless the server is + /// . + /// private void OnCallback(WitRequest? request) { if (request == null || m_isDisposed) return; + var scope = CallbackScope.CurrentScope; + var target = scope?.Recipients ?? CallbackTarget.Broadcast; + scope?.RecordRaise(); + + if (target.IsBroadcast && m_callbackDelivery.Mode == CallbackDeliveryMode.TargetedOnly) + { + Logger?.LogError("Event {EventName} was raised without a target on a targeted-only server; not delivered", request.MethodName); + scope?.Record(Id, Guid.Empty, CallbackDeliveryStatus.Refused, null); + return; + } + byte[] callback; try { @@ -488,19 +581,37 @@ private void OnCallback(WitRequest? request) return; } - foreach (var connection in m_connections.Values) + if (target.IsBroadcast) { - // Only clients that finished the handshake receive events, and the - // send goes through the connection's send lock so it never - // interleaves with a response on the same transport. + foreach (var connection in m_connections.Values) + { + // Only clients that finished the handshake receive events. + if (connection.IsAuthorized) + DispatchCallback(connection, callback, scope); + } + + return; + } + + foreach (var connectionId in target.ConnectionIds) + { + if (!m_connections.TryGetValue(connectionId, out var connection)) + { + scope?.Record(Id, connectionId, CallbackDeliveryStatus.UnknownConnection, null); + continue; + } + if (!connection.IsAuthorized) + { + scope?.Record(Id, connectionId, CallbackDeliveryStatus.NotAuthorized, null); continue; + } - _ = SendCallbackAsync(connection, callback); + DispatchCallback(connection, callback, scope); } } - private async Task SendCallbackAsync(ConnectionInfo connection, byte[] callback) + private void DispatchCallback(ConnectionInfo connection, byte[] callback, CallbackScope? scope) { var message = new WitMessage { @@ -509,23 +620,19 @@ private async Task SendCallbackAsync(ConnectionInfo connection, byte[] callback) Data = callback }; - var send = SendMessageAsync(connection, message); - - if (Timeout != null && Timeout != TimeSpan.Zero) + var outbox = connection.Outbox; + if (outbox == null) { - try - { - await send.WaitAsync(Timeout.Value).ConfigureAwait(false); - } - catch (TimeoutException) - { - Logger?.LogWarning("Callback to client {ClientId} timed out", connection.Id); - } + scope?.Record(Id, connection.Id, CallbackDeliveryStatus.SendFailed, null); + return; } + + // The queue enforces the per-connection bounds and the overflow policy; + // the writer reports the send's outcome through the completion task. + if (outbox.TryEnqueueCallback(message, out var completion, out var refusal)) + scope?.Record(Id, connection.Id, CallbackDeliveryStatus.Queued, completion); else - { - await send.ConfigureAwait(false); - } + scope?.Record(Id, connection.Id, refusal, null); } #endregion @@ -552,6 +659,13 @@ private void OnNewClientConnected(ITransportServer transport) // window in which a fast client's first frame is delivered to nobody. // The encryptor is built lazily on the processing loop for that reason. var connection = new ConnectionInfo(transport, EncryptorFactory); + connection.AttachOutbox(new ConnectionOutbox( + connection.Id, + message => WriteMessageAsync(connection, message), + m_callbackDelivery, + Timeout, + Logger, + reason => CloseConnection(connection, reason))); if (!m_connections.TryAdd(transport.Id, connection)) { @@ -667,6 +781,40 @@ public void Dispose() public IServerOptions Options => TransportFactory.Options; + /// + /// How this server delivers callbacks (see ). + /// + public CallbackDeliveryOptions CallbackDelivery => m_callbackDelivery; + + /// + /// Callbacks waiting in one connection's outbound queue; 0 for an unknown connection. + /// For diagnostics and tests. + /// + /// The connection. + /// The number of queued callbacks. + public long GetPendingCallbacks(Guid connectionId) + { + return m_connections.TryGetValue(connectionId, out var connection) ? connection.PendingCallbacks : 0; + } + + /// + /// Connections that finished the handshake, for diagnostics and tests. + /// + public int AuthorizedConnectionCount + { + get + { + var count = 0; + foreach (var connection in m_connections.Values) + { + if (connection.IsAuthorized) + count++; + } + + return count; + } + } + #endregion } } diff --git a/Communication/OutWit.Communication.Server/WitServerBuilder.cs b/Communication/OutWit.Communication.Server/WitServerBuilder.cs index 7017f9e..1a70b09 100644 --- a/Communication/OutWit.Communication.Server/WitServerBuilder.cs +++ b/Communication/OutWit.Communication.Server/WitServerBuilder.cs @@ -11,6 +11,7 @@ using OutWit.Communication.Processors; using OutWit.Communication.Serializers; using OutWit.Communication.Server.Authorization; +using OutWit.Communication.Server.Callbacks; using OutWit.Communication.Server.Discovery; using OutWit.Communication.Server.Encryption; @@ -43,7 +44,7 @@ public static WitServer Build(WitServerBuilderOptions options) return new WitServer(options.TransportFactory, options.EncryptorFactory, options.TokenValidator, options.ParametersSerializer, options.MessageSerializer, options.RequestProcessor, options.DiscoveryServer, options.Logger, options.Timeout, options.Name, options.Description, options.MaxConcurrentRequests, - options.HandshakeTimeout); + options.HandshakeTimeout, options.CallbackDelivery); } #region Transport @@ -308,6 +309,50 @@ public static WitServerBuilderOptions WithHandshakeTimeout(this WitServerBuilder } #endregion + + #region Callbacks + + /// + /// Configures how callbacks are delivered: whether an untargeted raise reaches every + /// authorized connection, how many callbacks may wait per connection, and what happens + /// beyond that. See ; the defaults reproduce the + /// pre-3.2 behaviour. + /// + /// The options. + /// Sets the delivery options. + /// The options, for chaining. + public static WitServerBuilderOptions WithCallbackDelivery(this WitServerBuilderOptions me, Action configure) + { + configure(me.CallbackDelivery); + return me; + } + + /// + /// Uses the given callback delivery options. + /// + /// The options. + /// The delivery options. + /// The options, for chaining. + public static WitServerBuilderOptions WithCallbackDelivery(this WitServerBuilderOptions me, CallbackDeliveryOptions callbackDelivery) + { + me.CallbackDelivery = callbackDelivery; + return me; + } + + /// + /// Makes the server refuse an event raised outside a : every + /// callback must name its target connection(s). For a server whose clients must never see + /// each other's events. + /// + /// The options. + /// The options, for chaining. + public static WitServerBuilderOptions WithTargetedCallbacksOnly(this WitServerBuilderOptions me) + { + me.CallbackDelivery.Mode = CallbackDeliveryMode.TargetedOnly; + return me; + } + + #endregion } /// diff --git a/Communication/OutWit.Communication.Server/WitServerBuilderOptions.cs b/Communication/OutWit.Communication.Server/WitServerBuilderOptions.cs index 8671d58..912e1c8 100644 --- a/Communication/OutWit.Communication.Server/WitServerBuilderOptions.cs +++ b/Communication/OutWit.Communication.Server/WitServerBuilderOptions.cs @@ -3,6 +3,7 @@ using OutWit.Communication.Interfaces; using OutWit.Communication.Serializers; using OutWit.Communication.Server.Authorization; +using OutWit.Communication.Server.Callbacks; using OutWit.Communication.Server.Encryption; namespace OutWit.Communication.Server @@ -70,6 +71,14 @@ public WitServerBuilderOptions() /// public TimeSpan? HandshakeTimeout { get; set; } = TimeSpan.FromSeconds(30); + /// + /// How callbacks are delivered: whether an untargeted raise reaches every + /// authorized connection, how many callbacks may wait per connection, and + /// what happens beyond that. The defaults reproduce the pre-3.2 behaviour. + /// See . + /// + public CallbackDeliveryOptions CallbackDelivery { get; set; } = new(); + #endregion } } diff --git a/Communication/OutWit.Communication.Tests/Callbacks/TargetedCallbackTests.cs b/Communication/OutWit.Communication.Tests/Callbacks/TargetedCallbackTests.cs new file mode 100644 index 0000000..47e413e --- /dev/null +++ b/Communication/OutWit.Communication.Tests/Callbacks/TargetedCallbackTests.cs @@ -0,0 +1,304 @@ +using System; +using System.Collections.Concurrent; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Castle.DynamicProxy; +using NUnit.Framework; +using OutWit.Communication.Client; +using OutWit.Communication.Client.Authorization; +using OutWit.Communication.Client.Encryption; +using OutWit.Communication.Interceptors; +using OutWit.Communication.Processors; +using OutWit.Communication.Serializers; +using OutWit.Communication.Server; +using OutWit.Communication.Server.Authorization; +using OutWit.Communication.Server.Callbacks; +using OutWit.Communication.Server.Encryption; +using OutWit.Communication.Tests.Mock; +using OutWit.Communication.Tests.Mock.Interfaces; + +namespace OutWit.Communication.Tests.Callbacks +{ + /// + /// Targeted callbacks over real transports (3.2): an event raised inside a + /// reaches the named connection(s) and nobody else, a raise + /// outside a scope still reaches everyone, and the scope reports what happened. + /// + [TestFixture] + public sealed class TargetedCallbackTests + { + #region Constants + + private const string TOKEN = "token"; + + private static readonly TimeSpan CONNECT_TIMEOUT = TimeSpan.FromSeconds(30); + + private static readonly TimeSpan RECEIVE_WAIT = TimeSpan.FromSeconds(5); + + private static readonly TimeSpan SILENCE = TimeSpan.FromMilliseconds(400); + + #endregion + + #region Fields + + private readonly string m_runId = Guid.NewGuid().ToString("N").Substring(0, 8); + + #endregion + + #region Delivery Tests + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + [TestCase(TransportType.Tcp)] + public async Task TargetedCallbackReachesOnlyTheTargetTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Cb_Caller_{transportType}_{m_runId}"); + using var first = await host.ConnectAsync(); + using var second = await host.ConnectAsync(); + using var third = await host.ConnectAsync(); + + first.Service.NotifyCaller("only me"); + + Assert.That(first.WaitFor("only me"), Is.True, "the caller receives its own event"); + Assert.That(second.ReceivedNothingFor(SILENCE), Is.True); + Assert.That(third.ReceivedNothingFor(SILENCE), Is.True); + + var report = await host.Service.Reports.Single(); + Assert.That(report.Raised, Is.EqualTo(1)); + Assert.That(report.Outcomes.Single().Status, Is.EqualTo(CallbackDeliveryStatus.Sent)); + Assert.That(report.Outcomes.Single().ConnectionId, Is.EqualTo(first.Service.WhoAmI())); + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task CallbackToASetReachesExactlyThatSetTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Cb_Set_{transportType}_{m_runId}"); + using var first = await host.ConnectAsync(); + using var second = await host.ConnectAsync(); + using var third = await host.ConnectAsync(); + + var ids = new[] { first.Service.WhoAmI(), second.Service.WhoAmI() }; + third.Service.NotifyMany(ids, "pair"); + + Assert.That(first.WaitFor("pair"), Is.True); + Assert.That(second.WaitFor("pair"), Is.True); + Assert.That(third.ReceivedNothingFor(SILENCE), Is.True, "the raiser is not in the set"); + + var report = await host.Service.Reports.Single(); + Assert.That(report.Outcomes.Select(outcome => outcome.Status), Is.All.EqualTo(CallbackDeliveryStatus.Sent)); + Assert.That(report.Outcomes.Select(outcome => outcome.ConnectionId), Is.EquivalentTo(ids)); + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task BroadcastWithoutAScopeIsUnchangedTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Cb_All_{transportType}_{m_runId}"); + using var first = await host.ConnectAsync(); + using var second = await host.ConnectAsync(); + using var third = await host.ConnectAsync(); + + first.Service.NotifyAll("everyone"); + + Assert.That(first.WaitFor("everyone"), Is.True); + Assert.That(second.WaitFor("everyone"), Is.True); + Assert.That(third.WaitFor("everyone"), Is.True); + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task UnknownConnectionIsReportedAndNothingIsSentTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Cb_Unknown_{transportType}_{m_runId}"); + using var first = await host.ConnectAsync(); + using var second = await host.ConnectAsync(); + + var unknown = Guid.NewGuid(); + first.Service.NotifyOne(unknown, "nobody"); + + Assert.That(first.ReceivedNothingFor(SILENCE), Is.True); + Assert.That(second.ReceivedNothingFor(SILENCE), Is.True); + + var report = await host.Service.Reports.Single(); + Assert.That(report.Raised, Is.EqualTo(1)); + Assert.That(report.Outcomes.Single().Status, Is.EqualTo(CallbackDeliveryStatus.UnknownConnection)); + Assert.That(report.Outcomes.Single().ConnectionId, Is.EqualTo(unknown)); + Assert.That(report.AnySent, Is.False); + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task TargetedOnlyServerRefusesABroadcastButDeliversTargetedTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Cb_TargetedOnly_{transportType}_{m_runId}", + new CallbackDeliveryOptions { Mode = CallbackDeliveryMode.TargetedOnly }); + using var first = await host.ConnectAsync(); + using var second = await host.ConnectAsync(); + + first.Service.NotifyAll("must not leave the server"); + + Assert.That(first.ReceivedNothingFor(SILENCE), Is.True); + Assert.That(second.ReceivedNothingFor(SILENCE), Is.True); + + second.Service.NotifyCaller("targeted still works"); + + Assert.That(second.WaitFor("targeted still works"), Is.True); + Assert.That(first.ReceivedNothingFor(SILENCE), Is.True); + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task ManyTargetedCallbacksArriveInOrderTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Cb_Order_{transportType}_{m_runId}"); + using var target = await host.ConnectAsync(); + using var other = await host.ConnectAsync(); + + var id = target.Service.WhoAmI(); + for (var i = 0; i < 200; i++) + other.Service.NotifyOne(id, i.ToString()); + + Assert.That(target.WaitForCount(200), Is.True); + Assert.That(other.ReceivedNothingFor(SILENCE), Is.True); + + // Wire order is guaranteed; the client hands each callback to a handler on its own + // task, so the handlers may complete out of order. What must hold is the set. + Assert.That(target.Received.OrderBy(int.Parse), Is.EqualTo(Enumerable.Range(0, 200).Select(i => i.ToString()))); + } + + #endregion + + #region Shared Service Tests + + [Test] + public async Task SharedServiceInTwoServersReportsPerServerTest() + { + var service = new MockConnectionAwareService(); + using var firstHost = await Host.StartAsync(TransportType.Pipes, $"Cb_Shared1_{m_runId}", service: service); + using var secondHost = await Host.StartAsync(TransportType.Pipes, $"Cb_Shared2_{m_runId}", service: service); + using var onFirst = await firstHost.ConnectAsync(); + using var onSecond = await secondHost.ConnectAsync(); + + // The same singleton is raised in both servers; only the one that owns the + // connection delivers, the other reports the id as unknown. + onFirst.Service.NotifyCaller("first server only"); + + Assert.That(onFirst.WaitFor("first server only"), Is.True); + Assert.That(onSecond.ReceivedNothingFor(SILENCE), Is.True); + + var report = await service.Reports.Single(); + Assert.That(report.Raised, Is.EqualTo(2), "each server saw the raise"); + Assert.That(report.Outcomes, Has.Count.EqualTo(2)); + Assert.That(report.Outcomes.Single(outcome => outcome.ServerId == firstHost.Server.Id).Status, Is.EqualTo(CallbackDeliveryStatus.Sent)); + Assert.That(report.Outcomes.Single(outcome => outcome.ServerId == secondHost.Server.Id).Status, Is.EqualTo(CallbackDeliveryStatus.UnknownConnection)); + Assert.That(report.AnySent, Is.True); + } + + #endregion + + #region Helpers + + private sealed class Host : IDisposable + { + private Host(WitServer server, MockConnectionAwareService service, TransportType transportType, string name) + { + Server = server; + Service = service; + TransportType = transportType; + Name = name; + } + + public WitServer Server { get; } + + public MockConnectionAwareService Service { get; } + + public TransportType TransportType { get; } + + public string Name { get; } + + public static Task StartAsync(TransportType transportType, string name, CallbackDeliveryOptions? options = null, MockConnectionAwareService? service = null) + { + service ??= new MockConnectionAwareService(); + var server = new WitServer( + Shared.GetServerTransport(transportType, 10, name), + new EncryptorServerFactory(), + new AccessTokenValidatorStatic(TOKEN), + new MessageSerializerJson(), + new MessageSerializerMemoryPack(), + new RequestProcessor(service), + null, null, null, name, null, int.MaxValue, null, options); + + server.StartWaitingForConnection(); + return Task.FromResult(new Host(server, service, transportType, name)); + } + + public async Task ConnectAsync() + { + var client = new WitClient( + Shared.GetClientTransport(TransportType, Name), + new EncryptorClientGeneral(), + new AccessTokenProviderStatic(TOKEN), + new MessageSerializerJson(), + new MessageSerializerMemoryPack(), + null, null); + + Assert.That(await client.ConnectAsync(CONNECT_TIMEOUT, CancellationToken.None), Is.True, "connect"); + return new Client(client); + } + + public void Dispose() + { + Server.StopWaitingForConnection(); + Server.Dispose(); + } + } + + private sealed class Client : IDisposable + { + private readonly ConcurrentQueue m_received = new(); + + public Client(WitClient client) + { + Raw = client; + Service = new ProxyGenerator().CreateInterfaceProxyWithoutTarget(new RequestInterceptorDynamic(client, true)); + Service.Notified += message => m_received.Enqueue(message); + } + + public WitClient Raw { get; } + + public IConnectionAwareService Service { get; } + + public string[] Received => m_received.ToArray(); + + public bool WaitFor(string message) + { + return SpinWait.SpinUntil(() => m_received.Contains(message), RECEIVE_WAIT); + } + + public bool WaitForCount(int count) + { + return SpinWait.SpinUntil(() => m_received.Count >= count, RECEIVE_WAIT); + } + + /// + /// True when no callback arrived during . + /// + public bool ReceivedNothingFor(TimeSpan silence) + { + var before = m_received.Count; + Thread.Sleep(silence); + return m_received.Count == before; + } + + public void Dispose() + { + Raw.Disconnect().GetAwaiter().GetResult(); + Raw.Dispose(); + } + } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Tests/Connections/ConnectionContextTests.cs b/Communication/OutWit.Communication.Tests/Connections/ConnectionContextTests.cs new file mode 100644 index 0000000..fc7c374 --- /dev/null +++ b/Communication/OutWit.Communication.Tests/Connections/ConnectionContextTests.cs @@ -0,0 +1,283 @@ +using System; +using System.Linq; +using System.Security.Claims; +using System.Threading; +using System.Threading.Tasks; +using Castle.DynamicProxy; +using NUnit.Framework; +using OutWit.Communication.Client; +using OutWit.Communication.Client.Authorization; +using OutWit.Communication.Client.Encryption; +using OutWit.Communication.Interceptors; +using OutWit.Communication.Interfaces; +using OutWit.Communication.Processors; +using OutWit.Communication.Serializers; +using OutWit.Communication.Server; +using OutWit.Communication.Server.Authorization; +using OutWit.Communication.Server.Connections; +using OutWit.Communication.Server.Encryption; +using OutWit.Communication.Tests.Mock; +using OutWit.Communication.Tests.Mock.Interfaces; + +namespace OutWit.Communication.Tests.Connections +{ + /// + /// The connection context a service reads during a request (3.2): the id of the connection + /// the request arrived on, the server, and the principal the connection authorized with. Per + /// async flow, never leaking across connections, gone outside a request. + /// + [TestFixture] + public sealed class ConnectionContextTests + { + #region Constants + + private const string TOKEN = "user:alice"; + + private static readonly TimeSpan CONNECT_TIMEOUT = TimeSpan.FromSeconds(30); + + #endregion + + #region Fields + + private readonly string m_runId = Guid.NewGuid().ToString("N").Substring(0, 8); + + #endregion + + #region Context Tests + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + [TestCase(TransportType.Tcp)] + public async Task ServiceSeesTheConnectionIdOfTheRequestTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Ctx_WhoAmI_{transportType}_{m_runId}"); + using var first = await host.ConnectAsync(); + using var second = await host.ConnectAsync(); + + var firstId = first.Service.WhoAmI(); + var secondId = second.Service.WhoAmI(); + + Assert.That(firstId, Is.Not.EqualTo(Guid.Empty)); + Assert.That(secondId, Is.Not.EqualTo(Guid.Empty)); + Assert.That(firstId, Is.Not.EqualTo(secondId), "each connection has its own id"); + Assert.That(first.Service.WhoAmI(), Is.EqualTo(firstId), "the id is stable across calls on one connection"); + Assert.That(first.Service.WhichServer(), Is.EqualTo(host.Server.Id)); + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task ContextFlowsAcrossAnAwaitInsideTheMethodTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Ctx_Await_{transportType}_{m_runId}"); + using var client = await host.ConnectAsync(); + + Assert.That(client.Service.WhoAmIAfterAwait(), Is.EqualTo(client.Service.WhoAmI())); + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task ConcurrentRequestsDoNotLeakContextAcrossConnectionsTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Ctx_Concurrent_{transportType}_{m_runId}"); + + var clients = new Client[6]; + try + { + for (var i = 0; i < clients.Length; i++) + clients[i] = await host.ConnectAsync(); + + var expected = clients.Select(client => client.Service.WhoAmI()).ToArray(); + Assert.That(expected.Distinct().Count(), Is.EqualTo(clients.Length)); + + // Every client fires a burst at once; every answer must be that client's own id. + var work = clients.Select((client, index) => Task.Run(() => + { + for (var call = 0; call < 20; call++) + Assert.That(client.Service.WhoAmI(), Is.EqualTo(expected[index]), $"client {index}, call {call}"); + })).ToArray(); + + await Task.WhenAll(work); + } + finally + { + foreach (var client in clients) + client?.Dispose(); + } + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task ReconnectedClientGetsANewConnectionIdTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Ctx_Reconnect_{transportType}_{m_runId}"); + + Guid firstId; + using (var first = await host.ConnectAsync()) + firstId = first.Service.WhoAmI(); + + using var second = await host.ConnectAsync(); + Assert.That(second.Service.WhoAmI(), Is.Not.EqualTo(firstId)); + } + + [Test] + public void ContextIsNullOutsideARequestTest() + { + var service = new MockConnectionAwareService(); + + Assert.That(ConnectionContext.Current, Is.Null); + Assert.That(service.WhoAmI(), Is.EqualTo(Guid.Empty)); + Assert.That(service.WhoAmIPrincipal(), Is.Null); + } + + [Test] + public void BeginScopeMakesAContextVisibleAndRestoresThePreviousOneTest() + { + var outer = new ConnectionContext(Guid.NewGuid(), Guid.NewGuid(), "outer", "Test", null, DateTimeOffset.UtcNow); + var inner = new ConnectionContext(Guid.NewGuid(), Guid.NewGuid(), "inner", "Test", null, DateTimeOffset.UtcNow); + var service = new MockConnectionAwareService(); + + using (ConnectionContext.BeginScope(outer)) + { + Assert.That(service.WhoAmI(), Is.EqualTo(outer.ConnectionId)); + + using (ConnectionContext.BeginScope(inner)) + Assert.That(service.WhoAmI(), Is.EqualTo(inner.ConnectionId)); + + Assert.That(service.WhoAmI(), Is.EqualTo(outer.ConnectionId), "the inner scope restored the outer context"); + } + + Assert.That(ConnectionContext.Current, Is.Null); + } + + #endregion + + #region Principal Tests + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task PrincipalIsSetWhenTheValidatorAuthenticatesTest(TransportType transportType) + { + var authenticator = new MockConnectionAuthenticator(); + using var host = await Host.StartAsync(transportType, $"Ctx_Principal_{transportType}_{m_runId}", authenticator); + using var client = await host.ConnectAsync(); + + Assert.That(client.Service.WhoAmIPrincipal(), Is.EqualTo("alice")); + Assert.That(client.Service.WhoAmIPrincipal(), Is.EqualTo("alice"), "established once, read on every request"); + Assert.That(authenticator.AuthenticateCalls, Is.EqualTo(1), "the principal is established at authorization, not per request"); + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task PrincipalIsNullForAPlainValidatorTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Ctx_NoPrincipal_{transportType}_{m_runId}", new AccessTokenValidatorStatic(TOKEN)); + using var client = await host.ConnectAsync(); + + Assert.That(client.Service.WhoAmI(), Is.Not.EqualTo(Guid.Empty)); + Assert.That(client.Service.WhoAmIPrincipal(), Is.Null); + } + + [TestCase(TransportType.Pipes)] + [TestCase(TransportType.WebSocket)] + public async Task AuthenticatorRefusalClosesTheConnectionTest(TransportType transportType) + { + using var host = await Host.StartAsync(transportType, $"Ctx_Refused_{transportType}_{m_runId}", new MockConnectionAuthenticator()); + + var client = host.CreateClient("not-a-user"); + try + { + Assert.That(await client.ConnectAsync(CONNECT_TIMEOUT, CancellationToken.None), Is.False); + Assert.That(client.IsAuthorized, Is.False); + } + finally + { + await client.Disconnect(); + client.Dispose(); + } + } + + #endregion + + #region Helpers + + private sealed class Host : IDisposable + { + private Host(WitServer server, MockConnectionAwareService service, TransportType transportType, string name) + { + Server = server; + Service = service; + TransportType = transportType; + Name = name; + } + + public WitServer Server { get; } + + public MockConnectionAwareService Service { get; } + + public TransportType TransportType { get; } + + public string Name { get; } + + public static Task StartAsync(TransportType transportType, string name, IAccessTokenValidator? validator = null) + { + var service = new MockConnectionAwareService(); + var server = new WitServer( + Shared.GetServerTransport(transportType, 10, name), + new EncryptorServerFactory(), + validator ?? new AccessTokenValidatorStatic(TOKEN), + new MessageSerializerJson(), + new MessageSerializerMemoryPack(), + new RequestProcessor(service), + null, null, null, name, null); + + server.StartWaitingForConnection(); + return Task.FromResult(new Host(server, service, transportType, name)); + } + + public WitClient CreateClient(string token) + { + return new WitClient( + Shared.GetClientTransport(TransportType, Name), + new EncryptorClientGeneral(), + new AccessTokenProviderStatic(token), + new MessageSerializerJson(), + new MessageSerializerMemoryPack(), + null, null); + } + + public async Task ConnectAsync() + { + var client = CreateClient(TOKEN); + Assert.That(await client.ConnectAsync(CONNECT_TIMEOUT, CancellationToken.None), Is.True, "connect"); + return new Client(client); + } + + public void Dispose() + { + Server.StopWaitingForConnection(); + Server.Dispose(); + } + } + + private sealed class Client : IDisposable + { + public Client(WitClient client) + { + Raw = client; + Service = new ProxyGenerator().CreateInterfaceProxyWithoutTarget(new RequestInterceptorDynamic(client, true)); + } + + public WitClient Raw { get; } + + public IConnectionAwareService Service { get; } + + public void Dispose() + { + Raw.Disconnect().GetAwaiter().GetResult(); + Raw.Dispose(); + } + } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Tests/Connections/ConnectionOutboxTests.cs b/Communication/OutWit.Communication.Tests/Connections/ConnectionOutboxTests.cs new file mode 100644 index 0000000..a10795e --- /dev/null +++ b/Communication/OutWit.Communication.Tests/Connections/ConnectionOutboxTests.cs @@ -0,0 +1,580 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using Microsoft.Extensions.Logging; +using NUnit.Framework; +using OutWit.Communication.Interfaces; +using OutWit.Communication.Messages; +using OutWit.Communication.Model; +using OutWit.Communication.Requests; +using OutWit.Communication.Responses; +using OutWit.Communication.Serializers; +using OutWit.Communication.Server; +using OutWit.Communication.Server.Authorization; +using OutWit.Communication.Server.Callbacks; +using OutWit.Communication.Server.Encryption; +using OutWit.Communication.Tests.Mock.Transports; + +namespace OutWit.Communication.Tests.Connections +{ + /// + /// The one outbound queue per connection (3.2): frames leave in the order they were + /// enqueued, a callback raised inside a handler precedes that handler's response, a + /// connection whose transport is stuck holds only its own queue, and the per-connection + /// bounds and the overflow policy do what they say. Observed on the transport's byte + /// sequence, not through a client. + /// + [TestFixture] + public sealed class ConnectionOutboxTests + { + #region Constants + + private static readonly TimeSpan WAIT = TimeSpan.FromSeconds(5); + + #endregion + + #region Ordering Tests + + [Test] + public void CallbackRaisedInsideAHandlerPrecedesTheResponseTest() + { + using var context = CreateContext(raiseCallbackInsideHandler: true); + var transport = context.Connect(); + + transport.RaiseDataReceived(context.RequestFrame("Work")); + + Assert.That(WaitUntil(() => transport.WrittenFrames == 4), Is.True, "handshake x2, callback, response"); + + var types = transport.Sent.Select(context.TypeOf).ToArray(); + Assert.That(types, Is.EqualTo(new[] + { + WitMessageType.Initialization, WitMessageType.Authorization, WitMessageType.Callback, WitMessageType.Request + })); + } + + [Test] + public void OneThousandCallbacksKeepTheirOrderTest() + { + using var context = CreateContext(); + var transport = context.Connect(); + + for (var i = 0; i < 1000; i++) + context.Processor.InvokeCallback(new WitRequest { MethodName = i.ToString() }); + + Assert.That(WaitUntil(() => transport.WrittenFrames == 1002), Is.True); + + var names = transport.Sent.Skip(2).Select(context.CallbackNameOf).ToArray(); + Assert.That(names, Is.EqualTo(Enumerable.Range(0, 1000).Select(i => i.ToString()).ToArray())); + } + + [Test] + public void ResponsesAndCallbacksShareOneOrderTest() + { + using var context = CreateContext(); + var transport = context.Connect(); + + // A request whose response is enqueued while callbacks are still flowing in + // behind it: the response keeps its place, nothing overtakes it. + transport.RaiseDataReceived(context.RequestFrame("First")); + Assert.That(WaitUntil(() => transport.WrittenFrames == 3), Is.True); + + transport.Block(); + context.Processor.InvokeCallback(new WitRequest { MethodName = "a" }); + transport.RaiseDataReceived(context.RequestFrame("Second")); + Assert.That(WaitUntil(() => context.Processor.ProcessCalls == 2), Is.True); + context.Processor.InvokeCallback(new WitRequest { MethodName = "b" }); + transport.Release(); + + Assert.That(WaitUntil(() => transport.WrittenFrames == 6), Is.True); + + var tail = transport.Sent.Skip(3).Select(context.TypeOf).ToArray(); + Assert.That(tail, Is.EqualTo(new[] { WitMessageType.Callback, WitMessageType.Request, WitMessageType.Callback })); + } + + #endregion + + #region Isolation Tests + + [Test] + public void SlowConnectionDoesNotDelayItsNeighbourTest() + { + using var context = CreateContext(); + var slow = context.Connect(); + var neighbour = context.Connect(); + + slow.Block(); + + for (var i = 0; i < 50; i++) + context.Processor.InvokeCallback(new WitRequest { MethodName = i.ToString() }); + + // The neighbour gets all fifty while the slow one still holds its first write. + Assert.That(WaitUntil(() => neighbour.WrittenFrames == 52), Is.True, "the neighbour must not wait for the slow connection"); + Assert.That(slow.WrittenFrames, Is.EqualTo(2), "the slow connection is stuck on its first callback"); + Assert.That(context.Server.AuthorizedConnectionCount, Is.EqualTo(2)); + Assert.That(slow.WaitForBlockedWrites(1, WAIT), Is.True); + + var pending = context.PendingCallbacksOf(slow); + Assert.That(pending, Is.EqualTo(49), "one callback is inside the blocked write, the rest wait in the queue"); + + slow.Release(); + Assert.That(WaitUntil(() => slow.WrittenFrames == 52), Is.True); + Assert.That(context.PendingCallbacksOf(slow), Is.EqualTo(0)); + } + + [Test] + public void DisconnectDuringASendDropsTheQueueQuietlyTest() + { + using var context = CreateContext(); + var slow = context.Connect(); + var neighbour = context.Connect(); + + slow.Block(); + for (var i = 0; i < 10; i++) + context.Processor.InvokeCallback(new WitRequest { MethodName = i.ToString() }); + + Assert.That(WaitUntil(() => neighbour.WrittenFrames == 12), Is.True); + + // The stuck client goes away: nothing hangs, nothing throws, the neighbour keeps working. + slow.Dispose(); + Assert.That(WaitUntil(() => context.Server.AuthorizedConnectionCount == 1), Is.True); + + context.Processor.InvokeCallback(new WitRequest { MethodName = "after" }); + Assert.That(WaitUntil(() => neighbour.WrittenFrames == 13), Is.True); + } + + #endregion + + #region Bound Tests + + [Test] + public void QueueFullClosesOnlyTheOffendingConnectionTest() + { + using var context = CreateContext(options: new CallbackDeliveryOptions + { + MaxPendingCallbacks = 2, + OverflowPolicy = CallbackOverflowPolicy.CloseConnection + }); + var slow = context.Connect(); + var neighbour = context.Connect(); + + slow.Block(); + + // 1 inside the blocked write, 2 in the queue, the 4th is one too many. The raises + // are paced by the neighbour's writes: the bound counts frames not yet written, and + // an idle connection must not trip it just because a burst outran its writer. + context.Processor.InvokeCallback(new WitRequest { MethodName = "0" }); + Assert.That(slow.WaitForBlockedWrites(1, WAIT), Is.True, "the first callback reached the transport and blocked"); + for (var i = 1; i < 4; i++) + { + Assert.That(WaitUntil(() => neighbour.WrittenFrames == 2 + i), Is.True); + context.Processor.InvokeCallback(new WitRequest { MethodName = i.ToString() }); + } + + Assert.That(WaitUntil(() => slow.IsDisposed), Is.True, "the slow connection is closed by the policy"); + Assert.That(WaitUntil(() => neighbour.WrittenFrames == 6), Is.True, "the neighbour gets all four"); + Assert.That(WaitUntil(() => context.Server.AuthorizedConnectionCount == 1), Is.True); + Assert.That(context.Logger.Contains(LogLevel.Warning, "callback queue full"), Is.True); + } + + [Test] + public void QueueFullLogsAndKeepsQueuingByDefaultTest() + { + using var context = CreateContext(options: new CallbackDeliveryOptions + { + MaxPendingCallbacks = 2 + }); + var slow = context.Connect(); + + slow.Block(); + context.Processor.InvokeCallback(new WitRequest { MethodName = "0" }); + Assert.That(slow.WaitForBlockedWrites(1, WAIT), Is.True, "the first callback reached the transport and blocked"); + for (var i = 1; i < 6; i++) + context.Processor.InvokeCallback(new WitRequest { MethodName = i.ToString() }); + + Thread.Sleep(200); + Assert.That(slow.IsDisposed, Is.False, "the default policy never closes"); + Assert.That(context.PendingCallbacksOf(slow), Is.EqualTo(5)); + Assert.That(context.Logger.Contains(LogLevel.Warning, "passed its bound"), Is.True); + + slow.Release(); + Assert.That(WaitUntil(() => slow.WrittenFrames == 8), Is.True, "everything queued is delivered"); + } + + [Test] + public void DropNewestDropsBeyondTheBoundTest() + { + using var context = CreateContext(options: new CallbackDeliveryOptions + { + MaxPendingCallbacks = 2, + OverflowPolicy = CallbackOverflowPolicy.DropNewest + }); + var slow = context.Connect(); + + slow.Block(); + context.Processor.InvokeCallback(new WitRequest { MethodName = "0" }); + Assert.That(slow.WaitForBlockedWrites(1, WAIT), Is.True, "the first callback reached the transport and blocked"); + for (var i = 1; i < 6; i++) + context.Processor.InvokeCallback(new WitRequest { MethodName = i.ToString() }); + + Assert.That(context.PendingCallbacksOf(slow), Is.EqualTo(2)); + slow.Release(); + + Assert.That(WaitUntil(() => slow.WrittenFrames == 5), Is.True, "handshake x2, the blocked one, the two queued"); + Thread.Sleep(100); + Assert.That(slow.WrittenFrames, Is.EqualTo(5)); + Assert.That(slow.IsDisposed, Is.False); + + var names = slow.Sent.Skip(2).Select(context.CallbackNameOf).ToArray(); + Assert.That(names, Is.EqualTo(new[] { "0", "1", "2" }), "the oldest survive, the newest are dropped"); + } + + [Test] + public void ByteBudgetCountsFrameSizesTest() + { + using var context = CreateContext(options: new CallbackDeliveryOptions + { + MaxPendingCallbackBytes = 1024, + OverflowPolicy = CallbackOverflowPolicy.DropNewest + }); + var slow = context.Connect(); + + slow.Block(); + var big = new string('x', 600); + context.Processor.InvokeCallback(new WitRequest { MethodName = big + 0 }); + Assert.That(slow.WaitForBlockedWrites(1, WAIT), Is.True, "the first callback reached the transport and blocked"); + for (var i = 1; i < 5; i++) + context.Processor.InvokeCallback(new WitRequest { MethodName = big + i }); + + // One frame is inside the blocked write; the queue holds one more 600+ byte frame + // before the second would exceed 1024 bytes. + Assert.That(context.PendingCallbacksOf(slow), Is.EqualTo(1)); + slow.Release(); + Assert.That(WaitUntil(() => slow.WrittenFrames == 4), Is.True); + } + + [Test] + public void ResponsesAreNeverCountedOrDroppedTest() + { + using var context = CreateContext(options: new CallbackDeliveryOptions + { + MaxPendingCallbacks = 1, + OverflowPolicy = CallbackOverflowPolicy.DropNewest + }); + var transport = context.Connect(); + + transport.Block(); + for (var i = 0; i < 5; i++) + transport.RaiseDataReceived(context.RequestFrame($"Request{i}")); + + // The connection loop handles one request at a time and waits for its response + // to be written, so only one response is queued at once; none is refused. + context.Processor.InvokeCallback(new WitRequest { MethodName = "c1" }); + context.Processor.InvokeCallback(new WitRequest { MethodName = "c2" }); + transport.Release(); + + Assert.That(WaitUntil(() => transport.WrittenFrames == 8), Is.True, "handshake x2, five responses, one callback"); + Assert.That(transport.Sent.Select(context.TypeOf).Count(type => type == WitMessageType.Request), Is.EqualTo(5)); + } + + #endregion + + #region Timeout Tests + + [Test] + public void SendTimeoutClosesTheConnectionUnderClosePolicyTest() + { + using var context = CreateContext( + options: new CallbackDeliveryOptions { OverflowPolicy = CallbackOverflowPolicy.CloseConnection }, + timeout: TimeSpan.FromMilliseconds(200)); + var slow = context.Connect(); + + slow.Block(); + context.Processor.InvokeCallback(new WitRequest { MethodName = "stuck" }); + + Assert.That(WaitUntil(() => slow.IsDisposed), Is.True, "a write that does not finish within the timeout closes the connection"); + Assert.That(context.Logger.Contains(LogLevel.Warning, "timed out"), Is.True); + } + + [Test] + public void SendTimeoutOnlyWarnsByDefaultTest() + { + using var context = CreateContext(timeout: TimeSpan.FromMilliseconds(200)); + var slow = context.Connect(); + + slow.Block(); + context.Processor.InvokeCallback(new WitRequest { MethodName = "stuck" }); + + Assert.That(WaitUntil(() => context.Logger.Contains(LogLevel.Warning, "timed out")), Is.True); + Thread.Sleep(200); + Assert.That(slow.IsDisposed, Is.False, "the default policy warns and waits"); + + slow.Release(); + Assert.That(WaitUntil(() => slow.WrittenFrames == 3), Is.True); + } + + #endregion + + #region Lifecycle Tests + + [Test] + public void DisposeWithPendingSendsDoesNotThrowTest() + { + var context = CreateContext(); + var slow = context.Connect(); + + slow.Block(); + for (var i = 0; i < 20; i++) + context.Processor.InvokeCallback(new WitRequest { MethodName = i.ToString() }); + + Assert.DoesNotThrow(() => context.Dispose()); + Assert.That(WaitUntil(() => slow.IsDisposed), Is.True); + } + + [Test] + public void TargetedOnlyServerRefusesABroadcastTest() + { + using var context = CreateContext(options: new CallbackDeliveryOptions { Mode = CallbackDeliveryMode.TargetedOnly }); + var transport = context.Connect(); + + context.Processor.InvokeCallback(new WitRequest { MethodName = "untargeted" }); + Thread.Sleep(200); + + Assert.That(transport.WrittenFrames, Is.EqualTo(2), "nothing beyond the handshake"); + Assert.That(context.Logger.Contains(LogLevel.Error, "without a target"), Is.True); + + using (CallbackScope.Target(transport.Id)) + context.Processor.InvokeCallback(new WitRequest { MethodName = "targeted" }); + + Assert.That(WaitUntil(() => transport.WrittenFrames == 3), Is.True); + } + + [Test] + public void ScopeReportsPerConnectionOutcomesTest() + { + using var context = CreateContext(options: new CallbackDeliveryOptions + { + MaxPendingCallbacks = 1, + OverflowPolicy = CallbackOverflowPolicy.DropNewest + }); + var first = context.Connect(); + var second = context.Connect(); + var unknown = Guid.NewGuid(); + + // The second connection ends up with one callback stuck in its write and one in its + // queue (the bound); the first has written everything before the probe. + second.Block(); + context.Processor.InvokeCallback(new WitRequest { MethodName = "fill" }); + Assert.That(second.WaitForBlockedWrites(1, WAIT), Is.True); + Assert.That(WaitUntil(() => context.PendingCallbacksOf(first) == 0), Is.True); + context.Processor.InvokeCallback(new WitRequest { MethodName = "fill" }); + Assert.That(WaitUntil(() => context.PendingCallbacksOf(first) == 0), Is.True); + Assert.That(context.PendingCallbacksOf(second), Is.EqualTo(1)); + + CallbackDeliveryReport report; + using (var scope = CallbackScope.Target(new[] { first.Id, second.Id, unknown })) + { + context.Processor.InvokeCallback(new WitRequest { MethodName = "probe" }); + second.Release(); + report = scope.Completion.GetAwaiter().GetResult(); + } + + Assert.That(report.Raised, Is.EqualTo(1)); + Assert.That(report.Outcomes.Single(outcome => outcome.ConnectionId == first.Id).Status, Is.EqualTo(CallbackDeliveryStatus.Sent)); + Assert.That(report.Outcomes.Single(outcome => outcome.ConnectionId == second.Id).Status, Is.EqualTo(CallbackDeliveryStatus.QueueFull)); + Assert.That(report.Outcomes.Single(outcome => outcome.ConnectionId == unknown).Status, Is.EqualTo(CallbackDeliveryStatus.UnknownConnection)); + Assert.That(report.Outcomes.Select(outcome => outcome.ServerId).Distinct().Single(), Is.EqualTo(context.Server.Id)); + } + + #endregion + + #region Helpers + + private static TestContext CreateContext(bool raiseCallbackInsideHandler = false, CallbackDeliveryOptions? options = null, TimeSpan? timeout = null) + { + var messageSerializer = new MessageSerializerMemoryPack(); + var logger = new CapturingLogger(); + var factory = new MockTransportServerFactory(); + var processor = new CallbackProcessor(raiseCallbackInsideHandler); + + var server = new WitServer( + factory, + new EncryptorServerFactory(), + new AccessTokenValidatorPlain(), + new MessageSerializerJson(), + messageSerializer, + processor, + discoveryServer: null, + logger, + timeout, + name: null, + description: null, + int.MaxValue, + handshakeTimeout: null, + options); + + return new TestContext(server, factory, processor, logger, messageSerializer); + } + + private static bool WaitUntil(Func condition) + { + return SpinWait.SpinUntil(condition, WAIT); + } + + private sealed class TestContext : IDisposable + { + public TestContext(WitServer server, MockTransportServerFactory factory, CallbackProcessor processor, CapturingLogger logger, IMessageSerializer messageSerializer) + { + Server = server; + Factory = factory; + Processor = processor; + Logger = logger; + MessageSerializer = messageSerializer; + } + + public WitServer Server { get; } + + public MockTransportServerFactory Factory { get; } + + public CallbackProcessor Processor { get; } + + public CapturingLogger Logger { get; } + + public IMessageSerializer MessageSerializer { get; } + + /// + /// Connects a transport and drives it through the handshake. + /// + public MockTransportServer Connect() + { + var transport = Factory.Connect(); + + transport.RaiseDataReceived(Frame(WitMessageType.Initialization, new WitRequestInitialization + { + PublicKey = new byte[] { 1, 2, 3 }, + ProtocolVersion = WitProtocol.VERSION + })); + transport.RaiseDataReceived(Frame(WitMessageType.Authorization, new WitRequestAuthorization { Token = string.Empty })); + + Assert.That(WaitUntil(() => transport.WrittenFrames == 2), Is.True, "handshake"); + return transport; + } + + public byte[] RequestFrame(string methodName) + { + return Frame(WitMessageType.Request, new WitRequest { Token = string.Empty, MethodName = methodName }); + } + + public WitMessageType TypeOf(byte[] frame) + { + return MessageSerializer.Deserialize(frame)!.Type; + } + + public string CallbackNameOf(byte[] frame) + { + var message = MessageSerializer.Deserialize(frame)!; + Assert.That(message.Type, Is.EqualTo(WitMessageType.Callback)); + return MessageSerializer.Deserialize(message.Data!)!.MethodName; + } + + public long PendingCallbacksOf(MockTransportServer transport) + { + return Server.GetPendingCallbacks(transport.Id); + } + + private byte[] Frame(WitMessageType type, object payload) + { + return MessageSerializer.Serialize(new WitMessage + { + Id = Guid.NewGuid(), + Type = type, + Data = MessageSerializer.Serialize(payload, payload.GetType()) + }); + } + + public void Dispose() + { + Server.Dispose(); + } + } + + private sealed class CallbackProcessor : IRequestProcessor + { + private readonly bool m_raiseInsideHandler; + + private int m_processCalls; + + private int m_raised; + + public CallbackProcessor(bool raiseInsideHandler) + { + m_raiseInsideHandler = raiseInsideHandler; + } + + public event RequestProcessorEventHandler Callback = delegate { }; + + public Task Process(WitRequest? request) + { + Interlocked.Increment(ref m_processCalls); + + if (m_raiseInsideHandler) + InvokeCallback(new WitRequest { MethodName = "inside" }); + + return Task.FromResult(WitResponse.Success(Array.Empty())); + } + + public void ResetSerializer(IMessageSerializer serializer) + { + } + + public void InvokeCallback(WitRequest request) + { + Interlocked.Increment(ref m_raised); + Callback(request); + } + + public int ProcessCalls => Volatile.Read(ref m_processCalls); + + public int RaisedCallbacks => Volatile.Read(ref m_raised); + } + + private sealed class CapturingLogger : ILogger + { + private readonly List<(LogLevel Level, string Message)> m_entries = new(); + + public IDisposable BeginScope(TState state) where TState : notnull + { + return NullScope.Instance; + } + + public bool IsEnabled(LogLevel logLevel) + { + return true; + } + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, Func formatter) + { + lock (m_entries) + m_entries.Add((logLevel, formatter(state, exception))); + } + + public bool Contains(LogLevel level, string part) + { + lock (m_entries) + return m_entries.Any(entry => entry.Level == level && entry.Message.Contains(part, StringComparison.OrdinalIgnoreCase)); + } + + private sealed class NullScope : IDisposable + { + public static readonly NullScope Instance = new(); + + public void Dispose() + { + } + } + } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Tests/_Mock/Interfaces/IConnectionAwareService.cs b/Communication/OutWit.Communication.Tests/_Mock/Interfaces/IConnectionAwareService.cs new file mode 100644 index 0000000..52d6ae1 --- /dev/null +++ b/Communication/OutWit.Communication.Tests/_Mock/Interfaces/IConnectionAwareService.cs @@ -0,0 +1,38 @@ +using System; + +namespace OutWit.Communication.Tests.Mock.Interfaces +{ + /// + /// A contract for the connection-context and targeted-callback tests: the service + /// answers who is calling and raises its one event to whoever the method names. + /// + public interface IConnectionAwareService + { + /// Raised by the Notify methods; the payload is the message given. + event Action Notified; + + /// The id of the connection the call arrived on, or without a context. + Guid WhoAmI(); + + /// The name of the principal the connection authorized with, or null. + string? WhoAmIPrincipal(); + + /// The id of the server the call arrived on. + Guid WhichServer(); + + /// Whether a context is visible after an await inside the method. + Guid WhoAmIAfterAwait(); + + /// Raises to every authorized connection. + void NotifyAll(string message); + + /// Raises to the calling connection only. + void NotifyCaller(string message); + + /// Raises to one connection. + void NotifyOne(Guid connectionId, string message); + + /// Raises to a set of connections. + void NotifyMany(Guid[] connectionIds, string message); + } +} diff --git a/Communication/OutWit.Communication.Tests/_Mock/MockConnectionAuthenticator.cs b/Communication/OutWit.Communication.Tests/_Mock/MockConnectionAuthenticator.cs new file mode 100644 index 0000000..8933b63 --- /dev/null +++ b/Communication/OutWit.Communication.Tests/_Mock/MockConnectionAuthenticator.cs @@ -0,0 +1,59 @@ +using System.Security.Claims; +using OutWit.Communication.Interfaces; +using OutWit.Communication.Server.Authorization; + +namespace OutWit.Communication.Tests.Mock +{ + /// + /// A token validator that also names the principal: any token of the form + /// user:<name> is valid and yields a principal with that name. What a + /// JWT validator does in a real server, without the JWT. + /// + public sealed class MockConnectionAuthenticator : IAccessTokenValidator, IConnectionAuthenticator + { + #region Constants + + public const string PREFIX = "user:"; + + #endregion + + #region IAccessTokenValidator + + public bool IsRequestTokenValid(string token) + { + return token.StartsWith(PREFIX, System.StringComparison.Ordinal); + } + + public bool IsAuthorizationTokenValid(string token) + { + return IsRequestTokenValid(token); + } + + #endregion + + #region IConnectionAuthenticator + + public bool TryAuthenticate(string token, out ClaimsPrincipal? principal) + { + AuthenticateCalls++; + + if (!IsAuthorizationTokenValid(token)) + { + principal = null; + return false; + } + + var identity = new ClaimsIdentity(new[] { new Claim(ClaimTypes.Name, token.Substring(PREFIX.Length)) }, "mock"); + principal = new ClaimsPrincipal(identity); + return true; + } + + #endregion + + #region Properties + + public int AuthenticateCalls { get; private set; } + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Tests/_Mock/MockConnectionAwareService.cs b/Communication/OutWit.Communication.Tests/_Mock/MockConnectionAwareService.cs new file mode 100644 index 0000000..5db0ff8 --- /dev/null +++ b/Communication/OutWit.Communication.Tests/_Mock/MockConnectionAwareService.cs @@ -0,0 +1,92 @@ +using System; +using System.Collections.Concurrent; +using System.Security.Claims; +using System.Threading.Tasks; +using OutWit.Communication.Server.Callbacks; +using OutWit.Communication.Server.Connections; +using OutWit.Communication.Tests.Mock.Interfaces; + +namespace OutWit.Communication.Tests.Mock +{ + /// + /// The service behind . It keeps the delivery reports + /// of its targeted raises so a test that holds the instance can read them. + /// + public sealed class MockConnectionAwareService : IConnectionAwareService + { + #region Events + + public event Action Notified = delegate { }; + + #endregion + + #region IConnectionAwareService + + public Guid WhoAmI() + { + return ConnectionContext.Current?.ConnectionId ?? Guid.Empty; + } + + public string? WhoAmIPrincipal() + { + return ConnectionContext.Current?.Principal?.FindFirst(ClaimTypes.Name)?.Value; + } + + public Guid WhichServer() + { + return ConnectionContext.Current?.ServerId ?? Guid.Empty; + } + + public Guid WhoAmIAfterAwait() + { + return WhoAmIAfterAwaitAsync().GetAwaiter().GetResult(); + } + + public void NotifyAll(string message) + { + Notified(message); + } + + public void NotifyCaller(string message) + { + using var scope = CallbackScope.TargetCaller(); + Notified(message); + Reports.Enqueue(scope.Completion); + } + + public void NotifyOne(Guid connectionId, string message) + { + using var scope = CallbackScope.Target(connectionId); + Notified(message); + Reports.Enqueue(scope.Completion); + } + + public void NotifyMany(Guid[] connectionIds, string message) + { + using var scope = CallbackScope.Target(connectionIds); + Notified(message); + Reports.Enqueue(scope.Completion); + } + + #endregion + + #region Tools + + private static async Task WhoAmIAfterAwaitAsync() + { + await Task.Delay(10).ConfigureAwait(false); + return ConnectionContext.Current?.ConnectionId ?? Guid.Empty; + } + + #endregion + + #region Properties + + /// + /// The completion of every targeted raise, in call order. + /// + public ConcurrentQueue> Reports { get; } = new(); + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Tests/_Mock/Transports/MockTransportServer.cs b/Communication/OutWit.Communication.Tests/_Mock/Transports/MockTransportServer.cs new file mode 100644 index 0000000..188d401 --- /dev/null +++ b/Communication/OutWit.Communication.Tests/_Mock/Transports/MockTransportServer.cs @@ -0,0 +1,153 @@ +using System; +using System.Collections.Concurrent; +using System.Threading; +using System.Threading.Tasks; +using OutWit.Communication.Interfaces; + +namespace OutWit.Communication.Tests.Mock.Transports +{ + /// + /// A server-side transport under the test's control: frames the server writes are + /// recorded in order, a write can be made to block until the test releases it, and the + /// test raises inbound frames and the disconnect itself. + /// + public sealed class MockTransportServer : ITransportServer + { + #region Events + + public event TransportDataEventHandler Callback = delegate { }; + + public event TransportEventHandler Disconnected = delegate { }; + + #endregion + + #region Fields + + private readonly ConcurrentQueue m_sent = new(); + + private volatile TaskCompletionSource? m_block; + + private int m_writtenFrames; + + private int m_blockedWrites; + + private int m_disconnectedRaised; + + #endregion + + #region Constructors + + public MockTransportServer() + { + Id = Guid.NewGuid(); + } + + #endregion + + #region Functions + + /// + /// Delivers one inbound frame the way the real transport would. + /// + public void RaiseDataReceived(byte[] data) + { + Callback(Id, data); + } + + /// + /// Makes every following write wait until . + /// + public void Block() + { + m_block = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + + /// + /// Lets the blocked writes through and stops blocking new ones. + /// + public void Release() + { + var block = Interlocked.Exchange(ref m_block, null); + block?.TrySetResult(true); + } + + /// + /// Waits until writes have hit the block. + /// + public bool WaitForBlockedWrites(int count, TimeSpan timeout) + { + return SpinWait.SpinUntil(() => BlockedWrites >= count, timeout); + } + + private void RaiseDisconnected() + { + if (Interlocked.Exchange(ref m_disconnectedRaised, 1) == 0) + Disconnected(Id); + } + + #endregion + + #region ITransportServer + + public Task InitializeConnectionAsync(CancellationToken token) + { + return Task.FromResult(true); + } + + public async Task SendBytesAsync(byte[] data) + { + if (IsDisposed) + throw new ObjectDisposedException(nameof(MockTransportServer)); + + var block = m_block; + if (block != null) + { + Interlocked.Increment(ref m_blockedWrites); + await block.Task.ConfigureAwait(false); + + if (IsDisposed) + throw new ObjectDisposedException(nameof(MockTransportServer)); + } + + m_sent.Enqueue(data); + Interlocked.Increment(ref m_writtenFrames); + } + + public void Dispose() + { + if (IsDisposed) + return; + + IsDisposed = true; + Release(); + RaiseDisconnected(); + } + + #endregion + + #region Properties + + public Guid Id { get; } + + public bool CanReinitialize => false; + + public bool IsDisposed { get; private set; } + + /// + /// The frames written to the transport, in write order. + /// + public ConcurrentQueue Sent => m_sent; + + /// + /// Frames written so far. + /// + public int WrittenFrames => Volatile.Read(ref m_writtenFrames); + + /// + /// Writes that hit the block and waited. + /// + public int BlockedWrites => Volatile.Read(ref m_blockedWrites); + + #endregion + } +} diff --git a/Communication/OutWit.Communication.Tests/_Mock/Transports/MockTransportServerFactory.cs b/Communication/OutWit.Communication.Tests/_Mock/Transports/MockTransportServerFactory.cs new file mode 100644 index 0000000..d1b1c7c --- /dev/null +++ b/Communication/OutWit.Communication.Tests/_Mock/Transports/MockTransportServerFactory.cs @@ -0,0 +1,70 @@ +using System.Collections.Generic; +using Microsoft.Extensions.Logging; +using OutWit.Communication.Interfaces; + +namespace OutWit.Communication.Tests.Mock.Transports +{ + /// + /// A transport factory the test drives: hands a + /// to the server as a new client. + /// + public sealed class MockTransportServerFactory : ITransportServerFactory + { + #region Events + + public event TransportFactoryEventHandler NewClientConnected = delegate { }; + + #endregion + + #region Functions + + /// + /// Presents a new connection to the server. + /// + public MockTransportServer Connect() + { + var transport = new MockTransportServer(); + NewClientConnected(transport); + return transport; + } + + #endregion + + #region ITransportServerFactory + + public void StartWaitingForConnection(ILogger? logger) + { + IsStarted = true; + } + + public void StopWaitingForConnection() + { + IsStarted = false; + } + + public void Dispose() + { + } + + #endregion + + #region Properties + + public IServerOptions Options { get; } = new MockServerOptions(); + + public bool IsStarted { get; private set; } + + #endregion + + #region Nested Types + + private sealed class MockServerOptions : IServerOptions + { + public string Transport => "Mock"; + + public Dictionary Data { get; } = new(); + } + + #endregion + } +} diff --git a/MIGRATION-3.md b/MIGRATION-3.md index 33c37aa..0f92d7a 100644 --- a/MIGRATION-3.md +++ b/MIGRATION-3.md @@ -412,6 +412,14 @@ of the gap between them is the S2S lookup. ### Step 4 — after the dust settles +**Server 3.2.0 (connection context, targeted events, per-connection outbound +queue) needs no consumer action.** It is additive on the server package only: +nothing on the wire, nothing in the core, the client packages or the DI package +(whose floor stays `Server >= 3.1.1`). A consumer takes it with an explicit +`OutWit.Communication.Server` pin and behaves exactly as before until it opens +a `CallbackScope`, reads `ConnectionContext.Current` or sets +`WithCallbackDelivery(...)`. + Lift `MaxConcurrentRequests` per service after the thread-safety audit; declare idempotent methods where retries are wanted; WebSocket restart hang in WitRPC. From the cutover itself: the six Simulator `Grid.ForEach` E2E tests that the diff --git a/OutWit.slnx b/OutWit.slnx index c4265ca..4f9c589 100644 --- a/OutWit.slnx +++ b/OutWit.slnx @@ -34,6 +34,7 @@ + diff --git a/ROADMAP-v3.md b/ROADMAP-v3.md index 4d9e0af..e67c3ce 100644 --- a/ROADMAP-v3.md +++ b/ROADMAP-v3.md @@ -475,6 +475,38 @@ else: - Consumer wave targets 3.1.0 directly — nobody in the workspace consumed 3.0.0, so the extra version costs no one a second bump. +### 3.2.0 follow-up — connection context and targeted events (Server only) + +What a shared server needed before one `WitServer` could serve many clients that +must not see each other's events (the WitCloud `/worker/v2` case: one endpoint +for every node instead of one server per node). Planned in +`WitCloud/@Docs/Roadmap/plan-witrpc-connection-context-and-targeted-callbacks.md`, +implemented on `feature/connection-context-targeted-callbacks` from `main`: + +- **`ConnectionContext.Current`** — an `AsyncLocal` set inside `WitServer.ProcessMessage` + (visible to the validator, the processor and the service method and whatever + they await or start; gone when the request completes) and around + `ProcessAuthorization`. Carries the connection id, the server and the + principal; the principal lives on `ConnectionInfo`, established once at + authorization through the opt-in `IConnectionAuthenticator` — `IAccessTokenValidator` + is untouched, a validator that does not opt in leaves it null. +- **`CallbackScope`** — the target of a raise, read in `WitServer.OnCallback` + (the raise chain `event → HandleEvent → Callback → OnCallback` is synchronous, + so the `AsyncLocal` is visible there without touching either processor or + `IRequestProcessor`). One connection or a set; a delivery report per server + and connection; `TargetedOnly` refuses an untargeted raise. Without a scope + the server broadcasts as before. +- **`ConnectionOutbox`** — one ordered outbound queue and one writer per + connection for responses, handshake replies and callbacks alike; replaces the + `_ = SendCallbackAsync(...)` fan-out. Wire order is a stated guarantee now, + the AEAD counter advances in wire order by construction, a stuck transport + holds only its own queue, and the per-connection bound with the + `Log | CloseConnection | DropNewest` policy is the outbound half of the + 2026-08-30 audit's P0.2 (the inbound bound stays for its Stage 1). +- Defaults reproduce 3.1 exactly; the wire, the core, every client package and + the DI package are untouched. Tests: 46 new, on Pipes / WebSocket / TCP and on + a stub transport that can block a write. + ### Published (2026-08-29) - **3.0.0** — all 23 packages, in dependency waves through the gated