diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index ed8dd036..cee3f326 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -343,6 +343,7 @@ Disconnect is detected at the **transport layer**, never via an application-leve - **In-flight command/control messages (gRPC call + deadline)**: When a connection drops while a request is in flight (e.g., a deployment sent but no response received), the gRPC call fails or hits its deadline and the caller receives a failure. There is **no automatic retry or buffering at central** — the engineer sees the failure in the UI and re-initiates the action. This is consistent with the design principle that central does not buffer messages. An in-progress deployment whose round-trip exceeds the timeout (default 120 s at `CommunicationService.DeployInstanceAsync`) surfaces as `DeploymentStatus.Failed` to the caller. - **Debug streams (gRPC)**: Any gRPC stream interruption is detected by the HTTP/2 keepalive PING (~25 s) and triggers reconnection logic in the `DebugStreamBridgeActor`. The bridge actor attempts to reconnect to the other site node endpoint (NodeB if NodeA failed, or vice versa), with up to 3 retries and 5-second backoff. If all retries fail, the consumer is notified via `OnStreamTerminated` and the bridge actor is stopped. Events during the reconnection gap are lost (acceptable for real-time debug view). On successful reconnection, the consumer can request a fresh snapshot to re-sync state. +- **Graceful end of stream is NOT a fault, but it IS a reconnect trigger.** When the site's `GrpcMaxStreamLifetime` (4 h) elapses — and on a graceful site shutdown — the server ends the RPC with status **OK**, so the client's read loop simply finishes with no exception. `SiteStreamGrpcClient` reports that through a dedicated `onCompleted` callback, distinct from `onError`, and both consuming actors treat it as a reconnect trigger that **does not spend (or refund) the error-retry budget** and does **not** flip the node — the peer that closed cleanly is healthy. `SiteAlarmAggregatorActor` marks the stream down and lets the periodic reconcile tick reopen it (which also re-seeds); `DebugStreamBridgeActor` reopens on the same endpoint after `ReconnectDelay`. Both stamp the stream generation on the callback so a completion racing out of an already-replaced stream is ignored, and both observe the subscription `Task` itself so a fault escaping the subscribe call can never go unobserved. Without the `onCompleted` leg an OK completion was invisible to the reconnect logic and the stream stayed silently dead until the central process restarted. ## Failover Behavior diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs index 7496b053..e0b26082 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs @@ -63,6 +63,14 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers private bool _stopped; private CancellationTokenSource? _grpcCts; + /// + /// Monotonic stream generation stamped on each opened gRPC stream and echoed back on its + /// error/completion callbacks: a late callback raced out of a previous (cancelled) stream + /// carries a stale generation and is ignored, so it can neither burn retry budget nor + /// open a duplicate stream. Mirrors SiteAlarmAggregatorActor. Actor-thread only. + /// + private int _streamGeneration; + /// /// Phase flag. until the initial /// has been delivered and the pre-snapshot buffer @@ -198,10 +206,33 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers // gRPC stream error — attempt reconnection Receive(msg => { + // Ignore a late error raced out of a previous (cancelled) stream: it must not + // burn retry budget or flip the node a second time. + if (msg.Generation != _streamGeneration) + { + _log.Debug("Ignoring stale gRPC error from stream generation {0} (current {1})", + msg.Generation, _streamGeneration); + return; + } _log.Warning("gRPC stream error for {0}: {1}", _instanceUniqueName, msg.Exception.Message); HandleGrpcError(); }); + // gRPC stream ended GRACEFULLY (server status OK) — the site's 4h max stream + // lifetime elapsing or a graceful site shutdown. Not a fault: reopen on the SAME + // node without spending retry budget. Without this the session went silently deaf. + Receive(msg => + { + if (_stopped) return; + if (msg.Generation != _streamGeneration) + { + _log.Debug("Ignoring stale gRPC completion from stream generation {0} (current {1})", + msg.Generation, _streamGeneration); + return; + } + HandleGrpcCompleted(); + }); + // Scheduled reconnection Receive(_ => OpenGrpcStream()); @@ -424,20 +455,57 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers // retry budget is recovered. Cancelled by HandleGrpcError. Timers.StartSingleTimer(StabilityTimerKey, new GrpcStreamStable(), StabilityWindow); + var generation = ++_streamGeneration; var client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint); var self = Self; var ct = _grpcCts.Token; - // Launch as background task — onEvent and onError marshal back to actor via Tell + // Launch as background task — the callbacks marshal back to the actor via Tell. + // The task itself is observed below: a fault escaping SubscribeAsync would otherwise + // leave the session waiting on a stream that does not exist, exception unobserved. Task.Run(async () => { await client.SubscribeAsync( _correlationId, _instanceUniqueName, evt => self.Tell(evt), - ex => self.Tell(new GrpcStreamError(ex)), + ex => self.Tell(new GrpcStreamError(ex, generation)), + () => self.Tell(new GrpcStreamCompleted(generation)), ct); - }, ct); + }, ct).ContinueWith(t => + { + if (t.IsFaulted) + self.Tell(new GrpcStreamError(t.Exception!.GetBaseException(), generation)); + else if (t.IsCanceled && !ct.IsCancellationRequested) + self.Tell(new GrpcStreamCompleted(generation)); + // RanToCompletion: SubscribeAsync already reported its own outcome. + }, TaskContinuationOptions.ExecuteSynchronously); + } + + /// + /// Handles a graceful end of stream (server status OK). The stream simply expired or the + /// site shut down cleanly, so the retry budget is left untouched and the endpoint is not + /// flipped; the reopen is scheduled through the existing reconnect timer, which also + /// rate-limits a pathological site that keeps closing streams immediately. + /// + private void HandleGrpcCompleted() + { + // The stream is gone, so its armed stability timer must not later "recover" a + // budget that its successor has since spent. + Timers.Cancel(StabilityTimerKey); + + _log.Info("gRPC stream for {0} completed gracefully (server end of stream); reopening", + _instanceUniqueName); + + // Release the site-side relay for the finished stream before reopening, so the site + // is not left with a zombie relay actor for this correlation id. + _grpcCts?.Cancel(); + _grpcCts?.Dispose(); + _grpcCts = null; + _grpcFactory.TryGet(_siteIdentifier, _useNodeA ? _grpcNodeAAddress : _grpcNodeBAddress) + ?.Unsubscribe(_correlationId); + + Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectGrpcStream(), ReconnectDelay); } private void HandleGrpcError() @@ -512,9 +580,17 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers public record StopDebugStream; /// -/// Internal message indicating a gRPC stream error occurred. +/// Internal message indicating a gRPC stream error occurred, stamped with the stream +/// generation it came from so a late error out of a cancelled stream can be ignored. /// -internal record GrpcStreamError(Exception Exception); +internal record GrpcStreamError(Exception Exception, int Generation); + +/// +/// Internal message indicating the gRPC stream ended gracefully (server status OK — the +/// site's max stream lifetime elapsed, or the site shut down cleanly), stamped with its +/// stream generation. +/// +internal record GrpcStreamCompleted(int Generation); /// /// Internal message to trigger gRPC stream reconnection. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs index a320479a..94fecd5e 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs @@ -78,12 +78,23 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers private bool _stopped; /// - /// True once the live gRPC stream has been given up (retry budget exhausted). Reconcile - /// snapshots keep serving in the meantime; the next reconcile tick self-heals the stream - /// by resetting the retry budget and reopening it, so a sustained site outage does not - /// permanently drop the live feed. Actor-thread only. + /// True while there is no live gRPC stream — either it was given up (retry budget + /// exhausted) or it ended gracefully (server status OK at the site's max stream + /// lifetime). Reconcile snapshots keep serving in the meantime; the next reconcile tick + /// self-heals the stream by resetting the retry budget and reopening it, so neither a + /// sustained site outage nor a routine 4h stream expiry permanently drops the live feed. + /// Actor-thread only. /// private bool _streamDown; + + /// + /// Why the stream is down: true = the retry budget was exhausted, so the + /// self-healing reopen must also reset it; false = it ended gracefully, and the + /// budget — which a completion neither spends nor refunds — is carried across the reopen + /// untouched (a stream flapping between faults and clean closes must still trip + /// MaxRetries). Actor-thread only. + /// + private bool _retryBudgetExhausted; private CancellationTokenSource? _grpcCts; private CancellationTokenSource? _lifetimeCts; @@ -218,6 +229,22 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers HandleGrpcError(); }); + // gRPC stream ended GRACEFULLY (server status OK) — the site's 4h max stream + // lifetime elapsing or a graceful site shutdown. Not a fault: the stream is marked + // down so the reconcile tick reopens it (which also re-seeds), but the retry budget + // is untouched and the node is not flipped. Same generation fence as the error path. + Receive(msg => + { + if (_stopped) return; + if (msg.Generation != _streamGeneration) + { + _log.Debug("Ignoring stale gRPC completion from stream generation {0} (current {1})", + msg.Generation, _streamGeneration); + return; + } + HandleGrpcCompleted(); + }); + Receive(_ => OpenGrpcStream()); // Owning service asks us to stop (last viewer left + linger elapsed). @@ -279,7 +306,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers if (_streamDown) { _log.Info("Site-alarm gRPC stream for {0} was down; reopening on reconcile tick", _siteIdentifier); - _retryCount = 0; + if (_retryBudgetExhausted) + { + _retryBudgetExhausted = false; + _retryCount = 0; + } // Telemetry: a reconcile-driven reopen after the stream was given up is a reconnect. ScadaBridgeTelemetry.RecordLiveAlarmStreamReconnect(); OpenGrpcStream(); @@ -514,14 +545,26 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers var self = Self; var ct = _grpcCts.Token; + // The subscription task itself is observed (below): a fault escaping + // SubscribeSiteAsync — or a Task.Run that never started because ct was already + // cancelled — would otherwise leave the actor waiting on a stream that does not + // exist, with the exception silently unobserved. Task.Run(async () => { await client.SubscribeSiteAsync( _correlationId, alarm => self.Tell(alarm), ex => self.Tell(new GrpcAlarmStreamError(ex, generation)), + () => self.Tell(new GrpcAlarmStreamCompleted(generation)), ct); - }, ct); + }, ct).ContinueWith(t => + { + if (t.IsFaulted) + self.Tell(new GrpcAlarmStreamError(t.Exception!.GetBaseException(), generation)); + else if (t.IsCanceled && !ct.IsCancellationRequested) + self.Tell(new GrpcAlarmStreamCompleted(generation)); + // RanToCompletion: SubscribeSiteAsync already reported its own outcome. + }, TaskContinuationOptions.ExecuteSynchronously); } private void HandleGrpcError() @@ -543,6 +586,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers "reconcile snapshots continue and the next reconcile tick will retry the stream", _siteIdentifier, MaxRetries); _streamDown = true; + _retryBudgetExhausted = true; CleanupGrpc(); return; } @@ -569,6 +613,26 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectAlarmStream(), _reconnectDelay); } + /// + /// Handles a graceful end of stream. The stream is torn down and left down for the + /// reconcile tick to reopen — deliberately NOT reopened inline, so a site that keeps + /// completing streams immediately can never spin this actor into a hot reconnect loop. + /// The retry budget is neither spent nor reset here (completion is not a fault), and the + /// endpoint is not flipped: the node that just closed a stream cleanly is healthy. + /// + private void HandleGrpcCompleted() + { + // The stream is gone, so its armed stability timer must not later "recover" a + // budget that its successor has since spent. + Timers.Cancel(StabilityTimerKey); + + _log.Info("Site-alarm gRPC stream for {0} completed gracefully (server end of stream); " + + "reopening on the next reconcile tick", _siteIdentifier); + + _streamDown = true; + CleanupGrpc(); + } + private void CleanupGrpc() { _grpcCts?.Cancel(); @@ -619,6 +683,11 @@ internal sealed record PublishCoalesced; /// generation it came from so a late error from a cancelled stream can be ignored (N7.2). internal sealed record GrpcAlarmStreamError(Exception Exception, int Generation); +/// Internal: the site-alarm gRPC stream ended gracefully (server status OK — max +/// stream lifetime or site shutdown), stamped with its stream generation so a late completion +/// from a cancelled stream can be ignored. +internal sealed record GrpcAlarmStreamCompleted(int Generation); + /// Internal: reconnect the site-alarm gRPC stream (flip node). internal sealed record ReconnectAlarmStream; diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs index 52584b40..5763c6c4 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcClient.cs @@ -167,13 +167,19 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable /// /// Opens a server-streaming subscription for a specific instance. /// This is a long-running async method; the caller launches it as a background task. - /// The callback delivers domain events, and - /// lets the caller handle reconnection. + /// The callback delivers domain events, + /// lets the caller handle reconnection after a fault, and + /// reports a graceful end of stream. /// /// Unique identifier for this subscription. /// Unique name of the instance to subscribe to. /// Callback invoked for each domain event received from the stream. /// Callback invoked when the subscription encounters an error. + /// + /// Callback invoked when the server ended the RPC with OK — the site's max stream + /// lifetime elapsing or a graceful site shutdown. Mutually exclusive with + /// ; see . + /// /// Cancellation token to stop the subscription. /// A task that represents the asynchronous operation. public virtual async Task SubscribeAsync( @@ -181,6 +187,7 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable string instanceUniqueName, Action onEvent, Action onError, + Action onCompleted, CancellationToken ct) { if (_client is null) @@ -195,30 +202,18 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable InstanceUniqueName = instanceUniqueName }; - try - { - using var call = _client.SubscribeInstance(request, cancellationToken: cts.Token); - - await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token)) + await ConsumeStreamAsync( + correlationId, + cts, + () => _client.SubscribeInstance(request, cancellationToken: cts.Token), + evt => { var domainEvent = ConvertToDomainEvent(evt); if (domainEvent != null) onEvent(domainEvent); - } - } - catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled) - { - // Normal cancellation — not an error - } - catch (Exception ex) - { - onError(ex); - } - finally - { - // Remove only our own entry -- a racing reconnect may already own the slot. - RemoveSubscription(correlationId, cts); - } + }, + onError, + onCompleted); } /// @@ -244,12 +239,18 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable /// Unique identifier for this subscription. /// Callback invoked for each alarm delta received from the site-wide stream. /// Callback invoked when the subscription encounters an error. + /// + /// Callback invoked when the server ended the RPC with OK — the site's max stream + /// lifetime elapsing or a graceful site shutdown. Mutually exclusive with + /// ; see . + /// /// Cancellation token to stop the subscription. /// A task that represents the asynchronous operation. public virtual async Task SubscribeSiteAsync( string correlationId, Action onAlarmEvent, Action onError, + Action onCompleted, CancellationToken ct) { if (_client is null) @@ -263,21 +264,73 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable CorrelationId = correlationId }; - try - { - using var call = _client.SubscribeSite(request, cancellationToken: cts.Token); - - await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token)) + await ConsumeStreamAsync( + correlationId, + cts, + () => _client.SubscribeSite(request, cancellationToken: cts.Token), + evt => { // Site-wide stream is alarm-only by contract; defensively ignore anything else. if (ConvertToAlarmEvent(evt) is { } alarm) onAlarmEvent(alarm); + }, + onError, + onCompleted); + } + + /// + /// Drains a server-streaming call to its end, classifying the outcome into exactly one + /// of three terminations: a fault (), a graceful server-side + /// end of stream (), or our own cancellation (neither). + /// + /// The graceful case is the one that matters operationally: the site server caps every + /// stream at CommunicationOptions.GrpcMaxStreamLifetime (4 h) and, when that + /// elapses, ends the RPC with status OK. The client-side await foreach then simply + /// finishes, so before onCompleted existed the consuming actor observed nothing at + /// all and the stream stayed silently dead until the process restarted. + /// + /// Internal so the completion/fault classification can be unit-tested with a fake + /// and no live channel. + /// + /// Unique identifier for this subscription. + /// The subscription's linked token source (owned by the caller). + /// + /// Opens the server-streaming call. Invoked inside the guarded region so a throw while + /// opening is reported through like any other stream fault. + /// + /// Invoked per wire event. + /// Invoked once if the stream faulted. + /// Invoked once if the server ended the stream with OK. + /// A task that completes when the stream has ended and its outcome been reported. + internal async Task ConsumeStreamAsync( + string correlationId, + CancellationTokenSource cts, + Func> openCall, + Action onEvent, + Action onError, + Action onCompleted) + { + var completedGracefully = false; + try + { + using (var call = openCall()) + { + await foreach (var evt in call.ResponseStream.ReadAllAsync(cts.Token)) + { + onEvent(evt); + } } + + completedGracefully = true; } catch (RpcException ex) when (ex.StatusCode == StatusCode.Cancelled) { // Normal cancellation — not an error } + catch (OperationCanceledException) when (cts.IsCancellationRequested) + { + // Our own Unsubscribe/reconnect cancelled the read — not an error either. + } catch (Exception ex) { onError(ex); @@ -287,6 +340,12 @@ public class SiteStreamGrpcClient : IAsyncDisposable, IDisposable // Remove only our own entry -- a racing reconnect may already own the slot. RemoveSubscription(correlationId, cts); } + + // Raised outside the try so a throwing callback is not misreported as a stream + // fault. Suppressed when our own token was cancelled as the stream ended: that is + // a teardown, and the caller has already moved on to a newer stream. + if (completedGracefully && !cts.IsCancellationRequested) + onCompleted(); } /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs index 7f79e3b7..c0832f16 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/DebugStreamBridgeActorTests.cs @@ -400,6 +400,67 @@ public class DebugStreamBridgeActorTests : TestKit Assert.Equal("corr-1", factory.ClientFor(GrpcNodeB).SubscribeCalls[0].CorrelationId); } + // ── WP1.1: graceful (status OK) stream completion is a reconnect trigger ── + + [Fact] + public void On_GracefulStreamCompletion_Reopens_On_The_Same_Node() + { + // The site caps every stream at GrpcMaxStreamLifetime and then ends the RPC with + // OK. Pre-fix the client's read loop just finished, the actor was told nothing, and + // the debug session went silently deaf for the rest of its life. + var (_, factory) = CreateBridgeWithTrackingFactory(); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).SubscribeCalls.Count == 1, + TimeSpan.FromSeconds(3)); + + factory.ClientFor(GrpcNodeA).SubscribeCalls[0].OnCompleted(); + + // Reopened on the SAME node — a clean close is not a fault, so there is nothing to + // fail over from — and the finished stream is released, not left zombie. + AwaitCondition(() => factory.ClientFor(GrpcNodeA).SubscribeCalls.Count == 2, + TimeSpan.FromSeconds(5)); + Assert.Empty(factory.ClientFor(GrpcNodeB).SubscribeCalls); + Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).UnsubscribedCorrelationIds); + } + + [Fact] + public void RepeatedGracefulCompletions_DoNotConsume_TheErrorRetryBudget() + { + // Five completions — two more than MaxRetries. Had completion been routed through + // the error path the session would have terminated on the fourth. + var ctx = CreateBridgeActor(); + ctx.CommProbe.ExpectMsg(); + AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3)); + + for (var i = 1; i <= 5; i++) + { + ctx.MockGrpcClient.SubscribeCalls[i - 1].OnCompleted(); + var expected = i + 1; + AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == expected, + TimeSpan.FromSeconds(5)); + } + + Assert.False(ctx.TerminatedFlag[0]); + } + + [Fact] + public void LateCompletionFromAPreviousStreamGeneration_IsIgnored() + { + var (_, factory) = CreateBridgeWithTrackingFactory(); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).SubscribeCalls.Count == 1, + TimeSpan.FromSeconds(3)); + + var firstSub = factory.ClientFor(GrpcNodeA).SubscribeCalls[0]; + firstSub.OnError(new Exception("NodeA down")); // gen 1 dies → gen 2 opens on NodeB + AwaitCondition(() => factory.ClientFor(GrpcNodeB).SubscribeCalls.Count == 1, + TimeSpan.FromSeconds(5)); + + // A completion racing out of the dead gen-1 stream must not tear down the live one. + firstSub.OnCompleted(); + Thread.Sleep(300); + Assert.DoesNotContain("corr-1", factory.ClientFor(GrpcNodeB).UnsubscribedCorrelationIds); + Assert.Single(factory.ClientFor(GrpcNodeB).SubscribeCalls); + } + // ── Task 6 (arch review 02, High): teardown/failover unsubscribe is endpoint-safe ── // Both paths use TryGet, never GetOrCreate, so cleanup can never open a fresh // channel or (with (site,endpoint) keying) touch another session's channel. @@ -894,9 +955,11 @@ internal class MockSiteStreamGrpcClient : SiteStreamGrpcClient string instanceUniqueName, Action onEvent, Action onError, + Action onCompleted, CancellationToken ct) { - var subscription = new MockSubscription(correlationId, instanceUniqueName, onEvent, onError, ct); + var subscription = new MockSubscription( + correlationId, instanceUniqueName, onEvent, onError, onCompleted, ct); lock (_lock) { _subscribeCalls.Add(subscription); } // Return a task that completes when cancelled (simulates long-running stream) @@ -916,6 +979,7 @@ internal record MockSubscription( string InstanceUniqueName, Action OnEvent, Action OnError, + Action OnCompleted, CancellationToken CancellationToken); /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs index 35c09662..98aa76a5 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs @@ -93,7 +93,8 @@ public class SiteAlarmAggregatorActorTests : TestKit } private sealed record SiteSub( - string CorrelationId, Action OnAlarm, Action OnError, CancellationToken Ct); + string CorrelationId, Action OnAlarm, Action OnError, + Action OnCompleted, CancellationToken Ct); private sealed class MockSiteAlarmStreamClient : SiteStreamGrpcClient { @@ -107,9 +108,10 @@ public class SiteAlarmAggregatorActorTests : TestKit public MockSiteAlarmStreamClient() : base() { } public override Task SubscribeSiteAsync( - string correlationId, Action onAlarmEvent, Action onError, CancellationToken ct) + string correlationId, Action onAlarmEvent, Action onError, + Action onCompleted, CancellationToken ct) { - lock (_lock) { _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, ct)); } + lock (_lock) { _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, onCompleted, ct)); } var tcs = new TaskCompletionSource(); ct.Register(() => tcs.TrySetResult()); return tcs.Task; // never completes until cancelled (simulates a live stream) @@ -445,6 +447,82 @@ public class SiteAlarmAggregatorActorTests : TestKit Assert.Equal(2, TotalSubs()); } + // ── WP1.1: graceful (status OK) stream completion is a reconnect trigger ── + + [Fact] + public void GracefulStreamCompletion_ReopensOnReconcileTick_OnTheSameNode() + { + // The site ends every stream with OK at its 4h max lifetime. Pre-fix the client's + // read loop just finished, nothing was told to the actor, and the site's alarm feed + // stayed silently dead until the central node restarted. + var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMilliseconds(300)); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + + factory.ClientFor(GrpcNodeA).Subs[0].OnCompleted(); + + // Reopened by the reconcile tick, on the SAME node — a clean close is not a fault, + // so there is nothing to fail over from. + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(3)); + Assert.Empty(factory.ClientFor(GrpcNodeB).Subs); + // The finished stream was released, so the site keeps no zombie relay actor. + Assert.Contains("corr-1", factory.ClientFor(GrpcNodeA).Unsubscribed); + } + + [Fact] + public void GracefulStreamCompletion_NeitherSpendsNorRefunds_TheErrorRetryBudget() + { + // Reconcile is far away; reopens are driven explicitly so the budget can be observed. + var (actor, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10)); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + + // Spend the whole error budget (MaxRetries = 3), each error flipping the node. + factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("1")); + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); + factory.ClientFor(GrpcNodeB).Subs[0].OnError(new Exception("2")); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5)); + factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3")); + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5)); + + // A graceful completion reopens without flipping the node (budget not spent) … + factory.ClientFor(GrpcNodeB).Subs[1].OnCompleted(); + actor.Tell(new RunReconcile()); + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 3, TimeSpan.FromSeconds(5)); + Assert.Equal(2, factory.ClientFor(GrpcNodeA).Subs.Count); + + int TotalSubs() => factory.ClientFor(GrpcNodeA).Subs.Count + factory.ClientFor(GrpcNodeB).Subs.Count; + var before = TotalSubs(); + + // … and without refunding it either: the next error is the 4th, so it exceeds + // MaxRetries and the stream is given up rather than reconnected. + factory.ClientFor(GrpcNodeB).Subs[2].OnError(new Exception("4")); + Thread.Sleep(400); + Assert.Equal(before, TotalSubs()); + } + + [Fact] + public void LateCompletionFromAPreviousStreamGeneration_IsIgnored() + { + var (_, seed, _, factory) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10)); + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + + var firstSub = factory.ClientFor(GrpcNodeA).Subs.Single(); + firstSub.OnError(new Exception("real fault")); // gen 1 dies → gen 2 opens on NodeB + AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5)); + + // A completion racing out of the dead gen-1 stream must not tear down the live + // gen-2 stream (which would go deltaless until the next reconcile tick). + firstSub.OnCompleted(); + Thread.Sleep(300); + Assert.DoesNotContain("corr-1", factory.ClientFor(GrpcNodeB).Unsubscribed); + Assert.Single(factory.ClientFor(GrpcNodeB).Subs); + } + // ── R2 T10: live-delta publish coalescing (N6) ── [Fact] diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs index 34b8563b..cb21d41b 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcClientTests.cs @@ -1,4 +1,5 @@ using Google.Protobuf.WellKnownTypes; +using Grpc.Core; using ZB.MOM.WW.ScadaBridge.Communication.Grpc; using ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming; using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums; @@ -277,7 +278,120 @@ public class SiteStreamGrpcClientTests var client = SiteStreamGrpcClient.CreateForTesting(); await Assert.ThrowsAsync(() => - client.SubscribeSiteAsync("corr", _ => { }, _ => { }, CancellationToken.None)); + client.SubscribeSiteAsync("corr", _ => { }, _ => { }, () => { }, CancellationToken.None)); + } + + // --- WP1.1: graceful (status OK) stream completion is reported, not swallowed --- + + [Fact] + public async Task ConsumeStream_ServerEndsStreamWithOk_InvokesOnCompleted_NotOnError() + { + // The site server caps every stream at GrpcMaxStreamLifetime (4h) and then ends the + // RPC with OK. That surfaces here as a read loop that simply runs out of events. + var client = SiteStreamGrpcClient.CreateForTesting(); + var cts = new CancellationTokenSource(); + var events = new List(); + Exception? error = null; + var completed = 0; + + await client.ConsumeStreamAsync( + "corr-ok", + cts, + () => FakeCall(new StubStreamReader(new SiteStreamEvent { CorrelationId = "corr-ok" })), + events.Add, + ex => error = ex, + () => completed++); + + Assert.Single(events); + Assert.Null(error); + Assert.Equal(1, completed); + } + + [Fact] + public async Task ConsumeStream_StreamFaults_InvokesOnError_NotOnCompleted() + { + var client = SiteStreamGrpcClient.CreateForTesting(); + var cts = new CancellationTokenSource(); + Exception? error = null; + var completed = 0; + + await client.ConsumeStreamAsync( + "corr-fault", + cts, + () => FakeCall(new StubStreamReader( + new RpcException(new Status(StatusCode.Unavailable, "site gone")))), + _ => { }, + ex => error = ex, + () => completed++); + + Assert.IsType(error); + Assert.Equal(0, completed); + } + + [Fact] + public async Task ConsumeStream_OwnCancellation_InvokesNeitherCallback() + { + // Our own Unsubscribe/reconnect is a teardown, not a fault and not a graceful end: + // the caller has already moved on to a newer stream. + var client = SiteStreamGrpcClient.CreateForTesting(); + var cts = new CancellationTokenSource(); + await cts.CancelAsync(); + Exception? error = null; + var completed = 0; + + await client.ConsumeStreamAsync( + "corr-cancel", + cts, + () => FakeCall(new StubStreamReader()), + _ => { }, + ex => error = ex, + () => completed++); + + Assert.Null(error); + Assert.Equal(0, completed); + } + + private static AsyncServerStreamingCall FakeCall(StubStreamReader reader) => + new(reader, + Task.FromResult(new Metadata()), + () => Status.DefaultSuccess, + () => new Metadata(), + () => { }); + + /// + /// Server stream stand-in: yields the queued events, then either ends the stream (the + /// status-OK completion the site's max-lifetime cap produces) or throws. + /// + private sealed class StubStreamReader : IAsyncStreamReader + { + private readonly Queue _events; + private readonly Exception? _fault; + + public StubStreamReader(params SiteStreamEvent[] events) + { + _events = new Queue(events); + } + + public StubStreamReader(Exception fault) + : this() + { + _fault = fault; + } + + public SiteStreamEvent Current { get; private set; } = null!; + + public Task MoveNext(CancellationToken cancellationToken) + { + cancellationToken.ThrowIfCancellationRequested(); + + if (_events.Count > 0) + { + Current = _events.Dequeue(); + return Task.FromResult(true); + } + + return _fault is null ? Task.FromResult(false) : Task.FromException(_fault); + } } // --- Communication-003 regression tests --- diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs index 533e203c..e386f760 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/SiteAlarmLiveCacheServiceTests.cs @@ -29,7 +29,7 @@ public class SiteAlarmLiveCacheServiceTests : TestKit public HangingClient() : base() { } public override Task SubscribeSiteAsync( string correlationId, Action onAlarmEvent, - Action onError, CancellationToken ct) + Action onError, Action onCompleted, CancellationToken ct) { var tcs = new TaskCompletionSource(); ct.Register(() => tcs.TrySetResult()); diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs index 4d51bba3..53f47c57 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs @@ -279,7 +279,8 @@ public class SiteAlarmStreamEndToEndTests : TestKit private sealed class NoopSiteStreamClient : SiteStreamGrpcClient { public override Task SubscribeSiteAsync( - string correlationId, Action onAlarmEvent, Action onError, CancellationToken ct) + string correlationId, Action onAlarmEvent, Action onError, + Action onCompleted, CancellationToken ct) { var tcs = new TaskCompletionSource(); ct.Register(() => tcs.TrySetResult());