fix(communication): P3-4 open-attempt deadline + P3-5 duplicate-subscribe cap order
P3-4 — SiteAlarmAggregatorActor pinned _openInFlight forever on a stream open that produced none of its three peer-driven outcomes. SiteStreamGrpcClient gives up on response headers after ConnectedHeaderTimeout (10s) and then treats the first received EVENT as its fallback connected signal, so an alarm-only site that stays quiet supplies neither — leaving _streamDown=true AND _openInFlight= true, which makes OnReconcileTick's reopen guard (_streamDown && !_openInFlight) skip every tick until the site's 4h max stream lifetime ends the call. Fix: bound one open attempt with a generation-fenced single-shot deadline (GrpcAlarmStreamOpenTimedOut, ctor-injected, production default 30s — above the 10s header timeout, below the 60s reconcile interval). Firing it releases the flag and nothing else: no retry-budget spend, no node flip, no liveness republish, and the abandoned subscription is left running so a merely-slow connect still comes up normally. Self-healing rather than reachability analysis. The deadline is cancelled on connect / error / graceful completion. P3-5 — SiteStreamGrpcServer cancelled the incumbent stream on a duplicate correlation id and only THEN checked the concurrency cap. 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 window a routine reconnect killed a healthy stream AND was refused with ResourceExhausted, leaving the correlation with ZERO streams until its next retry. Fix: claim the removal, consult the cap only for a genuinely NEW correlation id, then cancel. A replacement is net-neutral on the count so it never consults the cap; a refused new subscriber cancelled nothing (TryRemove found nothing), so the incumbent survives. Newest-wins is unchanged — a fresh authenticated subscribe still displaces a dead/wedged incumbent. Tests (both shown failing against the unfixed code): - OpenAttemptThatNeverResolves_ReleasesTheInFlightGuard_SoTheTickCanReopen — fails (reopen never happens) with the deadline arming disabled. - OpenDeadline_DoesNotFire_OnAStreamThatConnects_NorFlipNodeOrBurnBudget — inertness lock on every healthy path. - AtCapOvershoot_DuplicateReconnect_IsNotRefused_AndLeavesExactlyOneLiveStream — fails with ResourceExhausted against the old ordering; the overshoot is driven deterministically by parking both openers on the ServerCallContext token read, which sits between the cap check and the _activeStreams insert. - AtCap_AGenuinelyNewSubscriber_IsStillRefused_AndTheIncumbentSurvives. The P3-4 test's doc comment carries the honesty note: the aggregator-side starvation reproduces exactly, but how often the >10s header delay that gets you there occurs in production is NOT demonstrated — the site flushes headers explicitly as soon as its relay is attached, so the budget has to be consumed by connect + handshake + server scheduling instead.
This commit is contained in:
Joseph Doherty
@@ -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";
|
||||
|
||||
/// <summary>
|
||||
/// Production default for the stream open-attempt deadline (arch-review 2026-08-19 P3-4).
|
||||
/// Comfortably above <c>SiteStreamGrpcClient.ConnectedHeaderTimeout</c> (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.
|
||||
/// </summary>
|
||||
internal static readonly TimeSpan DefaultOpenAttemptTimeout = TimeSpan.FromSeconds(30);
|
||||
|
||||
/// <summary>True while a coalesced publish is armed (dirty deltas awaiting one tick). Actor-thread only.</summary>
|
||||
private bool _publishPending;
|
||||
@@ -179,10 +188,38 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
|
||||
/// <summary>
|
||||
/// 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 cref="_openAttemptTimeout"/> — see there for why an unbounded flag
|
||||
/// was a starvation hazard. Actor-thread only.
|
||||
/// </summary>
|
||||
private bool _openInFlight;
|
||||
|
||||
/// <summary>
|
||||
/// Deadline on ONE stream open attempt, after which <see cref="_openInFlight"/> is
|
||||
/// released so <see cref="OnReconcileTick"/> can reopen (arch-review 2026-08-19 P3-4).
|
||||
/// <para>
|
||||
/// 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
|
||||
/// <c>SiteStreamGrpcClient.ConnectedHeaderTimeout</c> (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 <c>_streamDown</c> is still true (nothing ever
|
||||
/// reported connected) but <c>_openInFlight</c> is also true, and the tick's reopen guard
|
||||
/// — <c>_streamDown && !_openInFlight</c> — 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.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// 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 (<c>_streamDown</c> was never cleared), and the abandoned subscription is
|
||||
/// left running — if it was merely slow and connects late, <c>GrpcAlarmStreamConnected</c>
|
||||
/// still matches the current generation and brings it up normally; if a reconcile tick
|
||||
/// reopens first, <see cref="OpenGrpcStream"/> cancels its token source and fences its
|
||||
/// generation out.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
private readonly TimeSpan _openAttemptTimeout;
|
||||
|
||||
/// <summary>Consecutive whole-fan-out failures, driving the seed-retry backoff. Actor-thread only.</summary>
|
||||
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.
|
||||
/// </param>
|
||||
/// <param name="openAttemptTimeout">
|
||||
/// Deadline on one gRPC stream open attempt (P3-4); <c>null</c> takes
|
||||
/// <see cref="DefaultOpenAttemptTimeout"/> (30 s) and <see cref="TimeSpan.Zero"/> or less
|
||||
/// disables the deadline. See <see cref="_openAttemptTimeout"/>.
|
||||
/// </param>
|
||||
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<GrpcAlarmStreamOpenTimedOut>(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<PublishCoalesced>(_ =>
|
||||
{
|
||||
@@ -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
|
||||
/// </summary>
|
||||
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;
|
||||
|
||||
/// <summary>Internal: the current site-alarm gRPC stream has stayed up long enough to recover the retry budget.</summary>
|
||||
internal sealed record GrpcAlarmStreamStable;
|
||||
|
||||
/// <summary>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.</summary>
|
||||
internal sealed record GrpcAlarmStreamOpenTimedOut(int Generation);
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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}");
|
||||
|
||||
|
||||
+99
-2
@@ -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 ───────────────────────
|
||||
|
||||
/// <summary>
|
||||
/// Proof of the P3-4 starvation and of its fix.
|
||||
/// <para>
|
||||
/// An open attempt is cleared by exactly three peer-produced outcomes: connected, error,
|
||||
/// or graceful completion. A stream that produces NONE of them pins <c>_openInFlight</c>.
|
||||
/// That is reachable: <c>SiteStreamGrpcClient</c> gives up on response headers after
|
||||
/// <c>ConnectedHeaderTimeout</c> (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
|
||||
/// <c>SiteStreamGrpcClientTests.ConsumeStream_HeaderTimeoutOnADeadSite_NeverReportsConnected</c>).
|
||||
/// The aggregator then sits with <c>_streamDown == true</c> AND <c>_openInFlight == true</c>,
|
||||
/// and <see cref="SiteAlarmAggregatorActor.OnReconcileTick"/>'s reopen guard
|
||||
/// (<c>_streamDown && !_openInFlight</c>) is skipped every tick until the site's
|
||||
/// 4 h max stream lifetime finally ends the call.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>Honesty note.</b> The mock reproduces the aggregator-side state machine exactly and
|
||||
/// deterministically (that is what <c>autoConnect: false</c> 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 (<c>SiteStreamGrpcServer.RunSubscriptionStreamAsync</c>), 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.
|
||||
/// </para>
|
||||
/// </summary>
|
||||
[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()
|
||||
{
|
||||
|
||||
@@ -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 ───────────────
|
||||
|
||||
/// <summary>
|
||||
/// A <see cref="ServerCallContext"/> whose <see cref="ServerCallContext.CancellationToken"/>
|
||||
/// read runs a caller-supplied hook. That read is the first statement after the concurrency
|
||||
/// cap check and before the <c>_activeStreams</c> insert in
|
||||
/// <c>SiteStreamGrpcServer.RunSubscriptionStreamAsync</c>, so parking there is how a test
|
||||
/// deterministically drives two subscribes into the cap's documented check-then-act window.
|
||||
/// </summary>
|
||||
private sealed class GatedServerCallContext : ServerCallContext
|
||||
{
|
||||
private readonly CancellationToken _token;
|
||||
private readonly Action _onCancellationTokenRead;
|
||||
private readonly Metadata _trailers = new();
|
||||
private readonly Dictionary<object, object> _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<string, List<AuthProperty>>());
|
||||
protected override IDictionary<object, object> 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<IServerStreamWriter<SiteStreamEvent>>(),
|
||||
new GatedServerCallContext(ctsA.Token, Park)));
|
||||
var streamB = Task.Run(() => server.SubscribeInstance(
|
||||
MakeRequest("corr-b"),
|
||||
Substitute.For<IServerStreamWriter<SiteStreamEvent>>(),
|
||||
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<IServerStreamWriter<SiteStreamEvent>>(),
|
||||
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<IServerStreamWriter<SiteStreamEvent>>(),
|
||||
CreateMockContext(cts1.Token)));
|
||||
|
||||
await WaitForConditionAsync(() => server.ActiveStreamCount == 1);
|
||||
|
||||
var ex = await Assert.ThrowsAsync<RpcException>(() => server.SubscribeInstance(
|
||||
MakeRequest("corr-new"),
|
||||
Substitute.For<IServerStreamWriter<SiteStreamEvent>>(),
|
||||
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;
|
||||
}
|
||||
|
||||
/// <summary>Total events carried across all captured frames (unpacking batch frames).</summary>
|
||||
private static int CountEvents(List<SiteStreamEvent> capture)
|
||||
{
|
||||
|
||||
@@ -224,7 +224,10 @@ public class SiteAlarmStreamEndToEndTests : TestKit
|
||||
var aggregator = Sys.ActorOf(Props.Create(() => new SiteAlarmAggregatorActor(
|
||||
"site-alpha", "e2e-parity", _ => Task.FromResult<IReadOnlyList<AlarmStateChanged>>(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);
|
||||
|
||||
Reference in New Issue
Block a user