From f27eb280639fd802191f8d006e788c45a2e1d576 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 7 Aug 2026 06:15:20 -0400 Subject: [PATCH] fix(GWC-28): stamp gateway envelope sequence at write, not construction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CreateEnvelope stamped Sequence with an interlocked increment when the envelope was built, so two concurrent InvokeAsync callers could take 1 and 2 and then enqueue in the order 2, 1 — non-monotonic on the wire, breaking gateway.md's "monotonic per sender" contract. Benign today (neither side validates inbound sequence, old GWC-10 still open) but it would fault healthy sessions the moment worker-side validation lands. WriteLoopAsync now stamps immediately before _writer.WriteAsync. It is the outbound channel's single consumer (SingleReader = true), so wire order and stamp order are the same thing by construction and _nextSequence drops to a plain ulong with no interlocking. This mirrors the worker's WRK-04 fix, which the gateway half never received. Also adds TST-28: a [Theory] pinning that GatewayHello.MaxFrameBytes carries the configured worker-frame maximum (default + 2 MiB override). The adoption half is asserted only in the Windows-only worker suite, so a regression to sending 0 — "older gateway, use default" to the worker — would silently downgrade the negotiated IPC-02 limit with every CI test still green. Mutation-checked (hard-coded 0 fails both cases). Tests: WorkerClientTests.ConcurrentInvokesEmitStrictlyIncreasingSequences OnTheWire (32 parallel invokes; failed 3/3 pre-fix) and .StartAsync_SendsGatewayHelloWithConfiguredMaxFrameBytes. --- gateway.md | 10 ++- .../Workers/WorkerClient.cs | 15 +++- .../Gateway/Workers/WorkerClientTests.cs | 70 +++++++++++++++++++ 3 files changed, 90 insertions(+), 5 deletions(-) diff --git a/gateway.md b/gateway.md index 1c15cd1..4280be2 100644 --- a/gateway.md +++ b/gateway.md @@ -348,9 +348,13 @@ messages, tagged from 10 upward: Rules: -- `sequence` is a monotonic per-sender counter used as a diagnostic aid; it is - not validated for gaps or ordering on receive (the named pipe already - guarantees FIFO delivery). +- `sequence` is a monotonic per-sender counter used as a diagnostic aid. Both + sides stamp it at the point of writing, inside their single write path — the + gateway on its outbound-channel write loop, the worker on its own writer — so + the numbers stay monotonic in wire order no matter how concurrent callers + interleave while building envelopes. It is not validated for gaps or ordering + on receive (the named pipe already guarantees FIFO delivery); if inbound + enforcement is ever added, this is the property it will rely on. - `correlation_id` links a command to its reply; it is authoritative on the envelope, and the inner `MxCommandReply.correlation_id` echoes it for MXAccess parity. diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index 207574e..3ad808c 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -40,7 +40,9 @@ public sealed class WorkerClient : IWorkerClient private readonly ConcurrentDictionary _pendingCommands = new(StringComparer.Ordinal); private readonly SemaphoreSlim _pendingCommandSlots; private readonly CancellationTokenSource _stopCts = new(); - private long _nextSequence; + // Touched only by WriteLoopAsync — the single consumer of _outboundEnvelopes — so it needs no + // interlocking. See WriteLoopAsync for why the stamp happens there rather than at construction. + private ulong _nextSequence; private WorkerClientState _state; private DateTimeOffset _lastHeartbeatAt; private int? _processId; @@ -404,6 +406,13 @@ public sealed class WorkerClient : IWorkerClient { await foreach (WorkerEnvelope envelope in _outboundEnvelopes.Reader.ReadAllAsync(_stopCts.Token).ConfigureAwait(false)) { + // GWC-28: stamp the sequence at the point of writing, not when the envelope is built. + // Stamping at construction let two concurrent InvokeAsync callers take 1 and 2 and then + // enqueue in the order 2, 1 — non-monotonic on the wire, breaking gateway.md's + // "monotonic per sender" contract. This loop is the channel's single consumer + // (SingleReader = true), so wire order and stamp order are the same thing here and + // _nextSequence needs no interlocking. Mirrors the worker's WRK-04 fix. + envelope.Sequence = unchecked(++_nextSequence); await _writer.WriteAsync(envelope, _stopCts.Token).ConfigureAwait(false); } } @@ -1072,11 +1081,13 @@ public sealed class WorkerClient : IWorkerClient string correlationId, Action setBody) { + // Sequence is deliberately left unset here: WriteLoopAsync stamps it immediately before the + // frame goes out, so the numbers are monotonic in wire order however the callers interleave + // between construction and enqueue (GWC-28, mirroring the worker's WRK-04 fix). WorkerEnvelope envelope = new() { ProtocolVersion = _connection.FrameOptions.ProtocolVersion, SessionId = SessionId, - Sequence = (ulong)Interlocked.Increment(ref _nextSequence), CorrelationId = correlationId, }; setBody(envelope); diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs index 53a0867..3be1668 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -4,6 +4,7 @@ using ZB.MOM.WW.MxGateway.Contracts; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Server.Metrics; using ZB.MOM.WW.MxGateway.Server.Workers; +using ZB.MOM.WW.MxGateway.Tests.Gateway.Workers.Fakes; using ZB.MOM.WW.MxGateway.Tests.TestSupport; namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Workers; @@ -29,6 +30,33 @@ public sealed class WorkerClientTests Assert.Equal(WorkerProcessId, client.ProcessId); } + /// + /// The GatewayHello carries the negotiated worker-frame maximum so the worker adopts the + /// configured limit instead of its own default (IPC-02); a regression to sending 0 means + /// "older gateway, use default" to the worker and would silently downgrade the negotiated limit. + /// The adoption half is asserted only in the Windows-only worker suite, so the gateway half is + /// pinned here, in the portable suite (TST-28). Both the default and an override are covered so + /// the assertion tracks configuration rather than a constant. + /// + /// Configured worker-frame maximum to negotiate. + /// A task that represents the asynchronous operation. + [Theory] + [InlineData(WorkerFrameProtocolOptions.DefaultMaxMessageBytes)] + [InlineData(2 * 1024 * 1024)] + public async Task StartAsync_SendsGatewayHelloWithConfiguredMaxFrameBytes(int maxMessageBytes) + { + await using FakeWorkerHarness harness = + await FakeWorkerHarness.CreateConnectedPairAsync(maxMessageBytes: maxMessageBytes); + await using WorkerClient client = harness.CreateClient(); + + Task startTask = client.StartAsync(CancellationToken.None); + WorkerEnvelope gatewayHello = await harness.CompleteStartupAsync().WaitAsync(TestTimeout); + await startTask.WaitAsync(TestTimeout); + + Assert.NotEqual(0u, gatewayHello.GatewayHello.MaxFrameBytes); + Assert.Equal((uint)maxMessageBytes, gatewayHello.GatewayHello.MaxFrameBytes); + } + /// Verifies that InvokeAsync completes a pending command when a matching reply arrives. /// A task that represents the asynchronous operation. [Fact] @@ -141,6 +169,48 @@ public sealed class WorkerClientTests Assert.Equal(MxCommandKind.GetWorkerInfo, reply.Reply.Kind); } + /// + /// The envelope sequence is a monotonic per-sender counter (gateway.md), so the values + /// observed on the pipe must be strictly increasing in wire order. Stamping the sequence when + /// the envelope is constructed lets two concurrent invokes stamp 1 and 2 but enqueue 2 then 1 + /// (GWC-28); stamping on the single-consumer write loop makes wire order and sequence order the + /// same thing by construction. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ConcurrentInvokesEmitStrictlyIncreasingSequencesOnTheWire() + { + const int commandCount = 32; + await using PipePair pipePair = await PipePair.CreateAsync(); + await using WorkerClient client = CreateClient(pipePair); + await CompleteHandshakeAsync(client, pipePair); + + Task[] invokeTasks = Enumerable.Range(0, commandCount) + .Select(_ => Task.Run(async () => await client.InvokeAsync( + CreateCommand(MxCommandKind.Ping), + TestTimeout, + CancellationToken.None))) + .ToArray(); + + ulong previousSequence = 0; + for (int index = 0; index < commandCount; index++) + { + WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase); + Assert.True( + commandEnvelope.Sequence > previousSequence, + $"Command {index} arrived with sequence {commandEnvelope.Sequence} after {previousSequence}; " + + "envelope sequences must be strictly increasing in wire order."); + previousSequence = commandEnvelope.Sequence; + + await pipePair.WorkerWriter.WriteAsync( + CreateCommandReplyEnvelope(commandEnvelope.CorrelationId, MxCommandKind.Ping)); + } + + await Task.WhenAll(invokeTasks).WaitAsync(TestTimeout); + Assert.Equal(WorkerClientState.Ready, client.State); + } + /// Verifies that ReadEventsAsync yields events in pipe order from the worker. /// A task that represents the asynchronous operation. [Fact]