fix(communication): streaming endpoint lookups are generation-fenced and cannot strand their in-flight guards (review 2026-10-21 03/R-2 follow-up)

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);
     }

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 creation inside the try: any fault must still report back, or the in-flight
-            cts.CancelAfter(EndpointResolveTimeout);
+            // 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);
     }

src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamEndpoints.cs

@@ -46,4 +46,8 @@ public delegate Task<SiteStreamEndpoints?> SiteStreamEndpointResolver(Cancellati
 /// <summary>Internal actor message: the result of a <see cref="SiteStreamEndpointResolver"/> lookup before a reopen.</summary>
 /// <param name="Endpoints">The resolved pair, or null when unavailable.</param>
 /// <param name="Error">The lookup failure, when it threw or timed out.</param>
-internal sealed record SiteStreamEndpointsResolved(SiteStreamEndpoints? Endpoints, Exception? Error);
+/// <param name="Generation">
+/// The consumer's stream generation when the lookup started; if a stream was opened since, the
+/// result must not open another one.
+/// </param>
+internal sealed record SiteStreamEndpointsResolved(SiteStreamEndpoints? Endpoints, Exception? Error, int Generation);

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));
     }
+
+    /// <summary>Verifies that an endpoint lookup result from before a newer open does not open a second stream.</summary>
+    [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"));
+    }
 }