diff --git a/docs/requirements/Component-StoreAndForward.md b/docs/requirements/Component-StoreAndForward.md index 53aa0564..f10f255f 100644 --- a/docs/requirements/Component-StoreAndForward.md +++ b/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) diff --git a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs index 1f8cc2ad..0d1474c2 100644 --- a/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs +++ b/src/ZB.MOM.WW.ScadaBridge.Commons/Observability/ScadaBridgeTelemetry.cs @@ -47,6 +47,14 @@ public static class ScadaBridgeTelemetry Meter.CreateCounter("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 _sfResyncCompleted = + Meter.CreateCounter("scadabridge.store_and_forward.resync.completed", unit: "1", + description: "S&F anti-entropy resyncs the standby acknowledged as applied"); + + private static readonly Counter _sfResyncAckMissing = + Meter.CreateCounter("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)"); + /// /// 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 /// Records one failed S&F replication dispatch. public static void RecordReplicationFailure() => _replicationFailures.Add(1); + /// Records one acknowledged (applied) S&F buffer resync. + public static void RecordSfResyncCompleted() => _sfResyncCompleted.Add(1); + + /// Records one resync whose ack window expired. + public static void RecordSfResyncAckMissing() => _sfResyncAckMissing.Add(1); + /// Records that a site connection opened (increments the up-count gauge). public static void SiteConnectionOpened() => Interlocked.Increment(ref _siteConnectionsUp); diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs index 49930016..63066d2d 100644 --- a/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs +++ b/src/ZB.MOM.WW.ScadaBridge.SiteRuntime/Actors/SiteReplicationActor.cs @@ -47,6 +47,15 @@ public class SiteReplicationActor : ReceiveActor, IWithTimers /// test seam; production default 30 s. private readonly TimeSpan _resyncAssemblyTimeout; + // ── Resync delivery confirmation (active side; actor-thread only) ── + private string? _pendingAckResyncId; + private const string ResyncAckTimerKey = "sf-resync-ack-timeout"; + + /// How long the active node waits for the standby's + /// before warning + counting the resync as unacknowledged (lost chunks / dead peer). Ctor + /// test seam; production default 60 s. + private readonly TimeSpan _resyncAckTimeout; + /// /// 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? 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(HandleSfResyncSnapshotLoaded); Receive(HandleSfBufferSnapshotChunk); Receive(HandleResyncAssemblyTimedOut); + Receive(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(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(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); } /// @@ -630,6 +665,9 @@ public class SiteReplicationActor : ReceiveActor, IWithTimers /// Internal: a partial chunk assembly for exceeded its window. internal sealed record ResyncAssemblyTimedOut(string ResyncId); + + /// Internal: the active node's ack window for expired. + internal sealed record ResyncAckTimedOut(string ResyncId); } /// diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs index 32a0a7b1..eb67ee4f 100644 --- a/tests/ZB.MOM.WW.ScadaBridge.SiteRuntime.Tests/Actors/SiteReplicationActorTests.cs +++ b/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(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(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)); + } + + /// Attaches a to a single ScadaBridge counter by name, + /// forwarding each recorded increment to . + private static MeterListener ListenCounter(string instrumentName, Action onMeasurement) + { + var listener = new MeterListener(); + listener.InstrumentPublished = (inst, l) => + { + if (inst.Meter.Name == ScadaBridgeTelemetry.MeterName && inst.Name == instrumentName) + l.EnableMeasurementEvents(inst); + }; + listener.SetMeasurementEventCallback((_, m, _, _) => onMeasurement(m)); + listener.Start(); + return listener; + } + private static StoreAndForwardMessage NewSfMessage(string id) => new() { Id = id, @@ -417,11 +482,12 @@ akka { }; /// Constructs a with the given active-node check - /// (the resync chunk/ack tests Tell to and expect from ). - private IActorRef CreateResyncActor(Func isActive) => + /// (the resync chunk/ack tests Tell to and expect from ). + /// is the active-side ack window seam (T7). + private IActorRef CreateResyncActor(Func isActive, TimeSpan? ackTimeout = null) => ActorOf(Props.Create(() => new ResyncTestActor( _storage, _sfStorage, _replicationService, SiteRole, - NullLogger.Instance, CreateTestProbe().Ref, isActive))); + NullLogger.Instance, CreateTestProbe().Ref, isActive, ackTimeout))); /// Test message: drives directly, /// standing in for the MemberUp→TryTrackPeer path (a single-node TestKit cannot form a real peer). @@ -438,9 +504,10 @@ akka { public ResyncTestActor( SiteStorageService storage, StoreAndForwardStorage sfStorage, ReplicationService replicationService, string siteRole, - ILogger logger, IActorRef peerProbe, Func isActive) + ILogger logger, IActorRef peerProbe, Func isActive, + TimeSpan? ackTimeout = null) : base(storage, sfStorage, replicationService, siteRole, logger, - configFetcher: null, isActiveOverride: isActive) + configFetcher: null, isActiveOverride: isActive, resyncAckTimeout: ackTimeout) { _peerProbe = peerProbe; Receive(_ => OnPeerTracked());