diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs index e2e0b439..ea2e201c 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Actors/SiteAlarmAggregatorActor.cs @@ -65,6 +65,15 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers private const string ReconcileTimerKey = "alarm-reconcile"; private const string PublishTimerKey = "alarm-publish-coalesce"; private const string SeedRetryTimerKey = "alarm-seed-retry"; + private const string OpenDeadlineTimerKey = "alarm-grpc-open-deadline"; + + /// + /// Production default for the stream open-attempt deadline (arch-review 2026-08-19 P3-4). + /// Comfortably above SiteStreamGrpcClient.ConnectedHeaderTimeout (10 s) so a + /// normally-slow connect is never pre-empted, and comfortably below the 60 s reconcile + /// interval so the tick that follows always finds the flag released. + /// + internal static readonly TimeSpan DefaultOpenAttemptTimeout = TimeSpan.FromSeconds(30); /// True while a coalesced publish is armed (dirty deltas awaiting one tick). Actor-thread only. private bool _publishPending; @@ -179,10 +188,38 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers /// /// True between issuing a stream open and its first outcome (connected / error / /// completed), so a reconcile tick never stacks a second open on an in-flight one. - /// Actor-thread only. + /// Bounded by — see there for why an unbounded flag + /// was a starvation hazard. Actor-thread only. /// private bool _openInFlight; + /// + /// Deadline on ONE stream open attempt, after which is + /// released so can reopen (arch-review 2026-08-19 P3-4). + /// + /// The three outcomes that clear the flag — connected, error, graceful completion — are + /// all events the peer has to produce. A stream that produces NONE of them pins the flag: + /// the client gives up on response headers after + /// SiteStreamGrpcClient.ConnectedHeaderTimeout (10 s) and then falls back to the + /// first received event as its connected signal, so an alarm-only site that is quiet + /// supplies neither. In that state _streamDown is still true (nothing ever + /// reported connected) but _openInFlight is also true, and the tick's reopen guard + /// — _streamDown && !_openInFlight — is skipped until the site's 4 h max + /// stream lifetime finally ends the call. The deadline makes the flag self-healing + /// instead of relying on that analysis staying true. + /// + /// + /// Firing it is deliberately CHEAP: it releases the flag and nothing else. The retry + /// budget is untouched (this is not a fault), the node is not flipped, liveness is not + /// republished (_streamDown was never cleared), and the abandoned subscription is + /// left running — if it was merely slow and connects late, GrpcAlarmStreamConnected + /// still matches the current generation and brings it up normally; if a reconcile tick + /// reopens first, cancels its token source and fences its + /// generation out. + /// + /// + private readonly TimeSpan _openAttemptTimeout; + /// Consecutive whole-fan-out failures, driving the seed-retry backoff. Actor-thread only. private int _consecutiveSeedFailures; @@ -246,6 +283,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers /// N per-site aggregators on one central node do not fan out in lockstep (production 0.2 = /// up to +20%). Zero disables jitter, which is what tests want for determinism. /// + /// + /// Deadline on one gRPC stream open attempt (P3-4); null takes + /// (30 s) and or less + /// disables the deadline. See . + /// public SiteAlarmAggregatorActor( string siteIdentifier, string correlationId, @@ -258,7 +300,8 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers TimeSpan publishCoalesce, TimeSpan reconnectDelay, TimeSpan stabilityWindow, - double reconcileJitterFraction = 0.0) + double reconcileJitterFraction = 0.0, + TimeSpan? openAttemptTimeout = null) { _siteIdentifier = siteIdentifier; _correlationId = correlationId; @@ -272,6 +315,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _reconnectDelay = reconnectDelay; _stabilityWindow = stabilityWindow; _reconcileJitterFraction = Math.Clamp(reconcileJitterFraction, 0.0, 1.0); + _openAttemptTimeout = openAttemptTimeout ?? DefaultOpenAttemptTimeout; // Live delta from the site-wide alarm stream (marshalled in via Self.Tell). // A received delta must NOT reset the retry budget (a flapping stream that @@ -306,6 +350,23 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers OnStreamConnected(); }); + // The open attempt produced none of its three outcomes inside the deadline (P3-4): + // release the in-flight flag so the reconcile tick's reopen guard can run again. + // Generation-fenced like every other stream callback, and a no-op once the attempt + // has already resolved (a late timer racing its own stream's connect). + Receive(msg => + { + if (_stopped) return; + if (msg.Generation != _streamGeneration) return; + if (!_openInFlight) return; + + _openInFlight = false; + _log.Warning( + "Site-alarm gRPC stream open for {0} did not report connected, error or completion within {1}; " + + "releasing the in-flight guard so the next reconcile tick can reopen", + _siteIdentifier, _openAttemptTimeout); + }); + // Coalesced-publish tick: one publish for a batch of dirtying deltas (N6). Receive(_ => { @@ -799,6 +860,15 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers _openInFlight = true; var generation = ++_streamGeneration; + + // Bound this attempt (P3-4). Single-shot on a shared key, so a re-open replaces the + // previous attempt's deadline rather than stacking one. + if (_openAttemptTimeout > TimeSpan.Zero) + { + Timers.StartSingleTimer( + OpenDeadlineTimerKey, new GrpcAlarmStreamOpenTimedOut(generation), _openAttemptTimeout); + } + var client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint); var self = Self; var ct = _grpcCts.Token; @@ -834,6 +904,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers /// private void OnStreamConnected() { + Timers.Cancel(OpenDeadlineTimerKey); _openInFlight = false; var wasDown = _streamDown; @@ -858,6 +929,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // Stream failed before the stability window — retry budget NOT recovered. Timers.Cancel(StabilityTimerKey); + Timers.Cancel(OpenDeadlineTimerKey); _openInFlight = false; var wasLive = !_streamDown; @@ -920,6 +992,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers // The stream is gone, so its armed stability timer must not later "recover" a // budget that its successor has since spent. Timers.Cancel(StabilityTimerKey); + Timers.Cancel(OpenDeadlineTimerKey); _log.Info("Site-alarm gRPC stream for {0} completed gracefully (server end of stream); " + "reopening on the next reconcile tick", _siteIdentifier); @@ -1005,3 +1078,8 @@ internal sealed record ReconnectAlarmStream; /// Internal: the current site-alarm gRPC stream has stayed up long enough to recover the retry budget. internal sealed record GrpcAlarmStreamStable; + +/// Internal: a stream OPEN attempt produced none of connected/error/completed inside its +/// deadline (P3-4), stamped with the generation it was armed for so a late timer from a superseded +/// attempt is ignored. Releases the in-flight guard only — it is not a fault. +internal sealed record GrpcAlarmStreamOpenTimedOut(int Generation); diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs index 9a257585..dd15e604 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/Grpc/SiteStreamGrpcServer.cs @@ -353,8 +353,35 @@ 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 -- REPLACE the existing stream for this correlationId. // + // The removal is claimed FIRST, but nothing is cancelled until the concurrency cap + // has had its say, and a REPLACEMENT never consults the cap at all (arch-review + // 2026-08-19 P3-5). Both halves matter: + // + // * A replacement is net-neutral on the stream count (one entry out, one back in), + // so refusing it can only ever destroy capacity, never protect it. The old code + // checked `Count >= max` after the removal, which is correct at exactly the cap + // but NOT above it — and above it is reachable, because the check below is + // deliberately check-then-act (see the note there). In that window a routine + // reconnect cancelled a healthy incumbent and was then refused with + // ResourceExhausted, leaving the correlation with ZERO streams until its next + // retry. Excluding the replacement from the cap removes that outcome entirely. + // * A genuinely NEW subscriber that the cap refuses must leave the existing streams + // untouched, which is automatic here: TryRemove found nothing, so nothing was + // cancelled before the throw. + // + // Newest-wins is preserved exactly: a fresh subscribe on a live correlation id + // always displaces the incumbent, whether that incumbent is healthy or wedged. + var replacingExistingStream = _activeStreams.TryRemove(correlationId, out var existingEntry); + + // Max concurrent streams, applied only to a genuinely new correlation id. + // Deliberately check-then-act: two concurrent subscribes at the boundary can + // both pass, overshooting the cap by at most (concurrent subscribers - 1). + // Bounded and benign — not worth a lock on this path. + if (!replacingExistingStream && _activeStreams.Count >= _maxConcurrentStreams) + throw new RpcException(new GrpcStatus(StatusCode.ResourceExhausted, "Max concurrent streams reached")); + // 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 @@ -365,11 +392,11 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase // 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)) + if (replacingExistingStream) { try { - existingEntry.Cts.Cancel(); + existingEntry!.Cts.Cancel(); } catch (ObjectDisposedException) { @@ -378,13 +405,6 @@ public class SiteStreamGrpcServer : SiteStreamService.SiteStreamServiceBase } } - // Check max concurrent streams after duplicate removal. - // Deliberately check-then-act: two concurrent subscribes at the boundary can - // both pass, overshooting the cap by at most (concurrent subscribers - 1). - // Bounded and benign — not worth a lock on this path. - if (_activeStreams.Count >= _maxConcurrentStreams) - throw new RpcException(new GrpcStatus(StatusCode.ResourceExhausted, "Max concurrent streams reached")); - using var streamCts = CancellationTokenSource.CreateLinkedTokenSource(context.CancellationToken); // Session timeout (design doc "gRPC Connection Keepalive": 4-hour third layer // of dead-client detection) — forces a long-lived zombie stream to terminate diff --git a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs index 49d55bf6..f920d558 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Communication/SiteAlarmLiveCacheService.cs @@ -245,7 +245,11 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache _options.LiveAlarmCachePublishCoalesce, TimeSpan.FromSeconds(5), // reconnect delay — former ReconnectDelay static default TimeSpan.FromSeconds(60), // stability window — former StabilityWindow static default - _options.LiveAlarmCacheReconcileJitterFraction)); + _options.LiveAlarmCacheReconcileJitterFraction, + // Open-attempt deadline (P3-4). Passed explicitly rather than defaulted: + // Props.Create takes an expression tree, which may not omit an optional + // argument (CS0854). + SiteAlarmAggregatorActor.DefaultOpenAttemptTimeout)); entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}"); 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 0c36cadd..867ba87f 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteAlarmAggregatorActorTests.cs @@ -166,7 +166,8 @@ public class SiteAlarmAggregatorActorTests : TestKit } private (IActorRef Actor, SeedStub Seed, PublishSink Sink, MockSiteAlarmStreamClientFactory Factory) CreateActor( - TimeSpan? reconcileInterval = null, TimeSpan? publishCoalesce = null, bool autoConnect = true) + TimeSpan? reconcileInterval = null, TimeSpan? publishCoalesce = null, bool autoConnect = true, + TimeSpan? openAttemptTimeout = null) { var seed = new SeedStub(); var sink = new PublishSink(); @@ -178,7 +179,11 @@ public class SiteAlarmAggregatorActorTests : TestKit publishCoalesce ?? TimeSpan.Zero, TestReconnectDelay, TestStabilityWindow, - 0.0)); // no reconcile jitter in tests — determinism + 0.0, // no reconcile jitter in tests — determinism + // Production default (30 s) unless a test overrides it: every other test in this + // class finishes far inside it, so the open-attempt deadline stays dormant and + // their behaviour is unchanged. + openAttemptTimeout ?? SiteAlarmAggregatorActor.DefaultOpenAttemptTimeout)); var actor = Sys.ActorOf(props); return (actor, seed, sink, factory); @@ -849,6 +854,98 @@ public class SiteAlarmAggregatorActorTests : TestKit AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); } + // ── P3-4: the stream open attempt is deadline-bounded ─────────────────────── + + /// + /// Proof of the P3-4 starvation and of its fix. + /// + /// An open attempt is cleared by exactly three peer-produced outcomes: connected, error, + /// or graceful completion. A stream that produces NONE of them pins _openInFlight. + /// That is reachable: SiteStreamGrpcClient gives up on response headers after + /// ConnectedHeaderTimeout (10 s) and then treats the FIRST RECEIVED EVENT as its + /// fallback connected signal — which an alarm-only site that happens to be quiet never + /// sends (proven independently by + /// SiteStreamGrpcClientTests.ConsumeStream_HeaderTimeoutOnADeadSite_NeverReportsConnected). + /// The aggregator then sits with _streamDown == true AND _openInFlight == true, + /// and 's reopen guard + /// (_streamDown && !_openInFlight) is skipped every tick until the site's + /// 4 h max stream lifetime finally ends the call. + /// + /// + /// Honesty note. The mock reproduces the aggregator-side state machine exactly and + /// deterministically (that is what autoConnect: false models). What it does NOT + /// prove is how often the >10 s header delay that gets you there actually happens in + /// production: the site flushes response headers explicitly as soon as its relay is + /// attached (SiteStreamGrpcServer.RunSubscriptionStreamAsync), so the 10 s budget + /// has to be consumed by channel connect + HTTP/2 handshake + server scheduling instead — + /// plausible under a WAN stall or a GC/thread-pool-starved site node, but not something + /// this suite can demonstrate. The deadline is therefore a self-healing guard rather than + /// a fix for a measured production incident. + /// + /// + [Fact] + public void OpenAttemptThatNeverResolves_ReleasesTheInFlightGuard_SoTheTickCanReopen() + { + // 250 ms deadline; the reconcile timer is parked 10 minutes away so the ONLY tick is + // the one this test issues by hand. + var (actor, seed, sink, factory) = CreateActor( + reconcileInterval: TimeSpan.FromMinutes(10), + autoConnect: false, + openAttemptTimeout: TimeSpan.FromMilliseconds(250)); + + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + + // The stream never reports connected, never faults, never completes — the quiet-site + // header-timeout shape. Liveness is (correctly) false the whole time. + AssertHoldsFor(TimeSpan.FromMilliseconds(300), + (() => !sink.StreamLive, "a stream that never connected must never read as live")); + + // Pre-fix this tick found _openInFlight still true and skipped the reopen, leaving + // node A on one subscription forever. Post-fix the deadline has released the flag. + actor.Tell(new RunReconcile()); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5)); + + // And the reopen is a real one: its connect brings the feed up and earns a re-seed. + factory.ClientFor(GrpcNodeA).Subs[1].OnConnected!(); + AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + AwaitCondition(() => sink.StreamLive, TimeSpan.FromSeconds(3)); + } + + [Fact] + public void OpenDeadline_DoesNotFire_OnAStreamThatConnects_NorFlipNodeOrBurnBudget() + { + // The deadline must be inert on every healthy path: a connect cancels it, so no + // spurious reopen, no node flip, and no retry-budget spend can be attributed to it. + // Timing-independent in the sense that matters — the assertions below would fail for + // ANY firing of the timer, and the window is 8x the deadline. + var (actor, seed, sink, factory) = CreateActor( + reconcileInterval: TimeSpan.FromMinutes(10), + autoConnect: true, + openAttemptTimeout: TimeSpan.FromMilliseconds(100)); + + AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); + AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3)); + seed.CompleteNext(); + AwaitCondition(() => sink.StreamLive, TimeSpan.FromSeconds(3)); + + AssertHoldsFor(TimeSpan.FromMilliseconds(800), + (() => sink.StreamLive, "a connected stream must stay live across the open deadline"), + (() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, "no second open on node A"), + (() => factory.ClientFor(GrpcNodeB).Subs.Count == 0, "no node flip"), + (() => seed.CallCount == 1, "no extra fan-out")); + + // A reconcile tick on a live stream still must not reopen. (First tick is consumed by + // the initial seed's skip flag; the second is the real fan-out.) + actor.Tell(new RunReconcile()); + actor.Tell(new RunReconcile()); + AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3)); // tick fan-out only + seed.CompleteNext(); + Assert.Single(factory.ClientFor(GrpcNodeA).Subs); + } + [Fact] public void FailedSeed_IsRetried_WithBackoff_BeforeTheNextReconcile() { diff --git a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs index 949e6218..abdfb87a 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.Communication.Tests/Grpc/SiteStreamGrpcServerTests.cs @@ -8,6 +8,7 @@ using Microsoft.Extensions.Logging.Abstractions; using NSubstitute; using ZB.MOM.WW.ScadaBridge.Commons.Observability; using ZB.MOM.WW.ScadaBridge.Communication.Grpc; +using GrpcStatus = Grpc.Core.Status; namespace ZB.MOM.WW.ScadaBridge.Communication.Tests.Grpc; @@ -794,6 +795,155 @@ public class SiteStreamGrpcServerTests : TestKit } } + // ── P3-5: the concurrency cap must never refuse a REPLACEMENT ─────────────── + + /// + /// A whose + /// read runs a caller-supplied hook. That read is the first statement after the concurrency + /// cap check and before the _activeStreams insert in + /// SiteStreamGrpcServer.RunSubscriptionStreamAsync, so parking there is how a test + /// deterministically drives two subscribes into the cap's documented check-then-act window. + /// + private sealed class GatedServerCallContext : ServerCallContext + { + private readonly CancellationToken _token; + private readonly Action _onCancellationTokenRead; + private readonly Metadata _trailers = new(); + private readonly Dictionary _userState = new(); + + public GatedServerCallContext(CancellationToken token, Action onCancellationTokenRead) + { + _token = token; + _onCancellationTokenRead = onCancellationTokenRead; + } + + protected override string MethodCore => "/sitestream.SiteStreamService/SubscribeInstance"; + protected override string HostCore => "localhost"; + protected override string PeerCore => "ipv4:127.0.0.1:0"; + protected override DateTime DeadlineCore => DateTime.MaxValue; + protected override Metadata RequestHeadersCore => Metadata.Empty; + protected override Metadata ResponseTrailersCore => _trailers; + protected override GrpcStatus StatusCore { get; set; } + protected override WriteOptions? WriteOptionsCore { get; set; } + protected override AuthContext AuthContextCore => + new(null, new Dictionary>()); + protected override IDictionary UserStateCore => _userState; + + protected override CancellationToken CancellationTokenCore + { + get + { + _onCancellationTokenRead(); + return _token; + } + } + + protected override ContextPropagationToken CreatePropagationTokenCore( + ContextPropagationOptions? options) => throw new NotSupportedException(); + + protected override Task WriteResponseHeadersAsyncCore(Metadata responseHeaders) => + Task.CompletedTask; + } + + [Fact] + public async Task AtCapOvershoot_DuplicateReconnect_IsNotRefused_AndLeavesExactlyOneLiveStream() + { + // P3-5. The duplicate-replacement path cancels the incumbent; the cap check then had + // to decide whether the REPLACEMENT may take its place. Checking `Count >= max` after + // the removal is right at exactly the cap but wrong above it — and above it is + // reachable, because the cap check is documented check-then-act: two subscribes that + // both clear it before either registers overshoot by one. In that state a routine + // reconnect killed a healthy stream AND was refused with ResourceExhausted, leaving + // the correlation with ZERO streams until its next retry. + // + // The overshoot is produced DETERMINISTICALLY here (no sleeps, no racing threads left + // to luck): both openers park on the CancellationToken read, which sits between the + // cap check and the _activeStreams insert, so both are past the cap with nothing + // registered. Releasing the gate then lands two entries under a cap of one. + var server = CreateServer(maxStreams: 1); + server.SetReady(Sys); + + using var gate = new ManualResetEventSlim(false); + using var parked = new CountdownEvent(2); + void Park() + { + parked.Signal(); + Assert.True(gate.Wait(TimeSpan.FromSeconds(15)), "gate was never released"); + } + + using var ctsA = new CancellationTokenSource(); + using var ctsB = new CancellationTokenSource(); + var streamA = Task.Run(() => server.SubscribeInstance( + MakeRequest("corr-a"), + Substitute.For>(), + new GatedServerCallContext(ctsA.Token, Park))); + var streamB = Task.Run(() => server.SubscribeInstance( + MakeRequest("corr-b"), + Substitute.For>(), + new GatedServerCallContext(ctsB.Token, Park))); + + Assert.True(parked.Wait(TimeSpan.FromSeconds(15)), "both openers should reach the token read"); + gate.Set(); + + // Both registered: the cap of 1 is overshot to 2, exactly as its comment allows. + await WaitForConditionAsync(() => server.ActiveStreamCount == 2); + + // Now "corr-a" reconnects. This is a REPLACEMENT — net-neutral on the stream count — + // so it must be accepted. Pre-fix it threw ResourceExhausted *after* cancelling the + // incumbent, so "corr-a" was left with no stream at all. + using var ctsA2 = new CancellationTokenSource(); + var streamA2 = Task.Run(() => server.SubscribeInstance( + MakeRequest("corr-a"), + Substitute.For>(), + CreateMockContext(ctsA2.Token))); + + // The incumbent unwinds through its normal cancellation path... + await streamA; + Assert.False(streamA2.IsFaulted, + "the replacement must not be refused: " + streamA2.Exception?.GetBaseException().Message); + + // ...and exactly one stream is live for "corr-a" (plus the untouched "corr-b"). + await WaitForConditionAsync(() => server.ActiveStreamCount == 2); + Assert.False(streamA2.IsCompleted, "the replacement stream should still be running"); + + ctsA2.Cancel(); + ctsB.Cancel(); + await streamA2; + await streamB; + await WaitForConditionAsync(() => server.ActiveStreamCount == 0); + } + + [Fact] + public async Task AtCap_AGenuinelyNewSubscriber_IsStillRefused_AndTheIncumbentSurvives() + { + // The other half of P3-5: excluding replacements from the cap must NOT weaken the cap + // for a new correlation id, and a refusal must leave the existing streams untouched + // (TryRemove found nothing, so nothing was cancelled before the throw). + var server = CreateServer(maxStreams: 1); + server.SetReady(Sys); + + using var cts1 = new CancellationTokenSource(); + var stream1 = Task.Run(() => server.SubscribeInstance( + MakeRequest("corr-keep"), + Substitute.For>(), + CreateMockContext(cts1.Token))); + + await WaitForConditionAsync(() => server.ActiveStreamCount == 1); + + var ex = await Assert.ThrowsAsync(() => server.SubscribeInstance( + MakeRequest("corr-new"), + Substitute.For>(), + CreateMockContext())); + Assert.Equal(StatusCode.ResourceExhausted, ex.StatusCode); + + // The incumbent is untouched by the refusal. + Assert.Equal(1, server.ActiveStreamCount); + Assert.False(stream1.IsCompleted, "a refused NEW subscriber must not disturb the incumbent"); + + cts1.Cancel(); + await stream1; + } + /// Total events carried across all captured frames (unpacking batch frames). private static int CountEvents(List capture) { diff --git a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs index 68dda8aa..de43d967 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs +++ b/tests/ZB.MOM.WW.ScadaBridge.IntegrationTests/Grpc/SiteAlarmStreamEndToEndTests.cs @@ -224,7 +224,10 @@ public class SiteAlarmStreamEndToEndTests : TestKit var aggregator = Sys.ActorOf(Props.Create(() => new SiteAlarmAggregatorActor( "site-alpha", "e2e-parity", _ => Task.FromResult>(new[] { seedRow }), sink.Publish, factory, "http://a:5100", "http://b:5100", TimeSpan.FromMinutes(10), - TimeSpan.Zero, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(60)))); + TimeSpan.Zero, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(60), + // Props.Create is an expression tree, which may not omit optional arguments + // (CS0854) — jitter and the P3-4 open-attempt deadline are passed explicitly. + 0.0, SiteAlarmAggregatorActor.DefaultOpenAttemptTimeout))); await WaitForConditionAsync(() => sink.Latest is { Count: 1 }); // seed applied aggregator.Tell(liveRow);