fix(communication): DebugStreamService drops the session on an InstanceNotFound snapshot (review 2026-10-21 03/M-3)
This commit is contained in:
@@ -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<SiteStreamEvent>` (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.
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Test/diagnostic accessor: the number of registered (attached) sessions.</summary>
|
||||
internal int SessionCount => _sessions.Count;
|
||||
|
||||
/// <summary>
|
||||
/// Sends one keepalive round to every attached session. Exposed (internal) so tests can
|
||||
/// drive the keepalive deterministically instead of waiting on the timer.
|
||||
|
||||
@@ -169,6 +169,61 @@ public class DebugStreamServiceTests : TestKit
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Verifies that an InstanceNotFound snapshot does not leave a registered session behind (review 2026-10-21 03/M-3).</summary>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
[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<ITemplateEngineRepository>();
|
||||
instanceRepo.GetInstanceByIdAsync(7, Arg.Any<CancellationToken>()).Returns(instance);
|
||||
var siteRepo = Substitute.For<ISiteRepository>();
|
||||
siteRepo.GetSiteByIdAsync(3, Arg.Any<CancellationToken>()).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<CommunicationService>.Instance);
|
||||
commService.SetCommunicationActor(commProbe.Ref);
|
||||
|
||||
using var grpcFactory = new Grpc.MockSiteStreamGrpcClientFactory(
|
||||
new Grpc.MockSiteStreamGrpcClient());
|
||||
using var service = new DebugStreamService(
|
||||
commService, provider, grpcFactory, NullLogger<DebugStreamService>.Instance);
|
||||
service.SetActorSystem(Sys);
|
||||
|
||||
var startTask = service.StartStreamAsync(instanceId: 7, onEvent: _ => { }, onTerminated: () => { });
|
||||
commProbe.ExpectMsg<SiteEnvelope>(TimeSpan.FromSeconds(5));
|
||||
var bridgeActor = commProbe.LastSender;
|
||||
Watch(bridgeActor);
|
||||
|
||||
bridgeActor.Tell(new ZB.MOM.WW.ScadaBridge.Commons.Messages.DebugView.DebugViewSnapshot(
|
||||
"Site1.Pump01",
|
||||
new List<ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming.AttributeValueChanged>(),
|
||||
new List<ZB.MOM.WW.ScadaBridge.Commons.Messages.Streaming.AlarmStateChanged>(),
|
||||
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 ---
|
||||
|
||||
/// <summary>Verifies that a single-endpoint site uses its sole address for both node slots.</summary>
|
||||
|
||||
Reference in New Issue
Block a user