From ebe6aeac988d97a5c12f410e0a2d1f91910c4618 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 09:09:14 -0400 Subject: [PATCH] feat(worker): adopt negotiated frame max, bound drain, priority write scheduler (IPC-02/04 + WRK-04/07 worker half) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Worker half of the Wave 3 size/backpressure + write-ordering pass: - IPC-02: the worker adopts GatewayHello.max_frame_bytes during the handshake (WorkerFrameProtocolOptions.AdoptNegotiatedMaxMessageBytes) instead of a hard-coded default; 0 keeps the default, a value above a 256 MiB ceiling is rejected. Reader and writer share the options instance, applied before the message loop. - IPC-04: DrainEvents caps each reply at MaxDrainEventsPerReply (10_000) and treats max_events = 0 as that cap rather than 'drain the entire queue', so one diagnostic drain cannot pack a session-killing reply frame. - WRK-04: WorkerFrameWriter stamps the envelope Sequence at the actual point of writing (under the write lock) instead of at envelope creation, so the on-wire order and the stamped sequence always agree under concurrent producers. - WRK-07: the writer is now a cooperative priority scheduler — callers enqueue at Control or Event priority and the draining lock-holder writes all control frames before any event frame, so replies/faults/heartbeats jump ahead of an event backlog. Per-frame validation/size rejections fail only that frame; a stream write failure fails all queued frames. Tests: monotonic gap-free sequence under concurrency, control-before-event priority (gated stream), negotiated-max adoption, DrainEvents zero-bound. Worker builds x86 only — verified on windev. --- .../Ipc/WorkerFrameProtocolTests.cs | 156 +++++++++++++++ .../Ipc/WorkerPipeSessionTests.cs | 38 ++++ .../TestSupport/FakeRuntimeSession.cs | 10 + .../Ipc/WorkerFrameProtocolOptions.cs | 42 +++- .../Ipc/WorkerFrameWritePriority.cs | 17 ++ .../Ipc/WorkerFrameWriter.cs | 182 ++++++++++++++++-- .../Ipc/WorkerPipeSession.cs | 35 +++- 7 files changed, 450 insertions(+), 30 deletions(-) create mode 100644 src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWritePriority.cs diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs index 468a81b..8921762 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerFrameProtocolTests.cs @@ -1,5 +1,7 @@ +using System; using System.IO; using System.Linq; +using System.Threading; using System.Threading.Tasks; using ZB.MOM.WW.MxGateway.Contracts; using ZB.MOM.WW.MxGateway.Contracts.Proto; @@ -305,6 +307,101 @@ public sealed class WorkerFrameProtocolTests Nonce); } + /// + /// Verifies that under concurrent writers every frame receives a distinct, gap-free sequence in + /// strictly increasing on-wire order — the sequence is stamped by the writer at write time, so the + /// wire order and the stamped sequence always agree (WRK-04). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_UnderConcurrentCalls_StampsGapFreeMonotonicSequence() + { + const int frameCount = 50; + WorkerFrameProtocolOptions options = CreateOptions(); + using MemoryStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + await Task.WhenAll( + Enumerable.Range(0, frameCount).Select(_ => writer.WriteAsync(CreateEventEnvelope()))); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + ulong[] sequences = new ulong[frameCount]; + for (int index = 0; index < frameCount; index++) + { + sequences[index] = (await reader.ReadAsync()).Sequence; + } + + // On-wire order is strictly increasing 1..frameCount with no gaps or duplicates. + Assert.Equal(Enumerable.Range(1, frameCount).Select(value => (ulong)value), sequences); + } + + /// + /// Verifies that when a control frame and an event frame are both queued behind an in-progress + /// write, the draining lock-holder writes the control frame first even though the event was queued + /// earlier (WRK-07). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task WriteAsync_WhenControlAndEventQueuedTogether_WritesControlFirst() + { + WorkerFrameProtocolOptions options = CreateOptions(); + using GatedWriteStream stream = new(); + WorkerFrameWriter writer = new(stream, options); + + // First write occupies the writer and blocks inside the stream, holding the write lock. + Task firstWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await AwaitWithTimeoutAsync(stream.FirstWriteStarted); + + // Queue an event first, then a control frame, while the writer is blocked. Both wait for the lock. + Task eventWrite = writer.WriteAsync(CreateEventEnvelope(), WorkerFrameWritePriority.Event); + Task controlWrite = writer.WriteAsync(CreateGatewayHelloEnvelope(), WorkerFrameWritePriority.Control); + await Task.Delay(50); + + stream.ReleaseFirstWrite(); + await AwaitWithTimeoutAsync(Task.WhenAll(firstWrite, eventWrite, controlWrite)); + + stream.Position = 0; + WorkerFrameReader reader = new(stream, options); + WorkerEnvelope frame1 = await reader.ReadAsync(); + WorkerEnvelope frame2 = await reader.ReadAsync(); + WorkerEnvelope frame3 = await reader.ReadAsync(); + + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame1.BodyCase); + // The control frame jumped ahead of the earlier-queued event. + Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, frame2.BodyCase); + Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerEvent, frame3.BodyCase); + } + + /// Verifies a zero negotiated frame maximum keeps the constructor default (IPC-02). + [Fact] + public void AdoptNegotiatedMaxMessageBytes_WithZero_KeepsDefault() + { + WorkerFrameProtocolOptions options = CreateOptions(); + int original = options.MaxMessageBytes; + options.AdoptNegotiatedMaxMessageBytes(0); + Assert.Equal(original, options.MaxMessageBytes); + } + + /// Verifies an in-range negotiated frame maximum is adopted (IPC-02). + [Fact] + public void AdoptNegotiatedMaxMessageBytes_WithInRangeValue_Adopts() + { + WorkerFrameProtocolOptions options = CreateOptions(); + options.AdoptNegotiatedMaxMessageBytes(4 * 1024 * 1024); + Assert.Equal(4 * 1024 * 1024, options.MaxMessageBytes); + } + + /// Verifies a negotiated frame maximum above the worker ceiling is rejected (IPC-02). + [Fact] + public void AdoptNegotiatedMaxMessageBytes_AboveCeiling_Throws() + { + WorkerFrameProtocolOptions options = CreateOptions(); + WorkerFrameProtocolException exception = Assert.Throws( + () => options.AdoptNegotiatedMaxMessageBytes((uint)WorkerFrameProtocolOptions.MaxNegotiableFrameBytes + 1)); + Assert.Equal(WorkerFrameProtocolErrorCode.InvalidConfiguration, exception.ErrorCode); + } + private static WorkerEnvelope CreateGatewayHelloEnvelope(ulong sequence = 1) { return new WorkerEnvelope @@ -321,4 +418,63 @@ public sealed class WorkerFrameProtocolTests }; } + // net48 has no Task.WaitAsync(TimeSpan); fail the test rather than hang if the writer misbehaves. + private static async Task AwaitWithTimeoutAsync(Task task) + { + Task completed = await Task.WhenAny(task, Task.Delay(TimeSpan.FromSeconds(5))); + if (completed != task) + { + throw new TimeoutException("Timed out waiting for the frame writer."); + } + + await task; + } + + private static WorkerEnvelope CreateEventEnvelope() + { + return new WorkerEnvelope + { + ProtocolVersion = GatewayContractInfo.WorkerProtocolVersion, + SessionId = SessionId, + WorkerEvent = new WorkerEvent + { + Event = new MxEvent { SessionId = SessionId }, + }, + }; + } + + // A MemoryStream whose first WriteAsync blocks until released, so a test can queue additional frames + // behind an in-progress write and observe the writer's priority ordering. + private sealed class GatedWriteStream : MemoryStream + { + private readonly SemaphoreSlim _release = new SemaphoreSlim(0); + private readonly TaskCompletionSource _firstWriteStarted = + new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + private int _writeCount; + + public Task FirstWriteStarted => _firstWriteStarted.Task; + + public void ReleaseFirstWrite() => _release.Release(); + + public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + if (Interlocked.Increment(ref _writeCount) == 1) + { + _firstWriteStarted.TrySetResult(true); + await _release.WaitAsync(cancellationToken); + } + + await base.WriteAsync(buffer, offset, count, cancellationToken); + } + + protected override void Dispose(bool disposing) + { + if (disposing) + { + _release.Dispose(); + } + + base.Dispose(disposing); + } + } } diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs index 117770a..3533efe 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -449,6 +449,44 @@ public sealed class WorkerPipeSessionTests await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); } + /// + /// Verifies that a DrainEvents control command with max_events = 0 is bounded by the + /// worker rather than draining the entire queue into one reply frame: the session passes a + /// capped, non-zero maximum to the runtime session (IPC-04). + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task RunAsync_DrainEventsWithZeroMaxEvents_BoundsTheDrain() + { + using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(5)); + using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); + FakeRuntimeSession runtime = new() { SuppressDrainForBatchSize = 128 }; + WorkerPipeSession session = CreatePipeSession(pipePair.WorkerStream, runtime); + runtime.EnqueueEvent(CreateWorkerEvent(sequence: 11)); + Task runTask = session.RunAsync(cancellation.Token); + await CompleteGatewayHandshakeAsync(pipePair, cancellation.Token); + + await pipePair.GatewayWriter + .WriteAsync( + CreateControlCommandEnvelope( + "drain-cap-1", + MxCommandKind.DrainEvents, + command => command.DrainEvents = new DrainEventsCommand { MaxEvents = 0 }), + cancellation.Token); + + await ReadUntilAsync( + pipePair.GatewayReader, + WorkerEnvelope.BodyOneofCase.WorkerCommandReply, + cancellation.Token); + + // The client asked for "all" (0) but the worker must pass a bounded, non-zero cap so the reply + // frame cannot grow without limit. + Assert.NotNull(runtime.LastDrainMaxEvents); + Assert.NotEqual(0u, runtime.LastDrainMaxEvents!.Value); + + await SendShutdownAndWaitAsync(pipePair, runTask, cancellation.Token); + } + /// /// Verifies that ShutdownWorker returns its OK reply BEFORE the graceful /// shutdown runs and disposes the runtime session, and that the message diff --git a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs index c1002a9..776c708 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/TestSupport/FakeRuntimeSession.cs @@ -125,6 +125,14 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession /// public uint? SuppressDrainForBatchSize { get; set; } + /// + /// Records the maxEvents argument of the most recent non-suppressed + /// call — i.e. the effective cap the session passed for an explicit + /// DrainEvents control command. Lets a test assert the worker bounds the drain (IPC-04) rather + /// than forwarding the client's raw max_events = 0. + /// + public uint? LastDrainMaxEvents { get; private set; } + /// public IReadOnlyList DrainEvents(uint maxEvents) { @@ -133,6 +141,8 @@ internal sealed class FakeRuntimeSession : IWorkerRuntimeSession return Array.Empty(); } + LastDrainMaxEvents = maxEvents; + lock (gate) { int drainCount = maxEvents == 0 diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameProtocolOptions.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameProtocolOptions.cs index d8332f5..b9e016e 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameProtocolOptions.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameProtocolOptions.cs @@ -10,6 +10,14 @@ public sealed class WorkerFrameProtocolOptions /// Default maximum message size in bytes (16 MB). public const int DefaultMaxMessageBytes = 16 * 1024 * 1024; + /// + /// Upper ceiling the worker will accept for a gateway-negotiated frame maximum + /// (GatewayHello.max_frame_bytes, IPC-02). Matches the gateway's own configuration ceiling + /// so a nonsensical negotiated value is rejected at the handshake rather than driving an absurd + /// per-frame allocation. 256 MiB. + /// + public const int MaxNegotiableFrameBytes = 256 * 1024 * 1024; + /// Initializes a new instance of the WorkerFrameProtocolOptions class from WorkerOptions. /// Worker initialization options. public WorkerFrameProtocolOptions(WorkerOptions options) @@ -98,6 +106,36 @@ public sealed class WorkerFrameProtocolOptions /// Gets the nonce for startup validation. public string Nonce { get; } - /// Gets the maximum message size in bytes. - public int MaxMessageBytes { get; } + /// + /// Gets the maximum worker-frame message size in bytes. Initialized from the constructor and + /// then adopted once from the gateway-negotiated value during the startup handshake + /// (GatewayHello.max_frame_bytes, IPC-02) via , + /// before the message loop starts. Not mutated afterwards, so the single-threaded handshake write + /// is safe for the reader/writer that share this instance. + /// + public int MaxMessageBytes { get; private set; } + + /// + /// Adopts the gateway-negotiated frame maximum conveyed in GatewayHello.max_frame_bytes + /// (IPC-02). A value of 0 (an older gateway that never set the field) is ignored and the + /// constructor default is kept. A value above is rejected. + /// + /// The gateway-negotiated maximum, or 0 for "keep default". + internal void AdoptNegotiatedMaxMessageBytes(uint negotiatedMaxFrameBytes) + { + if (negotiatedMaxFrameBytes == 0) + { + return; + } + + if (negotiatedMaxFrameBytes > MaxNegotiableFrameBytes) + { + throw new WorkerFrameProtocolException( + WorkerFrameProtocolErrorCode.InvalidConfiguration, + $"GatewayHello negotiated frame maximum {negotiatedMaxFrameBytes} exceeds the worker ceiling " + + $"of {MaxNegotiableFrameBytes} bytes."); + } + + MaxMessageBytes = (int)negotiatedMaxFrameBytes; + } } diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWritePriority.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWritePriority.cs new file mode 100644 index 0000000..46c60f8 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWritePriority.cs @@ -0,0 +1,17 @@ +namespace ZB.MOM.WW.MxGateway.Worker.Ipc; + +/// +/// Relative scheduling priority for an outbound worker frame. The single writer task drains all +/// pending frames before any frame, so a command reply, +/// fault, heartbeat, or shutdown acknowledgement is not delayed behind a backlog of queued events +/// (WRK-07). Priority only reorders the write; the frame sequence is stamped at actual write time, +/// so the on-wire order and the envelope Sequence always agree (WRK-04). +/// +public enum WorkerFrameWritePriority +{ + /// Control-plane frame (hello, ready, command reply, heartbeat, fault, shutdown ack). Written ahead of events. + Control = 0, + + /// Event frame. Written only when no control frame is pending. + Event = 1, +} diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs index a80d548..5ae2e58 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerFrameWriter.cs @@ -1,4 +1,5 @@ using System; +using System.Collections.Generic; using System.IO; using System.Threading; using System.Threading.Tasks; @@ -7,12 +8,40 @@ using ZB.MOM.WW.MxGateway.Contracts.Proto; namespace ZB.MOM.WW.MxGateway.Worker.Ipc; -/// Writes worker frames to a stream with length-prefixed protobuf serialization. +/// +/// Writes worker frames to a stream with length-prefixed protobuf serialization. Callers enqueue a +/// frame at a and then contend for a single write lock; whoever +/// holds the lock drains every queued frame, control frames first, so a reply, fault, or heartbeat is +/// never delayed behind an event backlog (WRK-07). The envelope Sequence is stamped by the +/// draining lock-holder at the moment of writing, so the on-wire order and the stamped sequence always +/// agree even under concurrent callers and priority reordering (WRK-04). +/// public sealed class WorkerFrameWriter { + private sealed class PendingFrame + { + public PendingFrame(WorkerEnvelope envelope) + { + Envelope = envelope; + Completion = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + } + + public WorkerEnvelope Envelope { get; } + + public TaskCompletionSource Completion { get; } + } + private readonly WorkerFrameProtocolOptions _options; - private readonly SemaphoreSlim _writeLock = new(1, 1); + private readonly SemaphoreSlim _writeLock = new SemaphoreSlim(1, 1); private readonly Stream _stream; + private readonly object _gate = new object(); + private readonly Queue _controlFrames = new Queue(); + private readonly Queue _eventFrames = new Queue(); + + // Only ever read/written by the current write-lock holder while draining, so no interlock is + // needed. Starts at 0 and is pre-incremented, so the first written frame carries sequence 1 + // (matching the previous behaviour). + private ulong _nextSequence; /// Initializes a new instance of the WorkerFrameWriter class. /// Stream to write frames to. @@ -25,12 +54,25 @@ public sealed class WorkerFrameWriter _options = options ?? throw new ArgumentNullException(nameof(options)); } - /// Writes a worker envelope frame to the stream with length prefix. + /// Writes a control-priority worker envelope frame to the stream with length prefix. /// Worker envelope to write. /// Token to cancel the asynchronous operation. - /// A task that represents the asynchronous operation. + /// A task that completes when the frame has been written and flushed. + public Task WriteAsync( + WorkerEnvelope envelope, + CancellationToken cancellationToken = default) + { + return WriteAsync(envelope, WorkerFrameWritePriority.Control, cancellationToken); + } + + /// Queues a worker envelope frame for writing at the given priority and drains the queue. + /// Worker envelope to write. + /// Scheduling priority; control frames are written ahead of event frames. + /// Token to cancel waiting for the write lock. + /// A task that completes when the frame has been written and flushed. public async Task WriteAsync( WorkerEnvelope envelope, + WorkerFrameWritePriority priority, CancellationToken cancellationToken = default) { if (envelope is null) @@ -38,8 +80,120 @@ public sealed class WorkerFrameWriter throw new ArgumentNullException(nameof(envelope)); } + PendingFrame frame = new PendingFrame(envelope); + lock (_gate) + { + if (priority == WorkerFrameWritePriority.Event) + { + _eventFrames.Enqueue(frame); + } + else + { + _controlFrames.Enqueue(frame); + } + } + + // Contend for the single writer: whoever wins drains every currently-queued frame in priority + // order, so this frame is written by this call or by a concurrent caller that got the lock + // first. Either way it completes via its own TaskCompletionSource. + await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await DrainQueuedFramesAsync().ConfigureAwait(false); + } + finally + { + _writeLock.Release(); + } + + await frame.Completion.Task.ConfigureAwait(false); + } + + // Runs only under _writeLock. Drains control frames before event frames, stamping and writing each. + // The stream write itself is not cancellable: a frame is written atomically or fails, never left + // half-written on the pipe because a caller gave up waiting. + private async Task DrainQueuedFramesAsync() + { + while (true) + { + PendingFrame? frame = DequeueNext(); + if (frame is null) + { + return; + } + + try + { + await WriteFrameAsync(frame.Envelope).ConfigureAwait(false); + frame.Completion.TrySetResult(true); + } + catch (WorkerFrameProtocolException exception) when (IsPerFrameRejection(exception)) + { + // Validation, empty-payload, and oversized-frame errors are specific to this frame and + // do not damage the stream; fail only this frame and keep draining the rest. + frame.Completion.TrySetException(exception); + } + catch (Exception exception) + { + // A stream write/flush failure means the pipe is broken; fail this frame and every frame + // still queued so no caller awaits forever, then stop draining. + frame.Completion.TrySetException(exception); + FailAllQueued(exception); + return; + } + } + } + + private static bool IsPerFrameRejection(WorkerFrameProtocolException exception) + { + return exception.ErrorCode is WorkerFrameProtocolErrorCode.InvalidEnvelope + or WorkerFrameProtocolErrorCode.MessageTooLarge + or WorkerFrameProtocolErrorCode.ProtocolVersionMismatch + or WorkerFrameProtocolErrorCode.SessionMismatch; + } + + private PendingFrame? DequeueNext() + { + lock (_gate) + { + if (_controlFrames.Count > 0) + { + return _controlFrames.Dequeue(); + } + + if (_eventFrames.Count > 0) + { + return _eventFrames.Dequeue(); + } + + return null; + } + } + + private void FailAllQueued(Exception exception) + { + lock (_gate) + { + while (_controlFrames.Count > 0) + { + _controlFrames.Dequeue().Completion.TrySetException(exception); + } + + while (_eventFrames.Count > 0) + { + _eventFrames.Dequeue().Completion.TrySetException(exception); + } + } + } + + private async Task WriteFrameAsync(WorkerEnvelope envelope) + { WorkerEnvelopeValidator.Validate(envelope, _options); + // Stamp the sequence at the actual point of writing, under the write lock, so the wire order + // and the stamped sequence agree regardless of caller concurrency or priority (WRK-04). + envelope.Sequence = unchecked(++_nextSequence); + int payloadLength = envelope.CalculateSize(); if (payloadLength == 0) { @@ -55,26 +209,16 @@ public sealed class WorkerFrameWriter $"Worker envelope payload length {payloadLength} exceeds the configured maximum of {_options.MaxMessageBytes} bytes."); } - // Serialize once into a single buffer that carries the 4-byte - // length prefix followed by the payload, then issue one stream write. - // This avoids a second serialization pass (envelope.ToByteArray() - // would re-run CalculateSize internally), a separate prefix array, - // and a separate prefix write. + // Serialize once into a single buffer that carries the 4-byte length prefix followed by the + // payload, then issue one stream write. This avoids a second serialization pass, a separate + // prefix array, and a separate prefix write. int frameLength = sizeof(uint) + payloadLength; byte[] frame = new byte[frameLength]; WriteUInt32LittleEndian(frame, (uint)payloadLength); envelope.WriteTo(new Span(frame, sizeof(uint), payloadLength)); - await _writeLock.WaitAsync(cancellationToken).ConfigureAwait(false); - try - { - await _stream.WriteAsync(frame, 0, frameLength, cancellationToken).ConfigureAwait(false); - await _stream.FlushAsync(cancellationToken).ConfigureAwait(false); - } - finally - { - _writeLock.Release(); - } + await _stream.WriteAsync(frame, 0, frameLength, CancellationToken.None).ConfigureAwait(false); + await _stream.FlushAsync(CancellationToken.None).ConfigureAwait(false); } private static void WriteUInt32LittleEndian( diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index b13b601..fea8594 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -18,6 +18,13 @@ public sealed class WorkerPipeSession private static readonly TimeSpan BackgroundTaskStopTimeout = TimeSpan.FromSeconds(1); private const uint EventDrainBatchSize = 128; + // Hard cap on how many events a single DrainEvents diagnostic reply may carry. DrainEvents is a + // non-streaming control command, so an unbounded drain (including the max_events = 0 "as many as + // available" request) could pack the whole queue into one session-killing reply frame (IPC-04). + // The gateway request validator rejects requests above its public ceiling; this worker-side cap is + // the backstop and defines the effective per-reply maximum. Kept in step with that public ceiling. + private const uint MaxDrainEventsPerReply = 10_000; + private readonly WorkerFrameProtocolOptions _options; private readonly Func _processIdProvider; private readonly Func _runtimeSessionFactory; @@ -28,7 +35,6 @@ public sealed class WorkerPipeSession private readonly object _commandTaskGate = new(); private readonly HashSet _activeCommandTasks = new(); private IWorkerRuntimeSession? _runtimeSession; - private long _nextSequence; // Mutated from the message loop, command tasks, the heartbeat loop and the // shutdown path; volatile so cross-thread reads observe the latest state @@ -225,6 +231,12 @@ public sealed class WorkerPipeSession WorkerFrameProtocolErrorCode.NonceMismatch, "GatewayHello nonce does not match the worker launch nonce."); } + + // Adopt the gateway-negotiated frame maximum so both ends frame to the same limit instead of + // matched compile-time defaults (IPC-02). Applied here, before the message loop, so every + // post-handshake frame is validated against the negotiated value; the reader and writer share + // this options instance. The hello frame itself was small and already read under the default. + _options.AdoptNegotiatedMaxMessageBytes(gatewayHello.MaxFrameBytes); } private Task WriteWorkerHelloAsync(CancellationToken cancellationToken) @@ -361,8 +373,11 @@ public sealed class WorkerPipeSession foreach (WorkerEvent workerEvent in events) { + // Events are the low-priority frame class: the writer holds them behind any pending + // control frame (reply, fault, heartbeat, shutdown ack) so those are not delayed + // behind an event backlog (WRK-07). await _writer - .WriteAsync(CreateEnvelope(workerEvent), cancellationToken) + .WriteAsync(CreateEnvelope(workerEvent), WorkerFrameWritePriority.Event, cancellationToken) .ConfigureAwait(false); } } @@ -527,7 +542,12 @@ public sealed class WorkerPipeSession IWorkerRuntimeSession? runtimeSession = _runtimeSession; if (runtimeSession is not null) { - uint maxEvents = command.DrainEvents?.MaxEvents ?? 0; + // Bound the diagnostic drain so max_events = 0 ("as many as available") or an over-large + // request cannot pack the whole queue into one session-killing reply frame (IPC-04). + uint requested = command.DrainEvents?.MaxEvents ?? 0; + uint maxEvents = requested == 0 || requested > MaxDrainEventsPerReply + ? MaxDrainEventsPerReply + : requested; foreach (WorkerEvent workerEvent in runtimeSession.DrainEvents(maxEvents)) { if (workerEvent.Event is not null) @@ -1004,19 +1024,16 @@ public sealed class WorkerPipeSession private WorkerEnvelope CreateBaseEnvelope() { + // Sequence is deliberately left unset here: the frame writer stamps it at the actual point of + // writing, under its single drain task, so the on-wire order and the stamped sequence agree + // even under concurrent producers and priority reordering (WRK-04). return new WorkerEnvelope { ProtocolVersion = _options.ProtocolVersion, SessionId = _options.SessionId, - Sequence = NextSequence(), }; } - private ulong NextSequence() - { - return unchecked((ulong)Interlocked.Increment(ref _nextSequence)); - } - private async Task InitializeMxAccessAsync(CancellationToken cancellationToken) { // RunAsync constructs the runtime session via _runtimeSessionFactory()