diff --git a/docs/GatewayConfiguration.md b/docs/GatewayConfiguration.md index 8621852..2b34d2d 100644 --- a/docs/GatewayConfiguration.md +++ b/docs/GatewayConfiguration.md @@ -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: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: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`, `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 maximally-sized accepted gRPC payload always fits one worker frame once wrapped 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 @@ -385,7 +393,7 @@ is **not yet available**. | Option | Default | Description | |--------|---------|-------------| | `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 configuration, not for compatibility negotiation. A mismatch fails validation diff --git a/docs/GatewayProcessDesign.md b/docs/GatewayProcessDesign.md index 0fe7cd9..035a9e1 100644 --- a/docs/GatewayProcessDesign.md +++ b/docs/GatewayProcessDesign.md @@ -593,9 +593,19 @@ Pending command handling: Timeouts should not assume the COM call stopped. A timed-out command may still finish inside the worker. -On timeout the client also forwards a `WorkerCancel` carrying the abandoned -correlation id, best-effort: the gateway has stopped waiting, but the worker has -not stopped working, and the worker owns a single STA. `WorkerPipeSession` routes +On timeout **and on caller cancellation** the client also forwards a +`WorkerCancel` carrying the abandoned correlation id, best-effort: the gateway +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 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 @@ -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 `MaxPendingCommands + 4`: the reserve above the pending-command limit is what -absorbs them, so a burst of timeouts stays bounded and cannot deadlock the -enqueue path. Failing to send the cancel is logged at debug and never replaces -the `CommandTimeout` the caller is owed. +absorbs them, so a burst of timeouts or cancellations stays bounded and cannot +deadlock the enqueue path. Failing to send the cancel is logged at debug and +never replaces the `CommandTimeout` — or the `OperationCanceledException` — the +caller is owed. Cancellation outranks the deadline. When a caller's token is canceled around the same time the timeout fires, the command is reported as canceled diff --git a/docs/Grpc.md b/docs/Grpc.md index e548e91..a09d5cc 100644 --- a/docs/Grpc.md +++ b/docs/Grpc.md @@ -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 diff --git a/docs/WorkerFrameProtocol.md b/docs/WorkerFrameProtocol.md index ebff8d7..132a226 100644 --- a/docs/WorkerFrameProtocol.md +++ b/docs/WorkerFrameProtocol.md @@ -43,9 +43,23 @@ guarantees hellos, heartbeats, acks, and faults fit. 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 will reject. `WorkerPipeSession` pre-sizes a `DrainEvents` reply below the -negotiated maximum (less a fixed envelope/reply-wrapper reserve) and reports the -truncation in the reply's `DiagnosticMessage`; the caller contract is to repeat -`DrainEvents` until it returns an empty reply. Should a reply still overshoot, +negotiated maximum and reports the truncation in the reply's +`DiagnosticMessage`; the caller contract is to repeat `DrainEvents` until it +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 `InvalidRequest` reply for that correlation, not with session teardown — no diagnostics command may kill a session. diff --git a/gateway.md b/gateway.md index 77c113f..08e03a0 100644 --- a/gateway.md +++ b/gateway.md @@ -567,9 +567,14 @@ Optional diagnostics: requests above the shared ceiling `GatewayContractInfo.MaxDrainEventsPerCommand` 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 - negotiated worker-frame maximum, so a byte-heavy queue truncates instead of - overshooting and events that do not fit stay queued. Truncation is reported in the - reply's `DiagnosticMessage`; callers drain iteratively until an empty reply. + negotiated worker-frame maximum less two envelope reserves, so a byte-heavy queue + truncates instead of overshooting and events that do not fit stay queued. The + 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` Do not compress MXAccess semantics into generic verbs too early. A command enum @@ -912,10 +917,19 @@ Timeouts: 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, - the worker should finish the COM call and discard or log the late reply if 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. ## Event Delivery And Backpressure diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs index e9a7b1f..a0d8496 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/GatewayOptionsValidator.cs @@ -606,6 +606,15 @@ public sealed class GatewayOptionsValidator : OptionsValidatorBase= {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) diff --git a/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs b/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs index 699a0e2..140f2e9 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Configuration/WorkerOptions.cs @@ -58,9 +58,12 @@ public sealed class WorkerOptions /// The maximum worker-frame (pipe) message size in bytes. Must stay at least /// above /// so a maximally-sized accepted gRPC payload - /// still fits one worker frame; the gateway conveys this value to the worker in the - /// handshake (GatewayHello.max_frame_bytes). Default is the 16 MB public gRPC cap - /// plus that reserve. + /// still fits one worker frame, and at most twice that reserve above it so a + /// DrainEvents reply budgeted against this maximum (frame max less two reserves) still + /// 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 + /// (GatewayHello.max_frame_bytes). Default is the 16 MB public gRPC cap plus one reserve. /// public int MaxMessageBytes { get; init; } = (16 * 1024 * 1024) + Workers.WorkerFrameProtocolOptions.EnvelopeOverheadReserveBytes; diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs index 4481c1d..a10699c 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs @@ -1051,6 +1051,7 @@ public sealed class MxAccessGatewayService( { WorkerClientErrorCode.CommandTimeout => StatusCode.DeadlineExceeded, WorkerClientErrorCode.GatewayShutdown => StatusCode.Cancelled, + WorkerClientErrorCode.CallerCancelled => StatusCode.Cancelled, WorkerClientErrorCode.InvalidState => StatusCode.FailedPrecondition, WorkerClientErrorCode.ProtocolViolation => StatusCode.Internal, WorkerClientErrorCode.CommandTooLarge => StatusCode.ResourceExhausted, diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs index bff47b1..41cf7bf 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClient.cs @@ -237,15 +237,22 @@ public sealed class WorkerClient : IWorkerClient // 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 // 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 - // pre-check stays a pre-check: it is conservative in the right direction. + // it as a knownSize would under-length the prefix and desync the worker's framing. + // + // 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(); - if (envelopeSize > _connection.FrameOptions.MaxMessageBytes) + int preCheckLimit = _connection.FrameOptions.MaxMessageBytes - MaxSequenceEncodingBytes; + if (envelopeSize > preCheckLimit) { throw new WorkerClientException( WorkerClientErrorCode.CommandTooLarge, $"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); @@ -284,7 +291,10 @@ public sealed class WorkerClient : IWorkerClient // throw, so it can never replace the timeout the caller is owed. if (removed) { - TrySendCancelForTimedOutCommand(correlationId, method, timeout); + TrySendCancelForAbandonedCommand( + correlationId, + method, + $"gateway command timeout after {timeout}"); } throw new WorkerClientException( @@ -293,11 +303,7 @@ public sealed class WorkerClient : IWorkerClient } catch (OperationCanceledException) { - RemovePendingCommandAsFailed( - correlationId, - pendingCommand, - WorkerClientErrorCode.GatewayShutdown, - "Command wait was canceled."); + AbandonCommandOnCallerCancellation(correlationId, pendingCommand, method); // WaitAsync surfaces TaskCanceledException; throwing through the token keeps the // 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 old wait classified this as cancellation, so this clause — reached only when // the filter above saw a canceled token — reproduces that treatment exactly. - RemovePendingCommandAsFailed( - correlationId, - pendingCommand, - WorkerClientErrorCode.GatewayShutdown, - "Command wait was canceled."); + AbandonCommandOnCallerCancellation(correlationId, pendingCommand, method); cancellationToken.ThrowIfCancellationRequested(); throw; } @@ -455,6 +457,18 @@ public sealed class WorkerClient : IWorkerClient _stopCts.Dispose(); } + /// + /// Worst-case number of bytes '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 CommandTooLarge pre-check subtracts this + /// from the negotiated frame maximum (IPC-39) because it measures the envelope while + /// Sequence 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. + /// + private const int MaxSequenceEncodingBytes = 11; + /// Manages writing envelopes to the worker pipe. private async Task WriteLoopAsync() { @@ -803,6 +817,45 @@ public sealed class WorkerClient : IWorkerClient return true; } + /// + /// 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 WorkerCancel for the correlation the caller has abandoned (IPC-34). + /// + /// 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. + /// + /// + /// 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 _stopCts 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. + /// + /// + /// Correlation id of the abandoned command. + /// The pending command entry to fail. + /// Command method name, for the cancel reason and diagnostics. + 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"); + } + } + /// /// Forwards a WorkerCancel for a correlation the gateway has given up waiting for, so the /// worker can drop it from its STA queue (WorkerPipeSession routes the envelope to @@ -819,9 +872,10 @@ public sealed class WorkerClient : IWorkerClient /// /// 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 - /// timeout, so anything escaping here — an envelope that failed to build, a TryWrite - /// against a disposed channel, a scheduler refusing the detached task — would replace the - /// the caller is owed with an unrelated + /// timeout or the cancellation, so anything escaping here — an envelope that failed to build, + /// a TryWrite against a disposed channel, a scheduler refusing the detached task — + /// would replace the (or the + /// ) the caller is owed with an unrelated /// 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 /// reason: its failures (including the 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 /// resource the shutdown path needs back, and the outbound channel is completed on close. /// - /// Correlation id of the command that timed out. - /// Command method name, for the cancel reason and diagnostics. - /// The elapsed command timeout, for the cancel reason. - private void TrySendCancelForTimedOutCommand( + /// Correlation id of the abandoned command. + /// Command method name, for diagnostics. + /// Why the gateway abandoned the command, carried on the WorkerCancel. + private void TrySendCancelForAbandonedCommand( string correlationId, string method, - TimeSpan timeout) + string reason) { try { @@ -843,7 +897,7 @@ public sealed class WorkerClient : IWorkerClient correlationId, envelope => envelope.WorkerCancel = new WorkerCancel { - Reason = $"gateway command timeout after {timeout}", + Reason = reason, }); if (_outboundEnvelopes.Writer.TryWrite(cancelEnvelope)) @@ -869,10 +923,10 @@ public sealed class WorkerClient : IWorkerClient } } - /// Records a cancel that could not be forwarded for a timed-out command. + /// Records a cancel that could not be forwarded for an abandoned command. /// The failure that stopped the cancel from being sent. - /// Command method name of the timed-out command. - /// Correlation id of the timed-out command. + /// Command method name of the abandoned command. + /// Correlation id of the abandoned command. private void LogCancelNotForwarded( Exception exception, string method, @@ -880,7 +934,7 @@ public sealed class WorkerClient : IWorkerClient { _logger.LogDebug( 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}.", method, SessionId, @@ -896,6 +950,21 @@ public sealed class WorkerClient : IWorkerClient CancellationToken cancellationToken) { 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) { throw new WorkerClientException( diff --git a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs index b16c58d..bcc8d48 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Workers/WorkerClientErrorCode.cs @@ -10,6 +10,14 @@ public enum WorkerClientErrorCode HeartbeatExpired, ShutdownTimeout, 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, PendingCommandLimitExceeded, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsTests.cs index 73594ec..966cec3 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsTests.cs @@ -86,6 +86,11 @@ public sealed class GatewayOptionsTests ["MxGateway:Events:QueueCapacity"] = "256", ["MxGateway:Dashboard:Enabled"] = "false", ["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. ["MxGateway:Ldap:ServiceAccountPassword"] = "test-bind-password" }); diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs index 89e4581..54be27a 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Configuration/GatewayOptionsValidatorTests.cs @@ -1,5 +1,6 @@ using Microsoft.Extensions.Options; using ZB.MOM.WW.MxGateway.Server.Configuration; +using ZB.MOM.WW.MxGateway.Server.Workers; using LdapTransport = ZB.MOM.WW.Auth.Abstractions.Ldap.LdapTransport; 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")); } + /// + /// 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 + /// Invoke, 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. + /// + [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)); + } + + /// + /// 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. + /// + /// Number of envelope reserves between the gRPC cap and the frame maximum. + [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); + } + /// /// Verifies the shipped worker event-queue capacity default passes validation and still matches /// the worker-side MxAccessEventQueue.DefaultCapacity the environment variable falls back diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs index 807249d..ad3761b 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Workers/WorkerClientTests.cs @@ -216,6 +216,151 @@ public sealed class WorkerClientTests Assert.Equal(WorkerClientState.Ready, client.State); } + /// + /// 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 WorkerCancel (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 + /// , driven + /// by the caller's token instead of the timer. + /// + /// A task that represents the asynchronous operation. + [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 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( + 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); + } + + /// + /// 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. + /// + /// A task that represents the asynchronous operation. + [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 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 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(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."); + } + } + + /// + /// A WorkerFault 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 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 + /// OpenFailed detail. + /// + /// A task that represents the asynchronous operation. + [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( + 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); + } + /// /// The envelope sequence is a monotonic per-sender counter (gateway.md), so the values /// observed on the pipe must be strictly increasing in wire order. Stamping the sequence when 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 dca5936..09df002 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker.Tests/Ipc/WorkerPipeSessionTests.cs @@ -692,7 +692,7 @@ public sealed class WorkerPipeSessionTests /// /// 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 - /// 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 /// 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. @@ -702,8 +702,9 @@ public sealed class WorkerPipeSessionTests public async Task DrainEvents_AtValidatorFloorFrameMax_MakesProgressAndTerminates() { // 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 - // here; the floored budget is half of the frame max (~33 KiB). + // (1024) plus the 64 KiB envelope-overhead reserve. The naive budget — frame max less the + // 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); using CancellationTokenSource cancellation = new(TimeSpan.FromSeconds(60)); using PipePair pipePair = await PipePair.CreateAsync(cancellation.Token); diff --git a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs index d52f8c4..60297c5 100644 --- a/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs +++ b/src/ZB.MOM.WW.MxGateway.Worker/Ipc/WorkerPipeSession.cs @@ -27,7 +27,8 @@ public sealed class WorkerPipeSession // Headroom subtracted from the negotiated frame maximum when budgeting a DrainEvents reply. It // covers the WorkerEnvelope/WorkerCommandReply/MxCommandReply wrapper the drained events are // 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 readonly WorkerFrameProtocolOptions _options; @@ -1021,20 +1022,33 @@ public sealed class WorkerPipeSession /// /// 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 - /// negotiated maximum. The lower bound must be a floor, not a step: a bare - /// subtract-then-guard-positive collapses the budget to a handful of bytes just above - /// the reserve (e.g. at the validator-permitted floor MaxMessageBytes = 1024 + 64 KiB the - /// subtraction leaves 1024, 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 the full reserve for the wrapper whenever the frame max is large enough - /// that the reserve is the smaller subtraction — which is every configuration above 128 KiB. + /// maximum less two reserves, but never below half the negotiated maximum. The + /// first reserve covers the envelope/reply wrapper the events are packed into so the reply + /// fits one pipe frame. The second reserve exists because that same + /// MxCommandReply does not stop at the pipe: the gateway returns it verbatim through + /// Invoke, where the public gRPC send cap applies, and the gateway's validator + /// couples the two limits so the pipe maximum never exceeds the gRPC cap by more than two + /// reserves (IPC-33). Subtracting both therefore keeps the packed events at or below the + /// gRPC cap and leaves the wrapper — correlation id, protocol status, kind, the + /// drain_events length prefix and, in the truncated case, the + /// DiagnosticMessage — 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. + /// + /// The lower bound must be a floor, not a step: a bare subtract-then-guard-positive + /// 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. + /// /// private int ResolveDrainReplyByteBudget() { return Math.Max( - _options.MaxMessageBytes - DrainReplyFrameHeadroomBytes, + _options.MaxMessageBytes - (2 * DrainReplyFrameHeadroomBytes), _options.MaxMessageBytes / 2); }