dohertj2/scadaproj
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(localdb): oplog store review fixes — ack clamping, age-based pruning, monotonicity tests

Browse Source 
Claude-Session: https://claude.ai/code/session_01BL2Vu1ESDQ9SCN4gVKkdts
This commit is contained in:
Joseph Doherty
2026-07-17 21:55:25 -04:00
parent 4f9982c53f
commit 3cf8cabaf9
2 changed files with 89 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/OplogStore.cs
+52 -13
View File
@@ -22,6 +22,11 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
private readonly Func<DateTimeOffset> _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow);
/// <summary>
/// Reads up to <paramref name="maxBatch"/> oplog entries with <c>seq &gt; afterSeq</c> in ascending
/// seq order. Caller contract: check <c>needs_snapshot</c> before trusting incremental reads —
/// after cap-pruning, reads silently resume past a gap.
/// </summary>
public Task<IReadOnlyList<OplogEntryRecord>> 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<D
r.GetInt64(4), r.GetString(5), r.GetInt64(6) != 0),
new { afterSeq, maxBatch }, ct);
/// <summary>
/// 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 <c>needs_snapshot</c> is set to force a resync rather than trust the peer.
/// </summary>
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);
}
/// <summary>Deletes row-version tombstones older than <see cref="ReplicationOptions.TombstoneRetention"/>; live rows and fresh tombstones are kept.</summary>
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);
/// <summary>Counts the unacked backlog: oplog entries above the peer's last-acked watermark.</summary>
public async Task<long> GetOplogDepthAsync(CancellationToken ct = default)
{
var rows = await db.QueryAsync(
@@ -56,6 +77,7 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
return rows[0];
}
/// <summary>The highest oplog seq present, or 0 when the oplog is empty.</summary>
public async Task<long> GetMaxSeqAsync(CancellationToken ct = default)
{
var rows = await db.QueryAsync(
@@ -63,31 +85,45 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
return rows[0];
}
/// <summary>
/// Enforces the backlog caps in one transaction; returns true if a snapshot resync was flagged.
/// Rows cap: backlog above <see cref="ReplicationOptions.MaxOplogRows"/> prunes to the newest
/// MaxOplogRows entries. Age cap: an oldest entry whose HLC physical time exceeds
/// <see cref="ReplicationOptions.MaxOplogAge"/> prunes every entry past the age cutoff — disk
/// safety over delta continuity. Either trigger sets <c>needs_snapshot</c>.
/// </summary>
public async Task<bool> 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;
}
/// <summary>Reads the singleton peer-state row.</summary>
public async Task<PeerState> GetPeerStateAsync(CancellationToken ct = default)
{
var rows = await db.QueryAsync(
@@ -100,12 +136,15 @@ internal sealed class OplogStore(ILocalDb db, ReplicationOptions options, Func<D
return rows[0];
}
/// <summary>Records the identity of the peer node this database syncs with.</summary>
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);
/// <summary>Sets or clears the snapshot-resync-required flag.</summary>
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);
/// <summary>Records the highest remote seq applied locally plus the peer HLC observed with it; also stamps <c>last_sync_utc</c>.</summary>
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",
ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/OplogStoreTests.cs
+37 -2
View File
@@ -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]