From 82976b7377581f1dee3acfd5ff90709ee67a5c37 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 22 Oct 2026 08:30:14 -0400 Subject: [PATCH] fix(communication): streaming endpoint lookups are generation-fenced and cannot strand their in-flight guards (review 2026-10-21 03/R-2 follow-up) --- .../Actors/DebugStreamBridgeActor.cs | 16 +++++++++++++--- .../Actors/SiteAlarmAggregatorActor.cs | 18 ++++++++++++++---- .../Grpc/SiteStreamEndpoints.cs | 6 +++++- .../Grpc/SiteAlarmAggregatorActorTests.cs | 15 +++++++++++++++ 4 files changed, 47 insertions(+), 8 deletions(-) diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs index c4ced0d8..ebc5cc32 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/DebugStreamBridgeActor.cs @@ -374,6 +374,8 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers _resolveInFlight = false; if (_stopped) return; ApplyResolvedEndpoints(msg); + // A stream was opened while the lookup ran (generation moved on): don't open a second. + if (msg.Generation != _streamGeneration) return; OpenGrpcStream(); }); @@ -649,16 +651,24 @@ public class DebugStreamBridgeActor : ReceiveActor, IWithTimers _resolveInFlight = true; var resolver = _endpointResolver; + var generation = _streamGeneration; Task.Run(async () => { - using var cts = new CancellationTokenSource(EndpointResolveTimeout); + // CTS creation inside the try: any fault must still report back, or the in-flight + // guards would never release. + CancellationTokenSource? cts = null; try { - return new SiteStreamEndpointsResolved(await resolver(cts.Token).ConfigureAwait(false), null); + cts = new CancellationTokenSource(EndpointResolveTimeout); + return new SiteStreamEndpointsResolved(await resolver(cts.Token).ConfigureAwait(false), null, generation); } catch (Exception ex) { - return new SiteStreamEndpointsResolved(null, ex); + return new SiteStreamEndpointsResolved(null, ex, generation); + } + finally + { + cts?.Dispose(); } }).PipeTo(Self); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs index a8007097..138e6852 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs @@ -479,6 +479,8 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _resolveInFlight = false; if (_stopped) return; ApplyResolvedEndpoints(msg); + // A stream was opened while the lookup ran (generation moved on): don't open a second. + if (msg.Generation != _streamGeneration) return; OpenGrpcStream(); }); @@ -933,17 +935,25 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _openInFlight = true; var resolver = _endpointResolver; var lifetime = _lifetimeCts?.Token ?? CancellationToken.None; + var generation = _streamGeneration; Task.Run(async () => { - using var cts = CancellationTokenSource.CreateLinkedTokenSource(lifetime); - cts.CancelAfter(EndpointResolveTimeout); + // CTS creation inside the try: any fault must still report back, or the in-flight + // guards would never release. + CancellationTokenSource? cts = null; try { - return new SiteStreamEndpointsResolved(await resolver(cts.Token).ConfigureAwait(false), null); + cts = CancellationTokenSource.CreateLinkedTokenSource(lifetime); + cts.CancelAfter(EndpointResolveTimeout); + return new SiteStreamEndpointsResolved(await resolver(cts.Token).ConfigureAwait(false), null, generation); } catch (Exception ex) { - return new SiteStreamEndpointsResolved(null, ex); + return new SiteStreamEndpointsResolved(null, ex, generation); + } + finally + { + cts?.Dispose(); } }).PipeTo(Self); } diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamEndpoints.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamEndpoints.cs index 115698d7..d57ebf48 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamEndpoints.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamEndpoints.cs @@ -46,4 +46,8 @@ public delegate Task SiteStreamEndpointResolver(Cancellati /// Internal actor message: the result of a lookup before a reopen. /// The resolved pair, or null when unavailable. /// The lookup failure, when it threw or timed out. -internal sealed record SiteStreamEndpointsResolved(SiteStreamEndpoints? Endpoints, Exception? Error); +/// +/// The consumer's stream generation when the lookup started; if a stream was opened since, the +/// result must not open another one. +/// +internal sealed record SiteStreamEndpointsResolved(SiteStreamEndpoints? Endpoints, Exception? Error, int Generation); 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 ea81176c..d6b743ef 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs @@ -1070,4 +1070,19 @@ public class SiteAlarmAggregatorActorTests : TestKit AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(3)); } + + /// Verifies that an endpoint lookup result from before a newer open does not open a second stream. + [Fact] + public void StaleEndpointLookupResult_DoesNotOpenASecondStream() + { + var (actor, seed, _, factory) = CreateActor(); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + + // A lookup that started at an older stream generation completes after a stream was opened. + actor.Tell(new SiteStreamEndpointsResolved(new SiteStreamEndpoints(GrpcNodeA, GrpcNodeB), null, Generation: -1)); + + AssertHoldsFor(TimeSpan.FromMilliseconds(300), + (() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, "a stale lookup must not open another stream")); + } }