fix(localdb): FailClosedOnDrift applies to snapshot rows (scadaproj#39)

The drift guard lived only in the delta path; SnapshotApplier applied every row unchecked, so a future-stamped snapshot row was applied, observed by the clock and persisted as last_seen_hlc - moving the receiver's HLC ahead across restarts despite FailClosedOnDrift (LocalDb-004). SnapshotApplier now takes ReplicationOptions and runs the same per-row check (warn once per snapshot; dead-letter under FailClosedOnDrift via the shared HlcDriftGuard, with the row's HLC kept out of last_seen_hlc). Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
2026-11-09 15:01:01 -05:00
parent 15a6c7c710
commit c207573595
9 changed files with 258 additions and 33 deletions
@@ -10,6 +10,7 @@ Shared embedded SQLite cache + 2-node LWW sync library for the **ZB.MOM.WW SCADA
 - **#36 LocalDb-001:** the passive `LocalDbSyncService` now enforces `LocalDb:Replication:ApiKey` itself (Bearer, constant-time, checked before the handshake and before newest-wins takeover; `Unauthenticated` otherwise). Blank key = unchanged behaviour plus a startup warning. ScadaBridge's and OtOpcUa's own `LocalDbSyncAuthInterceptor`s use the same key and header, so they keep working; they are now redundant except for their fail-closed-when-blank rule.
 - **#37 LocalDb-002:** snapshots deliver FK-linked rows in any order. `SnapshotStreamer` sends referenced tables first (Kahn order, name tie-break); `SnapshotApplier` defers rows that fail only SQLITE_CONSTRAINT_FOREIGNKEY (`LwwApplier.ApplyBatchAsync(..., deferForeignKeyViolations: true)`), retries them as rows land, and dead-letters true orphans at `SnapshotComplete`. Before, a child that arrived first was dead-lettered and acked over permanently.
 - **#38 LocalDb-003:** the applier upserts with `INSERT … ON CONFLICT(<pk>) DO UPDATE` (`DO NOTHING` when every column is PK) instead of `INSERT OR REPLACE`. A secondary-`UNIQUE` clash is now a dead-letter instead of a silent delete of another row; as a side effect a remote update of a parent with `ON DELETE RESTRICT` children applies instead of being dead-lettered.
+- **#39 LocalDb-004:** `SnapshotApplier` now takes `ReplicationOptions` (wired by `SyncSessionFactory`) and runs the delta path's drift guard per snapshot row (shared `HlcDriftGuard` dead-letter); a dead-lettered row's HLC is kept out of `last_seen_hlc`. Before, `FailClosedOnDrift` was skipped on snapshots and one future-stamped row moved the receiver's HLC ahead across restarts.

 ## Index history (migrated from the scadaproj root CLAUDE.md)

@@ -283,7 +283,11 @@ All options are validated at startup (`ZB.MOM.WW.Configuration` validators).
 - **Clock-drift guard** — warn-once per session when a peer entry's HLC physical time is
  more than `MaxHlcDriftAhead` ahead. With `FailClosedOnDrift`, each over-drift entry is
  dead-lettered; otherwise it is applied (the HLC merge absorbs the skew). The **session
-  never dies from drift** — no livelock on reconnect.
+  never dies from drift** — no livelock on reconnect. Since `0.3.6` the guard covers
+  **snapshot rows** too (warn once per snapshot; dead-lettered under `FailClosedOnDrift`, and
+  their HLC is neither observed nor persisted as `last_seen_hlc`). Up to `0.3.5` a snapshot
+  applied every row unchecked, so one future-stamped row could push the receiver's clock
+  ahead permanently.
 - **Writes never block on the peer**; local commit is independent of replication.
 - **All timestamps are UTC** (HLC physical component is UTC Unix milliseconds).

@@ -0,0 +1,33 @@
+using System.Globalization;
+using ZB.MOM.WW.LocalDb.Internal;
+
+namespace ZB.MOM.WW.LocalDb.Replication.Internal;
+
+/// <summary>
+/// The fail-closed disposition for an inbound row whose HLC runs further ahead of local UTC than
+/// <see cref="ReplicationOptions.MaxHlcDriftAhead"/> allows. Shared by the delta path
+/// (<see cref="SyncSession"/>) and the snapshot path (<see cref="SnapshotApplier"/>) so both write
+/// the same dead-letter row (LocalDb-004).
+/// </summary>
+internal static class HlcDriftGuard
+{
+    /// <summary>The marker every drift dead-letter's <c>error</c> starts with.</summary>
+    public const string ErrorPrefix = "HLC drift ahead";
+
+    public static Task DeadLetterAsync(
+        SqliteLocalDb db, OplogEntryRecord entry, long physicalMs, long nowMs, DateTimeOffset now,
+        TimeSpan maxDriftAhead, CancellationToken ct) =>
+        db.ExecuteAsync(
+            "INSERT INTO __localdb_dead_letter (received_utc, table_name, pk_json, row_json, hlc, node_id, error) " +
+            "VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)",
+            new
+            {
+                utc = now.UtcDateTime.ToString(OplogStore.TombstoneUtcFormat, CultureInfo.InvariantCulture),
+                t = entry.TableName,
+                pk = entry.PkJson,
+                row = entry.RowJson,
+                hlc = entry.Hlc,
+                node = entry.NodeId,
+                err = $"{ErrorPrefix}: entry physical time {physicalMs} ms (Unix UTC) exceeds local now {nowMs} ms + MaxHlcDriftAhead {maxDriftAhead} (fail-closed).",
+            }, ct);
+}
@@ -1,5 +1,6 @@
 using Microsoft.Extensions.Logging;
 using ZB.MOM.WW.LocalDb.Contracts;
+using ZB.MOM.WW.LocalDb.Hlc;
 using ZB.MOM.WW.LocalDb.Internal;
 using ZB.MOM.WW.LocalDb.Registration;

@@ -246,9 +247,13 @@ internal sealed class SnapshotStreamer(SqliteLocalDb db, OplogStore store, Repli
 /// </summary>
 internal sealed class SnapshotApplier : ISnapshotApplier
 {
+    private readonly SqliteLocalDb _db;
    private readonly LwwApplier _applier;
    private readonly OplogStore _store;
    private readonly ILogger _logger;
+    private readonly ReplicationOptions _options;
+    private readonly Func<DateTimeOffset> _utcNow;
+    private bool _driftWarned;
    private long _asOfSeq;
    private long _rows;
    private long _maxHlc;
@@ -260,14 +265,27 @@ internal sealed class SnapshotApplier : ISnapshotApplier
    // sender's debt clears only on the SnapshotAck, so the whole snapshot is re-streamed.
    private List<OplogEntryRecord> _pending = [];

-    // db is part of the collaborator's canonical shape (mirrors the streamer) even though the merge
-    // reaches the database through the applier and store; kept for construction symmetry in the factory.
-    public SnapshotApplier(SqliteLocalDb db, LwwApplier applier, OplogStore store, ILogger logger)
+    /// <param name="db">The receiving database (dead-letter target for over-drift rows).</param>
+    /// <param name="applier">The shared LWW applier.</param>
+    /// <param name="store">The oplog/peer-state store.</param>
+    /// <param name="logger">Logger.</param>
+    /// <param name="options">
+    /// Supplies the drift guard (<see cref="ReplicationOptions.MaxHlcDriftAhead"/> /
+    /// <see cref="ReplicationOptions.FailClosedOnDrift"/>), applied to snapshot rows exactly as the
+    /// delta path applies it (LocalDb-004). Defaults to <see cref="ReplicationOptions"/>' defaults:
+    /// warn, never dead-letter.
+    /// </param>
+    /// <param name="utcNow">Test seam for the drift clock.</param>
+    public SnapshotApplier(
+        SqliteLocalDb db, LwwApplier applier, OplogStore store, ILogger logger,
+        ReplicationOptions? options = null, Func<DateTimeOffset>? utcNow = null)
    {
-        _ = db;
+        _db = db;
        _applier = applier;
        _store = store;
        _logger = logger;
+        _options = options ?? new ReplicationOptions();
+        _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
    }

    public Task OnBeginAsync(SnapshotBegin begin, CancellationToken ct)
@@ -276,6 +294,7 @@ internal sealed class SnapshotApplier : ISnapshotApplier
        _rows = 0;
        _maxHlc = 0;
        _pending = [];
+        _driftWarned = false;
        // A non-empty table list scopes the snapshot; empty (which is all a pre-0.3.0 sender can
        // emit) means every registered table. Read here rather than on complete because
        // SnapshotComplete does not repeat it — the receive loop guarantees a Begin precedes both.
@@ -291,12 +310,43 @@ internal sealed class SnapshotApplier : ISnapshotApplier
        if (batch.Rows.Count == 0)
            return;

+        // The same drift guard the delta path runs (SyncSession.ApplyDeltaEntriesAsync). Without it a
+        // row stamped by a peer whose clock ran far ahead was applied, LwwApplier observed its HLC,
+        // and OnCompleteAsync persisted it as last_seen_hlc — every later local write on this node
+        // then carried the future stamp, across restarts, defeating FailClosedOnDrift (LocalDb-004).
+        var now = _utcNow();
+        var nowMs = now.ToUnixTimeMilliseconds();
+        var driftLimitMs = nowMs + (long)_options.MaxHlcDriftAhead.TotalMilliseconds;
+
        var entries = new List<OplogEntryRecord>(batch.Rows.Count);
        foreach (var row in batch.Rows)
        {
+            var entry = new OplogEntryRecord(
+                0, batch.TableName, row.PkJson, row.HasRowJson ? row.RowJson : null, row.Hlc, row.NodeId, row.IsTombstone);
+            var physicalMs = HybridLogicalClock.PhysicalMs(row.Hlc);
+            if (physicalMs > driftLimitMs)
+            {
+                if (!_driftWarned)
+                {
+                    _driftWarned = true;
+                    _logger.LogWarning(
+                        "Snapshot row HLC physical time {EntryMs} ms is ahead of local UTC {NowMs} ms by more than MaxHlcDriftAhead ({Drift}). Further drift in this snapshot is not re-logged.",
+                        physicalMs, nowMs, _options.MaxHlcDriftAhead);
+                }
+                if (_options.FailClosedOnDrift)
+                {
+                    // Skipped rows contribute nothing to _maxHlc, so last_seen_hlc stays honest.
+                    await HlcDriftGuard.DeadLetterAsync(_db, entry, physicalMs, nowMs, now, _options.MaxHlcDriftAhead, ct);
+                    continue;
+                }
+            }
            if (row.Hlc > _maxHlc) _maxHlc = row.Hlc;
-            entries.Add(new OplogEntryRecord(
-                0, batch.TableName, row.PkJson, row.HasRowJson ? row.RowJson : null, row.Hlc, row.NodeId, row.IsTombstone));
+            entries.Add(entry);
+        }
+        if (entries.Count == 0)
+        {
+            _rows += batch.Rows.Count;
+            return;
        }
        var result = await _applier.ApplyBatchAsync(entries, ct, deferForeignKeyViolations: true);
        _rows += batch.Rows.Count;
@@ -1,4 +1,3 @@
-using System.Globalization;
 using System.Threading.Channels;
 using Microsoft.Extensions.Logging;
 using ZB.MOM.WW.LocalDb.Contracts;
@@ -649,19 +648,7 @@ internal sealed class SyncSession
    // dead-letters the same entry again (at-least-once — a benign duplicate dead-letter row).
    private Task DeadLetterDriftedAsync(
        OplogEntryRecord entry, long physicalMs, long nowMs, DateTimeOffset now, CancellationToken ct) =>
-        _db.ExecuteAsync(
-            "INSERT INTO __localdb_dead_letter (received_utc, table_name, pk_json, row_json, hlc, node_id, error) " +
-            "VALUES (@utc, @t, @pk, @row, @hlc, @node, @err)",
-            new
-            {
-                utc = now.UtcDateTime.ToString(OplogStore.TombstoneUtcFormat, CultureInfo.InvariantCulture),
-                t = entry.TableName,
-                pk = entry.PkJson,
-                row = entry.RowJson,
-                hlc = entry.Hlc,
-                node = entry.NodeId,
-                err = $"HLC drift ahead: entry physical time {physicalMs} ms (Unix UTC) exceeds local now {nowMs} ms + MaxHlcDriftAhead {_options.MaxHlcDriftAhead} (fail-closed).",
-            }, ct);
+        HlcDriftGuard.DeadLetterAsync(_db, entry, physicalMs, nowMs, now, _options.MaxHlcDriftAhead, ct);

    private async Task DrainSecondaryAsync(Task task)
    {
@@ -31,7 +31,7 @@ internal static class SyncSessionFactory

        var streamer = new SnapshotStreamer(db, store, options, loggerFactory.CreateLogger<SnapshotStreamer>());
        session.SnapshotSender = streamer.SendAsync;
-        session.SnapshotApplier = new SnapshotApplier(db, applier, store, loggerFactory.CreateLogger<SnapshotApplier>());
+        session.SnapshotApplier = new SnapshotApplier(db, applier, store, loggerFactory.CreateLogger<SnapshotApplier>(), options);
        return session;
    }
 }
@@ -0,0 +1,135 @@
+using System.Threading.Channels;
+using Microsoft.Extensions.Logging.Abstractions;
+using ZB.MOM.WW.LocalDb.Contracts;
+using ZB.MOM.WW.LocalDb.Internal;
+using ZB.MOM.WW.LocalDb.Replication;
+using ZB.MOM.WW.LocalDb.Replication.Internal;
+
+namespace ZB.MOM.WW.LocalDb.Tests;
+
+/// <summary>
+/// LocalDb-004 (scadaproj#39): <see cref="ReplicationOptions.FailClosedOnDrift"/> must hold on the
+/// snapshot path too. It was checked only for deltas, so a future-stamped snapshot row was applied,
+/// pushed the receiver's HLC a year ahead and persisted that in <c>last_seen_hlc</c>.
+/// </summary>
+public sealed class SnapshotDriftTests : IDisposable
+{
+    private static readonly TimeSpan RunTimeout = TimeSpan.FromSeconds(15);
+
+    private readonly List<string> _paths = [];
+    private readonly List<IDisposable> _disposables = [];
+
+    public void Dispose()
+    {
+        foreach (var d in _disposables)
+            d.Dispose();
+        Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools();
+        foreach (var p in _paths)
+        {
+            if (File.Exists(p)) File.Delete(p);
+            if (File.Exists(p + "-wal")) File.Delete(p + "-wal");
+            if (File.Exists(p + "-shm")) File.Delete(p + "-shm");
+        }
+    }
+
+    private sealed record Side(SqliteLocalDb Db, OplogStore Store, SyncSession Session);
+
+    private async Task<SqliteLocalDb> NewDbAsync()
+    {
+        var path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
+        _paths.Add(path);
+        var db = new SqliteLocalDb(new LocalDbOptions { Path = path });
+        _disposables.Add(db);
+        await db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)");
+        db.RegisterReplicated("orders");
+        return db;
+    }
+
+    // Built exactly as production does (SyncSessionFactory), so the test covers the real wiring of
+    // the snapshot applier — the place the options were missing.
+    private static Side SessionOver(SqliteLocalDb db, ReplicationOptions options) =>
+        new(db, new OplogStore(db, options), SyncSessionFactory.Create(db, options, NullLoggerFactory.Instance));
+
+    private static (SyncDuplex A, SyncDuplex B) DuplexPair()
+    {
+        var aToB = Channel.CreateUnbounded<SyncMessage>();
+        var bToA = Channel.CreateUnbounded<SyncMessage>();
+        return (
+            new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader },
+            new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader });
+    }
+
+    private static long HlcAt(DateTimeOffset utc) => utc.ToUnixTimeMilliseconds() << 16;
+
+    private static async Task<List<long>> Ids(SqliteLocalDb db) =>
+        (await db.QueryAsync("SELECT id FROM orders ORDER BY id", r => r.GetInt64(0))).ToList();
+
+    private static async Task<long> Scalar(SqliteLocalDb db, string sql) =>
+        (await db.QueryAsync(sql, r => r.GetInt64(0)))[0];
+
+    private static async Task WaitForAsync(Func<Task<bool>> predicate, TimeSpan timeout)
+    {
+        var deadline = DateTime.UtcNow + timeout;
+        while (DateTime.UtcNow < deadline)
+        {
+            if (await predicate()) return;
+            await Task.Delay(20);
+        }
+        throw new TimeoutException("Condition not reached within " + timeout);
+    }
+
+    private static async Task SwallowAsync(Task task)
+    {
+        try { await task; }
+        catch (OperationCanceledException) { }
+    }
+
+    /// <summary>
+    /// A wiped B is healed by a snapshot from A, one of whose rows A stamped while its clock ran a
+    /// year ahead. With FailClosedOnDrift that row must be dead-lettered on B and must not move B's
+    /// clock; the honest rows still arrive.
+    /// </summary>
+    [Theory]
+    [InlineData(true)]
+    [InlineData(false)]
+    public async Task FutureStampedSnapshotRow_HonoursFailClosedOnDrift(bool failClosed)
+    {
+        var options = new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), FailClosedOnDrift = failClosed };
+        var dbA = await NewDbAsync();
+        await dbA.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'ok', 1)");
+        await dbA.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (2, 'future', 2)");
+        var future = HlcAt(DateTimeOffset.UtcNow.AddDays(365));
+        await dbA.ExecuteAsync(
+            "UPDATE __localdb_row_version SET hlc = @hlc WHERE table_name = 'orders' AND pk_json = '{\"id\":2}'",
+            new { hlc = future });
+        var a = SessionOver(dbA, options);
+        await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync());
+
+        var dbB = await NewDbAsync();
+        var b = SessionOver(dbB, options);
+
+        var (da, dbx) = DuplexPair();
+        using var cts = new CancellationTokenSource(RunTimeout);
+        var runA = a.Session.RunAsync(da, cts.Token);
+        var runB = b.Session.RunAsync(dbx, cts.Token);
+        await WaitForAsync(async () => await Scalar(dbB, "SELECT last_applied_remote_seq FROM __localdb_peer_state") > 0, RunTimeout);
+        cts.Cancel();
+        await SwallowAsync(runA);
+        await SwallowAsync(runB);
+
+        var lastSeen = await Scalar(dbB, "SELECT last_seen_hlc FROM __localdb_peer_state");
+        if (failClosed)
+        {
+            Assert.Equal(new[] { 1L }, await Ids(dbB));
+            Assert.Equal(1, await Scalar(dbB, "SELECT COUNT(*) FROM __localdb_dead_letter WHERE error LIKE '%drift%'"));
+            Assert.True(dbB.Clock.Current < future, "the drifted row advanced the receiver's clock");
+            Assert.True(lastSeen < future, "the drifted row's HLC was persisted as last_seen_hlc");
+        }
+        else
+        {
+            // Warn-only mode applies it, exactly like the delta path.
+            Assert.Equal(new[] { 1L, 2L }, await Ids(dbB));
+            Assert.Equal(0, await Scalar(dbB, "SELECT COUNT(*) FROM __localdb_dead_letter"));
+        }
+    }
+}
@@ -10,7 +10,7 @@
 | Last reviewed | 2026-11-09 |
 | Reviewer | Claude |
 | Commit reviewed | `91fb2b4` |
-| Open findings | 4 |
+| Open findings | 3 |

 ## Summary

@@ -303,7 +303,7 @@ waits for the watermark.
 | Severity | Medium |
 | Priority | P2 |
 | Category | Spec & shared-contract adherence |
-| Status | Open |
+| Status | Resolved |
 | Location | `ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SnapshotStreamer.cs:249` |

 **Description**
@@ -336,7 +336,24 @@ check into `LwwApplier` so both paths share it. Add a snapshot-drift test.

 **Resolution**

-_Unresolved._
+Resolved 2026-11-09 (scadaproj#39, LocalDb `0.3.6`, unpublished). Re-verified with a 2-node test: A served a
+snapshot to a wiped B. One of A's rows had a ledger HLC 365 days ahead. With `FailClosedOnDrift = true`, B applied
+that row and wrote no dead-letter.
+
+The fix: `SnapshotApplier` now takes `ReplicationOptions` and a `utcNow` seam, and `SyncSessionFactory` passes the
+options. `OnBatchAsync` runs the same per-row check as `SyncSession.ApplyDeltaEntriesAsync`, against
+`MaxHlcDriftAhead` and with one warning per snapshot. Under `FailClosedOnDrift`, it dead-letters an over-drift row
+through `HlcDriftGuard.DeadLetterAsync`, a new helper that the delta path now uses too. That row's HLC is kept out of
+`_maxHlc`, so `OnCompleteAsync` never persists it as `last_seen_hlc`. In warn-only mode, rows are applied, as on the
+delta path.
+
+The check stays in the session and snapshot layers rather than moving into `LwwApplier`. That keeps the delta path's
+existing metrics and ack accounting unchanged.
+
+Tests (`SnapshotDriftTests`): the test is a theory built through `SyncSessionFactory`. In fail-closed mode it checks
+that the row is absent and that a drift dead-letter was written. It also checks that the receiver's `Clock.Current`
+and `last_seen_hlc` stay below the future stamp; this case fails pre-fix. In warn mode it checks that the row is
+applied and that no dead-letter is written. README §Semantics is updated.

 ### LocalDb-005 — Auto-assigned rowid / AUTOINCREMENT keys collide across nodes and LWW silently drops one row

@@ -40,7 +40,7 @@ code-reviews/

 ## Summary

-10 of 10 libraries reviewed. 20 pending findings across all libraries.
+10 of 10 libraries reviewed. 19 pending findings across all libraries.

 Coverage: Secrets, LocalDb, GalaxyRepository and Overview first reviewed, and Auth, Health,
 Theme, Audit, Configuration and Telemetry re-reviewed, on 2026-11-09 (#15). Each finding also
@@ -50,9 +50,9 @@ carries a P1/P2/P3 priority (P1 = Critical/High, P2 = Medium, P3 = Low).
 |----------|---------------|
 | Critical | 0 |
 | High | 0 |
-| Medium | 1 |
+| Medium | 0 |
 | Low | 19 |
-| **Total** | **20** |
+| **Total** | **19** |

 ## Library Status

@@ -63,7 +63,7 @@ carries a P1/P2/P3 priority (P1 = Critical/High, P2 = Medium, P3 = Low).
 | [Configuration](Configuration/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 4 |
 | [GalaxyRepository](GalaxyRepository/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 8 |
 | [Health](Health/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 7 |
-| [LocalDb](LocalDb/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/1/3 | 4 | 7 |
+| [LocalDb](LocalDb/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/3 | 3 | 7 |
 | [Overview](Overview/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 5 |
 | [Secrets](Secrets/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/13 | 13 | 19 |
 | [Telemetry](Telemetry/findings.md) | 2026-11-09 | `91fb2b4` | 0/0/0/0 | 0 | 11 |
@@ -84,11 +84,9 @@ _None open._

 _None open._

-### Medium (1)
+### Medium (0)

-| ID | Library | Title |
-|----|---------|-------|
-| LocalDb-004 | [LocalDb](LocalDb/findings.md) | `FailClosedOnDrift` is bypassed on the snapshot path, so a future-stamped snapshot row moves the local HLC forward permanently |
+_None open._

 ### Low (19)