fix(comm): stop the duplicate-stream replacement disposing another handler's CTS

Root-caused from a full-suite-load-only failure of
GrpcStreamIntegrationTests.Pipeline_DuplicateCorrelationId_ReplacesStream:
ObjectDisposedException escaping SubscribeInstance.

SubscribeInstance's duplicate-prevention path cancelled AND disposed the
replaced stream's CancellationTokenSource. That CTS belongs to the replaced
handler's own `using var streamCts`, which is still running and still has to
read `streamCts.Token` — the Dispose raced that read. The race is pre-existing
and independent of R2: the same Dispose sat against the same first-token-read
when the handler still used `ReadAllAsync(streamCts.Token)`; it only ever
loses under enough scheduling pressure to land the replacement inside the
first stream's setup window.

Cancel only. The owning handler's `using` still disposes it exactly once on
every exit path, and cancellation is all replacement ever needed. The Cancel
is wrapped for the converse race (owner already finished and disposed),
mirroring CancelAllStreams().

Regression test makes the race deterministic by gating the first stream inside
its setup window — the _activeStreams entry is registered before Subscribe is
called, so the replacement always lands before the first stream reads its
token. Verified fail-before (ObjectDisposedException) / pass-after by
reinstating the Dispose as a negative control.

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

@@ -353,11 +353,29 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase
                 StatusCode.InvalidArgument, "correlation_id is missing or not a valid identifier"));
         }
 
-        // Duplicate prevention -- cancel existing stream for this correlationId
+        // Duplicate prevention -- cancel existing stream for this correlationId.
+        //
+        // CANCEL ONLY, never Dispose. The replaced stream's CTS belongs to its own
+        // handler's `using var streamCts`, which is still running and still reads
+        // `streamCts.Token` (at the pump call below, and previously at the
+        // `ReadAllAsync(streamCts.Token)` it replaced). Disposing it from here raced that
+        // read and surfaced as an unhandled ObjectDisposedException escaping the RPC —
+        // observed as a full-suite-load-only failure of
+        // GrpcStreamIntegrationTests.Pipeline_DuplicateCorrelationId_ReplacesStream, and a
+        // pre-existing hazard (the same Dispose + the same first-token-read relationship
+        // exist unchanged before R2). Cancellation alone is what replacement needs; the
+        // owning handler's `using` still disposes it exactly once on every exit path.
         if (_activeStreams.TryRemove(correlationId, out var existingEntry))
         {
-            existingEntry.Cts.Cancel();
-            existingEntry.Cts.Dispose();
+            try
+            {
+                existingEntry.Cts.Cancel();
+            }
+            catch (ObjectDisposedException)
+            {
+                // Its owner finished and disposed it between the TryRemove and here —
+                // already terminal, nothing to cancel. Mirrors CancelAllStreams().
+            }
         }
 
         // Check max concurrent streams after duplicate removal.

tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs

@@ -581,6 +581,67 @@ public class SiteStreamGrpcServerTests : TestKit
         Assert.Equal(0, server.DroppedStreamEventCount);
     }
 
+    [Fact]
+    public async Task DuplicateReplacement_CancelsTheReplacedStream_WithoutDisposingItsCts()
+    {
+        // Regression: the duplicate-replacement path used to Cancel AND Dispose the
+        // replaced stream's CancellationTokenSource. That CTS belongs to the replaced
+        // handler's own `using var streamCts`, which is still running and still has to
+        // read `streamCts.Token` — so the Dispose raced that read and escaped the RPC as
+        // an unhandled ObjectDisposedException. It surfaced only under full-suite load
+        // (GrpcStreamIntegrationTests.Pipeline_DuplicateCorrelationId_ReplacesStream) and
+        // predates R2: the same Dispose and the same first-token-read relationship existed
+        // when the handler still used `ReadAllAsync(streamCts.Token)`.
+        //
+        // The race is made DETERMINISTIC here by gating the first stream inside its setup
+        // window (its _activeStreams entry is registered before Subscribe is called), so
+        // the replacement always lands before the first stream reads its token.
+        using var gate = new ManualResetEventSlim(false);
+        var calls = 0;
+        var subscriber = Substitute.For<ISiteStreamSubscriber>();
+        subscriber.Subscribe(Arg.Any<string>(), Arg.Any<IActorRef>())
+            .Returns(ci =>
+            {
+                var n = Interlocked.Increment(ref calls);
+                if (n == 1)
+                    gate.Wait(TimeSpan.FromSeconds(15));
+                return $"sub-dup-race-{n}";
+            });
+
+        var server = new SiteStreamGrpcServer(subscriber, _logger);
+        server.SetReady(Sys);
+
+        using var cts1 = new CancellationTokenSource();
+        var stream1 = Task.Run(() => server.SubscribeInstance(
+            MakeRequest("corr-dup-race"),
+            Substitute.For<IServerStreamWriter<SiteStreamEvent>>(),
+            CreateMockContext(cts1.Token)));
+
+        await WaitForConditionAsync(() => server.ActiveStreamCount == 1);
+        await WaitForConditionAsync(() => Volatile.Read(ref calls) == 1);
+
+        using var cts2 = new CancellationTokenSource();
+        var stream2 = Task.Run(() => server.SubscribeInstance(
+            MakeRequest("corr-dup-race"),
+            Substitute.For<IServerStreamWriter<SiteStreamEvent>>(),
+            CreateMockContext(cts2.Token)));
+
+        // The replacement has taken the slot (and cancelled stream 1's CTS) by the time
+        // its own Subscribe has been called.
+        await WaitForConditionAsync(() => Volatile.Read(ref calls) == 2);
+
+        gate.Set();
+
+        // Pre-fix this threw ObjectDisposedException out of the RPC. Post-fix the replaced
+        // stream observes a plain cancellation and unwinds through its normal finally.
+        await stream1;
+
+        cts2.Cancel();
+        await stream2;
+
+        Assert.Equal(0, server.ActiveStreamCount);
+    }
+
     // ── R2: gRPC event batching, and its negotiation ────────────────────────────
 
     [Fact]