Merge fix/c3-ipc-boundaries: IPC-33 drain budget bounded under the gRPC send cap + validator upper coupling, IPC-39 pre-check sequence allowance, IPC-34 WorkerCancel on caller cancellation + CallerCancelled code, IPC-35 handshake WorkerFault surfaces category

This commit is contained in:
Joseph Doherty
2026-08-20 09:33:55 -04:00
15 changed files with 430 additions and 62 deletions
+11 -3
View File
@@ -123,7 +123,7 @@ launch CWD (SEC-01, SEC-33).
| `MxGateway:Worker:ShutdownTimeoutSeconds` | `10` | Grace period for worker shutdown before the gateway treats shutdown as failed and may kill the worker process tree. | | `MxGateway:Worker:ShutdownTimeoutSeconds` | `10` | Grace period for worker shutdown before the gateway treats shutdown as failed and may kill the worker process tree. |
| `MxGateway:Worker:HeartbeatIntervalSeconds` | `5` | Worker heartbeat send interval and gateway heartbeat check cadence input. | | `MxGateway:Worker:HeartbeatIntervalSeconds` | `5` | Worker heartbeat send interval and gateway heartbeat check cadence input. |
| `MxGateway:Worker:HeartbeatGraceSeconds` | `15` | Maximum age of the last worker heartbeat before the gateway faults the worker. This must be greater than or equal to `HeartbeatIntervalSeconds`. | | `MxGateway:Worker:HeartbeatGraceSeconds` | `15` | Maximum age of the last worker heartbeat before the gateway faults the worker. This must be greater than or equal to `HeartbeatIntervalSeconds`. |
| `MxGateway:Worker:MaxMessageBytes` | `16842752` | Maximum worker IPC frame payload size in bytes. The validator allows values from `1024` through `268435456`, and additionally requires this to be at least `MxGateway:Protocol:MaxGrpcMessageBytes` plus a 65536-byte (64 KiB) envelope-overhead reserve. The default is the 16 MiB public gRPC cap plus that reserve. The gateway conveys the negotiated value to the worker in the handshake (`GatewayHello.max_frame_bytes`), so the worker frames to the same limit rather than a hard-coded default. | | `MxGateway:Worker:MaxMessageBytes` | `16842752` | Maximum worker IPC frame payload size in bytes. The validator allows values from `1024` through `268435456`, and additionally requires this to be at least `MxGateway:Protocol:MaxGrpcMessageBytes` plus a 65536-byte (64 KiB) envelope-overhead reserve **and at most that cap plus twice the reserve**. The default is the 16 MiB public gRPC cap plus one reserve. The gateway conveys the negotiated value to the worker in the handshake (`GatewayHello.max_frame_bytes`), so the worker frames to the same limit rather than a hard-coded default. The band is narrow by design, so changing either message-size knob alone no longer validates — move `MxGateway:Protocol:MaxGrpcMessageBytes` and this one together, keeping this one one to two reserves above it. |
`StartupProbeRetryAttempts`, `StartupProbeRetryDelayMilliseconds`, `StartupProbeRetryAttempts`, `StartupProbeRetryDelayMilliseconds`,
`PipeConnectAttemptTimeoutMilliseconds`, timeout values, heartbeat values, and `PipeConnectAttemptTimeoutMilliseconds`, timeout values, heartbeat values, and
@@ -132,7 +132,15 @@ to avoid accidental large allocations from malformed or oversized frames. It
must also stay above `MaxGrpcMessageBytes` by the envelope-overhead reserve so a must also stay above `MaxGrpcMessageBytes` by the envelope-overhead reserve so a
maximally-sized accepted gRPC payload always fits one worker frame once wrapped maximally-sized accepted gRPC payload always fits one worker frame once wrapped
in a `WorkerEnvelope`; otherwise the oversized outbound write would fault the in a `WorkerEnvelope`; otherwise the oversized outbound write would fault the
whole session. Startup validation fails fast when the headroom is violated. whole session. The coupling is two-sided: `MaxMessageBytes` may exceed
`MaxGrpcMessageBytes` by at most *twice* the reserve, because the worker budgets
a `DrainEvents` reply against the frame maximum less two reserves and the
gateway then returns that same `MxCommandReply` through `Invoke`, where the gRPC
send cap applies. Without the upper bound a legal configuration (large pipe
maximum, stock gRPC cap) lets the worker pack a reply the pipe carries and the
gRPC layer refuses to send — with the drained events already removed from the
worker queue and therefore lost to both the caller and the event stream. Startup
validation fails fast when either half of the headroom rule is violated.
## Session Options ## Session Options
@@ -385,7 +393,7 @@ is **not yet available**.
| Option | Default | Description | | Option | Default | Description |
|--------|---------|-------------| |--------|---------|-------------|
| `MxGateway:Protocol:WorkerProtocolVersion` | `1` | Worker IPC protocol version expected by the gateway and worker. This must match `GatewayContractInfo.WorkerProtocolVersion`. | | `MxGateway:Protocol:WorkerProtocolVersion` | `1` | Worker IPC protocol version expected by the gateway and worker. This must match `GatewayContractInfo.WorkerProtocolVersion`. |
| `MxGateway:Protocol:MaxGrpcMessageBytes` | `16777216` | Public gRPC max send and receive message size in bytes. The same default is used by official clients. The validator allows values from `1024` through `268435456`, and requires `MxGateway:Worker:MaxMessageBytes` to exceed this by at least the 64 KiB worker-frame envelope reserve (see the Worker options). | | `MxGateway:Protocol:MaxGrpcMessageBytes` | `16777216` | Public gRPC max send and receive message size in bytes. The same default is used by official clients. The validator allows values from `1024` through `268435456`, and requires `MxGateway:Worker:MaxMessageBytes` to exceed this by at least the 64 KiB worker-frame envelope reserve and by no more than twice that reserve, so a `DrainEvents` reply budgeted at the pipe maximum still fits this send cap (see the Worker options). |
The protocol option is exposed for diagnostics and explicit deployment The protocol option is exposed for diagnostics and explicit deployment
configuration, not for compatibility negotiation. A mismatch fails validation configuration, not for compatibility negotiation. A mismatch fails validation
+17 -6
View File
@@ -593,9 +593,19 @@ Pending command handling:
Timeouts should not assume the COM call stopped. A timed-out command may still Timeouts should not assume the COM call stopped. A timed-out command may still
finish inside the worker. finish inside the worker.
On timeout the client also forwards a `WorkerCancel` carrying the abandoned On timeout **and on caller cancellation** the client also forwards a
correlation id, best-effort: the gateway has stopped waiting, but the worker has `WorkerCancel` carrying the abandoned correlation id, best-effort: the gateway
not stopped working, and the worker owns a single STA. `WorkerPipeSession` routes has stopped waiting, but the worker has not stopped working, and the worker owns
a single STA. Caller cancellation — a per-call deadline shorter than the
gateway's command timeout, or a client disconnect cancelling the RPC's token — is
the more common of the two paths, and under a retry storm it is exactly the
backlog the cancel exists to shorten, so both arms send it; the pending command
is failed as `CallerCancelled` rather than `GatewayShutdown` so metrics and logs
name the real cause (both still map to gRPC `Cancelled`). The send is gated on
the same ownership check as the timeout arm — only the arm that removed the
pending entry sends, so a reply that won the race is never chased — and is
skipped when the client's own stop token has fired, since the outbound channel is
being completed and every pending command is failed anyway. `WorkerPipeSession` routes
the cancel to `CancelCommand`, which drops the correlation from the STA queue if the cancel to `CancelCommand`, which drops the correlation from the STA queue if
it has not started and replies `Canceled` for it. A cancel that arrives after the it has not started and replies `Canceled` for it. A cancel that arrives after the
command reached MXAccess is a no-op — there is no way to abort an in-flight COM command reached MXAccess is a no-op — there is no way to abort an in-flight COM
@@ -606,9 +616,10 @@ the worker reads the command and then its cancel and drops it before execution.
Cancels ride the same outbound channel as commands, whose capacity is Cancels ride the same outbound channel as commands, whose capacity is
`MaxPendingCommands + 4`: the reserve above the pending-command limit is what `MaxPendingCommands + 4`: the reserve above the pending-command limit is what
absorbs them, so a burst of timeouts stays bounded and cannot deadlock the absorbs them, so a burst of timeouts or cancellations stays bounded and cannot
enqueue path. Failing to send the cancel is logged at debug and never replaces deadlock the enqueue path. Failing to send the cancel is logged at debug and
the `CommandTimeout` the caller is owed. never replaces the `CommandTimeout` — or the `OperationCanceledException` — the
caller is owed.
Cancellation outranks the deadline. When a caller's token is canceled around the Cancellation outranks the deadline. When a caller's token is canceled around the
same time the timeout fires, the command is reported as canceled same time the timeout fires, the command is reported as canceled
+1 -1
View File
@@ -282,7 +282,7 @@ StatusCode statusCode = exception.ErrorCode switch
}; };
``` ```
`WorkerClientException` follows the same pattern: `CommandTimeout` becomes `DeadlineExceeded`, `GatewayShutdown` becomes `Cancelled`, `InvalidState` becomes `FailedPrecondition`, `ProtocolViolation` becomes `Internal`, `CommandTooLarge` becomes `ResourceExhausted`, and unmapped codes fall through to `Unavailable`. `WorkerClientException` follows the same pattern: `CommandTimeout` becomes `DeadlineExceeded`, `GatewayShutdown` and `CallerCancelled` both become `Cancelled`, `InvalidState` becomes `FailedPrecondition`, `ProtocolViolation` becomes `Internal`, `CommandTooLarge` becomes `ResourceExhausted`, and unmapped codes fall through to `Unavailable`. `CallerCancelled` is the caller's own token firing — a per-call deadline shorter than the gateway command timeout, or a disconnect — and is separate from `GatewayShutdown` only so metrics and logs stop attributing those calls to a shutdown that did not happen; the status the client observes is the same.
## Event Streaming Model ## Event Streaming Model
+17 -3
View File
@@ -43,9 +43,23 @@ guarantees hellos, heartbeats, acks, and faults fit.
Every worker-to-gateway frame must serialize within this limit, control replies Every worker-to-gateway frame must serialize within this limit, control replies
included, so reply builders truncate to fit rather than emit a frame the writer included, so reply builders truncate to fit rather than emit a frame the writer
will reject. `WorkerPipeSession` pre-sizes a `DrainEvents` reply below the will reject. `WorkerPipeSession` pre-sizes a `DrainEvents` reply below the
negotiated maximum (less a fixed envelope/reply-wrapper reserve) and reports the negotiated maximum and reports the truncation in the reply's
truncation in the reply's `DiagnosticMessage`; the caller contract is to repeat `DiagnosticMessage`; the caller contract is to repeat `DrainEvents` until it
`DrainEvents` until it returns an empty reply. Should a reply still overshoot, returns an empty reply. The event budget is `max(MaxMessageBytes - 2 x 64 KiB,
MaxMessageBytes / 2)`: the reserve is subtracted **twice** because the reply has
two boundaries to clear, not one. The first reserve covers the
`WorkerEnvelope`/`WorkerCommandReply`/`MxCommandReply` wrapper on the pipe; the
second exists because the gateway returns that same `MxCommandReply` verbatim
through `Invoke`, where the public gRPC send cap applies. `DrainEvents` removes
the events it packs from the worker queue, so a reply the pipe carries and the
gRPC layer then refuses would lose them for good — the gateway's validator
therefore also caps `MxGateway:Worker:MaxMessageBytes` at
`MxGateway:Protocol:MaxGrpcMessageBytes` plus **two** reserves, which is what
makes the doubled subtraction sufficient rather than merely generous. The
`MaxMessageBytes / 2` floor keeps the budget usable at frame maxima small enough
that two reserves would consume them (below 256 KiB), where a collapsed budget
would truncate every drain and never let the drain-until-empty caller finish.
Should a reply still overshoot,
`MessageTooLarge` at the reply-write seam is answered with a small `MessageTooLarge` at the reply-write seam is answered with a small
`InvalidRequest` reply for that correlation, not with session teardown — no `InvalidRequest` reply for that correlation, not with session teardown — no
diagnostics command may kill a session. diagnostics command may kill a session.
+18 -4
View File
@@ -567,9 +567,14 @@ Optional diagnostics:
requests above the shared ceiling `GatewayContractInfo.MaxDrainEventsPerCommand` requests above the shared ceiling `GatewayContractInfo.MaxDrainEventsPerCommand`
and the worker clamps to the same value, treating `max_events = 0` as "the default and the worker clamps to the same value, treating `max_events = 0` as "the default
cap". By **bytes**: the worker sizes the reply while draining, against the cap". By **bytes**: the worker sizes the reply while draining, against the
negotiated worker-frame maximum, so a byte-heavy queue truncates instead of negotiated worker-frame maximum less two envelope reserves, so a byte-heavy queue
overshooting and events that do not fit stay queued. Truncation is reported in the truncates instead of overshooting and events that do not fit stay queued. The
reply's `DiagnosticMessage`; callers drain iteratively until an empty reply. reserve is subtracted twice because the reply has two boundaries to clear — the
pipe frame, then the public gRPC send cap it is returned through by `Invoke` — and
the validator keeps `Worker:MaxMessageBytes` within two reserves of
`Protocol:MaxGrpcMessageBytes` so that budget always fits both. Truncation is
reported in the reply's `DiagnosticMessage`; callers drain iteratively until an
empty reply.
- `ShutdownWorker` - `ShutdownWorker`
Do not compress MXAccess semantics into generic verbs too early. A command enum Do not compress MXAccess semantics into generic verbs too early. A command enum
@@ -912,10 +917,19 @@ Timeouts:
Cancellation: Cancellation:
- canceling the gRPC call should stop waiting in the gateway, - canceling the gRPC call stops the gateway waiting **and** forwards a
`WorkerCancel` for the abandoned correlation, the same best-effort frame the
command timeout sends — a client deadline or a disconnect is the common
abandonment path, and leaving the correlation on the worker's single STA queue
makes live callers queue behind work nobody is waiting for,
- the cancel is sent only by the caller that owned the pending-entry removal (so
a reply that won the race is never chased) and is skipped once the client is
stopping, where every pending command fails anyway,
- it cannot safely abort an in-flight COM call on the STA, - it cannot safely abort an in-flight COM call on the STA,
- the worker should finish the COM call and discard or log the late reply if - the worker should finish the COM call and discard or log the late reply if
the correlation was canceled, the correlation was canceled,
- caller cancellation is reported as `CallerCancelled` (mapped to gRPC
`Cancelled`), distinct from a gateway shutdown, so metrics name the real cause,
- hard cancellation means killing the worker process. - hard cancellation means killing the worker process.
## Event Delivery And Backpressure ## Event Delivery And Backpressure
@@ -606,6 +606,15 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
// wrapped in a WorkerEnvelope and the outbound write faults the whole session. Fail fast // wrapped in a WorkerEnvelope and the outbound write faults the whole session. Fail fast
// at startup rather than mid-traffic. Only checked when both knobs are themselves in range so the // at startup rather than mid-traffic. Only checked when both knobs are themselves in range so the
// message is not doubled up with the individual range errors. // message is not doubled up with the individual range errors.
//
// IPC-33: the coupling also has an upper half. The worker budgets a DrainEvents reply against the
// negotiated frame maximum less two reserves, and that same MxCommandReply is then returned
// verbatim through Invoke, where the gateway's gRPC MaxSendMessageSize (= MaxGrpcMessageBytes)
// applies. Letting the pipe maximum run arbitrarily far above the gRPC cap lets the worker pack a
// reply the pipe carries and the gRPC layer then refuses to send — with the drained events already
// removed from the worker queue, so they are lost to both the caller and the event stream. Keeping
// the pipe maximum within two reserves of the gRPC cap makes "frame max - 2x reserve" a budget that
// is always at or below the gRPC cap, so the wrapper always has a full reserve of margin.
private static void ValidateFrameSizeHeadroom( private static void ValidateFrameSizeHeadroom(
WorkerOptions worker, WorkerOptions worker,
ProtocolOptions protocol, ProtocolOptions protocol,
@@ -628,6 +637,20 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase<GatewayOption
+ $"{WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes}-byte worker-frame envelope reserve " + $"{WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes}-byte worker-frame envelope reserve "
+ $"(>= {requiredWorkerMax})."); + $"(>= {requiredWorkerMax}).");
} }
long permittedWorkerMax =
(long)protocol.MaxGrpcMessageBytes + (2L * WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes);
if (worker.MaxMessageBytes > permittedWorkerMax)
{
builder.Add(
$"MxGateway:Worker:MaxMessageBytes ({worker.MaxMessageBytes}) must be at most "
+ $"MxGateway:Protocol:MaxGrpcMessageBytes ({protocol.MaxGrpcMessageBytes}) plus twice the "
+ $"{WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes}-byte worker-frame envelope reserve "
+ $"(<= {permittedWorkerMax}). The worker budgets a DrainEvents reply against the negotiated "
+ "frame maximum less two reserves; a pipe maximum further above the gRPC cap lets the worker "
+ "pack a reply the pipe carries but the gRPC send cap refuses, losing the already-drained "
+ "events. Lower MxGateway:Worker:MaxMessageBytes or raise MxGateway:Protocol:MaxGrpcMessageBytes.");
}
} }
private static void AddIfBlank(string? value, string message, ValidationBuilder builder) private static void AddIfBlank(string? value, string message, ValidationBuilder builder)
@@ -58,9 +58,12 @@ public sealed class WorkerOptions
/// The maximum worker-frame (pipe) message size in bytes. Must stay at least /// The maximum worker-frame (pipe) message size in bytes. Must stay at least
/// <see cref="Workers.WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes"/> above /// <see cref="Workers.WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes"/> above
/// <see cref="ProtocolOptions.MaxGrpcMessageBytes"/> so a maximally-sized accepted gRPC payload /// <see cref="ProtocolOptions.MaxGrpcMessageBytes"/> so a maximally-sized accepted gRPC payload
/// still fits one worker frame; the gateway conveys this value to the worker in the /// still fits one worker frame, and at most <em>twice</em> that reserve above it so a
/// handshake (<c>GatewayHello.max_frame_bytes</c>). Default is the 16 MB public gRPC cap /// <c>DrainEvents</c> reply budgeted against this maximum (frame max less two reserves) still
/// plus that reserve. /// fits the gRPC send cap it is returned through — otherwise the worker can pack a reply the
/// pipe carries and the gRPC layer refuses, losing the already-drained events (IPC-33). The
/// gateway conveys this value to the worker in the handshake
/// (<c>GatewayHello.max_frame_bytes</c>). Default is the 16 MB public gRPC cap plus one reserve.
/// </summary> /// </summary>
public int MaxMessageBytes { get; init; } = public int MaxMessageBytes { get; init; } =
(16 * 1024 * 1024) + Workers.WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes; (16 * 1024 * 1024) + Workers.WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes;
@@ -1051,6 +1051,7 @@ public sealed class MxAccessGatewayService(
{ {
WorkerClientErrorCode.CommandTimeout => StatusCode.DeadlineExceeded, WorkerClientErrorCode.CommandTimeout => StatusCode.DeadlineExceeded,
WorkerClientErrorCode.GatewayShutdown => StatusCode.Cancelled, WorkerClientErrorCode.GatewayShutdown => StatusCode.Cancelled,
WorkerClientErrorCode.CallerCancelled => StatusCode.Cancelled,
WorkerClientErrorCode.InvalidState => StatusCode.FailedPrecondition, WorkerClientErrorCode.InvalidState => StatusCode.FailedPrecondition,
WorkerClientErrorCode.ProtocolViolation => StatusCode.Internal, WorkerClientErrorCode.ProtocolViolation => StatusCode.Internal,
WorkerClientErrorCode.CommandTooLarge => StatusCode.ResourceExhausted, WorkerClientErrorCode.CommandTooLarge => StatusCode.ResourceExhausted,
@@ -237,15 +237,22 @@ public sealed class WorkerClient : IWorkerClient
// CalculateSize. WriteLoopAsync stamps envelope.Sequence immediately before the write // CalculateSize. WriteLoopAsync stamps envelope.Sequence immediately before the write
// (GWC-28), and a non-zero varint field grows the encoding — so the number computed here // (GWC-28), and a non-zero varint field grows the encoding — so the number computed here
// is a lower bound on the frame the writer actually emits, never the frame length. Passing // is a lower bound on the frame the writer actually emits, never the frame length. Passing
// it as a knownSize would under-length the prefix and desync the worker's framing. The // it as a knownSize would under-length the prefix and desync the worker's framing.
// pre-check stays a pre-check: it is conservative in the right direction. //
// IPC-39: being a lower bound is conservative for the knownSize question but the wrong
// direction for this pre-check, which must not admit an envelope the writer will then
// reject with a session-fatal MessageTooLarge. Compare against the frame maximum less the
// MaxSequenceEncodingBytes the stamp can add, so every envelope this check accepts still
// fits once the sequence is written.
int envelopeSize = commandEnvelope.CalculateSize(); int envelopeSize = commandEnvelope.CalculateSize();
if (envelopeSize > _connection.FrameOptions.MaxMessageBytes) int preCheckLimit = _connection.FrameOptions.MaxMessageBytes - MaxSequenceEncodingBytes;
if (envelopeSize > preCheckLimit)
{ {
throw new WorkerClientException( throw new WorkerClientException(
WorkerClientErrorCode.CommandTooLarge, WorkerClientErrorCode.CommandTooLarge,
$"Worker command {method} serializes to {envelopeSize} bytes, exceeding the negotiated " $"Worker command {method} serializes to {envelopeSize} bytes, exceeding the negotiated "
+ $"worker-frame maximum of {_connection.FrameOptions.MaxMessageBytes} bytes."); + $"worker-frame maximum of {_connection.FrameOptions.MaxMessageBytes} bytes less the "
+ $"{MaxSequenceEncodingBytes}-byte sequence-stamp allowance.");
} }
await EnqueueAsync(commandEnvelope, cancellationToken).ConfigureAwait(false); await EnqueueAsync(commandEnvelope, cancellationToken).ConfigureAwait(false);
@@ -284,7 +291,10 @@ public sealed class WorkerClient : IWorkerClient
// throw, so it can never replace the timeout the caller is owed. // throw, so it can never replace the timeout the caller is owed.
if (removed) if (removed)
{ {
TrySendCancelForTimedOutCommand(correlationId, method, timeout); TrySendCancelForAbandonedCommand(
correlationId,
method,
$"gateway command timeout after {timeout}");
} }
throw new WorkerClientException( throw new WorkerClientException(
@@ -293,11 +303,7 @@ public sealed class WorkerClient : IWorkerClient
} }
catch (OperationCanceledException) catch (OperationCanceledException)
{ {
RemovePendingCommandAsFailed( AbandonCommandOnCallerCancellation(correlationId, pendingCommand, method);
correlationId,
pendingCommand,
WorkerClientErrorCode.GatewayShutdown,
"Command wait was canceled.");
// WaitAsync surfaces TaskCanceledException; throwing through the token keeps the // WaitAsync surfaces TaskCanceledException; throwing through the token keeps the
// exception the caller observes exactly what the hand-rolled wait produced. // exception the caller observes exactly what the hand-rolled wait produced.
@@ -309,11 +315,7 @@ public sealed class WorkerClient : IWorkerClient
// The deadline and the caller's cancellation raced and WaitAsync picked the timer. // The deadline and the caller's cancellation raced and WaitAsync picked the timer.
// The old wait classified this as cancellation, so this clause — reached only when // The old wait classified this as cancellation, so this clause — reached only when
// the filter above saw a canceled token — reproduces that treatment exactly. // the filter above saw a canceled token — reproduces that treatment exactly.
RemovePendingCommandAsFailed( AbandonCommandOnCallerCancellation(correlationId, pendingCommand, method);
correlationId,
pendingCommand,
WorkerClientErrorCode.GatewayShutdown,
"Command wait was canceled.");
cancellationToken.ThrowIfCancellationRequested(); cancellationToken.ThrowIfCancellationRequested();
throw; throw;
} }
@@ -455,6 +457,18 @@ public sealed class WorkerClient : IWorkerClient
_stopCts.Dispose(); _stopCts.Dispose();
} }
/// <summary>
/// Worst-case number of bytes <see cref="WriteLoopAsync"/>'s sequence stamp adds to an
/// envelope that was sized before it: one tag byte plus a ten-byte varint, the maximum
/// encoding of a 64-bit protobuf field. The <c>CommandTooLarge</c> pre-check subtracts this
/// from the negotiated frame maximum (IPC-39) because it measures the envelope while
/// <c>Sequence</c> is still 0 — an unset field costs nothing to encode, so the measured size
/// is a lower bound on the frame the writer emits, and without the allowance an envelope in
/// the last eleven bytes of the range would pass the pre-check and then fault the session on
/// the write.
/// </summary>
private const int MaxSequenceEncodingBytes = 11;
/// <summary>Manages writing envelopes to the worker pipe.</summary> /// <summary>Manages writing envelopes to the worker pipe.</summary>
private async Task WriteLoopAsync() private async Task WriteLoopAsync()
{ {
@@ -803,6 +817,45 @@ public sealed class WorkerClient : IWorkerClient
return true; return true;
} }
/// <summary>
/// Fails the pending command for a caller whose own token fired — a client deadline shorter
/// than the gateway's command timeout, or a disconnect that cancelled the RPC's token — and
/// forwards a <c>WorkerCancel</c> for the correlation the caller has abandoned (IPC-34).
/// <para>
/// Caller cancellation is the common abandonment path, not the rare one, and it leaves the
/// worker in exactly the state the timeout arm's cancel exists for: the correlation is still
/// on the worker's single STA queue and would execute in its turn, behind live callers, for
/// a caller that is already gone. Under a burst of deadline-cancelled calls that backlog is
/// precisely what the cancel was added to shorten, so both arms send it.
/// </para>
/// <para>
/// Gated the same way as the timeout arm: only the caller that owned the removal sends, so a
/// reply that won the race is never chased. Additionally skipped once <c>_stopCts</c> has
/// fired — the client is closing, the outbound channel is being completed and every pending
/// command failed anyway, so a cancel would be at best pointless and at worst a write racing
/// teardown.
/// </para>
/// </summary>
/// <param name="correlationId">Correlation id of the abandoned command.</param>
/// <param name="pendingCommand">The pending command entry to fail.</param>
/// <param name="method">Command method name, for the cancel reason and diagnostics.</param>
private void AbandonCommandOnCallerCancellation(
string correlationId,
PendingCommand pendingCommand,
string method)
{
bool removed = RemovePendingCommandAsFailed(
correlationId,
pendingCommand,
WorkerClientErrorCode.CallerCancelled,
"Command wait was canceled.");
if (removed && !_stopCts.IsCancellationRequested)
{
TrySendCancelForAbandonedCommand(correlationId, method, "caller cancelled the command");
}
}
/// <summary> /// <summary>
/// Forwards a <c>WorkerCancel</c> for a correlation the gateway has given up waiting for, so the /// Forwards a <c>WorkerCancel</c> for a correlation the gateway has given up waiting for, so the
/// worker can drop it from its STA queue (<c>WorkerPipeSession</c> routes the envelope to /// worker can drop it from its STA queue (<c>WorkerPipeSession</c> routes the envelope to
@@ -819,9 +872,10 @@ public sealed class WorkerClient : IWorkerClient
/// <remarks> /// <remarks>
/// The whole body sits under one catch-all that debug-logs, because the guarantee this method /// The whole body sits under one catch-all that debug-logs, because the guarantee this method
/// owes its caller is structural, not incidental: the caller is on the throw path for the /// owes its caller is structural, not incidental: the caller is on the throw path for the
/// timeout, so anything escaping here — an envelope that failed to build, a <c>TryWrite</c> /// timeout or the cancellation, so anything escaping here — an envelope that failed to build,
/// against a disposed channel, a scheduler refusing the detached task — would replace the /// a <c>TryWrite</c> against a disposed channel, a scheduler refusing the detached task —
/// <see cref="WorkerClientErrorCode.CommandTimeout"/> the caller is owed with an unrelated /// would replace the <see cref="WorkerClientErrorCode.CommandTimeout"/> (or the
/// <see cref="OperationCanceledException"/>) the caller is owed with an unrelated
/// exception. Losing the cancel costs the worker one wasted command; losing the timeout /// exception. Losing the cancel costs the worker one wasted command; losing the timeout
/// misreports why the call failed. The detached task carries its own handler for the same /// misreports why the call failed. The detached task carries its own handler for the same
/// reason: its failures (including the <see cref="ObjectDisposedException"/> from /// reason: its failures (including the <see cref="ObjectDisposedException"/> from
@@ -829,13 +883,13 @@ public sealed class WorkerClient : IWorkerClient
/// and would otherwise be unobserved. It is deliberately not tracked or awaited — it holds no /// and would otherwise be unobserved. It is deliberately not tracked or awaited — it holds no
/// resource the shutdown path needs back, and the outbound channel is completed on close. /// resource the shutdown path needs back, and the outbound channel is completed on close.
/// </remarks> /// </remarks>
/// <param name="correlationId">Correlation id of the command that timed out.</param> /// <param name="correlationId">Correlation id of the abandoned command.</param>
/// <param name="method">Command method name, for the cancel reason and diagnostics.</param> /// <param name="method">Command method name, for diagnostics.</param>
/// <param name="timeout">The elapsed command timeout, for the cancel reason.</param> /// <param name="reason">Why the gateway abandoned the command, carried on the <c>WorkerCancel</c>.</param>
private void TrySendCancelForTimedOutCommand( private void TrySendCancelForAbandonedCommand(
string correlationId, string correlationId,
string method, string method,
TimeSpan timeout) string reason)
{ {
try try
{ {
@@ -843,7 +897,7 @@ public sealed class WorkerClient : IWorkerClient
correlationId, correlationId,
envelope => envelope.WorkerCancel = new WorkerCancel envelope => envelope.WorkerCancel = new WorkerCancel
{ {
Reason = $"gateway command timeout after {timeout}", Reason = reason,
}); });
if (_outboundEnvelopes.Writer.TryWrite(cancelEnvelope)) if (_outboundEnvelopes.Writer.TryWrite(cancelEnvelope))
@@ -869,10 +923,10 @@ public sealed class WorkerClient : IWorkerClient
} }
} }
/// <summary>Records a cancel that could not be forwarded for a timed-out command.</summary> /// <summary>Records a cancel that could not be forwarded for an abandoned command.</summary>
/// <param name="exception">The failure that stopped the cancel from being sent.</param> /// <param name="exception">The failure that stopped the cancel from being sent.</param>
/// <param name="method">Command method name of the timed-out command.</param> /// <param name="method">Command method name of the abandoned command.</param>
/// <param name="correlationId">Correlation id of the timed-out command.</param> /// <param name="correlationId">Correlation id of the abandoned command.</param>
private void LogCancelNotForwarded( private void LogCancelNotForwarded(
Exception exception, Exception exception,
string method, string method,
@@ -880,7 +934,7 @@ public sealed class WorkerClient : IWorkerClient
{ {
_logger.LogDebug( _logger.LogDebug(
exception, exception,
"Could not forward a cancel for timed-out worker command {Method} on session {SessionId} " "Could not forward a cancel for abandoned worker command {Method} on session {SessionId} "
+ "and correlation {CorrelationId}.", + "and correlation {CorrelationId}.",
method, method,
SessionId, SessionId,
@@ -896,6 +950,21 @@ public sealed class WorkerClient : IWorkerClient
CancellationToken cancellationToken) CancellationToken cancellationToken)
{ {
WorkerEnvelope envelope = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false); WorkerEnvelope envelope = await _reader.ReadAsync(cancellationToken).ConfigureAwait(false);
// IPC-35: a fault frame during the handshake is the worker reporting a structured startup
// failure — most often MXAccess COM creation (not installed, not registered, access denied)
// — not a protocol problem. Flattening it to ProtocolViolation discarded the category and
// diagnostic exactly where they matter most: first session open on a freshly deployed host,
// where the message reaches the operator log and the client's OpenFailed detail. Use the
// same formatter the steady-state fault path uses so both windows read alike.
if (envelope.BodyCase == WorkerEnvelope.BodyOneofCase.WorkerFault
&& expectedBody != WorkerEnvelope.BodyOneofCase.WorkerFault)
{
throw new WorkerClientException(
WorkerClientErrorCode.WorkerFaulted,
CreateWorkerFaultMessage(envelope.WorkerFault));
}
if (envelope.BodyCase != expectedBody) if (envelope.BodyCase != expectedBody)
{ {
throw new WorkerClientException( throw new WorkerClientException(
@@ -10,6 +10,14 @@ public enum WorkerClientErrorCode
HeartbeatExpired, HeartbeatExpired,
ShutdownTimeout, ShutdownTimeout,
GatewayShutdown, GatewayShutdown,
// The caller's own cancellation token fired while it waited for the reply — a client deadline
// shorter than the gateway's command timeout, or a disconnect that cancelled the RPC's token.
// Distinct from GatewayShutdown (which these arms used to report) so metrics and logs stop
// recording a shutdown that did not happen; both map to Cancelled at the gRPC boundary, so the
// status the client observes is unchanged.
CallerCancelled,
WriteFailed, WriteFailed,
PendingCommandLimitExceeded, PendingCommandLimitExceeded,
@@ -86,6 +86,11 @@ public sealed class GatewayOptionsTests
["MxGateway:Events:QueueCapacity"] = "256", ["MxGateway:Events:QueueCapacity"] = "256",
["MxGateway:Dashboard:Enabled"] = "false", ["MxGateway:Dashboard:Enabled"] = "false",
["MxGateway:Protocol:MaxGrpcMessageBytes"] = "8388608", ["MxGateway:Protocol:MaxGrpcMessageBytes"] = "8388608",
// The frame-headroom rule is two-sided (IPC-33): the worker-frame maximum must sit
// between one and two 64 KiB envelope reserves above the gRPC cap, so moving the
// gRPC cap without moving this one no longer validates. Kept at one reserve above,
// the same shape as the shipped default.
["MxGateway:Worker:MaxMessageBytes"] = "8454144",
// Blank-by-default bind password must be supplied so enabled LDAP validates on bind. // Blank-by-default bind password must be supplied so enabled LDAP validates on bind.
["MxGateway:Ldap:ServiceAccountPassword"] = "test-bind-password" ["MxGateway:Ldap:ServiceAccountPassword"] = "test-bind-password"
}); });
@@ -1,5 +1,6 @@
using Microsoft.Extensions.Options; using Microsoft.Extensions.Options;
using ZB.MOM.WW.MxGateway.Server.Configuration; using ZB.MOM.WW.MxGateway.Server.Configuration;
using ZB.MOM.WW.MxGateway.Server.Workers;
using LdapTransport = ZB.MOM.WW.Auth.Abstractions.Ldap.LdapTransport; using LdapTransport = ZB.MOM.WW.Auth.Abstractions.Ldap.LdapTransport;
namespace ZB.MOM.WW.MxGateway.Tests.Configuration; namespace ZB.MOM.WW.MxGateway.Tests.Configuration;
@@ -1256,6 +1257,57 @@ public sealed class GatewayOptionsValidatorTests
Assert.Contains(result.Failures!, f => f.Contains("MaxMessageBytes") && f.Contains("reserve")); Assert.Contains(result.Failures!, f => f.Contains("MaxMessageBytes") && f.Contains("reserve"));
} }
/// <summary>
/// Verifies the headroom rule's upper half (IPC-33): a worker-frame maximum more than two
/// envelope reserves above the gRPC cap is rejected. The worker budgets a DrainEvents reply at
/// the frame maximum less two reserves and the gateway returns that same reply through
/// <c>Invoke</c>, so a wider gap lets the worker pack a reply the pipe carries and the gRPC send
/// cap refuses — with the drained events already removed from the worker queue.
/// </summary>
[Fact]
public void Validate_Fails_WhenWorkerFrameMaxExceedsGrpcMaxByMoreThanTwoReserves()
{
const int grpcMax = 16 * 1024 * 1024;
int tooLarge = grpcMax + (2 * WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes) + 1;
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(
null,
WithWorkerAndProtocol(
new WorkerOptions { MaxMessageBytes = tooLarge },
new ProtocolOptions { MaxGrpcMessageBytes = grpcMax }));
Assert.True(result.Failed);
Assert.Contains(
result.Failures!,
f => f.Contains("MaxMessageBytes", StringComparison.Ordinal)
&& f.Contains("at most", StringComparison.Ordinal)
&& f.Contains("DrainEvents", StringComparison.Ordinal));
}
/// <summary>
/// Verifies both edges of the two-sided headroom rule are themselves accepted: exactly one
/// reserve above the gRPC cap (the shipped default's shape) and exactly two reserves above it
/// (the widest legal gap). Pinning the boundaries keeps a future retune of the reserve from
/// silently making the shipped default illegal.
/// </summary>
/// <param name="reserveMultiple">Number of envelope reserves between the gRPC cap and the frame maximum.</param>
[Theory]
[InlineData(1)]
[InlineData(2)]
public void Validate_Succeeds_AtBothEdgesOfTheFrameHeadroomBand(int reserveMultiple)
{
const int grpcMax = 16 * 1024 * 1024;
int workerMax = grpcMax + (reserveMultiple * WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes);
ValidateOptionsResult result = new GatewayOptionsValidator().Validate(
null,
WithWorkerAndProtocol(
new WorkerOptions { MaxMessageBytes = workerMax },
new ProtocolOptions { MaxGrpcMessageBytes = grpcMax }));
Assert.True(result.Succeeded);
}
/// <summary> /// <summary>
/// Verifies the shipped worker event-queue capacity default passes validation and still matches /// Verifies the shipped worker event-queue capacity default passes validation and still matches
/// the worker-side <c>MxAccessEventQueue.DefaultCapacity</c> the environment variable falls back /// the worker-side <c>MxAccessEventQueue.DefaultCapacity</c> the environment variable falls back
@@ -216,6 +216,151 @@ public sealed class WorkerClientTests
Assert.Equal(WorkerClientState.Ready, client.State); Assert.Equal(WorkerClientState.Ready, client.State);
} }
/// <summary>
/// Caller cancellation abandons the correlation exactly as a gateway timeout does — a client
/// deadline shorter than the gateway's command timeout, or a disconnect, is the common
/// abandonment path — so it forwards the same <c>WorkerCancel</c> (IPC-34). Without it a
/// retry storm of deadline-cancelled calls leaves the worker's single STA working through
/// commands nobody is waiting for while live callers queue behind them. Twin of
/// <see cref="InvokeAsync_WhenCommandTimesOut_SendsWorkerCancelForThatCorrelation"/>, driven
/// by the caller's token instead of the timer.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task InvokeAsync_WhenCallerCancels_SendsWorkerCancelForThatCorrelation()
{
await using PipePair pipePair = await PipePair.CreateAsync();
await using WorkerClient client = CreateClient(pipePair);
await CompleteHandshakeAsync(client, pipePair);
using CancellationTokenSource callerCts = new();
// Advise rather than a control command: control commands bypass the worker's STA queue, so a
// data command is the case the cancel actually exists for. The timeout is the full test
// timeout so the cancellation, not the timer, is unambiguously what abandons the command.
Task<WorkerCommandReply> invokeTask = client.InvokeAsync(
CreateCommand(MxCommandKind.Advise),
TestTimeout,
callerCts.Token);
WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase);
await callerCts.CancelAsync();
await Assert.ThrowsAnyAsync<OperationCanceledException>(
async () => await invokeTask.WaitAsync(TestTimeout));
WorkerEnvelope cancelEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCancel, cancelEnvelope.BodyCase);
Assert.Equal(commandEnvelope.CorrelationId, cancelEnvelope.CorrelationId);
Assert.False(string.IsNullOrWhiteSpace(cancelEnvelope.WorkerCancel.Reason));
Assert.True(
cancelEnvelope.Sequence > commandEnvelope.Sequence,
$"The cancel arrived with sequence {cancelEnvelope.Sequence} after {commandEnvelope.Sequence}; "
+ "envelope sequences must be strictly increasing in wire order.");
// One caller gave up; the session did not fault.
Assert.Equal(WorkerClientState.Ready, client.State);
}
/// <summary>
/// The cancel is forwarded for a caller that abandoned its command, never for the client's own
/// teardown (IPC-34): once the stop token has fired the outbound channel is being completed and
/// every pending command is failed anyway, so a cancel would at best be pointless and at worst
/// a write racing teardown. Asserted on the wire — the guard in the cancellation arm is the
/// code-level expression of it, and the completed outbound channel is the second line of
/// defense, so this pins the observable contract rather than either mechanism.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task InvokeAsync_WhenClientStopsUnderInFlightCommand_DoesNotSendWorkerCancel()
{
await using PipePair pipePair = await PipePair.CreateAsync();
await using WorkerClient client = CreateClient(pipePair);
await CompleteHandshakeAsync(client, pipePair);
using CancellationTokenSource callerCts = new();
Task<WorkerCommandReply> invokeTask = client.InvokeAsync(
CreateCommand(MxCommandKind.Advise),
TestTimeout,
callerCts.Token);
WorkerEnvelope commandEnvelope = await pipePair.WorkerReader.ReadAsync().AsTask().WaitAsync(TestTimeout);
Assert.Equal(WorkerEnvelope.BodyOneofCase.WorkerCommand, commandEnvelope.BodyCase);
// Read ahead before tearing down so anything the teardown path puts on the wire is caught.
// Exceptions are swallowed: the read is expected to be still pending at the assertion, and
// the stream is disposed underneath it when the test ends.
Task<WorkerEnvelope?> pendingRead = Task.Run(async () =>
{
try
{
return await pipePair.WorkerReader.ReadAsync();
}
catch (Exception)
{
return null;
}
});
// Kill fires the client's stop token and then fails the pending command; a caller token that
// is wired to gateway shutdown (as the session's is) cancels in the same window.
client.Kill("test teardown");
await callerCts.CancelAsync();
await Assert.ThrowsAnyAsync<Exception>(async () => await invokeTask.WaitAsync(TestTimeout));
await Task.Delay(TimeSpan.FromMilliseconds(250));
if (pendingRead.IsCompleted)
{
WorkerEnvelope? unexpected = await pendingRead;
Assert.True(
unexpected is null || unexpected.BodyCase != WorkerEnvelope.BodyOneofCase.WorkerCancel,
"A stopping worker client must not put a WorkerCancel on the wire.");
}
}
/// <summary>
/// A <c>WorkerFault</c> arriving in place of a handshake frame is the worker reporting a
/// structured startup failure — most often MXAccess COM creation — not a protocol problem, so
/// the gateway surfaces <see cref="WorkerClientErrorCode.WorkerFaulted"/> with the fault's
/// category and diagnostic rather than a generic protocol violation naming only the body type
/// (IPC-35). This is the window where the information matters most: first session open on a
/// freshly deployed host, where the message reaches the operator log and the client's
/// <c>OpenFailed</c> detail.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[Fact]
public async Task StartAsync_WhenWorkerFaultsDuringHandshake_SurfacesFaultCategoryAndDiagnostic()
{
const string diagnostic = "CoCreateInstance for LMXProxy.LMXProxyServer.1 failed (0x80040154).";
await using FakeWorkerHarness harness = await FakeWorkerHarness.CreateConnectedPairAsync();
await using WorkerClient client = harness.CreateClient();
Task startTask = client.StartAsync(CancellationToken.None);
WorkerEnvelope gatewayHello = await harness.ReadGatewayEnvelopeAsync().WaitAsync(TestTimeout);
Assert.Equal(WorkerEnvelope.BodyOneofCase.GatewayHello, gatewayHello.BodyCase);
// The hello succeeds; the fault replaces the WorkerReady, which is where a COM creation
// failure actually lands (the worker creates MXAccess after answering the hello).
await harness.SendWorkerHelloAsync();
await harness.EmitFaultAsync(WorkerFaultCategory.MxaccessCreationFailed, diagnostic);
WorkerClientException exception = await Assert.ThrowsAsync<WorkerClientException>(
async () => await startTask.WaitAsync(TestTimeout));
Assert.Equal(WorkerClientErrorCode.WorkerFaulted, exception.ErrorCode);
Assert.Contains(
WorkerFaultCategory.MxaccessCreationFailed.ToString(),
exception.Message,
StringComparison.Ordinal);
Assert.Contains(diagnostic, exception.Message, StringComparison.Ordinal);
}
/// <summary> /// <summary>
/// The envelope <c>sequence</c> is a monotonic per-sender counter (gateway.md), so the values /// 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 /// observed on the pipe must be strictly increasing in wire order. Stamping the sequence when
@@ -692,7 +692,7 @@ public sealed class WorkerPipeSessionTests
/// <summary> /// <summary>
/// Regression for the reserve-boundary budget bug. The gateway validator accepts a Worker /// Regression for the reserve-boundary budget bug. The gateway validator accepts a Worker
/// frame maximum as low as 1024 + 64 KiB, and just above that boundary a naive /// frame maximum as low as 1024 + 64 KiB, and just above that boundary a naive
/// subtract-then-guard budget collapses to ~1024 bytes — too small to move even one /// subtract-then-guard budget collapses to nothing — too small to move even one
/// byte-heavy event, so every drain reports truncation with the same head blocked and the /// byte-heavy event, so every drain reports truncation with the same head blocked and the
/// drain-until-empty loop never terminates. The budget is now a floor (never below half the /// drain-until-empty loop never terminates. The budget is now a floor (never below half the
/// negotiated maximum), so a byte-heavy queue drains to empty even at the validator floor. /// negotiated maximum), so a byte-heavy queue drains to empty even at the validator floor.
@@ -702,8 +702,9 @@ public sealed class WorkerPipeSessionTests
public async Task DrainEvents_AtValidatorFloorFrameMax_MakesProgressAndTerminates() public async Task DrainEvents_AtValidatorFloorFrameMax_MakesProgressAndTerminates()
{ {
// The lowest Worker.MaxMessageBytes GatewayOptionsValidator permits: the public gRPC floor // The lowest Worker.MaxMessageBytes GatewayOptionsValidator permits: the public gRPC floor
// (1024) plus the 64 KiB envelope-overhead reserve. The naive budget would be exactly 1024 // (1024) plus the 64 KiB envelope-overhead reserve. The naive budget — frame max less the
// here; the floored budget is half of the frame max (~33 KiB). // two reserves the DrainEvents budget subtracts (IPC-33) — would be negative here; the
// floored budget is half of the frame max (~33 KiB).
const uint validatorFloorFrameMax = 1024 + (64 * 1024); const uint validatorFloorFrameMax = 1024 + (64 * 1024);
using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(60)); using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(60));
using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token);
@@ -27,7 +27,8 @@ public sealed class WorkerPipeSession
// Headroom subtracted from the negotiated frame maximum when budgeting a DrainEvents reply. It // Headroom subtracted from the negotiated frame maximum when budgeting a DrainEvents reply. It
// covers the WorkerEnvelope/WorkerCommandReply/MxCommandReply wrapper the drained events are // covers the WorkerEnvelope/WorkerCommandReply/MxCommandReply wrapper the drained events are
// packed into — the same envelope-overhead reserve rationale docs/WorkerFrameProtocol.md // packed into — the same envelope-overhead reserve rationale docs/WorkerFrameProtocol.md
// records for the frame max itself. // records for the frame max itself. ResolveDrainReplyByteBudget subtracts it twice: once for the
// pipe frame, once so the reply also clears the public gRPC send cap it is forwarded through.
private const int DrainReplyFrameHeadroomBytes = 64 * 1024; private const int DrainReplyFrameHeadroomBytes = 64 * 1024;
private readonly WorkerFrameProtocolOptions _options; private readonly WorkerFrameProtocolOptions _options;
@@ -1021,20 +1022,33 @@ public sealed class WorkerPipeSession
/// <summary> /// <summary>
/// Byte budget for the events packed into one DrainEvents reply: the negotiated frame /// Byte budget for the events packed into one DrainEvents reply: the negotiated frame
/// maximum less a fixed reserve for the envelope/reply wrapper, but never below half the /// maximum less <em>two</em> reserves, but never below half the negotiated maximum. The
/// negotiated maximum. The lower bound must be a floor, not a step: a bare /// first reserve covers the envelope/reply wrapper the events are packed into so the reply
/// <c>subtract-then-guard-positive</c> collapses the budget to a handful of bytes just above /// fits one pipe frame. The second reserve exists because that same
/// the reserve (e.g. at the validator-permitted floor MaxMessageBytes = 1024 + 64 KiB the /// <c>MxCommandReply</c> does not stop at the pipe: the gateway returns it verbatim through
/// subtraction leaves 1024, too small to move even one byte-heavy event, so every drain /// <c>Invoke</c>, where the public gRPC send cap applies, and the gateway's validator
/// truncates and the drain-until-empty caller never terminates). Taking the max with /// couples the two limits so the pipe maximum never exceeds the gRPC cap by more than two
/// half the negotiated maximum keeps the budget monotonic across the reserve boundary while /// reserves (IPC-33). Subtracting both therefore keeps the packed events at or below the
/// still leaving the full reserve for the wrapper whenever the frame max is large enough /// gRPC cap and leaves the wrapper — correlation id, protocol status, kind, the
/// that the reserve is the smaller subtraction — which is every configuration above 128 KiB. /// <c>drain_events</c> length prefix and, in the truncated case, the
/// <c>DiagnosticMessage</c> — a full reserve of margin under it. Budgeting against the pipe
/// maximum alone let a legal configuration build a reply the pipe carried and the gRPC layer
/// then refused to send, with the events already dequeued and therefore lost.
/// <para>
/// The lower bound must be a floor, not a step: a bare <c>subtract-then-guard-positive</c>
/// collapses the budget to a handful of bytes (or below zero) once the reserves approach the
/// frame maximum (e.g. at the validator-permitted floor MaxMessageBytes = 1024 + 64 KiB),
/// too small to move even one byte-heavy event, so every drain truncates and the
/// drain-until-empty caller never terminates. Taking the max with half the negotiated
/// maximum keeps the budget monotonic across the reserve boundary while still leaving both
/// reserves for the wrapper whenever the frame max is large enough that the subtraction is
/// the smaller of the two — which is every configuration above 256 KiB.
/// </para>
/// </summary> /// </summary>
private int ResolveDrainReplyByteBudget() private int ResolveDrainReplyByteBudget()
{ {
return Math.Max( return Math.Max(
_options.MaxMessageBytes - DrainReplyFrameHeadroomBytes, _options.MaxMessageBytes - (2 * DrainReplyFrameHeadroomBytes),
_options.MaxMessageBytes / 2); _options.MaxMessageBytes / 2);
} }