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()