feat(saf): resync ack confirmation + completed/ack-missing telemetry; doc resync rewrite (plan R2-02 T7)

docs/requirements/Component-StoreAndForward.md

@@ -80,7 +80,7 @@ There is **no maximum buffer size**. Messages accumulate in the buffer until del
 - The standby node applies the same operations to its own local SQLite database but is **passive**: it never runs the delivery sweep. The retry sweep is **gated to the active node** (the oldest Up member / singleton host, re-evaluated every sweep tick), so only one node delivers at a time. The standby applies replicated operations purely to keep its copy warm for a future failover.
 - On failover, the new active node has a near-complete copy of the buffer. In rare cases, the most recent operations may not have been replicated (e.g., a message added or removed just before failover). This can result in a few **duplicate deliveries** (message delivered but its `Remove` not yet replicated) or a few **missed retries** (message added but not replicated). Duplicate deliveries are therefore confined to the **failover window** — an in-flight delivery whose `Remove` had not yet replicated — and never occur in steady-state operation (the standby's gate keeps it from delivering the same rows). Both are acceptable trade-offs for the latency benefit.
 - On failover, the new active node's gate flips to active within one sweep interval and it resumes delivery from its local copy.
-- **Peer-join anti-entropy resync.** Asynchronous, no-ack replication keeps the standby warm in steady state, but a standby that was **down for an extended period** (a crash, a long maintenance window) misses every operation replicated while it was gone and would otherwise diverge from the active node's buffer forever. To close that gap, whenever a node **(re)tracks its peer**, a **standby** requests a full-buffer snapshot (`RequestSfBufferResync`); the **active** node answers with up to `MaxResyncRows` (10 000) of its oldest rows (`SfBufferSnapshot`), and the standby **replaces its entire local buffer** with that snapshot (`ReplaceAllAsync`, one transaction). Only the active node answers; only a standby applies (each side checks the repo-standard oldest-Up member active-node predicate — singleton-host semantics via the shared `ActiveNodeEvaluator`, the same predicate as the S&F delivery gate; N1 — safe-by-default to standby). Because replicated applies are **upserts** (see the replication apply path), any Add/Remove/Park that lands after the resync merges cleanly onto the resynced state — no primary-key conflict, no lost delta. If the buffer exceeds the 10 000-row cap the snapshot is flagged `Truncated` and the standby logs a Warning; the residual divergence beyond the cap drains naturally as the active node delivers.
+- **Peer-join anti-entropy resync (chunked, ack-confirmed).** Asynchronous, no-ack replication keeps the standby warm in steady state, but a standby that was **down for an extended period** (a crash, a long maintenance window) misses every operation replicated while it was gone and would otherwise diverge from the active node's buffer forever. To close that gap, whenever a node **(re)tracks its peer**, a **standby** requests a full-buffer snapshot (`RequestSfBufferResync`); the **active** node loads up to `MaxResyncRows` (10 000) of its oldest rows and answers with a **sequence of byte-budgeted chunks** (`SfBufferSnapshotChunk`), each carrying a shared `ResyncId`, a 1-based `Sequence`, and the `TotalChunks` count. Chunking is mandatory because the monolithic snapshot exceeds Akka remoting's **default 128 000-byte frame** for any realistic backlog and `BuildHocon` sets no override — a single oversized message is silently undeliverable (review 02 round 2, **N2 High**). Rows accumulate into a chunk until the estimated payload budget (`MaxResyncChunkBytes` = 64 000, ≈50% frame headroom) or the row cap (`MaxResyncChunkRows` = 200) is hit; a single row whose payload alone exceeds the budget ships solo with a Warning. The standby **assembles all chunks of one `ResyncId`** (a new `ResyncId` discards any stale partial assembly; a partial that never completes is dropped after `resyncAssemblyTimeout`, default 30 s, and counted a replication failure), then **replaces its entire local buffer** with the assembled snapshot (`ReplaceAllAsync`, one transaction) and returns a delivery confirmation (`SfBufferResyncAck`). The active node arms an ack window (`resyncAckTimeout`, default 60 s): an acknowledged resync increments `scadabridge.store_and_forward.resync.completed`; an unacknowledged one (lost chunks / dead peer) logs a Warning and increments `scadabridge.store_and_forward.resync.ack_missing` — closing N2's silent-loss mode (previously nothing counted a lost snapshot and nothing retried until the next peer-track). Only the active node answers; only a standby applies, and both sides re-check at apply time (a mid-flight active-flip aborts the wipe) — each side checks the repo-standard **oldest-Up member** active-node predicate (singleton-host semantics via the shared `ActiveNodeEvaluator`, the **same predicate as the S&F delivery gate**; review 02 round 2, **N1 Critical** — using cluster *leadership* here let a rolling restart of the lower-address node make the delivering node wipe its own live buffer). Because replicated applies are **upserts** (see the replication apply path), any Add/Remove/Park that lands after the resync merges cleanly onto the resynced state — no primary-key conflict, no lost delta; the one accepted exception is the rare **N5** orphan-row race (a replicated `Remove` ordered before the snapshot chunks can re-add the removed row, re-delivered once and self-corrected at the next resync — inherent to no-ack replication). If the buffer exceeds the 10 000-row cap the snapshot is flagged `Truncated` and the standby logs a Warning; the residual divergence beyond the cap drains naturally as the active node delivers. The legacy monolithic `SfBufferSnapshot` message + standby handler are **retained** for rolling-upgrade compatibility (an old active node's monolithic snapshot is still applied by a new standby).
 
 ### Operation Tracking Table (lives in Site Runtime, not here)
 

src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs

@@ -47,6 +47,14 @@ public static class ScadaBridgeTelemetry
         Meter.CreateCounter<long>("scadabridge.store_and_forward.replication.failures", unit: "1",
             description: "S&F buffer replication operations that failed to dispatch/deliver to the peer node");
 
+    private static readonly Counter<long> _sfResyncCompleted =
+        Meter.CreateCounter<long>("scadabridge.store_and_forward.resync.completed", unit: "1",
+            description: "S&F anti-entropy resyncs the standby acknowledged as applied");
+
+    private static readonly Counter<long> _sfResyncAckMissing =
+        Meter.CreateCounter<long>("scadabridge.store_and_forward.resync.ack_missing", unit: "1",
+            description: "S&F resyncs answered by the active node but never acknowledged by the standby within the ack window (lost chunks / dead peer)");
+
     /// <summary>
     /// Incremented each time a per-site live-alarm aggregator re-establishes its site-wide
     /// gRPC stream — a NodeA↔NodeB failover flip or a reconcile-driven reopen after the
@@ -107,6 +115,12 @@ public static class ScadaBridgeTelemetry
     /// <summary>Records one failed S&amp;F replication dispatch.</summary>
     public static void RecordReplicationFailure() => _replicationFailures.Add(1);
 
+    /// <summary>Records one acknowledged (applied) S&amp;F buffer resync.</summary>
+    public static void RecordSfResyncCompleted() => _sfResyncCompleted.Add(1);
+
+    /// <summary>Records one resync whose ack window expired.</summary>
+    public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.Add(1);
+
     /// <summary>Records that a site connection opened (increments the up-count gauge).</summary>
     public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp);
 

src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs

@@ -47,6 +47,15 @@ public class SiteReplicationActor : ReceiveActor, IWithTimers
     /// test seam; production default 30 s.</summary>
     private readonly TimeSpan _resyncAssemblyTimeout;
 
+    // ── Resync delivery confirmation (active side; actor-thread only) ──
+    private string? _pendingAckResyncId;
+    private const string ResyncAckTimerKey = "sf-resync-ack-timeout";
+
+    /// <summary>How long the active node waits for the standby's <see cref="SfBufferResyncAck"/>
+    /// before warning + counting the resync as unacknowledged (lost chunks / dead peer). Ctor
+    /// test seam; production default 60 s.</summary>
+    private readonly TimeSpan _resyncAckTimeout;
+
     /// <summary>
     /// Maximum rows an active node returns in a single anti-entropy resync snapshot.
     /// A standby whose buffer exceeded this (Truncated snapshot) resyncs the oldest
@@ -129,7 +138,8 @@ public class SiteReplicationActor : ReceiveActor, IWithTimers
         Func<bool>? isActiveOverride = null,
         SiteRuntimeOptions? options = null,
         TimeSpan? configFetchRetryDelay = null,
-        TimeSpan? resyncAssemblyTimeout = null)
+        TimeSpan? resyncAssemblyTimeout = null,
+        TimeSpan? resyncAckTimeout = null)
     {
         _storage = storage;
         _sfStorage = sfStorage;
@@ -140,6 +150,7 @@ public class SiteReplicationActor : ReceiveActor, IWithTimers
         _cluster = Cluster.Get(Context.System);
         _isActive = isActiveOverride ?? DefaultIsActive;
         _resyncAssemblyTimeout = resyncAssemblyTimeout ?? TimeSpan.FromSeconds(30);
+        _resyncAckTimeout = resyncAckTimeout ?? TimeSpan.FromSeconds(60);
         // UA2: bound the standby's replicated-config fetch retries. At least one
         // attempt always runs; the fixed inter-attempt delay is a test seam
         // (production default 2 s).
@@ -173,6 +184,26 @@ public class SiteReplicationActor : ReceiveActor, IWithTimers
         Receive<SfResyncSnapshotLoaded>(HandleSfResyncSnapshotLoaded);
         Receive<SfBufferSnapshotChunk>(HandleSfBufferSnapshotChunk);
         Receive<ResyncAssemblyTimedOut>(HandleResyncAssemblyTimedOut);
+        Receive<SfBufferResyncAck>(msg =>
+        {
+            if (msg.ResyncId == _pendingAckResyncId)
+            {
+                _pendingAckResyncId = null;
+                Timers.Cancel(ResyncAckTimerKey);
+            }
+            ScadaBridgeTelemetry.RecordSfResyncCompleted();
+            _logger.LogInformation("S&F resync {ResyncId} acknowledged by standby: {Rows} row(s) applied",
+                msg.ResyncId, msg.RowCount);
+        });
+        Receive<ResyncAckTimedOut>(msg =>
+        {
+            if (msg.ResyncId != _pendingAckResyncId) return;
+            _pendingAckResyncId = null;
+            ScadaBridgeTelemetry.RecordSfResyncAckMissing();
+            _logger.LogWarning(
+                "S&F resync {ResyncId} was never acknowledged within {Window} — snapshot chunks may have been lost (frame drop / dead peer); the next peer-track retries",
+                msg.ResyncId, _resyncAckTimeout);
+        });
         Receive<SfBufferSnapshot>(HandleSfBufferSnapshot); // legacy monolithic handler — retained for rolling compat
     }
 
@@ -491,7 +522,11 @@ public class SiteReplicationActor : ReceiveActor, IWithTimers
         _logger.LogInformation(
             "Answered S&F resync request with {Rows} row(s) in {Chunks} chunk(s), resyncId={ResyncId}",
             msg.Messages.Count, chunks.Count, resyncId);
-        // Task 7 registers the pending-ack entry here.
+        // Arm the ack window: absence of an SfBufferResyncAck within it surfaces as a
+        // Warning + counter (the silent-loss mode N2 flagged). Single-outstanding-resync
+        // bookkeeping: a new request supersedes by overwriting the id and restarting the timer.
+        _pendingAckResyncId = resyncId;
+        Timers.StartSingleTimer(ResyncAckTimerKey, new ResyncAckTimedOut(resyncId), _resyncAckTimeout);
     }
 
     /// <summary>
@@ -630,6 +665,9 @@ public class SiteReplicationActor : ReceiveActor, IWithTimers
 
     /// <summary>Internal: a partial chunk assembly for <paramref name="ResyncId"/> exceeded its window.</summary>
     internal sealed record ResyncAssemblyTimedOut(string ResyncId);
+
+    /// <summary>Internal: the active node's ack window for <paramref name="ResyncId"/> expired.</summary>
+    internal sealed record ResyncAckTimedOut(string ResyncId);
 }
 
 /// <summary>

tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs

@@ -1,4 +1,5 @@
 using System.Collections.Concurrent;
+using System.Diagnostics.Metrics;
 using Akka.Actor;
 using Akka.TestKit.Xunit2;
 using Microsoft.Extensions.Logging;
@@ -9,6 +10,7 @@ using ZB.MOM.WW.ScadaBridge.SiteRuntime.Deployment;
 using ZB.MOM.WW.ScadaBridge.SiteRuntime.Messages;
 using ZB.MOM.WW.ScadaBridge.SiteRuntime.Persistence;
 using ZB.MOM.WW.ScadaBridge.StoreAndForward;
+using ZB.MOM.WW.ScadaBridge.Commons.Observability;
 using ZB.MOM.WW.ScadaBridge.Commons.Types.Enums;
 
 namespace ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests.Actors;
@@ -386,6 +388,69 @@ akka {
         ExpectNoMsg(TimeSpan.FromMilliseconds(300));
     }
 
+    // ── R2 T7: active-side resync ack confirmation + telemetry ──
+    //
+    // NOTE (deviation from plan): the actor logs via Microsoft ILogger (NullLogger in
+    // tests), NOT Akka's EventStream, so the plan's EventFilter.Warning assertions can
+    // never observe these warnings. We observe the two OTel counters via a MeterListener
+    // instead — the equivalent, and stronger, observable signal.
+
+    [Fact]
+    public async Task ActiveNode_ReceivingAck_CountsResyncCompleted()
+    {
+        long completed = 0;
+        using var listener = ListenCounter("scadabridge.store_and_forward.resync.completed",
+            m => Interlocked.Add(ref completed, m));
+
+        await _sfStorage.EnqueueAsync(NewMessage("m1"));
+        var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromSeconds(30));
+        actor.Tell(new RequestSfBufferResync(), TestActor);
+        var chunk = ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
+
+        actor.Tell(new SfBufferResyncAck(chunk.ResyncId, 1), TestActor);
+
+        await AwaitAssertAsync(() =>
+        {
+            Assert.True(Interlocked.Read(ref completed) >= 1); // ack recorded the completion
+            return Task.CompletedTask;
+        }, TimeSpan.FromSeconds(5));
+    }
+
+    [Fact]
+    public async Task ActiveNode_MissingAck_WarnsAfterAckTimeout()
+    {
+        long ackMissing = 0;
+        using var listener = ListenCounter("scadabridge.store_and_forward.resync.ack_missing",
+            m => Interlocked.Add(ref ackMissing, m));
+
+        await _sfStorage.EnqueueAsync(NewMessage("m1"));
+        var actor = CreateResyncActor(isActive: () => true, ackTimeout: TimeSpan.FromMilliseconds(200));
+        actor.Tell(new RequestSfBufferResync(), TestActor);
+        ExpectMsg<SfBufferSnapshotChunk>(TimeSpan.FromSeconds(5));
+        // No ack is sent → the ack window expires and the resync is counted unacknowledged.
+
+        await AwaitAssertAsync(() =>
+        {
+            Assert.True(Interlocked.Read(ref ackMissing) >= 1);
+            return Task.CompletedTask;
+        }, TimeSpan.FromSeconds(5));
+    }
+
+    /// <summary>Attaches a <see cref="MeterListener"/> to a single ScadaBridge counter by name,
+    /// forwarding each recorded increment to <paramref name="onMeasurement"/>.</summary>
+    private static MeterListener ListenCounter(string instrumentName, Action<long> onMeasurement)
+    {
+        var listener = new MeterListener();
+        listener.InstrumentPublished = (inst, l) =>
+        {
+            if (inst.Meter.Name == ScadaBridgeTelemetry.MeterName && inst.Name == instrumentName)
+                l.EnableMeasurementEvents(inst);
+        };
+        listener.SetMeasurementEventCallback<long>((_, m, _, _) => onMeasurement(m));
+        listener.Start();
+        return listener;
+    }
+
     private static StoreAndForwardMessage NewSfMessage(string id) => new()
     {
         Id = id,
@@ -417,11 +482,12 @@ akka {
     };
 
     /// <summary>Constructs a <see cref="ResyncTestActor"/> with the given active-node check
-    /// (the resync chunk/ack tests Tell to and expect from <see cref="TestKit.TestActor"/>).</summary>
-    private IActorRef CreateResyncActor(Func<bool> isActive) =>
+    /// (the resync chunk/ack tests Tell to and expect from <see cref="TestKit.TestActor"/>).
+    /// <paramref name="ackTimeout"/> is the active-side ack window seam (T7).</summary>
+    private IActorRef CreateResyncActor(Func<bool> isActive, TimeSpan? ackTimeout = null) =>
         ActorOf(Props.Create(() => new ResyncTestActor(
             _storage, _sfStorage, _replicationService, SiteRole,
-            NullLogger<SiteReplicationActor>.Instance, CreateTestProbe().Ref, isActive)));
+            NullLogger<SiteReplicationActor>.Instance, CreateTestProbe().Ref, isActive, ackTimeout)));
 
     /// <summary>Test message: drives <see cref="SiteReplicationActor.OnPeerTracked"/> directly,
     /// standing in for the MemberUp→TryTrackPeer path (a single-node TestKit cannot form a real peer).</summary>
@@ -438,9 +504,10 @@ akka {
         public ResyncTestActor(
             SiteStorageService storage, StoreAndForwardStorage sfStorage,
             ReplicationService replicationService, string siteRole,
-            ILogger<SiteReplicationActor> logger, IActorRef peerProbe, Func<bool> isActive)
+            ILogger<SiteReplicationActor> logger, IActorRef peerProbe, Func<bool> isActive,
+            TimeSpan? ackTimeout = null)
             : base(storage, sfStorage, replicationService, siteRole, logger,
-                   configFetcher: null, isActiveOverride: isActive)
+                   configFetcher: null, isActiveOverride: isActive, resyncAckTimeout: ackTimeout)
         {
             _peerProbe = peerProbe;
             Receive<TriggerPeerTracked>(_ => OnPeerTracked());