fix(GWC-28): stamp gateway envelope sequence at write, not construction
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.
This commit is contained in:
+7
-3
@@ -348,9 +348,13 @@ messages, tagged from 10 upward:
|
|||||||
|
|
||||||
Rules:
|
Rules:
|
||||||
|
|
||||||
- `sequence` is a monotonic per-sender counter used as a diagnostic aid; it is
|
- `sequence` is a monotonic per-sender counter used as a diagnostic aid. Both
|
||||||
not validated for gaps or ordering on receive (the named pipe already
|
sides stamp it at the point of writing, inside their single write path — the
|
||||||
guarantees FIFO delivery).
|
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
|
- `correlation_id` links a command to its reply; it is authoritative on the
|
||||||
envelope, and the inner `MxCommandReply.correlation_id` echoes it for
|
envelope, and the inner `MxCommandReply.correlation_id` echoes it for
|
||||||
MXAccess parity.
|
MXAccess parity.
|
||||||
|
|||||||
@@ -40,7 +40,9 @@ public sealed class WorkerClient : IWorkerClient
|
|||||||
private readonly ConcurrentDictionary<string, PendingCommand> _pendingCommands = new(StringComparer.Ordinal);
|
private readonly ConcurrentDictionary<string, PendingCommand> _pendingCommands = new(StringComparer.Ordinal);
|
||||||
private readonly SemaphoreSlim _pendingCommandSlots;
|
private readonly SemaphoreSlim _pendingCommandSlots;
|
||||||
private readonly CancellationTokenSource _stopCts = new();
|
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 WorkerClientState _state;
|
||||||
private DateTimeOffset _lastHeartbeatAt;
|
private DateTimeOffset _lastHeartbeatAt;
|
||||||
private int? _processId;
|
private int? _processId;
|
||||||
@@ -404,6 +406,13 @@ public sealed class WorkerClient : IWorkerClient
|
|||||||
{
|
{
|
||||||
await foreach (WorkerEnvelope envelope in _outboundEnvelopes.Reader.ReadAllAsync(_stopCts.Token).ConfigureAwait(false))
|
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);
|
await _writer.WriteAsync(envelope, _stopCts.Token).ConfigureAwait(false);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1072,11 +1081,13 @@ public sealed class WorkerClient : IWorkerClient
|
|||||||
string correlationId,
|
string correlationId,
|
||||||
Action<WorkerEnvelope> setBody)
|
Action<WorkerEnvelope> 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()
|
WorkerEnvelope envelope = new()
|
||||||
{
|
{
|
||||||
ProtocolVersion = _connection.FrameOptions.ProtocolVersion,
|
ProtocolVersion = _connection.FrameOptions.ProtocolVersion,
|
||||||
SessionId = SessionId,
|
SessionId = SessionId,
|
||||||
Sequence = (ulong)Interlocked.Increment(ref _nextSequence),
|
|
||||||
CorrelationId = correlationId,
|
CorrelationId = correlationId,
|
||||||
};
|
};
|
||||||
setBody(envelope);
|
setBody(envelope);
|
||||||
|
|||||||
@@ -4,6 +4,7 @@ using ZB.MOM.WW.MxGateway.Contracts;
|
|||||||
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
||||||
using ZB.MOM.WW.MxGateway.Server.Metrics;
|
using ZB.MOM.WW.MxGateway.Server.Metrics;
|
||||||
using ZB.MOM.WW.MxGateway.Server.Workers;
|
using ZB.MOM.WW.MxGateway.Server.Workers;
|
||||||
|
using ZB.MOM.WW.MxGateway.Tests.Gateway.Workers.Fakes;
|
||||||
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
|
using ZB.MOM.WW.MxGateway.Tests.TestSupport;
|
||||||
|
|
||||||
namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Workers;
|
namespace ZB.MOM.WW.MxGateway.Tests.Gateway.Workers;
|
||||||
@@ -29,6 +30,33 @@ public sealed class WorkerClientTests
|
|||||||
Assert.Equal(WorkerProcessId, client.ProcessId);
|
Assert.Equal(WorkerProcessId, client.ProcessId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The <c>GatewayHello</c> carries the negotiated worker-frame maximum so the worker adopts the
|
||||||
|
/// configured limit instead of its own default (IPC-02); a regression to sending <c>0</c> 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <param name="maxMessageBytes">Configured worker-frame maximum to negotiate.</param>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[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);
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>Verifies that InvokeAsync completes a pending command when a matching reply arrives.</summary>
|
/// <summary>Verifies that InvokeAsync completes a pending command when a matching reply arrives.</summary>
|
||||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
[Fact]
|
[Fact]
|
||||||
@@ -141,6 +169,48 @@ public sealed class WorkerClientTests
|
|||||||
Assert.Equal(MxCommandKind.GetWorkerInfo, reply.Reply.Kind);
|
Assert.Equal(MxCommandKind.GetWorkerInfo, reply.Reply.Kind);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// <summary>
|
||||||
|
/// The envelope <c>sequence</c> 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.
|
||||||
|
/// </summary>
|
||||||
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
|
[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<WorkerCommandReply>[] 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);
|
||||||
|
}
|
||||||
|
|
||||||
/// <summary>Verifies that ReadEventsAsync yields events in pipe order from the worker.</summary>
|
/// <summary>Verifies that ReadEventsAsync yields events in pipe order from the worker.</summary>
|
||||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||||
[Fact]
|
[Fact]
|
||||||
|
|||||||
Reference in New Issue
Block a user