perf(comms): alarms-only seed, capped buffers, at-least-once audit pull

2026-08-14 21:10:19 -04:00
parent 1040dc0fcc
commit 2ce0ad7ed1
30 changed files with 1941 additions and 482 deletions
@@ -920,6 +920,149 @@ public class DebugStreamBridgeActorTests : TestKit
            DebugStreamBridgeActor.StabilityWindow = TimeSpan.FromSeconds(30);
        }
    }
+
+    // ── WP2.3: bounded pre-snapshot buffer, hard snapshot deadline, timeout hygiene ──
+
+    [Fact]
+    public void PreSnapshotBuffer_IsCapped_DropsOldest_AndCountsTheDrops()
+    {
+        // Before the cap a snapshot that never arrived buffered every live event on the
+        // CENTRAL node without limit — one wedged session on a chatty instance was enough
+        // to grow unbounded. Now the oldest are evicted and counted.
+        var ctx = CreateBridgeActor();
+        ctx.CommProbe.ExpectMsg<SiteEnvelope>(); // subscribe envelope; no snapshot is ever sent
+
+        const int cap = 20_000;
+        const int overflow = 250;
+        var before = Interlocked.Read(ref DebugStreamBridgeActor.TotalPreSnapshotDropped);
+
+        var t = DateTimeOffset.UtcNow;
+        for (var i = 0; i < cap + overflow; i++)
+        {
+            ctx.BridgeActor.Tell(new AttributeValueChanged(
+                InstanceName, "Modules.IO", $"Attr{i}", i, "Good", t.AddMilliseconds(i)));
+        }
+
+        // The overflow was evicted (drop-oldest) and counted.
+        AwaitCondition(
+            () => Interlocked.Read(ref DebugStreamBridgeActor.TotalPreSnapshotDropped) - before >= overflow,
+            TimeSpan.FromSeconds(10));
+
+        // The session is still healthy: the snapshot can still arrive and flush the
+        // (capped) buffer — the newest events, which the snapshot may predate, survived.
+        var snapshot = new DebugViewSnapshot(
+            InstanceName,
+            new List<AttributeValueChanged>(),
+            new List<AlarmStateChanged>(),
+            t.AddMilliseconds(-1));
+        ctx.BridgeActor.Tell(snapshot);
+
+        AwaitCondition(() =>
+        {
+            lock (ctx.ReceivedEvents) { return ctx.ReceivedEvents.Count >= cap; }
+        }, TimeSpan.FromSeconds(10));
+
+        lock (ctx.ReceivedEvents)
+        {
+            // Snapshot + exactly the retained (capped) events, and the newest survived.
+            Assert.Equal(cap + 1, ctx.ReceivedEvents.Count);
+            var lastAttr = ctx.ReceivedEvents.OfType<AttributeValueChanged>().Last();
+            Assert.Equal($"Attr{cap + overflow - 1}", lastAttr.AttributeName);
+        }
+    }
+
+    [Fact]
+    public void NoSnapshotWithinDeadline_FailsTheSession_InsteadOfBufferingForever()
+    {
+        // Nothing else ends a session wedged in the buffering phase: a lost site reply
+        // raises no gRPC error, and stream events no longer reset the orphan timeout.
+        DebugStreamBridgeActor.SnapshotTimeout = TimeSpan.FromMilliseconds(300);
+        try
+        {
+            var ctx = CreateBridgeActor();
+            ctx.CommProbe.ExpectMsg<SiteEnvelope>(); // subscribe request — never answered
+
+            Watch(ctx.BridgeActor);
+            ExpectTerminated(ctx.BridgeActor, TimeSpan.FromSeconds(5));
+
+            // The consumer is told, so the UI can surface the failure and reopen.
+            Assert.True(ctx.TerminatedFlag[0]);
+            // And the site-side relay was released rather than left as a zombie.
+            AwaitCondition(
+                () => ctx.MockGrpcClient.UnsubscribedCorrelationIds.Contains("corr-1"),
+                TimeSpan.FromSeconds(3));
+        }
+        finally
+        {
+            DebugStreamBridgeActor.SnapshotTimeout = TimeSpan.FromSeconds(60);
+        }
+    }
+
+    [Fact]
+    public void SnapshotArrival_StandsDownTheDeadline()
+    {
+        // The deadline must not fire after a healthy snapshot — a live session would
+        // otherwise be killed mid-stream.
+        DebugStreamBridgeActor.SnapshotTimeout = TimeSpan.FromMilliseconds(300);
+        try
+        {
+            var ctx = CreateBridgeActor();
+            ctx.CommProbe.ExpectMsg<SiteEnvelope>();
+
+            ctx.BridgeActor.Tell(new DebugViewSnapshot(
+                InstanceName,
+                new List<AttributeValueChanged>(),
+                new List<AlarmStateChanged>(),
+                DateTimeOffset.UtcNow));
+
+            Thread.Sleep(700); // well past the deadline
+            Assert.False(ctx.TerminatedFlag[0]);
+
+            // Still serving: a post-snapshot event passes straight through.
+            ctx.MockGrpcClient.SubscribeCalls[0].OnEvent(new AlarmStateChanged(
+                InstanceName, "PumpFault", Commons.Types.Enums.AlarmState.Active, 500,
+                DateTimeOffset.UtcNow));
+            AwaitCondition(() =>
+            {
+                lock (ctx.ReceivedEvents) { return ctx.ReceivedEvents.OfType<AlarmStateChanged>().Any(); }
+            }, TimeSpan.FromSeconds(3));
+        }
+        finally
+        {
+            DebugStreamBridgeActor.SnapshotTimeout = TimeSpan.FromSeconds(60);
+        }
+    }
+
+    [Fact]
+    public void StreamEvents_AreWrapped_SoTheyDoNotResetTheOrphanReceiveTimeout()
+    {
+        // Structural pin for the fix: the gRPC callback wraps every event in an envelope
+        // marked INotInfluenceReceiveTimeout, so a busy site can no longer keep an
+        // abandoned session alive indefinitely by feeding it events.
+        Assert.True(typeof(Akka.Actor.INotInfluenceReceiveTimeout)
+            .IsAssignableFrom(typeof(LiveDebugStreamEvent)));
+
+        var ctx = CreateBridgeActor();
+        ctx.CommProbe.ExpectMsg<SiteEnvelope>();
+        AwaitCondition(() => ctx.MockGrpcClient.SubscribeCalls.Count == 1, TimeSpan.FromSeconds(3));
+
+        ctx.BridgeActor.Tell(new DebugViewSnapshot(
+            InstanceName,
+            new List<AttributeValueChanged>(),
+            new List<AlarmStateChanged>(),
+            DateTimeOffset.UtcNow));
+
+        // An event delivered through the real gRPC callback path still reaches the
+        // consumer — the wrapper is transparent to delivery.
+        var evt = new AttributeValueChanged(
+            InstanceName, "Modules.IO", "Temperature", 42.5, "Good", DateTimeOffset.UtcNow);
+        ctx.MockGrpcClient.SubscribeCalls[0].OnEvent(evt);
+
+        AwaitCondition(() =>
+        {
+            lock (ctx.ReceivedEvents) { return ctx.ReceivedEvents.OfType<AttributeValueChanged>().Any(); }
+        }, TimeSpan.FromSeconds(3));
+    }
 }

 /// <summary>
@@ -79,11 +79,14 @@ public class SiteAlarmAggregatorActorTests : TestKit
        private readonly object _lock = new();
        public List<IReadOnlyList<AlarmStateChanged>> Snapshots { get; } = new();

-        public void Publish(IReadOnlyList<AlarmStateChanged> snapshot)
+        public void Publish(IReadOnlyList<AlarmStateChanged> snapshot, bool streamLive)
        {
-            lock (_lock) { Snapshots.Add(snapshot); }
+            lock (_lock) { Snapshots.Add(snapshot); StreamLive = streamLive; }
        }

+        /// <summary>Liveness reported with the most recent publish.</summary>
+        public bool StreamLive { get; private set; }
+
        public IReadOnlyList<AlarmStateChanged>? Latest
        {
            get { lock (_lock) { return Snapshots.Count == 0 ? null : Snapshots[^1]; } }
@@ -94,7 +97,7 @@ public class SiteAlarmAggregatorActorTests : TestKit

    private sealed record SiteSub(
        string CorrelationId, Action<AlarmStateChanged> OnAlarm, Action<Exception> OnError,
-        Action OnCompleted, CancellationToken Ct);
+        Action OnCompleted, CancellationToken Ct, Action? OnConnected);

    private sealed class MockSiteAlarmStreamClient : SiteStreamGrpcClient
    {
@@ -107,11 +110,21 @@ public class SiteAlarmAggregatorActorTests : TestKit

        public MockSiteAlarmStreamClient() : base() { }

+        /// <summary>When false the stream never reports connected — the site accepted the TCP
+        /// call but never answered, so the aggregator's connect-driven re-seed must not run.</summary>
+        public bool AutoConnect { get; set; } = true;
+
        public override Task SubscribeSiteAsync(
            string correlationId, Action<AlarmStateChanged> onAlarmEvent, Action<Exception> onError,
-            Action onCompleted, CancellationToken ct)
+            Action onCompleted, CancellationToken ct, Action? onConnected = null)
        {
-            lock (_lock) { _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, onCompleted, ct)); }
+            lock (_lock)
+            {
+                _subs.Add(new SiteSub(correlationId, onAlarmEvent, onError, onCompleted, ct, onConnected));
+            }
+            // A healthy site accepts the subscription immediately (it flushes response headers
+            // as soon as its relay is attached), which is the aggregator's connect signal.
+            if (AutoConnect) onConnected?.Invoke();
            var tcs = new TaskCompletionSource();
            ct.Register(() => tcs.TrySetResult());
            return tcs.Task; // never completes until cancelled (simulates a live stream)
@@ -130,8 +143,11 @@ public class SiteAlarmAggregatorActorTests : TestKit
        public MockSiteAlarmStreamClientFactory()
            : base(Microsoft.Extensions.Logging.Abstractions.NullLoggerFactory.Instance) { }

+        /// <summary>Applied to every client this factory hands out (set before the actor starts).</summary>
+        public bool AutoConnect { get; set; } = true;
+
        public MockSiteAlarmStreamClient ClientFor(string endpoint) =>
-            _byEndpoint.GetOrAdd(endpoint, _ => new MockSiteAlarmStreamClient());
+            _byEndpoint.GetOrAdd(endpoint, _ => new MockSiteAlarmStreamClient { AutoConnect = AutoConnect });

        public override SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint)
            => ClientFor(grpcEndpoint);
@@ -141,18 +157,19 @@ public class SiteAlarmAggregatorActorTests : TestKit
    }

    private (IActorRef Actor, SeedStub Seed, PublishSink Sink, MockSiteAlarmStreamClientFactory Factory) CreateActor(
-        TimeSpan? reconcileInterval = null, TimeSpan? publishCoalesce = null)
+        TimeSpan? reconcileInterval = null, TimeSpan? publishCoalesce = null, bool autoConnect = true)
    {
        var seed = new SeedStub();
        var sink = new PublishSink();
-        var factory = new MockSiteAlarmStreamClientFactory();
+        var factory = new MockSiteAlarmStreamClientFactory { AutoConnect = autoConnect };

        var props = Props.Create(() => new SiteAlarmAggregatorActor(
            SiteId, "corr-1", seed.Seed, sink.Publish, factory, GrpcNodeA, GrpcNodeB,
            reconcileInterval ?? TimeSpan.FromMinutes(10),
            publishCoalesce ?? TimeSpan.Zero,
            TestReconnectDelay,
-            TestStabilityWindow));
+            TestStabilityWindow,
+            0.0)); // no reconcile jitter in tests — determinism

        var actor = Sys.ActorOf(props);
        return (actor, seed, sink, factory);
@@ -414,12 +431,16 @@ public class SiteAlarmAggregatorActorTests : TestKit
        AwaitAssert(() => Assert.Equal(1, sink.Count));
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));

+        // The first tick after a fan-out is deliberately skipped (WP2.3: the initial seed
+        // already covered this window); the second one actually fans out.
+        actor.Tell(new RunReconcile());            // skipped — consumes the "already fanned out" flag
        actor.Tell(new RunReconcile());            // reconcile fan-out now in flight (CallCount 2)
        AwaitAssert(() => Assert.Equal(2, seed.CallCount));

-        // Stream error while the reconcile is in flight → the failover re-seed must not be
-        // silently skipped. Pre-fix: StartFanout no-ops and CallCount stays 2 until the next
-        // 60s reconcile tick.
+        // Stream error while the reconcile is in flight: the reconnect that follows
+        // reconnects (the mock accepts immediately) and its connect-driven re-seed must not
+        // be silently swallowed. Pre-fix: StartFanout no-ops and CallCount stays 2 until the
+        // next 60s reconcile tick.
        factory.ClientFor(GrpcNodeA).Subs.Last().OnError(new Exception("stream fault"));

        seed.CompleteNext();                       // finish the in-flight reconcile
@@ -577,4 +598,141 @@ public class SiteAlarmAggregatorActorTests : TestKit
        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Unsubscribed.Contains("corr-1"),
            TimeSpan.FromSeconds(3));
    }
+
+    // -- WP2.3: seed once per SUCCESSFUL (re)connect, and stream-truthful liveness --
+
+    [Fact]
+    public void ReconnectAttempts_DoNotEachReSeed_OnlyTheSuccessfulConnectDoes()
+    {
+        // Finding #10's per-attempt re-fan-out: every reconnect ATTEMPT used to kick a
+        // whole-site snapshot fan-out - against a site that is, by definition of the
+        // reconnect, unreachable. Now the re-seed is owed to the connect that succeeds.
+        var (_, seed, _, factory) = CreateActor(
+            reconcileInterval: TimeSpan.FromMinutes(10), autoConnect: false);
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3)); // initial seed
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
+        seed.CompleteNext();
+
+        // Three failed attempts, each flipping the node - none of them may re-seed.
+        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("1"));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 1, TimeSpan.FromSeconds(5));
+        factory.ClientFor(GrpcNodeB).Subs[0].OnError(new Exception("2"));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5));
+        factory.ClientFor(GrpcNodeA).Subs[1].OnError(new Exception("3"));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeB).Subs.Count == 2, TimeSpan.FromSeconds(5));
+
+        Thread.Sleep(300);
+        Assert.Equal(1, seed.CallCount); // pre-fix: 4
+
+        // The stream finally comes up - exactly ONE re-seed, on the connect.
+        factory.ClientFor(GrpcNodeB).Subs[1].OnConnected!();
+        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5));
+        seed.CompleteNext();
+        Thread.Sleep(300);
+        Assert.Equal(2, seed.CallCount); // and no second one
+    }
+
+    [Fact]
+    public void ReconcileTick_IsSkipped_WhenAFanoutAlreadyRanInTheWindow()
+    {
+        // The reconcile is a BACKSTOP, not an unconditional 60s whole-site snapshot: a
+        // window already covered by a seed costs nothing. Staleness stays bounded because
+        // the skip consumes the flag, so the next tick always fans out.
+        var (actor, seed, _, _) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+        seed.CompleteNext();
+        Thread.Sleep(200);
+
+        actor.Tell(new RunReconcile());   // skipped - the initial seed covered this window
+        Thread.Sleep(300);
+        Assert.Equal(1, seed.CallCount);
+
+        actor.Tell(new RunReconcile());   // and the next one runs
+        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
+    }
+
+    [Fact]
+    public void UnchangedReconcileSnapshot_DoesNotRepublish()
+    {
+        // A reconcile that finds nothing changed used to wake every viewer's render path
+        // once a minute regardless. It now publishes as a diff.
+        var (actor, seed, sink, _) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+
+        var t = DateTimeOffset.UtcNow;
+        seed.CompleteNext(Alarm("PumpFault", "", 500, t));
+        AwaitAssert(() => Assert.Equal(1, sink.Count));
+
+        actor.Tell(new RunReconcile()); // skipped (seed covered the window)
+        actor.Tell(new RunReconcile()); // real fan-out
+        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(3));
+        seed.CompleteNext(Alarm("PumpFault", "", 500, t)); // identical snapshot
+
+        Thread.Sleep(300);
+        Assert.Equal(1, sink.Count); // no second publish
+
+        // A genuinely changed snapshot still publishes.
+        actor.Tell(new RunReconcile());
+        actor.Tell(new RunReconcile());
+        AwaitCondition(() => seed.CallCount == 3, TimeSpan.FromSeconds(3));
+        seed.CompleteNext(Alarm("PumpFault", "", 900, t.AddSeconds(1)));
+        AwaitAssert(() => Assert.Equal(2, sink.Count));
+    }
+
+    [Fact]
+    public void StreamCompletion_DropsLiveness_AndTheReopenRestoresIt()
+    {
+        // WP2.3 carried residual: a completed (or given-up) stream kept reporting live
+        // until the next reconcile publish, so the page grafted a freezing snapshot over
+        // fresh poll data.
+        // Reconcile is held far away so the drop can be observed before the reopen
+        // restores liveness; the reopen is then driven explicitly.
+        var (actor, seed, sink, factory) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
+        seed.CompleteNext();
+        AwaitCondition(() => sink.Count >= 1 && sink.StreamLive, TimeSpan.FromSeconds(3));
+
+        // The site ends the stream cleanly (its 4h max lifetime).
+        factory.ClientFor(GrpcNodeA).Subs[0].OnCompleted();
+
+        // Liveness drops IMMEDIATELY - not at the next reconcile.
+        AwaitCondition(() => !sink.StreamLive, TimeSpan.FromSeconds(3));
+
+        // The reconcile tick reopens; the connect restores liveness.
+        actor.Tell(new RunReconcile());
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 2, TimeSpan.FromSeconds(5));
+        AwaitCondition(() => sink.StreamLive, TimeSpan.FromSeconds(5));
+    }
+
+    [Fact]
+    public void StreamFault_DropsLiveness_Immediately()
+    {
+        var (_, seed, sink, factory) = CreateActor(
+            reconcileInterval: TimeSpan.FromMinutes(10), autoConnect: false);
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+        AwaitCondition(() => factory.ClientFor(GrpcNodeA).Subs.Count == 1, TimeSpan.FromSeconds(3));
+        factory.ClientFor(GrpcNodeA).Subs[0].OnConnected!();
+        seed.CompleteNext();
+        AwaitCondition(() => sink.StreamLive, TimeSpan.FromSeconds(3));
+
+        factory.ClientFor(GrpcNodeA).Subs[0].OnError(new Exception("site gone"));
+        AwaitCondition(() => !sink.StreamLive, TimeSpan.FromSeconds(3));
+    }
+
+    [Fact]
+    public void FailedSeed_IsRetried_WithBackoff_BeforeTheNextReconcile()
+    {
+        // The seed leg had no backoff: a failing fan-out simply waited a full reconcile
+        // interval. Now it retries on its own timer (reconnectDelay, doubling).
+        var (_, seed, _, _) = CreateActor(reconcileInterval: TimeSpan.FromMinutes(10));
+        AwaitCondition(() => seed.CallCount == 1, TimeSpan.FromSeconds(3));
+
+        seed.FaultNext();
+        // TestReconnectDelay is 50 ms, so the first retry lands far inside the 10-minute
+        // reconcile interval - pre-fix nothing would run until that interval elapsed.
+        AwaitCondition(() => seed.CallCount == 2, TimeSpan.FromSeconds(5));
+        seed.FaultNext();
+        AwaitCondition(() => seed.CallCount == 3, TimeSpan.FromSeconds(5));
+    }
 }
@@ -530,4 +530,54 @@ public class SiteStreamGrpcServerTests : TestKit

        Assert.True(condition(), $"Condition not met within {timeoutMs}ms");
    }
+
+    // ── WP2.3: the site-wide alarm feed gets its OWN, larger send channel ──
+
+    [Fact]
+    public void SiteAlarmStream_HasItsOwnLargerChannel_ThanTheDebugView()
+    {
+        // Sharing the Debug View's 1000-slot DropOldest channel meant an alarm burst during
+        // a WAN stall silently evicted operator-visible transitions to make room for
+        // diagnostics traffic. The two feeds are now sized independently.
+        var options = Microsoft.Extensions.Options.Options.Create(new CommunicationOptions());
+        var server = new SiteStreamGrpcServer(_subscriber, _logger, options);
+
+        Assert.Equal(1000, server.InstanceChannelCapacity);
+        Assert.Equal(20_000, server.SiteAlarmChannelCapacity);
+        Assert.True(server.SiteAlarmChannelCapacity > server.InstanceChannelCapacity);
+    }
+
+    [Fact]
+    public void ChannelCapacities_AreBoundFromOptions_AndFloorAtOne()
+    {
+        var options = Microsoft.Extensions.Options.Options.Create(new CommunicationOptions
+        {
+            GrpcInstanceStreamChannelCapacity = 42,
+            GrpcSiteAlarmStreamChannelCapacity = 4242,
+        });
+        var server = new SiteStreamGrpcServer(_subscriber, _logger, options);
+
+        Assert.Equal(42, server.InstanceChannelCapacity);
+        Assert.Equal(4242, server.SiteAlarmChannelCapacity);
+
+        // A misconfigured zero/negative capacity must not throw at channel-construction
+        // time deep inside a live RPC — it floors at one instead.
+        var degenerate = new SiteStreamGrpcServer(_subscriber, _logger,
+            Microsoft.Extensions.Options.Options.Create(new CommunicationOptions
+            {
+                GrpcInstanceStreamChannelCapacity = 0,
+                GrpcSiteAlarmStreamChannelCapacity = -5,
+            }));
+        Assert.Equal(1, degenerate.InstanceChannelCapacity);
+        Assert.Equal(1, degenerate.SiteAlarmChannelCapacity);
+    }
+
+    [Fact]
+    public void DroppedStreamEventCount_StartsAtZero()
+    {
+        // The raw counter behind scadabridge.site.stream.events_dropped — a fresh node has
+        // evicted nothing.
+        var server = CreateServer();
+        Assert.Equal(0, server.DroppedStreamEventCount);
+    }
 }
@@ -26,11 +26,34 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
    // A mock site-wide alarm stream client whose subscription hangs until cancelled.
    private sealed class HangingClient : SiteStreamGrpcClient
    {
+        private readonly object _lock = new();
+        private readonly List<Action> _connects = new();
+        private readonly List<Action> _completions = new();
+
        public HangingClient() : base() { }
+
+        /// <summary>When false the site never accepts the subscription (no connect signal).</summary>
+        public bool AutoConnect { get; set; } = true;
+
+        /// <summary>Connect callbacks captured from every subscribe, for manual firing.</summary>
+        public List<Action> Connects { get { lock (_lock) { return _connects.ToList(); } } }
+
+        /// <summary>Graceful-completion callbacks captured from every subscribe.</summary>
+        public List<Action> Completions { get { lock (_lock) { return _completions.ToList(); } } }
+
        public override Task SubscribeSiteAsync(
            string correlationId, Action<Commons.Messages.Streaming.AlarmStateChanged> onAlarmEvent,
-            Action<Exception> onError, Action onCompleted, CancellationToken ct)
+            Action<Exception> onError, Action onCompleted, CancellationToken ct,
+            Action? onConnected = null)
        {
+            lock (_lock)
+            {
+                if (onConnected is not null) _connects.Add(onConnected);
+                _completions.Add(onCompleted);
+            }
+            // A healthy site accepts the subscription immediately (headers flushed on
+            // subscribe), which is what makes the aggregator report the stream live.
+            if (AutoConnect) onConnected?.Invoke();
            var tcs = new TaskCompletionSource();
            ct.Register(() => tcs.TrySetResult());
            return tcs.Task;
@@ -43,6 +66,9 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
        private readonly HangingClient _client = new();
        public int GetOrCreateCount;

+        /// <summary>The single client this factory hands out, for connect/complete control.</summary>
+        public HangingClient Client => _client;
+
        public CountingFactory() : base(NullLoggerFactory.Instance) { }

        public override SiteStreamGrpcClient GetOrCreate(string siteIdentifier, string grpcEndpoint)
@@ -338,4 +364,43 @@ public class SiteAlarmLiveCacheServiceTests : TestKit
        Thread.Sleep(300);
        Assert.Equal(startsAfterStop, factory.GetOrCreateCount); // no self-heal restart fired
    }
+
+    // ── WP2.3: IsLive tracks the STREAM, not merely "a snapshot was published once" ──
+
+    [Fact]
+    public void IsLive_StaysFalse_UntilTheSiteAcceptsTheStream()
+    {
+        var service = CreateService(TimeSpan.FromMilliseconds(200), out var factory);
+        factory.Client.AutoConnect = false; // site never answers the subscription
+
+        using var sub = service.Subscribe(SiteId, () => { });
+
+        // The (empty) seed completes and publishes, but with no accepted stream behind it
+        // the cache is not live — the page must keep polling. Pre-fix IsLive flipped true
+        // here and the page grafted a never-updating snapshot over fresh poll data.
+        AwaitCondition(() => service.GetCurrentAlarms(SiteId) is not null, TimeSpan.FromSeconds(5));
+        Thread.Sleep(400);
+        Assert.False(service.IsLive(SiteId));
+
+        // The site accepts it → live.
+        AwaitCondition(() => factory.Client.Connects.Count >= 1, TimeSpan.FromSeconds(5));
+        factory.Client.Connects[0]();
+        AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5));
+    }
+
+    [Fact]
+    public void IsLive_DropsWhenTheStreamCompletes()
+    {
+        // The carried residual from Phase 1: a stream that ended gracefully (the site's 4h
+        // max lifetime) left IsLive true until the next reconcile publish.
+        var service = CreateService(TimeSpan.FromMilliseconds(200), out var factory);
+
+        using var sub = service.Subscribe(SiteId, () => { });
+        AwaitCondition(() => service.IsLive(SiteId), TimeSpan.FromSeconds(5));
+
+        AwaitCondition(() => factory.Client.Completions.Count >= 1, TimeSpan.FromSeconds(5));
+        factory.Client.Completions[0]();
+
+        AwaitCondition(() => !service.IsLive(SiteId), TimeSpan.FromSeconds(5));
+    }
 }
@@ -13,9 +13,10 @@ using ZB.MOM.WW.ScadaBridge.Communication.Grpc;
 namespace ZB.MOM.WW.ScadaBridge.Communication.Tests;

 /// <summary>
-/// Bundle A A2 tests for <see cref="SiteStreamGrpcServer.PullAuditEvents"/>.
-/// Verifies the request → ISiteAuditQueue.ReadPendingSinceAsync → response →
-/// MarkReconciledAsync round-trip through the gRPC handler. The queue is an
+/// Tests for <see cref="SiteStreamGrpcServer.PullAuditEvents"/>: the request →
+/// <c>ISiteAuditQueue.ReadPendingSinceAsync</c> → response round-trip, plus the WP2.3
+/// at-least-once contract — rows are retired by the NEXT pull's cursor
+/// (<c>MarkReconciledUpToAsync</c>), never by the act of serving them. The queue is an
 /// NSubstitute stub so the tests never touch SQLite.
 /// </summary>
 public class SiteStreamPullAuditEventsTests : TestKit
@@ -63,11 +64,12 @@ public class SiteStreamPullAuditEventsTests : TestKit
    }

    [Fact]
-    public async Task PullAuditEvents_With5PendingRows_ReturnsAllFiveDtos_AndFlipsToReconciled()
+    public async Task PullAuditEvents_With5PendingRows_ReturnsAllFiveDtos_AndDoesNotFlipThem()
    {
        var queue = Substitute.For<ISiteAuditQueue>();
        var events = Enumerable.Range(0, 5).Select(_ => NewEvent()).ToList();
-        queue.ReadPendingSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+        queue.ReadPendingSinceAsync(
+                Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
            .Returns((IReadOnlyList<AuditEvent>)events);

        var server = CreateServer();
@@ -86,11 +88,100 @@ public class SiteStreamPullAuditEventsTests : TestKit
        var expectedIds = events.Select(e => e.EventId.ToString()).ToHashSet();
        Assert.True(expectedIds.SetEquals(response.Events.Select(d => d.EventId).ToHashSet()));

-        // Verify MarkReconciledAsync received the same 5 ids (best-effort flip).
-        await queue.Received(1).MarkReconciledAsync(
-            Arg.Is<IReadOnlyList<Guid>>(ids => ids.Count == 5 &&
-                                               ids.ToHashSet().SetEquals(events.Select(e => e.EventId))),
-            Arg.Any<CancellationToken>());
+        // AT-LEAST-ONCE: serving rows is NOT proof of receipt. The per-id flip is gone
+        // entirely; only a later cursor retires rows.
+        await queue.DidNotReceive().MarkReconciledAsync(
+            Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task PullAuditEvents_FaultBetweenResponseAndNextPull_ReservesTheSameRows()
+    {
+        // The failure this closes: central receives the batch, then dies before committing
+        // it, so its cursor never advances. Pre-fix the site had already flipped the rows to
+        // Reconciled while serving them, and ReadPendingSinceAsync would never return them
+        // again — the rows were silently lost. Now the unchanged cursor means no flip, and
+        // the identical batch is served again.
+        var queue = Substitute.For<ISiteAuditQueue>();
+        var since = DateTime.SpecifyKind(new DateTime(2026, 5, 20, 9, 30, 0), DateTimeKind.Utc);
+        var events = Enumerable.Range(0, 3).Select(_ => NewEvent()).ToList();
+        queue.ReadPendingSinceAsync(
+                Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+            .Returns((IReadOnlyList<AuditEvent>)events);
+
+        var server = CreateServer();
+        server.SetSiteAuditQueue(queue);
+
+        var request = new PullAuditEventsRequest
+        {
+            SinceUtc = Timestamp.FromDateTime(since),
+            BatchSize = 100,
+        };
+
+        var first = await server.PullAuditEvents(request, NewContext());
+        // …central faults here; it never commits, so it re-pulls with the SAME cursor.
+        var second = await server.PullAuditEvents(request, NewContext());
+
+        Assert.Equal(3, first.Events.Count);
+        Assert.Equal(
+            first.Events.Select(e => e.EventId).ToHashSet(),
+            second.Events.Select(e => e.EventId).ToHashSet());
+
+        // Neither pull retired anything past the (unchanged) cursor: the flip is bounded by
+        // the cursor value, so replaying the same cursor can never retire the served rows.
+        await queue.Received(2).MarkReconciledUpToAsync(
+            since, null, Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task PullAuditEvents_AdvancedCursor_RetiresEverythingUpToIt_BeforeReading()
+    {
+        // The cursor central sends back IS the receipt: everything at or before it has been
+        // ingested, so those rows are flipped — and flipped BEFORE the read, so they do not
+        // consume this batch's budget.
+        var queue = Substitute.For<ISiteAuditQueue>();
+        queue.ReadPendingSinceAsync(
+                Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+            .Returns((IReadOnlyList<AuditEvent>)Array.Empty<AuditEvent>());
+
+        var server = CreateServer();
+        server.SetSiteAuditQueue(queue);
+
+        var cursorTime = DateTime.SpecifyKind(new DateTime(2026, 5, 20, 10, 0, 0), DateTimeKind.Utc);
+        var cursorId = Guid.NewGuid().ToString();
+        var request = new PullAuditEventsRequest
+        {
+            SinceUtc = Timestamp.FromDateTime(cursorTime),
+            BatchSize = 100,
+            AfterId = cursorId,
+        };
+
+        await server.PullAuditEvents(request, NewContext());
+
+        await queue.Received(1).MarkReconciledUpToAsync(
+            cursorTime, cursorId, Arg.Any<CancellationToken>());
+        // The keyset cursor is passed straight through to the read as well.
+        await queue.Received(1).ReadPendingSinceAsync(
+            cursorTime, 100, cursorId, Arg.Any<CancellationToken>());
+    }
+
+    [Fact]
+    public async Task PullAuditEvents_FirstEverPull_DoesNotFlipAnything()
+    {
+        // since == MinValue means "from the beginning of recorded history" — central has
+        // consumed nothing yet, so there is nothing to retire.
+        var queue = Substitute.For<ISiteAuditQueue>();
+        queue.ReadPendingSinceAsync(
+                Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
+            .Returns((IReadOnlyList<AuditEvent>)Array.Empty<AuditEvent>());
+
+        var server = CreateServer();
+        server.SetSiteAuditQueue(queue);
+
+        await server.PullAuditEvents(new PullAuditEventsRequest { BatchSize = 10 }, NewContext());
+
+        await queue.DidNotReceive().MarkReconciledUpToAsync(
+            Arg.Any<DateTime>(), Arg.Any<string?>(), Arg.Any<CancellationToken>());
    }

    [Fact]
@@ -102,7 +193,8 @@ public class SiteStreamPullAuditEventsTests : TestKit
        // yields an empty gRPC response.
        var queue = Substitute.For<ISiteAuditQueue>();
        var capturedSince = DateTime.MinValue;
-        queue.ReadPendingSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+        queue.ReadPendingSinceAsync(
+                Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
            .Returns(call =>
            {
                capturedSince = call.ArgAt<DateTime>(0);
@@ -124,9 +216,6 @@ public class SiteStreamPullAuditEventsTests : TestKit
        Assert.Empty(response.Events);
        Assert.False(response.MoreAvailable);
        Assert.Equal(since, capturedSince);
-        // Empty result → no MarkReconciledAsync call (no rows to flip).
-        await queue.DidNotReceive().MarkReconciledAsync(
-            Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>());
    }

    [Fact]
@@ -134,7 +223,8 @@ public class SiteStreamPullAuditEventsTests : TestKit
    {
        var queue = Substitute.For<ISiteAuditQueue>();
        var events = Enumerable.Range(0, 3).Select(_ => NewEvent()).ToList();
-        queue.ReadPendingSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+        queue.ReadPendingSinceAsync(
+                Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
            .Returns((IReadOnlyList<AuditEvent>)events);

        var server = CreateServer();
@@ -154,16 +244,17 @@ public class SiteStreamPullAuditEventsTests : TestKit
    }

    [Fact]
-    public async Task PullAuditEvents_MarkReconciledThrows_ResponseStillReturned()
+    public async Task PullAuditEvents_MarkReconciledUpToThrows_ResponseStillReturned()
    {
-        // The Reconciled flip is best-effort — if it fails, the response must
-        // still surface so central can ingest the rows (and dedup on EventId
-        // when it pulls them again).
+        // The retire step is best-effort — if it fails, the pull must still serve rows.
+        // Worst case the same rows are shipped again and central dedups on EventId.
        var queue = Substitute.For<ISiteAuditQueue>();
        var events = Enumerable.Range(0, 2).Select(_ => NewEvent()).ToList();
-        queue.ReadPendingSinceAsync(Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<CancellationToken>())
+        queue.ReadPendingSinceAsync(
+                Arg.Any<DateTime>(), Arg.Any<int>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
            .Returns((IReadOnlyList<AuditEvent>)events);
-        queue.MarkReconciledAsync(Arg.Any<IReadOnlyList<Guid>>(), Arg.Any<CancellationToken>())
+        queue.MarkReconciledUpToAsync(
+                Arg.Any<DateTime>(), Arg.Any<string?>(), Arg.Any<CancellationToken>())
            .ThrowsAsync(new InvalidOperationException("SQLite disposed mid-call"));

        var server = CreateServer();
@@ -175,8 +266,6 @@ public class SiteStreamPullAuditEventsTests : TestKit
            BatchSize = 100,
        };

-        // Must NOT throw — the response is built before the flip and returned
-        // regardless of the flip outcome.
        var response = await server.PullAuditEvents(request, NewContext());

        Assert.Equal(2, response.Events.Count);