diff --git a/docs/requirements/Component-Communication.md b/docs/requirements/Component-Communication.md index 2bf52562..f22bd0e0 100644 --- a/docs/requirements/Component-Communication.md +++ b/docs/requirements/Component-Communication.md @@ -70,6 +70,7 @@ Both central and site clusters. Each side has communication actors that handle m - **Backpressure is lossy-by-design and now observable:** the per-session `Channel` (bounded `GrpcInstanceStreamChannelCapacity`, default 1000, `DropOldest`) silently evicts the oldest event when a slow consumer falls behind. The eviction is counted via the channel's `itemDropped` callback, logged at Warning (first eviction, then every 500th) and exported as `scadabridge.site.stream.events_dropped` (tagged `stream=instance`), so real event loss on a debug stream is visible instead of silent. The **site-wide alarm feed has its own, larger channel** — see §6.1. - **Bridge-session hardening (WP2.3):** - The pre-snapshot buffer is **bounded (20 000 events, drop-oldest)** and its evictions counted (`scadabridge.central.debug_view.presnapshot_dropped`). It was previously unbounded, so a session whose snapshot never arrived grew without limit on the CENTRAL node. Dropping the oldest is correct here: the snapshot that ends the buffering phase is authoritative for anything that old. + - An **`InstanceNotFound` snapshot** (instance not deployed on the site) stops the bridge actor, and `DebugStreamService.StartStreamAsync` returns the flagged snapshot **without registering a session** — so a consumer that never calls `StopStream` (the SignalR hub only forwards the flag) leaves no registry row for the keepalive timer to Tell a dead actor. - A **hard snapshot deadline** (`DebugStreamBridgeActor.SnapshotTimeout`, 60 s) fails the session if no `DebugViewSnapshot` arrives. Nothing else ended a session wedged in the buffering phase — a lost site reply raises no gRPC error. - **The orphan net measures the CONSUMER, not the mailbox.** A session self-terminates after `DebugStreamBridgeActor.ConsumerIdleTimeout` (5 min) without a sign of life from its consumer, checked by a periodic self-tick (`ConsumerLivenessCheckInterval`, 30 s) against a consumer-last-seen stamp. The stamp is renewed only by `DebugStreamConsumerAlive`, which `DebugStreamService` Tells to every session still in its registry on a shared 30 s timer — holding a session there IS what "a consumer is attached" means, since both consumers (the Blazor debug view and the SignalR hub) release it on dispose/disconnect. This replaced an Akka `SetReceiveTimeout(5 min)`. That version measured the MAILBOX, which conflates site chatter with consumer liveness: a busy site kept an abandoned session alive forever. Marking stream events `INotInfluenceReceiveTimeout` fixed that but left the timeout with nothing recurring to reset it — the snapshot lands once and the stability tick once — so **every healthy session self-terminated ~6 min in and the operator was told "Site disconnected"**. Both failure directions are pinned by tests: streaming events with keepalives survive many windows; an orphaned session terminates while events keep arriving. diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs index bade9cea..02a01cd8 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/DebugStreamService.cs @@ -207,6 +207,19 @@ public class DebugStreamService : IDisposable ex); } + if (snapshot.InstanceNotFound) + { + // The bridge actor already stopped itself on the not-found snapshot. Do not keep a + // live-looking session for it (review 2026-10-21 03/M-3): a consumer that never + // calls StopStream (the SignalR hub just forwards the flag) left the row in the + // registry until disconnect, and every keepalive round Telling a dead actor. + _sessions.TryRemove(sessionId, out _); + _logger.LogInformation( + "Debug stream for {Instance} on site {Site} not started: instance is not deployed there", + instanceUniqueName, siteIdentifier); + return new DebugStreamSession(sessionId, snapshot); + } + _logger.LogInformation("Debug stream {SessionId} started for {Instance} on site {Site}", sessionId, instanceUniqueName, siteIdentifier); @@ -241,6 +254,9 @@ public class DebugStreamService : IDisposable } } + /// Test/diagnostic accessor: the number of registered (attached) sessions. + internal int SessionCount => _sessions.Count; + /// /// Sends one keepalive round to every attached session. Exposed (internal) so tests can /// drive the keepalive deterministically instead of waiting on the timer. diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs index a6975a16..14b31bab 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/DebugStreamServiceTests.cs @@ -169,6 +169,61 @@ public class DebugStreamServiceTests : TestKit } } + /// Verifies that an InstanceNotFound snapshot does not leave a registered session behind (review 2026-10-21 03/M-3). + /// A task that represents the asynchronous operation. + [Fact] + public async Task StartStreamAsync_InstanceNotFoundSnapshot_DoesNotRegisterASession() + { + var instance = new Instance("Site1.Pump01") { Id = 7, SiteId = 3 }; + var site = new Site("Site One", "site-1") + { + Id = 3, + GrpcNodeAAddress = "http://localhost:5100", + GrpcNodeBAddress = "http://localhost:5200" + }; + + var instanceRepo = Substitute.For(); + instanceRepo.GetInstanceByIdAsync(7, Arg.Any()).Returns(instance); + var siteRepo = Substitute.For(); + siteRepo.GetSiteByIdAsync(3, Arg.Any()).Returns(site); + + var services = new ServiceCollection(); + services.AddScoped(_ => instanceRepo); + services.AddScoped(_ => siteRepo); + using var provider = services.BuildServiceProvider(); + + var commProbe = CreateTestProbe(); + var commService = new CommunicationService( + Options.Create(new CommunicationOptions()), + NullLogger.Instance); + commService.SetCommunicationActor(commProbe.Ref); + + using var grpcFactory = new Grpc.MockSiteStreamGrpcClientFactory( + new Grpc.MockSiteStreamGrpcClient()); + using var service = new DebugStreamService( + commService, provider, grpcFactory, NullLogger.Instance); + service.SetActorSystem(Sys); + + var startTask = service.StartStreamAsync(instanceId: 7, onEvent: _ => { }, onTerminated: () => { }); + commProbe.ExpectMsg(TimeSpan.FromSeconds(5)); + var bridgeActor = commProbe.LastSender; + Watch(bridgeActor); + + bridgeActor.Tell(new ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView.DebugViewSnapshot( + "Site1.Pump01", + new List(), + new List(), + DateTimeOffset.UtcNow, + InstanceNotFound: true)); + var session = await startTask; + + // The caller still sees the flag, but the bridge actor has stopped itself — the + // service must not keep a live-looking session for it (keepalives to a dead actor). + Assert.True(session.InitialSnapshot.InstanceNotFound); + Assert.Equal(0, service.SessionCount); + ExpectTerminated(bridgeActor, TimeSpan.FromSeconds(3)); + } + // --- single-endpoint sites --- /// Verifies that a single-endpoint site uses its sole address for both node slots.