perf(comm): coalesce live-alarm delta publishes (250ms window, 0 = legacy) (plan R2-02 T10)

2026-07-13 10:14:54 -04:00
parent cc4ff7029d
commit bb8be55382
6 changed files with 123 additions and 6 deletions
@@ -52,11 +52,16 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
    private readonly string _grpcNodeAAddress;
    private readonly string _grpcNodeBAddress;
    private readonly TimeSpan _reconcileInterval;
+    private readonly TimeSpan _publishCoalesce;

    private const int MaxRetries = 3;
    private const string ReconnectTimerKey = "alarm-grpc-reconnect";
    private const string StabilityTimerKey = "alarm-grpc-stability";
    private const string ReconcileTimerKey = "alarm-reconcile";
+    private const string PublishTimerKey = "alarm-publish-coalesce";
+
+    /// <summary>True while a coalesced publish is armed (dirty deltas awaiting one tick). Actor-thread only.</summary>
+    private bool _publishPending;

    /// <summary>Delay between gRPC reconnection attempts. Settable for tests.</summary>
    internal static TimeSpan ReconnectDelay { get; set; } = TimeSpan.FromSeconds(5);
@@ -118,6 +123,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
    /// <param name="grpcNodeAAddress">gRPC address of the site's node A.</param>
    /// <param name="grpcNodeBAddress">gRPC address of the site's node B.</param>
    /// <param name="reconcileInterval">Periodic reconcile snapshot cadence.</param>
+    /// <param name="publishCoalesce">
+    /// Publish-coalescing window for live deltas: a positive value batches a delta storm
+    /// into one publish per window (review 02 round 2, N6); <see cref="TimeSpan.Zero"/>
+    /// restores per-delta publishing (legacy). Seed/reconcile publishes stay immediate.
+    /// </param>
    public SiteAlarmAggregatorActor(
        string siteIdentifier,
        string correlationId,
@@ -126,7 +136,8 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        SiteStreamGrpcClientFactory grpcFactory,
        string grpcNodeAAddress,
        string grpcNodeBAddress,
-        TimeSpan reconcileInterval)
+        TimeSpan reconcileInterval,
+        TimeSpan publishCoalesce)
    {
        _siteIdentifier = siteIdentifier;
        _correlationId = correlationId;
@@ -136,6 +147,7 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        _grpcNodeAAddress = grpcNodeAAddress;
        _grpcNodeBAddress = grpcNodeBAddress;
        _reconcileInterval = reconcileInterval;
+        _publishCoalesce = publishCoalesce;

        // 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
@@ -152,6 +164,13 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        // Periodic reconcile tick (and the re-seed kicked after a reconnect).
        Receive<RunReconcile>(_ => OnReconcileTick());

+        // Coalesced-publish tick: one publish for a batch of dirtying deltas (N6).
+        Receive<PublishCoalesced>(_ =>
+        {
+            _publishPending = false;
+            if (!_stopped) Publish();
+        });
+
        // Stream stayed up for StabilityWindow — recover the retry budget.
        Receive<GrpcAlarmStreamStable>(_ =>
        {
@@ -300,6 +319,10 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        _log.Debug("Site-alarm {0} {1} complete: {2} alarm row(s)",
            _siteIdentifier, msg.IsInitial ? "seed" : "reconcile", _cache.Count);

+        // The fresh snapshot already carries the buffered deltas; drop any armed coalesce
+        // tick so we publish once, immediately.
+        Timers.Cancel(PublishTimerKey);
+        _publishPending = false;
        Publish();
    }

@@ -319,7 +342,11 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
        // Only publish if we already had a seed (so IsLive doesn't flip true on a
        // failed initial seed — the page keeps its poll fallback until we truly seed).
        if (_seeded)
+        {
+            Timers.Cancel(PublishTimerKey);
+            _publishPending = false;
            Publish();
+        }
    }

    /// <summary>
@@ -362,9 +389,23 @@ public sealed class SiteAlarmAggregatorActor : ReceiveActor, IWithTimers
            return;
        }

-        // Pass-through: apply and publish only if the cache actually changed.
+        // Pass-through: apply and (coalesced) publish only if the cache actually changed.
        if (ApplyDelta(delta, requireStrictlyNewer: false))
-            Publish();
+            SchedulePublish();
+    }
+
+    /// <summary>
+    /// Coalesced publish: with a positive window, the first dirtying delta arms a
+    /// single-shot timer and further deltas ride the same tick — one snapshot copy and
+    /// one viewer fan-out per window instead of per transition (N6). Zero = legacy
+    /// immediate publish. Last write wins, so batching never changes final state.
+    /// </summary>
+    private void SchedulePublish()
+    {
+        if (_publishCoalesce <= TimeSpan.Zero) { Publish(); return; }
+        if (_publishPending) return;
+        _publishPending = true;
+        Timers.StartSingleTimer(PublishTimerKey, new PublishCoalesced(), _publishCoalesce);
    }

    /// <summary>
@@ -520,6 +561,9 @@ internal sealed record SeedFailed(Exception Exception, bool IsInitial);
 /// <summary>Internal: periodic reconcile tick (and the re-seed kicked after a reconnect).</summary>
 internal sealed record RunReconcile;

+/// <summary>Internal: coalesced-publish tick — flush the dirty cache to viewers once (N6).</summary>
+internal sealed record PublishCoalesced;
+
 /// <summary>Internal: site-alarm gRPC stream error occurred.</summary>
 internal sealed record GrpcAlarmStreamError(Exception Exception);

@@ -122,4 +122,13 @@ public class CommunicationOptions
    /// count; this just bounds the subscriber list. Default 200.
    /// </summary>
    public int LiveAlarmCacheMaxSubscribersPerSite { get; set; } = 200;
+
+    /// <summary>
+    /// Publish-coalescing window for live alarm deltas: an applied delta marks the cache
+    /// dirty and one publish (fresh snapshot + per-viewer onChanged fan-out) fires after
+    /// this window, batching an alarm storm into ~4 publishes/second instead of one per
+    /// transition (review 02 round 2, N6). Zero = publish per delta (legacy). Seed and
+    /// reconcile publishes are always immediate. Default 250 ms.
+    /// </summary>
+    public TimeSpan LiveAlarmCachePublishCoalesce { get; set; } = TimeSpan.FromMilliseconds(250);
 }
@@ -85,5 +85,10 @@ public sealed class CommunicationOptionsValidator : OptionsValidatorBase<Communi
        // Per-site viewer cap must admit at least one viewer, else the page could never go live.
        builder.RequireThat(options.LiveAlarmCacheMaxSubscribersPerSite >= 1,
            $"Communication:LiveAlarmCacheMaxSubscribersPerSite must be at least 1 (was {options.LiveAlarmCacheMaxSubscribersPerSite}).");
+
+        // Publish-coalescing window drives a single-shot timer; TimeSpan.Zero is valid
+        // (publish per delta — legacy), only a negative value is invalid.
+        builder.RequireThat(options.LiveAlarmCachePublishCoalesce >= TimeSpan.Zero,
+            $"Communication:LiveAlarmCachePublishCoalesce must be zero or a positive duration (was {options.LiveAlarmCachePublishCoalesce}).");
    }
 }
@@ -233,7 +233,8 @@ public sealed class SiteAlarmLiveCacheService : ISiteAlarmLiveCache
                    _grpcFactory,
                    grpcA,
                    grpcB,
-                    _options.LiveAlarmCacheReconcileInterval));
+                    _options.LiveAlarmCacheReconcileInterval,
+                    _options.LiveAlarmCachePublishCoalesce));

                entry.Actor = system.ActorOf(props, $"site-alarm-aggregator-{entry.SiteId}-{Guid.NewGuid():N}");
                entry.StartRetryTimer?.Dispose();
@@ -86,4 +86,22 @@ public class CommunicationOptionsValidatorTests
        Assert.True(result.Failed);
        Assert.Contains("LiveAlarmCacheMaxSubscribersPerSite", result.FailureMessage);
    }
+
+    // ── R2 T10: live-delta publish-coalescing window (N6) ────────────────────────
+
+    [Fact]
+    public void ZeroLiveAlarmCachePublishCoalesce_IsValid()
+    {
+        // Zero = publish per delta (legacy behavior).
+        var result = Validate(new CommunicationOptions { LiveAlarmCachePublishCoalesce = TimeSpan.Zero });
+        Assert.True(result.Succeeded, result.FailureMessage);
+    }
+
+    [Fact]
+    public void NegativeLiveAlarmCachePublishCoalesce_IsRejected()
+    {
+        var result = Validate(new CommunicationOptions { LiveAlarmCachePublishCoalesce = TimeSpan.FromMilliseconds(-1) });
+        Assert.True(result.Failed);
+        Assert.Contains("LiveAlarmCachePublishCoalesce", result.FailureMessage);
+    }
 }
@@ -137,7 +137,7 @@ public class SiteAlarmAggregatorActorTests : TestKit
    }

    private (IActorRef Actor, SeedStub Seed, PublishSink Sink, MockSiteAlarmStreamClientFactory Factory) CreateActor(
-        TimeSpan? reconcileInterval = null)
+        TimeSpan? reconcileInterval = null, TimeSpan? publishCoalesce = null)
    {
        var seed = new SeedStub();
        var sink = new PublishSink();
@@ -145,7 +145,8 @@ public class SiteAlarmAggregatorActorTests : TestKit

        var props = Props.Create(() => new SiteAlarmAggregatorActor(
            SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB,
-            reconcileInterval ?? TimeSpan.FromMinutes(10)));
+            reconcileInterval ?? TimeSpan.FromMinutes(10),
+            publishCoalesce ?? TimeSpan.Zero));

        var actor = Sys.ActorOf(props);
        return (actor, seed, sink, factory);
@@ -396,6 +397,45 @@ public class SiteAlarmAggregatorActorTests : TestKit
        AwaitCondition(() => TotalSubs() > before, TimeSpan.FromSeconds(3));
    }

+    // ── R2 T10: live-delta publish coalescing (N6) ──
+
+    [Fact]
+    public void DeltaBurst_IsCoalesced_IntoFewPublishes_ContainingEveryRow()
+    {
+        var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.FromMilliseconds(150));
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+        seed.CompleteNext(); // empty initial seed
+        AwaitAssert(() => Assert.Equal(1, sink.Count)); // seed publish, immediate
+
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
+        var sub = factory.ClientFor(GrpcNodeA).Subs.Single();
+        var now = DateTimeOffset.UtcNow;
+        for (var i = 0; i < 50; i++)
+            sub.OnAlarm(Alarm($"A{i}", "", 900, now)); // 50 distinct keys, one burst
+
+        AwaitAssert(() =>
+        {
+            Assert.Equal(50, sink.Latest!.Count);   // nothing lost
+            Assert.InRange(sink.Count, 2, 4);       // pre-fix: 51 publishes (1 seed + 1 per delta)
+        }, TimeSpan.FromSeconds(3));
+    }
+
+    [Fact]
+    public void ZeroCoalesce_PreservesLegacyPerDeltaPublish()
+    {
+        var (_, seed, sink, factory) = CreateActor(publishCoalesce: TimeSpan.Zero);
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+        seed.CompleteNext();
+        AwaitAssert(() => Assert.Equal(1, sink.Count));
+
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
+        var sub = factory.ClientFor(GrpcNodeA).Subs.Single();
+        var now = DateTimeOffset.UtcNow;
+        sub.OnAlarm(Alarm("A1", "", 900, now));
+        sub.OnAlarm(Alarm("A2", "", 900, now));
+        AwaitAssert(() => Assert.Equal(3, sink.Count)); // 1 seed + 1 per delta
+    }
+
    [Fact]
    public void Stop_Message_TearsDown_Grpc_And_Stops_Actor()
    {