diff --git a/docs/Grpc.md b/docs/Grpc.md index 28d1cc4..5c0bf4b 100644 --- a/docs/Grpc.md +++ b/docs/Grpc.md @@ -82,6 +82,8 @@ return mapper.MapCommandReply(workerReply); Carrying the enqueue timestamp into the worker layer is what lets queue-wait time be measured separately from worker-side execution time when troubleshooting timeouts. +Every `MxCommandReply` echoes the request's `client_correlation_id` in `correlation_id` — success, MXAccess failure, and constraint-denied replies alike, and an empty string when the caller sent none. The handler stamps it after mapping the worker reply, overwriting the internal gateway↔worker envelope id the worker reply carries: that id is a pipe-routing detail with no meaning to a client, and clients that sent no correlation id previously received it while clients that sent one received it only on denied replies. + An accepted gRPC command payload can still be too large for the worker pipe: the envelope built around it must fit `MxGateway:Worker:MaxMessageBytes`, which is validated at startup to sit at least a fixed envelope-overhead reserve above `MaxGrpcMessageBytes` (see the headroom rule in [Gateway Configuration](./GatewayConfiguration.md)) so this should not occur for a conformant payload, but if it does, `WorkerClient` raises `WorkerClientException(CommandTooLarge)` and `Invoke` reports `ResourceExhausted` for that command — the session is not faulted, so a client can retry with a smaller payload without reopening the session. Selected control commands have additional public behavior beyond their protobuf shape: @@ -94,6 +96,8 @@ Selected control commands have additional public behavior beyond their protobuf `StreamEvents` is a server-streaming RPC. The handler delegates the full pipeline to `IEventStreamService` and just forwards each `MxEvent` onto the response stream. Keeping the channel and producer/consumer machinery out of the handler means cancellation, exception mapping, and metric bookkeeping live in one place. +Each response write receives the call's cancellation token, so a write parked on a slow or vanished client is released when the call is cancelled rather than waiting on the transport. A write that then fails with an `IOException` or `OperationCanceledException` while the call token is cancelled is a routine client disconnect: it maps to `Cancelled` and logs nothing. The same `IOException` on a live call keeps the generic `Unavailable` + warning mapping, because that is a real transport failure. `StreamAlarms` applies the same rule. + ### `AcknowledgeAlarm` `AcknowledgeAlarm` is a unary, **session-less** RPC that acknowledges a single alarm. The handler validates `alarm_full_reference` and `operator_user` inline (it does not run through `MxAccessGrpcRequestValidator`) and delegates to `IGatewayAlarmService.AcknowledgeAsync`. The always-on `GatewayAlarmMonitor` routes the ack over its own gateway-managed worker session — clients no longer open a session to acknowledge an alarm. A reference that parses as a canonical GUID forwards to `AcknowledgeAlarmCommand`; a `Provider!Group.Tag` reference forwards to `AcknowledgeAlarmByNameCommand`. @@ -125,6 +129,8 @@ The worker calls `AuthenticateUser` as the operator, then writes `operator_comme `StreamAlarms` is a server-streaming, **session-less** RPC that attaches to the gateway's central alarm feed. The handler delegates to `IGatewayAlarmService.StreamAsync`. The stream opens with a `provider_status` and a `snapshot_status` `AlarmFeedMessage` (the current provider mode and snapshot-completeness verdict), then one `AlarmFeedMessage` carrying an `active_alarm` per currently-active alarm (the ConditionRefresh snapshot), then a single `snapshot_complete`, then a `transition` for every subsequent raise / acknowledge / clear — interleaved with a further `provider_status` on each failover/failback and a further `snapshot_status` on each change of the truncation verdict. It is served by the always-on `GatewayAlarmMonitor`, which owns a single gateway-managed worker session and fans out to every attached client — clients no longer open a session of their own. `alarm_filter_prefix`, when set, scopes the stream to a sub-tree. +A subscriber's opening snapshot is taken from the monitor's memoized alarm-set projection (the same one `QueryActiveAlarms` and the dashboard read), captured under the broadcast lock in the same critical section that registers the subscriber, so no transition can fall between the snapshot and the live stream; the prefix filter runs afterwards, outside the lock. Each subscriber has a bounded queue (2048 messages). A subscriber that falls that far behind is dropped and its stream ends with `ResourceExhausted` ("Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot."), the same status a `StreamEvents` overflow carries; the remedy is to reconnect, which re-sends the snapshot. The drop is not logged as a gateway failure. + ### `QueryActiveAlarms` `QueryActiveAlarms` is a server-streaming, **session-less** RPC that returns a point-in-time snapshot of the currently-active alarm set. Unlike `StreamAlarms`, it does not stay attached for live transitions — it writes the snapshot and completes, which clients use after a reconnect to reseed alarm state or to reconcile alarms that may have been missed during a transport blip. It is served from the always-on `GatewayAlarmMonitor` cache (`IGatewayAlarmService.CurrentAlarms`), so no worker session is opened. The handler does not run through `MxAccessGrpcRequestValidator`; it only null-checks the request inline, then streams one `ActiveAlarmSnapshot` per cached alarm, honouring cancellation between writes. When `alarm_filter_prefix` is non-empty, only alarms whose `alarm_full_reference` starts with that prefix (ordinal `StartsWith`) are emitted; an empty prefix returns the full set. Exceptions are translated by the same `MapException` path as the other handlers. The RPC requires the `events:read` scope (`GatewayScopes.EventsRead`), the same scope as `StreamEvents` and `StreamAlarms`. @@ -294,6 +300,8 @@ Every handler wraps its body in `try { ... } catch (Exception exception) when (e - `SessionManagerException` is mapped by its `ErrorCode`. - `WorkerClientException` is mapped by its `ErrorCode`. +Two stream-specific cases are handled before `MapException`: on `StreamEvents` and `StreamAlarms` a write failure after the client cancelled becomes `Cancelled` without a log entry, and on `StreamAlarms` an `AlarmFeedSubscriberOverflowException` (slow-subscriber drop) becomes `ResourceExhausted` without a log entry. + Anything else is logged at warning and surfaced as `Unavailable` with a generic message — clients see a retryable status, and the unexpected exception is captured in gateway logs rather than leaked over the wire. ```csharp diff --git a/gateway.md b/gateway.md index c5ae6b4..9a6a8ec 100644 --- a/gateway.md +++ b/gateway.md @@ -446,7 +446,9 @@ Rules: enforcement is ever added, this is the property it will rely on. - `correlation_id` links a command to its reply; it is authoritative on the envelope, and the inner `MxCommandReply.correlation_id` echoes it for - MXAccess parity. + MXAccess parity. The envelope id never reaches public clients: the gateway + overwrites the public `MxCommandReply.correlation_id` with the request's + `client_correlation_id` (empty when the client sent none). - Events carry their own correlation id or leave it empty. - Replies must preserve MXAccess HRESULT/status information even when the command is also represented as a protocol-level failure. diff --git a/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs b/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs index ed0acf1..d411a84 100644 --- a/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs +++ b/src/ZB.MOM.WW.MxGateway.IntegrationTests/WorkerLiveMxAccessSmokeTests.cs @@ -1704,6 +1704,20 @@ public sealed class WorkerLiveMxAccessSmokeTests(ITestOutputHelper output) /// public WriteOptions? WriteOptions { get; set; } + /// + /// Records the message unless the token is already cancelled. + /// + /// The message to write. + /// Token that cancels the write. + /// A task that represents the asynchronous operation. + public Task WriteAsync(T message, CancellationToken cancellationToken) + { + // The interface default throws NotSupportedException for a cancelable token; the + // gateway passes the call's token on every streamed write. + cancellationToken.ThrowIfCancellationRequested(); + return WriteAsync(message); + } + /// /// Records the message and signals any pending waiter. /// diff --git a/src/ZB.MOM.WW.MxGateway.Server/Alarms/AlarmFeedSubscriberOverflowException.cs b/src/ZB.MOM.WW.MxGateway.Server/Alarms/AlarmFeedSubscriberOverflowException.cs new file mode 100644 index 0000000..07c0952 --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Server/Alarms/AlarmFeedSubscriberOverflowException.cs @@ -0,0 +1,39 @@ +namespace ZB.MOM.WW.MxGateway.Server.Alarms; + +/// +/// Completes an alarm-feed subscriber that fell behind its bounded queue and was dropped. +/// +/// +/// A typed exception rather than a bare so the gRPC +/// layer can map it to ResourceExhausted — the status a StreamEvents overflow +/// carries — without treating a slow client as a gateway failure (an unrecognized exception is +/// logged as a warning with its stack and surfaced as Unavailable). The remedy is the +/// same as for an event-stream overflow: reconnect, which re-snapshots the alarm set. +/// +public sealed class AlarmFeedSubscriberOverflowException : Exception +{ + /// The message every dropped subscriber is completed with. + public const string DefaultMessage = + "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot."; + + /// Initializes a new instance of the class. + public AlarmFeedSubscriberOverflowException() + : base(DefaultMessage) + { + } + + /// Initializes a new instance of the class. + /// The error message. + public AlarmFeedSubscriberOverflowException(string message) + : base(message) + { + } + + /// Initializes a new instance of the class. + /// The error message. + /// The exception that caused this one. + public AlarmFeedSubscriberOverflowException(string message, Exception innerException) + : base(message, innerException) + { + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs b/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs index 76fb408..4a6be85 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Alarms/GatewayAlarmMonitor.cs @@ -967,7 +967,9 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic } // Caller holds _sync. Pushes a feed message to every matching subscriber; - // a subscriber that has fallen behind is completed with an error and dropped. + // a subscriber that has fallen behind is completed with an AlarmFeedSubscriberOverflowException + // and dropped. The typed fault is what lets StreamAlarms answer ResourceExhausted instead of + // treating a slow client as a gateway failure. private void Broadcast(AlarmFeedMessage message, string reference) { for (int index = _subscribers.Count - 1; index >= 0; index--) @@ -980,8 +982,7 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic if (!subscriber.Channel.Writer.TryWrite(message)) { - subscriber.Channel.Writer.TryComplete(new InvalidOperationException( - "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot.")); + subscriber.Channel.Writer.TryComplete(new AlarmFeedSubscriberOverflowException()); _subscribers.RemoveAt(index); } } @@ -997,8 +998,7 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic Subscriber subscriber = _subscribers[index]; if (!subscriber.Channel.Writer.TryWrite(message)) { - subscriber.Channel.Writer.TryComplete(new InvalidOperationException( - "Alarm feed subscriber fell behind and was dropped; reconnect to re-snapshot.")); + subscriber.Channel.Writer.TryComplete(new AlarmFeedSubscriberOverflowException()); _subscribers.RemoveAt(index); } } @@ -1046,9 +1046,14 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic _subscribers.Add(subscriber); providerStatus = BuildProviderStatus(); snapshotTruncated = _snapshotTruncated; - snapshot = _alarms.Values - .Where(alarm => prefix.Length == 0 - || alarm.AlarmFullReference.StartsWith(prefix, StringComparison.Ordinal)) + + // Take the memoized projection rather than cloning the whole set per attach: the + // clone runs at most once per alarm-set generation (shared with CurrentAlarms), so a + // burst of reconnecting subscribers no longer holds the broadcast lock for one full + // copy each. The array is never mutated once built — a write to _alarms replaces it + // with null — so capturing the reference here pins exactly the generation this + // subscriber registered against, which is what keeps snapshot-then-live ordering. + snapshot = _currentAlarmsProjection ??= _alarms.Values .Select(alarm => alarm.Clone()) .ToArray(); } @@ -1068,8 +1073,15 @@ public sealed class GatewayAlarmMonitor : BackgroundService, IGatewayAlarmServic SnapshotStatus = new AlarmSnapshotStatus { Truncated = snapshotTruncated }, }; + // The prefix filter runs here, outside _sync, over the immutable projection. foreach (ActiveAlarmSnapshot alarm in snapshot) { + if (prefix.Length != 0 + && !alarm.AlarmFullReference.StartsWith(prefix, StringComparison.Ordinal)) + { + continue; + } + yield return new AlarmFeedMessage { ActiveAlarm = alarm }; } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs index 96688a8..5ed29ea 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Grpc/MxAccessGatewayService.cs @@ -144,6 +144,12 @@ public sealed class MxAccessGatewayService( publicReply = bulkConstraintPlan.MergeDeniedInto(publicReply); } + // MxCommandReply.correlation_id always echoes the caller's client_correlation_id — + // the same value the constraint-denied replies carry — so a client can match every + // reply to its request. The worker reply carries the gateway's internal envelope id, + // which is a gateway↔worker routing detail and is overwritten here, empty included. + publicReply.CorrelationId = request.ClientCorrelationId; + session.TrackCommandReply(commandToInvoke, publicReply); return publicReply; } @@ -172,7 +178,10 @@ public sealed class MxAccessGatewayService( // volume gateway path. Stopwatch.GetTimestamp/GetElapsedTime measure the exact // same wall-clock span the former Stopwatch.StartNew()/.Elapsed did. long sendStartTimestamp = Stopwatch.GetTimestamp(); - await responseStream.WriteAsync(publicEvent).ConfigureAwait(false); + + // The call token releases a write parked on a slow or vanished client when the + // call is cancelled, instead of leaving it to the transport's own timeout. + await responseStream.WriteAsync(publicEvent, context.CancellationToken).ConfigureAwait(false); metrics.RecordEventStreamSend( publicEvent.Family.ToString(), Stopwatch.GetElapsedTime(sendStartTimestamp)); @@ -180,6 +189,11 @@ public sealed class MxAccessGatewayService( } catch (Exception exception) when (exception is not RpcException) { + if (IsClientAbort(exception, context)) + { + throw ClientCancelled(); + } + throw MapException(exception); } } @@ -225,11 +239,24 @@ public sealed class MxAccessGatewayService( .WithCancellation(context.CancellationToken) .ConfigureAwait(false)) { - await responseStream.WriteAsync(message).ConfigureAwait(false); + await responseStream.WriteAsync(message, context.CancellationToken).ConfigureAwait(false); } } + catch (AlarmFeedSubscriberOverflowException overflow) + { + // The monitor dropped this subscriber for falling behind its bounded queue. That is a + // client-side slowness condition with a client-side remedy (reconnect and + // re-snapshot), so it maps to ResourceExhausted like a StreamEvents overflow rather + // than reaching MapException's warning-with-stack + Unavailable fallback. + throw new RpcException(new Status(StatusCode.ResourceExhausted, overflow.Message)); + } catch (Exception exception) when (exception is not RpcException) { + if (IsClientAbort(exception, context)) + { + throw ClientCancelled(); + } + throw MapException(exception); } } @@ -264,6 +291,17 @@ public sealed class MxAccessGatewayService( } } + // A response write fails with an IOException (or an operation-cancelled exception) when the + // client disconnects mid-stream. With the call token already cancelled that is a routine + // client abort, not a gateway fault: it maps to Cancelled with no warning. The same exception + // on a live call token is a genuine transport failure and keeps the MapException path. + private static bool IsClientAbort(Exception exception, ServerCallContext context) => + context.CancellationToken.IsCancellationRequested + && exception is IOException or OperationCanceledException; + + private static RpcException ClientCancelled() => + new(new Status(StatusCode.Cancelled, "gRPC request was canceled.")); + private string? ResolveClientIdentity() { return identityAccessor.Current?.DisplayName ?? identityAccessor.Current?.KeyId; diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmTruncationSignalTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmTruncationSignalTests.cs index 6aa2177..2852da7 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmTruncationSignalTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Alarms/AlarmTruncationSignalTests.cs @@ -372,6 +372,105 @@ public sealed class AlarmTruncationSignalTests await monitor.StopAsync(CancellationToken.None); } + /// + /// A subscriber that stops reading is dropped once its bounded queue fills, and the fault it + /// observes is the typed — the signal the + /// gRPC layer maps to ResourceExhausted — rather than a bare InvalidOperationException. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task SubscriberThatStopsReading_IsDroppedWithTypedOverflowFault() + { + // Comfortably past the monitor's per-subscriber queue capacity (2048). + const int ProbeCount = 2300; + using GatewayMetrics metrics = new(); + StubSessionManager sessions = new(); + using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics); + + using CancellationTokenSource cts = new(); + await monitor.StartAsync(cts.Token); + await sessions.WaitForReconcileAsync(WaitTimeout); + + // The slow subscriber: registered (first frame read) and then never read again. + using CancellationTokenSource slowCts = new(WaitTimeout); + await using IAsyncEnumerator slow = + monitor.StreamAsync(null, slowCts.Token).GetAsyncEnumerator(slowCts.Token); + Assert.True(await slow.MoveNextAsync()); + + // Raises on distinct references: each one is broadcast (no dedup) without forcing a + // reconcile, which keeps the burst cheap under a loaded parallel test run. + for (ulong sequence = 1; sequence <= ProbeCount; sequence++) + { + sessions.EmitEvent(UniqueRaise(sequence)); + } + + // The monitor caches each raise and broadcasts it under the same lock, so once the cache + // holds every raise, every broadcast has been offered to the slow subscriber. (A draining + // reader is no barrier here: under a loaded test run it can fall 2048 frames behind the + // burst and be dropped itself.) + await WaitUntilAsync(() => monitor.CurrentAlarms.Count >= ProbeCount, WaitTimeout); + + // Drain what the slow subscriber had buffered; the terminal fault is the typed overflow. + await Assert.ThrowsAsync(async () => + { + while (await slow.MoveNextAsync()) + { + } + }); + + await cts.CancelAsync(); + await monitor.StopAsync(CancellationToken.None); + } + + /// + /// A StreamAsync attach reuses the memoized alarm-set projection that CurrentAlarms serves + /// (no per-attach clone of the whole set under the broadcast lock) and still applies the + /// subscriber's prefix filter to the snapshot it emits. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task SubscriberAttach_ReusesMemoizedProjectionAndAppliesPrefixFilter() + { + using GatewayMetrics metrics = new(); + StubSessionManager sessions = new() + { + Snapshots = + [ + NewSnapshot("Galaxy!Area.Tank01.Level.HiHi", fromTruncatedSnapshot: false), + NewSnapshot("Galaxy!Area.Pump02.Flow.Lo", fromTruncatedSnapshot: false), + ], + }; + using GatewayAlarmMonitor monitor = CreateMonitor(sessions, metrics); + + using CancellationTokenSource cts = new(); + await monitor.StartAsync(cts.Token); + await sessions.WaitForReconcileAsync(WaitTimeout); + await WaitUntilAsync(() => monitor.CurrentAlarms.Count == 2, WaitTimeout); + IReadOnlyList projection = monitor.CurrentAlarms; + + List streamed = []; + using CancellationTokenSource streamCts = new(WaitTimeout); + await foreach (AlarmFeedMessage message in monitor.StreamAsync("Galaxy!Area.Tank01", streamCts.Token)) + { + if (message.PayloadCase == AlarmFeedMessage.PayloadOneofCase.ActiveAlarm) + { + streamed.Add(message.ActiveAlarm); + } + + if (message.PayloadCase == AlarmFeedMessage.PayloadOneofCase.SnapshotComplete) + { + break; + } + } + + ActiveAlarmSnapshot only = Assert.Single(streamed); + Assert.Equal("Galaxy!Area.Tank01.Level.HiHi", only.AlarmFullReference); + Assert.Contains(projection, alarm => ReferenceEquals(alarm, only)); + + await cts.CancelAsync(); + await monitor.StopAsync(CancellationToken.None); + } + private static int CountOf(List received, AlarmFeedMessage.PayloadOneofCase payloadCase) { lock (received) @@ -442,6 +541,23 @@ public sealed class AlarmTruncationSignalTests }, }; + // A Raise on a reference unique to the sequence, so the monitor never dedups it. + private static MxEvent UniqueRaise(ulong sequence) => new() + { + Family = MxEventFamily.OnAlarmTransition, + WorkerSequence = sequence, + OnAlarmTransition = new OnAlarmTransitionEvent + { + AlarmFullReference = $"Galaxy!Area.Tank{sequence}.Level.Hi", + SourceObjectReference = $"Tank{sequence}.Level", + AlarmTypeName = "AnalogLimitAlarm.Hi", + TransitionKind = AlarmTransitionKind.Raise, + Severity = 500, + SourceProvider = AlarmProviderMode.Alarmmgr, + TransitionTimestamp = Google.Protobuf.WellKnownTypes.Timestamp.FromDateTimeOffset(DateTimeOffset.UtcNow), + }, + }; + private static ActiveAlarmSnapshot NewSnapshot(string reference, bool fromTruncatedSnapshot) { return new ActiveAlarmSnapshot diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs index 1b473aa..ce4cb9d 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Grpc/MxAccessGatewayServiceTests.cs @@ -2,9 +2,11 @@ using System.Diagnostics.Metrics; using System.Runtime.CompilerServices; using Google.Protobuf.WellKnownTypes; using Grpc.Core; +using Microsoft.Extensions.Logging; using Microsoft.Extensions.Logging.Abstractions; using ZB.MOM.WW.MxGateway.Contracts; using ZB.MOM.WW.MxGateway.Contracts.Proto; +using ZB.MOM.WW.MxGateway.Server.Alarms; using ZB.MOM.WW.MxGateway.Server.Grpc; using ZB.MOM.WW.MxGateway.Server.Metrics; using ZB.MOM.WW.MxGateway.Server.Security.Authentication; @@ -213,6 +215,8 @@ public sealed class MxAccessGatewayServiceTests Assert.Equal(42, reply.AddItem.ItemHandle); Assert.Equal("status detail", Assert.Single(reply.Statuses).DiagnosticText); Assert.Equal("mxaccess diagnostic", reply.DiagnosticMessage); + // The reply echoes the caller's client_correlation_id, not the internal worker envelope id. + Assert.Equal("client-correlation", reply.CorrelationId); } /// Verifies that StreamEvents writes only events after the specified worker sequence. @@ -463,11 +467,151 @@ public sealed class MxAccessGatewayServiceTests Assert.Contains("server-stream-active-alarms", reply.Capabilities); } + /// + /// Every reply echoes the request's client_correlation_id, including when the caller + /// sent none: the internal worker envelope id never reaches the client. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task Invoke_WithoutClientCorrelationId_DoesNotLeakWorkerEnvelopeId() + { + FakeSessionManager sessionManager = new() + { + InvokeReply = new WorkerCommandReply + { + Reply = new MxCommandReply + { + SessionId = "session-1", + CorrelationId = "4f1d1b0e-internal-envelope-guid", + Kind = MxCommandKind.AddItem, + ProtocolStatus = MxAccessGrpcMapper.Ok(), + AddItem = new AddItemReply { ItemHandle = 42 }, + }, + }, + }; + MxAccessGatewayService service = CreateService(sessionManager); + MxCommandRequest request = new() + { + SessionId = "session-1", + Command = new MxCommand + { + Kind = MxCommandKind.AddItem, + AddItem = new AddItemCommand { ServerHandle = 12, ItemDefinition = "Galaxy.Tag.Value" }, + }, + }; + + MxCommandReply reply = await service.Invoke(request, new TestServerCallContext()); + + Assert.Equal(string.Empty, reply.CorrelationId); + } + + /// + /// StreamEvents hands the call's cancellation token to every response write, so a write + /// blocked on a slow or vanished client is released when the call is cancelled. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEvents_PassesCallCancellationTokenToResponseWrites() + { + FakeSessionManager sessionManager = new(); + sessionManager.Events.Add(CreateWorkerEvent("session-1", workerSequence: 1)); + MxAccessGatewayService service = CreateService(sessionManager); + RecordingServerStreamWriter writer = new(); + using CancellationTokenSource callCts = new(); + + await service.StreamEvents( + new StreamEventsRequest { SessionId = "session-1" }, + writer, + new TestServerCallContext(cancellationToken: callCts.Token)); + + Assert.Single(writer.Messages); + Assert.Equal(callCts.Token, writer.LastWriteCancellationToken); + } + + /// + /// A response write that fails with an because the client went + /// away (the call token is cancelled) maps to Cancelled and logs no warning: a client + /// disconnect is routine, not a gateway failure. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEvents_WhenWriteFailsAfterClientCancel_ReturnsCancelledWithoutWarning() + { + FakeSessionManager sessionManager = new(); + sessionManager.Events.Add(CreateWorkerEvent("session-1", workerSequence: 1)); + RecordingLogger logger = new(); + MxAccessGatewayService service = CreateService(sessionManager, logger: logger); + using CancellationTokenSource callCts = new(); + ThrowingStreamWriter writer = new(() => + { + callCts.Cancel(); + return new IOException("The client reset the request stream."); + }); + + RpcException exception = await Assert.ThrowsAsync( + () => service.StreamEvents( + new StreamEventsRequest { SessionId = "session-1" }, + writer, + new TestServerCallContext(cancellationToken: callCts.Token))); + + Assert.Equal(StatusCode.Cancelled, exception.StatusCode); + Assert.DoesNotContain(logger.Entries, entry => entry.Level >= LogLevel.Warning); + } + + /// + /// The same IOException with a live call token is a genuine transport failure and keeps the + /// existing Unavailable + warning mapping. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamEvents_WhenWriteFailsWithoutCancel_KeepsUnavailableMapping() + { + FakeSessionManager sessionManager = new(); + sessionManager.Events.Add(CreateWorkerEvent("session-1", workerSequence: 1)); + RecordingLogger logger = new(); + MxAccessGatewayService service = CreateService(sessionManager, logger: logger); + ThrowingStreamWriter writer = new(() => new IOException("Transport failure.")); + + RpcException exception = await Assert.ThrowsAsync( + () => service.StreamEvents( + new StreamEventsRequest { SessionId = "session-1" }, + writer, + new TestServerCallContext())); + + Assert.Equal(StatusCode.Unavailable, exception.StatusCode); + Assert.Contains(logger.Entries, entry => entry.Level == LogLevel.Warning); + } + + /// + /// A StreamAlarms subscriber the monitor drops for falling behind receives ResourceExhausted + /// (the same status a StreamEvents overflow carries), and the gateway logs no + /// warning-with-stack for what is a client-side slowness condition. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task StreamAlarms_WhenSubscriberOverflows_ReturnsResourceExhaustedWithoutWarning() + { + RecordingLogger logger = new(); + FakeGatewayAlarmService alarms = new() { StreamFault = new AlarmFeedSubscriberOverflowException() }; + MxAccessGatewayService service = CreateService(new FakeSessionManager(), alarmService: alarms, logger: logger); + + RpcException exception = await Assert.ThrowsAsync( + () => service.StreamAlarms( + new StreamAlarmsRequest(), + new RecordingServerStreamWriter(), + new TestServerCallContext())); + + Assert.Equal(StatusCode.ResourceExhausted, exception.StatusCode); + Assert.DoesNotContain(logger.Entries, entry => entry.Level >= LogLevel.Warning); + } + private static MxAccessGatewayService CreateService( FakeSessionManager sessionManager, IGatewayRequestIdentityAccessor? identityAccessor = null, GatewayMetrics? metrics = null, - IConstraintEnforcer? constraintEnforcer = null) + IConstraintEnforcer? constraintEnforcer = null, + IGatewayAlarmService? alarmService = null, + ILogger? logger = null) { return new MxAccessGatewayService( sessionManager, @@ -477,8 +621,8 @@ public sealed class MxAccessGatewayServiceTests new MxAccessGrpcMapper(), new FakeEventStreamService(sessionManager), metrics ?? new GatewayMetrics(), - NullLogger.Instance, - new FakeGatewayAlarmService()); + logger ?? NullLogger.Instance, + alarmService ?? new FakeGatewayAlarmService()); } private static ApiKeyIdentity CreateIdentity() @@ -786,4 +930,18 @@ public sealed class MxAccessGatewayServiceTests } } + + /// A stream writer whose token-bearing write throws the supplied exception. + /// The streamed message type. + private sealed class ThrowingStreamWriter(Func createException) : IServerStreamWriter + { + /// + public WriteOptions? WriteOptions { get; set; } + + /// + public Task WriteAsync(T message) => throw createException(); + + /// + public Task WriteAsync(T message, CancellationToken cancellationToken) => throw createException(); + } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeGatewayAlarmService.cs b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeGatewayAlarmService.cs index 369b672..38f873d 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeGatewayAlarmService.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/FakeGatewayAlarmService.cs @@ -40,8 +40,18 @@ public sealed class FakeGatewayAlarmService : IGatewayAlarmService yield return new AlarmFeedMessage { SnapshotComplete = true }; await Task.CompletedTask.ConfigureAwait(false); + if (StreamFault is not null) + { + throw StreamFault; + } } + /// + /// Gets or sets an exception the feed throws after its snapshot, standing in for the monitor + /// completing a subscriber with a fault (for example a slow-subscriber overflow). + /// + public Exception? StreamFault { get; set; } + /// public Task AcknowledgeAsync( AcknowledgeAlarmRequest request, diff --git a/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/RecordingLogger.cs b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/RecordingLogger.cs new file mode 100644 index 0000000..847f71d --- /dev/null +++ b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/RecordingLogger.cs @@ -0,0 +1,48 @@ +using Microsoft.Extensions.Logging; + +namespace ZB.MOM.WW.MxGateway.Tests.TestSupport; + +/// +/// that records every entry's level and formatted message, +/// so a test can assert that a path logs (or does not log) at a given level. +/// +/// The logger category type. +public sealed class RecordingLogger : ILogger +{ + private readonly object _syncRoot = new(); + private readonly List<(LogLevel Level, string Message, Exception? Exception)> _entries = []; + + /// Gets a snapshot of the recorded entries, in order. + public IReadOnlyList<(LogLevel Level, string Message, Exception? Exception)> Entries + { + get + { + lock (_syncRoot) + { + return _entries.ToArray(); + } + } + } + + /// + public IDisposable? BeginScope(TState state) + where TState : notnull => null; + + /// + public bool IsEnabled(LogLevel logLevel) => true; + + /// + public void Log( + LogLevel logLevel, + EventId eventId, + TState state, + Exception? exception, + Func formatter) + { + ArgumentNullException.ThrowIfNull(formatter); + lock (_syncRoot) + { + _entries.Add((logLevel, formatter(state, exception), exception)); + } + } +} diff --git a/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/RecordingServerStreamWriter.cs b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/RecordingServerStreamWriter.cs index 06539b7..c359c38 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/RecordingServerStreamWriter.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/TestSupport/RecordingServerStreamWriter.cs @@ -29,6 +29,25 @@ public sealed class RecordingServerStreamWriter : IServerStreamWriter /// Gets or sets options for writing messages to the stream. public WriteOptions? WriteOptions { get; set; } + /// Gets the cancellation token passed to the most recent token-bearing write. + public CancellationToken LastWriteCancellationToken { get; private set; } + + /// Records the supplied message unless the token is already cancelled. + /// The message to record. + /// Token that cancels the write. + /// A completed task. + /// + /// The interface default throws for a cancelable token, + /// and the gateway passes the call's token on every streamed write, so a test writer must + /// override this overload the way the ASP.NET Core writer does. + /// + public Task WriteAsync(T message, CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + LastWriteCancellationToken = cancellationToken; + return WriteAsync(message); + } + /// Records the supplied message. /// The message to record. /// A completed task.