From 3cf8cabaf9a473b4c18f8ccda52877985a3fcfa6 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Fri, 17 Jul 2026 21:55:25 -0400 Subject: [PATCH] =?UTF-8?q?fix(localdb):=20oplog=20store=20review=20fixes?= =?UTF-8?q?=20=E2=80=94=20ack=20clamping,=20age-based=20pruning,=20monoton?= =?UTF-8?q?icity=20tests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts --- .../Internal/OplogStore.cs | 65 +++++++++++++++---- .../OplogStoreTests.cs | 39 ++++++++++- 2 files changed, 89 insertions(+), 15 deletions(-) diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs index 76749b9..ae844c3 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs @@ -22,6 +22,11 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow); + /// + /// Reads up to oplog entries with seq > afterSeq in ascending + /// seq order. Caller contract: check needs_snapshot before trusting incremental reads — + /// after cap-pruning, reads silently resume past a gap. + /// public Task> ReadBatchAboveAsync(long afterSeq, int maxBatch, CancellationToken ct = default) => db.QueryAsync( "SELECT seq, table_name, pk_json, row_json, hlc, node_id, is_tombstone FROM __localdb_oplog " + @@ -31,23 +36,39 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func + /// Records the peer's delivery ack and prunes oplog rows at or below the watermark, in one + /// transaction. The watermark is monotonic (a late out-of-order ack never regresses it) and + /// clamped to the highest seq this node has ever produced: an ack beyond that is a protocol + /// violation, so needs_snapshot is set to force a resync rather than trust the peer. + /// public async Task RecordPeerAckAsync(long ackedSeq, CancellationToken ct = default) { await using var tx = await db.BeginTransactionAsync(ct); + // Known-max includes last_acked_seq: acked rows are already pruned, so a duplicate ack of + // a pruned seq is benign — only an ack beyond anything ever produced is a violation. + var knownMax = (await tx.QueryAsync( + "SELECT MAX((SELECT COALESCE(MAX(seq), 0) FROM __localdb_oplog), " + + "(SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1))", + static r => r.GetInt64(0), null, ct))[0]; + if (ackedSeq > knownMax) + await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1", null, ct); await tx.ExecuteAsync( "UPDATE __localdb_peer_state SET last_acked_seq = MAX(last_acked_seq, @acked), last_sync_utc = @now WHERE id = 1", - new { acked = ackedSeq, now = FormatUtc(_utcNow()) }, ct); + new { acked = Math.Min(ackedSeq, knownMax), now = FormatUtc(_utcNow()) }, ct); await tx.ExecuteAsync( "DELETE FROM __localdb_oplog WHERE seq <= (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)", null, ct); await tx.CommitAsync(ct); } + /// Deletes row-version tombstones older than ; live rows and fresh tombstones are kept. public Task PruneTombstonesAsync(CancellationToken ct = default) => db.ExecuteAsync( "DELETE FROM __localdb_row_version WHERE is_tombstone = 1 AND tombstone_utc < @cutoff", new { cutoff = FormatUtc(_utcNow() - options.TombstoneRetention) }, ct); + /// Counts the unacked backlog: oplog entries above the peer's last-acked watermark. public async Task GetOplogDepthAsync(CancellationToken ct = default) { var rows = await db.QueryAsync( @@ -56,6 +77,7 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, FuncThe highest oplog seq present, or 0 when the oplog is empty. public async Task GetMaxSeqAsync(CancellationToken ct = default) { var rows = await db.QueryAsync( @@ -63,31 +85,45 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func + /// Enforces the backlog caps in one transaction; returns true if a snapshot resync was flagged. + /// Rows cap: backlog above prunes to the newest + /// MaxOplogRows entries. Age cap: an oldest entry whose HLC physical time exceeds + /// prunes every entry past the age cutoff — disk + /// safety over delta continuity. Either trigger sets needs_snapshot. + /// public async Task EnforceCapsAsync(CancellationToken ct = default) { - var overRows = await GetOplogDepthAsync(ct) > options.MaxOplogRows; + var ageCutoffMs = (_utcNow() - options.MaxOplogAge).ToUnixTimeMilliseconds(); - var overAge = false; - var oldest = await db.QueryAsync( + // Condition reads and prunes share the transaction so the decision cannot go stale (TOCTOU). + await using var tx = await db.BeginTransactionAsync(ct); + + var depth = (await tx.QueryAsync( + "SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)", + static r => r.GetInt64(0), null, ct))[0]; + var overRows = depth > options.MaxOplogRows; + + var oldest = await tx.QueryAsync( "SELECT hlc FROM __localdb_oplog ORDER BY seq LIMIT 1", static r => r.GetInt64(0), null, ct); - if (oldest.Count > 0) - { - var cutoffMs = (_utcNow() - options.MaxOplogAge).ToUnixTimeMilliseconds(); - overAge = HybridLogicalClock.PhysicalMs(oldest[0]) < cutoffMs; - } + var overAge = oldest.Count > 0 && HybridLogicalClock.PhysicalMs(oldest[0]) < ageCutoffMs; if (!overRows && !overAge) return false; - await using var tx = await db.BeginTransactionAsync(ct); await tx.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = 1 WHERE id = 1", null, ct); - await tx.ExecuteAsync( - "DELETE FROM __localdb_oplog WHERE seq < ((SELECT MAX(seq) FROM __localdb_oplog) - @keep + 1)", - new { keep = options.MaxOplogRows }, ct); + if (overRows) + await tx.ExecuteAsync( + "DELETE FROM __localdb_oplog WHERE seq < ((SELECT MAX(seq) FROM __localdb_oplog) - @keep + 1)", + new { keep = options.MaxOplogRows }, ct); + if (overAge) + await tx.ExecuteAsync( + "DELETE FROM __localdb_oplog WHERE (hlc >> 16) < @ageCutoffMs", new { ageCutoffMs }, ct); await tx.CommitAsync(ct); return true; } + /// Reads the singleton peer-state row. public async Task GetPeerStateAsync(CancellationToken ct = default) { var rows = await db.QueryAsync( @@ -100,12 +136,15 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, FuncRecords the identity of the peer node this database syncs with. public Task SetPeerNodeIdAsync(string peerNodeId, CancellationToken ct = default) => db.ExecuteAsync("UPDATE __localdb_peer_state SET peer_node_id = @v WHERE id = 1", new { v = peerNodeId }, ct); + /// Sets or clears the snapshot-resync-required flag. public Task SetNeedsSnapshotAsync(bool value, CancellationToken ct = default) => db.ExecuteAsync("UPDATE __localdb_peer_state SET needs_snapshot = @v WHERE id = 1", new { v = value ? 1 : 0 }, ct); + /// Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps last_sync_utc. public Task SetLastAppliedRemoteSeqAsync(long seq, long lastSeenHlc, CancellationToken ct = default) => db.ExecuteAsync( "UPDATE __localdb_peer_state SET last_applied_remote_seq = @seq, last_seen_hlc = @hlc, last_sync_utc = @now WHERE id = 1", diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs index fa4fd82..becfd3e 100644 --- a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs @@ -90,6 +90,38 @@ public sealed class OplogStoreTests : IDisposable Assert.Equal(third.Seq, peer.LastAckedSeq); } + [Fact] + public async Task RecordPeerAck_OutOfOrderAck_DoesNotRegressWatermark() + { + using var db = await NewOrdersDb(); + await WriteRows(db, 5); + var store = new OplogStore(db, new ReplicationOptions()); + var batch = await store.ReadBatchAboveAsync(0, 100, default); + + await store.RecordPeerAckAsync(batch[4].Seq, default); + await store.RecordPeerAckAsync(batch[1].Seq, default); + + var peer = await store.GetPeerStateAsync(default); + Assert.Equal(batch[4].Seq, peer.LastAckedSeq); + // A late duplicate/out-of-order ack of an already-acked seq is benign, not a violation. + Assert.False(peer.NeedsSnapshot); + } + + [Fact] + public async Task RecordPeerAck_AckBeyondMaxSeq_ClampsAndFlagsSnapshot() + { + using var db = await NewOrdersDb(); + await WriteRows(db, 3); + var store = new OplogStore(db, new ReplicationOptions()); + var maxSeq = await store.GetMaxSeqAsync(default); + + await store.RecordPeerAckAsync(maxSeq + 100, default); + + var peer = await store.GetPeerStateAsync(default); + Assert.Equal(maxSeq, peer.LastAckedSeq); + Assert.True(peer.NeedsSnapshot); + } + [Fact] public async Task Prune_DropsExpiredTombstoneRowVersions_KeepsLive() { @@ -137,12 +169,12 @@ public sealed class OplogStoreTests : IDisposable Assert.True(flagged); var count = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", r => r.GetInt64(0)); - Assert.True(count[0] <= 3, $"oplog should be pruned to <= 3, was {count[0]}"); + Assert.Equal(3, count[0]); Assert.True((await store.GetPeerStateAsync(default)).NeedsSnapshot); } [Fact] - public async Task CapExceeded_ByAge_SetsNeedsSnapshot() + public async Task CapExceeded_ByAge_SetsNeedsSnapshot_AndPrunesExpired() { using var db = await NewOrdersDb(); await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2)"); @@ -154,6 +186,9 @@ public sealed class OplogStoreTests : IDisposable Assert.True(flagged); Assert.True((await store.GetPeerStateAsync(default)).NeedsSnapshot); + // Age-only trigger must actually prune: both entries are past the age cutoff. + var count = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", r => r.GetInt64(0)); + Assert.Equal(0, count[0]); } [Fact]