chore(comm): ctor-inject aggregator reconnect/stability tuning, remove process-global statics (plan R2-02 T12)

This commit is contained in:
Joseph Doherty
2026-07-13 10:19:51 -04:00
parent c29e588905
commit b7f8632db8
3 changed files with 26 additions and 12 deletions
@@ -63,15 +63,15 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
/// <summary>True while a coalesced publish is armed (dirty deltas awaiting one tick). Actor-thread only.</summary> /// <summary>True while a coalesced publish is armed (dirty deltas awaiting one tick). Actor-thread only.</summary>
private bool _publishPending; private bool _publishPending;
/// <summary>Delay between gRPC reconnection attempts. Settable for tests.</summary> /// <summary>Delay between gRPC reconnection attempts (ctor-injected; production default 5s).</summary>
internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5); private readonly TimeSpan _reconnectDelay;
/// <summary> /// <summary>
/// How long a freshly-opened gRPC stream must stay up before its retry budget is /// How long a freshly-opened gRPC stream must stay up before its retry budget is
/// considered recovered (mirrors <see cref="DebugStreamBridgeActor.StabilityWindow"/>). /// considered recovered (mirrors <see cref="DebugStreamBridgeActor.StabilityWindow"/>);
/// Settable for tests. /// ctor-injected (production default 60s).
/// </summary> /// </summary>
internal static TimeSpan StabilityWindow { get; set; } = TimeSpan.FromSeconds(60); private readonly TimeSpan _stabilityWindow;
private int _retryCount; private int _retryCount;
private bool _useNodeA = true; private bool _useNodeA = true;
@@ -143,6 +143,10 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
/// into one publish per window (review 02 round 2, N6); <see cref="TimeSpan.Zero"/> /// into one publish per window (review 02 round 2, N6); <see cref="TimeSpan.Zero"/>
/// restores per-delta publishing (legacy). Seed/reconcile publishes stay immediate. /// restores per-delta publishing (legacy). Seed/reconcile publishes stay immediate.
/// </param> /// </param>
/// <param name="reconnectDelay">Delay between gRPC reconnection attempts (production 5s).</param>
/// <param name="stabilityWindow">
/// How long a fresh gRPC stream must stay up before its retry budget recovers (production 60s).
/// </param>
public SiteAlarmAggregatorActor( public SiteAlarmAggregatorActor(
string siteIdentifier, string siteIdentifier,
string correlationId, string correlationId,
@@ -152,7 +156,9 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
string grpcNodeAAddress, string grpcNodeAAddress,
string grpcNodeBAddress, string grpcNodeBAddress,
TimeSpan reconcileInterval, TimeSpan reconcileInterval,
TimeSpan publishCoalesce) TimeSpan publishCoalesce,
TimeSpan reconnectDelay,
TimeSpan stabilityWindow)
{ {
_siteIdentifier = siteIdentifier; _siteIdentifier = siteIdentifier;
_correlationId = correlationId; _correlationId = correlationId;
@@ -163,6 +169,8 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
_grpcNodeBAddress = grpcNodeBAddress; _grpcNodeBAddress = grpcNodeBAddress;
_reconcileInterval = reconcileInterval; _reconcileInterval = reconcileInterval;
_publishCoalesce = publishCoalesce; _publishCoalesce = publishCoalesce;
_reconnectDelay = reconnectDelay;
_stabilityWindow = stabilityWindow;
// Live delta from the site-wide alarm stream (marshalled in via Self.Tell). // 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 // A received delta must NOT reset the retry budget (a flapping stream that
@@ -499,7 +507,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
_grpcCts?.Dispose(); _grpcCts?.Dispose();
_grpcCts = new CancellationTokenSource(); _grpcCts = new CancellationTokenSource();
Timers.StartSingleTimer(StabilityTimerKey, new GrpcAlarmStreamStable(), StabilityWindow); Timers.StartSingleTimer(StabilityTimerKey, new GrpcAlarmStreamStable(), _stabilityWindow);
var generation = ++_streamGeneration; var generation = ++_streamGeneration;
var client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint); var client = _grpcFactory.GetOrCreate(_siteIdentifier, endpoint);
@@ -558,7 +566,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
if (_retryCount == 1) if (_retryCount == 1)
Self.Tell(new ReconnectAlarmStream()); Self.Tell(new ReconnectAlarmStream());
else else
Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectAlarmStream(), ReconnectDelay); Timers.StartSingleTimer(ReconnectTimerKey, new ReconnectAlarmStream(), _reconnectDelay);
} }
private void CleanupGrpc() private void CleanupGrpc()
@@ -234,7 +234,9 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache
grpcA, grpcA,
grpcB, grpcB,
_options.LiveAlarmCacheReconcileInterval, _options.LiveAlarmCacheReconcileInterval,
_options.LiveAlarmCachePublishCoalesce)); _options.LiveAlarmCachePublishCoalesce,
TimeSpan.FromSeconds(5), // reconnect delay — former ReconnectDelay static default
TimeSpan.FromSeconds(60))); // stability window — former StabilityWindow static default
entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}"); entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}");
entry.StartRetryTimer?.Dispose(); entry.StartRetryTimer?.Dispose();
@@ -24,10 +24,12 @@ public class SiteAlarmAggregatorActorTests : TestKit
public SiteAlarmAggregatorActorTests() : base(@"akka.loglevel = WARNING") public SiteAlarmAggregatorActorTests() : base(@"akka.loglevel = WARNING")
{ {
SiteAlarmAggregatorActor.ReconnectDelay = TimeSpan.FromMilliseconds(50);
SiteAlarmAggregatorActor.StabilityWindow = TimeSpan.FromSeconds(30);
} }
// Former process-global static test seams, now ctor-injected per actor (R2 T12).
private static readonly TimeSpan TestReconnectDelay = TimeSpan.FromMilliseconds(50);
private static readonly TimeSpan TestStabilityWindow = TimeSpan.FromSeconds(30);
// ── Test doubles ──────────────────────────────────────────────────────────── // ── Test doubles ────────────────────────────────────────────────────────────
/// <summary> /// <summary>
@@ -146,7 +148,9 @@ public class SiteAlarmAggregatorActorTests : TestKit
var props = Props.Create(() => new SiteAlarmAggregatorActor( var props = Props.Create(() => new SiteAlarmAggregatorActor(
SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB, SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB,
reconcileInterval ?? TimeSpan.FromMinutes(10), reconcileInterval ?? TimeSpan.FromMinutes(10),
publishCoalesce ?? TimeSpan.Zero)); publishCoalesce ?? TimeSpan.Zero,
TestReconnectDelay,
TestStabilityWindow));
var actor = Sys.ActorOf(props); var actor = Sys.ActorOf(props);
return (actor, seed, sink, factory); return (actor, seed, sink, factory);