Joseph Doherty
be5e9fd436
0.3.7 (unpublished), owner-chosen option A. A tombstone is now pruned only when it is past TombstoneRetention AND the peer is known to hold it: - held while database-wide or per-table snapshot debt is outstanding (the snapshot is built from the row-version ledger, so a pruned tombstone is a delete it can no longer carry — the issue's second path); - a local tombstone held while its delete is still in the unacked oplog. Delivery (delta ack covering the delete, or snapshot debt discharged) restamps tombstone_utc so retention runs from delivery. New optional LocalDb:Replication:TombstoneHardCap (default off, >= TombstoneRetention) bounds an undelivered tombstone's life. No proto, lib_schema_version or on-disk schema change. Regression test TombstoneResurrectionTests (two real nodes over loopback gRPC): offline update on B, newer delete on A, A's clock +8 days (oplog age cap + tombstone prune) -> reconnect. Fails on 0.3.6 (row resurrected), passes now; plus a no-prune positive control and a pruned-after-delivery control, and five controlled-clock OplogStore tests. Each guard clause mutation-killed. 259/259. Claude-Session: https://claude.ai/code/session_01GT67AEHd2Acv1YPwGfXEoV
489 lines
20 KiB
C#
489 lines
20 KiB
C#
using Microsoft.Data.Sqlite;
|
|
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;
|
|
|
|
public sealed class OplogStoreTests : IDisposable
|
|
{
|
|
private readonly string _path = Path.Combine(Path.GetTempPath(), Guid.NewGuid() + ".db");
|
|
|
|
public void Dispose()
|
|
{
|
|
SqliteConnection.ClearAllPools();
|
|
if (File.Exists(_path))
|
|
File.Delete(_path);
|
|
}
|
|
|
|
private async Task<SqliteLocalDb> NewOrdersDb()
|
|
{
|
|
var db = new SqliteLocalDb(new LocalDbOptions { Path = _path });
|
|
await db.ExecuteAsync("CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)");
|
|
db.RegisterReplicated("orders");
|
|
return db;
|
|
}
|
|
|
|
private async Task WriteRows(SqliteLocalDb db, int count)
|
|
{
|
|
for (var i = 1; i <= count; i++)
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'S', @id)", new { id = i });
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadBatchAbove_ReturnsInSeqOrder_RespectsMaxBatch()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await WriteRows(db, 5);
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
|
|
var first = await store.ReadBatchAboveAsync(0, 3, default);
|
|
Assert.Equal(3, first.Count);
|
|
Assert.True(first.Select(e => e.Seq).SequenceEqual(first.Select(e => e.Seq).OrderBy(x => x)));
|
|
|
|
var rest = await store.ReadBatchAboveAsync(first[^1].Seq, 100, default);
|
|
Assert.Equal(2, rest.Count);
|
|
Assert.All(rest, e => Assert.True(e.Seq > first[^1].Seq));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task ReadBatch_MapsAllFields()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'ABC', 5)");
|
|
await db.ExecuteAsync("DELETE FROM orders WHERE id = 1");
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
|
|
var batch = await store.ReadBatchAboveAsync(0, 100, default);
|
|
Assert.Equal(2, batch.Count);
|
|
|
|
var ins = batch[0];
|
|
Assert.Equal("orders", ins.TableName);
|
|
Assert.Equal("{\"id\":1}", ins.PkJson);
|
|
Assert.Equal("{\"id\":1,\"sku\":\"ABC\",\"qty\":5}", ins.RowJson);
|
|
Assert.Equal(db.NodeId, ins.NodeId);
|
|
Assert.False(ins.IsTombstone);
|
|
Assert.True(ins.Hlc > 0);
|
|
|
|
var tomb = batch[1];
|
|
Assert.Equal("{\"id\":1}", tomb.PkJson);
|
|
Assert.Null(tomb.RowJson);
|
|
Assert.True(tomb.IsTombstone);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task RecordPeerAck_PrunesBelowWatermark()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await WriteRows(db, 5);
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
var batch = await store.ReadBatchAboveAsync(0, 100, default);
|
|
var third = batch[2];
|
|
|
|
await store.RecordPeerAckAsync(third.Seq, default);
|
|
|
|
var remaining = await store.ReadBatchAboveAsync(0, 100, default);
|
|
Assert.Equal(2, remaining.Count);
|
|
Assert.All(remaining, e => Assert.True(e.Seq > third.Seq));
|
|
|
|
var peer = await store.GetPeerStateAsync(default);
|
|
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);
|
|
}
|
|
|
|
// ---- scadaproj#16: a tombstone is pruned only once the peer holds it --------------------
|
|
|
|
private static async Task<long> TombstoneCount(SqliteLocalDb db) =>
|
|
(await db.QueryAsync(
|
|
"SELECT COUNT(*) FROM __localdb_row_version WHERE is_tombstone = 1", static r => r.GetInt64(0)))[0];
|
|
|
|
[Fact]
|
|
public async Task Prune_KeepsUndeliveredLocalTombstonePastRetention_ThenAgesItFromDelivery()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 1)");
|
|
await db.ExecuteAsync("DELETE FROM orders WHERE id = 1");
|
|
var now = DateTimeOffset.UtcNow;
|
|
var store = new OplogStore(db, new ReplicationOptions(), () => now);
|
|
|
|
// Peer offline 8 days: past the 7-day retention, but the delete is still unacked.
|
|
now += TimeSpan.FromDays(8);
|
|
Assert.Equal(0, await store.PruneTombstonesAsync(default));
|
|
Assert.Equal(1, await TombstoneCount(db));
|
|
|
|
// The peer comes back and acks it: retention restarts at the ack, not at the delete.
|
|
await store.RecordPeerAckAsync(await store.GetMaxSeqAsync(), default);
|
|
now += TimeSpan.FromDays(6);
|
|
Assert.Equal(0, await store.PruneTombstonesAsync(default));
|
|
Assert.Equal(1, await TombstoneCount(db));
|
|
|
|
now += TimeSpan.FromDays(2);
|
|
Assert.Equal(1, await store.PruneTombstonesAsync(default));
|
|
Assert.Equal(0, await TombstoneCount(db));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Prune_KeepsTombstonesWhileDatabaseWideSnapshotDebtIsOutstanding()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 1)");
|
|
await db.ExecuteAsync("DELETE FROM orders WHERE id = 1");
|
|
var now = DateTimeOffset.UtcNow;
|
|
var store = new OplogStore(db, new ReplicationOptions(), () => now);
|
|
|
|
// The age cap drops the delete from the oplog (never delivered) and flags a snapshot — the
|
|
// snapshot is built from the ledger, so the tombstone is the only carrier of the delete.
|
|
now += TimeSpan.FromDays(8);
|
|
Assert.True(await store.EnforceCapsAsync(default));
|
|
Assert.Equal(0, await store.GetMaxSeqAsync());
|
|
Assert.Equal(0, await store.PruneTombstonesAsync(default));
|
|
Assert.Equal(1, await TombstoneCount(db));
|
|
|
|
// The snapshot is acknowledged: debt cleared, tombstone restamped, aged from there.
|
|
await store.ClearSnapshotDebtAsync([], default);
|
|
now += TimeSpan.FromDays(6);
|
|
Assert.Equal(0, await store.PruneTombstonesAsync(default));
|
|
now += TimeSpan.FromDays(2);
|
|
Assert.Equal(1, await store.PruneTombstonesAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Prune_KeepsTombstonesOfATableWithPerTableSnapshotDebt()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 1)");
|
|
await db.ExecuteAsync("DELETE FROM orders WHERE id = 1");
|
|
var now = DateTimeOffset.UtcNow;
|
|
var store = new OplogStore(db, new ReplicationOptions(), () => now);
|
|
await store.RecordPeerAckAsync(await store.GetMaxSeqAsync(), default);
|
|
await db.ExecuteAsync(
|
|
"INSERT INTO __localdb_snapshot_state (table_name, needs_snapshot) VALUES ('orders', 1)");
|
|
|
|
now += TimeSpan.FromDays(8);
|
|
Assert.Equal(0, await store.PruneTombstonesAsync(default));
|
|
|
|
await store.ClearSnapshotDebtAsync(["orders"], default);
|
|
now += TimeSpan.FromDays(8);
|
|
Assert.Equal(1, await store.PruneTombstonesAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Prune_RemoteOriginTombstone_AgesOnRetentionAlone_EvenWithLocalBacklog()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
// A local write the peer has not acked — it must not hold the PEER's own delete.
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (9, 'LOCAL', 1)");
|
|
await db.ExecuteAsync(
|
|
"INSERT INTO __localdb_row_version (table_name, pk_json, hlc, node_id, is_tombstone, tombstone_utc) " +
|
|
"VALUES ('orders', '{\"id\":1}', 9223372036854775000, 'the-peer', 1, @utc)",
|
|
new { utc = DateTimeOffset.UtcNow.UtcDateTime.ToString(OplogStore.TombstoneUtcFormat) });
|
|
var now = DateTimeOffset.UtcNow + TimeSpan.FromDays(8);
|
|
var store = new OplogStore(db, new ReplicationOptions(), () => now);
|
|
|
|
Assert.Equal(1, await store.PruneTombstonesAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Prune_HardCap_DropsAnUndeliveredTombstoneOnceExceeded()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A', 1)");
|
|
await db.ExecuteAsync("DELETE FROM orders WHERE id = 1");
|
|
var now = DateTimeOffset.UtcNow;
|
|
var store = new OplogStore(db, new ReplicationOptions { TombstoneHardCap = TimeSpan.FromDays(30) }, () => now);
|
|
|
|
now += TimeSpan.FromDays(29);
|
|
Assert.Equal(0, await store.PruneTombstonesAsync(default));
|
|
now += TimeSpan.FromDays(2);
|
|
Assert.Equal(1, await store.PruneTombstonesAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Prune_DropsExpiredTombstoneRowVersions_KeepsLive()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2),(3,'C',3)");
|
|
await db.ExecuteAsync("DELETE FROM orders WHERE id = 1"); // will be backdated -> expired
|
|
await db.ExecuteAsync("DELETE FROM orders WHERE id = 3"); // fresh tombstone -> kept
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
// The peer has both deletes (scadaproj#16: an undelivered tombstone is never pruned on age);
|
|
// backdate AFTER the ack, which restamps delivered tombstones.
|
|
await store.RecordPeerAckAsync(await store.GetMaxSeqAsync(), default);
|
|
await db.ExecuteAsync(
|
|
"UPDATE __localdb_row_version SET tombstone_utc = '2000-01-01T00:00:00.000Z' WHERE pk_json = '{\"id\":1}'");
|
|
|
|
await store.PruneTombstonesAsync(default);
|
|
|
|
var rv = await db.QueryAsync(
|
|
"SELECT pk_json, is_tombstone FROM __localdb_row_version WHERE table_name='orders'",
|
|
r => (Pk: r.GetString(0), Tomb: r.GetInt64(1)));
|
|
|
|
Assert.DoesNotContain(rv, x => x.Pk == "{\"id\":1}");
|
|
Assert.Contains(rv, x => x.Pk == "{\"id\":2}" && x.Tomb == 0);
|
|
Assert.Contains(rv, x => x.Pk == "{\"id\":3}" && x.Tomb == 1);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OplogDepth_ReportsBacklog()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await WriteRows(db, 5);
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
Assert.Equal(5, await store.GetOplogDepthAsync(default));
|
|
|
|
var batch = await store.ReadBatchAboveAsync(0, 100, default);
|
|
await store.RecordPeerAckAsync(batch[1].Seq, default); // ack 2nd -> prunes 2, backlog 3
|
|
|
|
Assert.Equal(3, await store.GetOplogDepthAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task CapExceeded_SetsNeedsSnapshotAndPrunes()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await WriteRows(db, 5);
|
|
var store = new OplogStore(db, new ReplicationOptions { MaxOplogRows = 3 });
|
|
|
|
var flagged = await store.EnforceCapsAsync(default);
|
|
|
|
Assert.True(flagged);
|
|
var count = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", r => r.GetInt64(0));
|
|
Assert.Equal(3, count[0]);
|
|
Assert.True((await store.GetPeerStateAsync(default)).NeedsSnapshot);
|
|
}
|
|
|
|
[Fact]
|
|
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)");
|
|
// Inject a clock 30 days ahead so the fresh entries exceed the 7d MaxOplogAge.
|
|
var future = DateTimeOffset.UtcNow.AddDays(30);
|
|
var store = new OplogStore(db, new ReplicationOptions(), () => future);
|
|
|
|
var flagged = await store.EnforceCapsAsync(default);
|
|
|
|
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]
|
|
public async Task CapExceeded_ByAge_MixedFixture_FreshRowSurvives()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1,'A',1),(2,'B',2)");
|
|
// Backdate only the first entry's HLC past the 7d MaxOplogAge; the second stays fresh.
|
|
var expiredHlc = DateTimeOffset.UtcNow.AddDays(-30).ToUnixTimeMilliseconds() << 16;
|
|
await db.ExecuteAsync(
|
|
"UPDATE __localdb_oplog SET hlc = @h WHERE seq = (SELECT MIN(seq) FROM __localdb_oplog)",
|
|
new { h = expiredHlc });
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
|
|
var flagged = await store.EnforceCapsAsync(default);
|
|
|
|
Assert.True(flagged);
|
|
Assert.True((await store.GetPeerStateAsync(default)).NeedsSnapshot);
|
|
var remaining = await store.ReadBatchAboveAsync(0, 100, default);
|
|
Assert.Single(remaining);
|
|
Assert.Equal("{\"id\":2}", remaining[0].PkJson);
|
|
}
|
|
|
|
// ---- O(1) backlog depth --------------------------------------------------------------
|
|
|
|
// The definition the O(1) formula has to reproduce. Every scenario below is asserted against
|
|
// this, not against a hand-computed number, so the two can never drift apart.
|
|
private static async Task<long> CountingDepth(SqliteLocalDb db) =>
|
|
(await db.QueryAsync(
|
|
"SELECT COUNT(*) FROM __localdb_oplog WHERE seq > (SELECT last_acked_seq FROM __localdb_peer_state WHERE id = 1)",
|
|
r => r.GetInt64(0)))[0];
|
|
|
|
private static async Task AssertDepthMatchesCount(SqliteLocalDb db, OplogStore store)
|
|
{
|
|
var expected = await CountingDepth(db);
|
|
Assert.Equal(expected, await store.GetOplogDepthAsync(default));
|
|
Assert.Equal(expected, store.GetOplogDepthSync());
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OplogDepth_Empty_IsZero()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
|
|
await AssertDepthMatchesCount(db, store);
|
|
Assert.Equal(0, await store.GetOplogDepthAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OplogDepth_AllAcked_EmptiesToZero_NotToTheSeqRange()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await WriteRows(db, 5);
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
await store.RecordPeerAckAsync(await store.GetMaxSeqAsync(default), default);
|
|
|
|
await AssertDepthMatchesCount(db, store);
|
|
Assert.Equal(0, await store.GetOplogDepthAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OplogDepth_AfterCapPrune_ReportsRetainedRows_NotTheWholeSeqRange()
|
|
{
|
|
// The reason the formula clamps its floor to MIN(seq) - 1. A cap prune leaves the retained
|
|
// rows far above an ack watermark that never moved, so a naive MAX(seq) - last_acked would
|
|
// report the entire seq range forever and re-flag the cap on every maintenance tick.
|
|
using var db = await NewOrdersDb();
|
|
await WriteRows(db, 40);
|
|
var store = new OplogStore(db, new ReplicationOptions { MaxOplogRows = 5 });
|
|
|
|
Assert.True(await store.EnforceCapsAsync(default));
|
|
|
|
await AssertDepthMatchesCount(db, store);
|
|
Assert.Equal(5, await store.GetOplogDepthAsync(default));
|
|
// And the next tick is clean: depth is not over cap any more.
|
|
Assert.False(await store.EnforceCapsAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OplogDepth_AfterAgePrune_MatchesCount()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await WriteRows(db, 6);
|
|
// Backdate the oldest three past the age cap; HLCs rise with seq, so this is a prefix.
|
|
var expired = DateTimeOffset.UtcNow.AddDays(-30).ToUnixTimeMilliseconds() << 16;
|
|
await db.ExecuteAsync(
|
|
"UPDATE __localdb_oplog SET hlc = @h WHERE seq IN (SELECT seq FROM __localdb_oplog ORDER BY seq LIMIT 3)",
|
|
new { h = expired });
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
|
|
Assert.True(await store.EnforceCapsAsync(default));
|
|
|
|
await AssertDepthMatchesCount(db, store);
|
|
Assert.Equal(3, await store.GetOplogDepthAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OplogDepth_PartialAck_MatchesCount()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await WriteRows(db, 9);
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
var batch = await store.ReadBatchAboveAsync(0, 100, default);
|
|
|
|
await store.RecordPeerAckAsync(batch[3].Seq, default);
|
|
|
|
await AssertDepthMatchesCount(db, store);
|
|
Assert.Equal(5, await store.GetOplogDepthAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OplogDepth_AckPruneThenMoreWrites_MatchesCount()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
await WriteRows(db, 4);
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
await store.RecordPeerAckAsync(await store.GetMaxSeqAsync(default), default);
|
|
|
|
// The steady state of a converged pair: oplog pruned empty, watermark high, new writes above it.
|
|
await db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (91,'S',1),(92,'S',2)");
|
|
|
|
await AssertDepthMatchesCount(db, store);
|
|
Assert.Equal(2, await store.GetOplogDepthAsync(default));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task OplogDepth_LargeBacklog_IsConstantTime()
|
|
{
|
|
// 200k rows: a COUNT(*) over the backlog runs ~200k VM steps per call (measured: 100,020 at
|
|
// 50k rows), the seek-based formula runs 58 regardless of size. The loop below costs tens of
|
|
// milliseconds at seek shape and seconds at scan shape, so the budget separates them by a
|
|
// wide margin in both directions.
|
|
using var db = await NewOrdersDb();
|
|
await db.ExecuteAsync(
|
|
"INSERT INTO __localdb_oplog (table_name, pk_json, row_json, hlc, node_id, is_tombstone) " +
|
|
"WITH RECURSIVE c(x) AS (SELECT 1 UNION ALL SELECT x + 1 FROM c WHERE x < 200000) " +
|
|
"SELECT 'orders', '{\"id\":' || x || '}', '{\"id\":' || x || '}', x << 16, 'n', 0 FROM c");
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
|
|
Assert.Equal(200_000, await store.GetOplogDepthAsync(default));
|
|
|
|
using var conn = db.CreateConnection();
|
|
using var cmd = conn.CreateCommand();
|
|
cmd.CommandText = OplogStore.DepthSql;
|
|
var sw = System.Diagnostics.Stopwatch.StartNew();
|
|
for (var i = 0; i < 2_000; i++)
|
|
Assert.Equal(200_000L, (long)cmd.ExecuteScalar()!);
|
|
sw.Stop();
|
|
|
|
Assert.True(sw.ElapsedMilliseconds < 1_500,
|
|
$"2000 depth reads over a 200k backlog took {sw.ElapsedMilliseconds} ms — that is scan-shaped, not seek-shaped.");
|
|
}
|
|
|
|
[Fact]
|
|
public async Task PeerState_RoundTrips()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
|
|
await store.SetPeerNodeIdAsync("node-xyz", default);
|
|
await store.SetNeedsSnapshotAsync(true, default);
|
|
await store.SetLastAppliedRemoteSeqAsync(42, 987654, default);
|
|
|
|
var p = await store.GetPeerStateAsync(default);
|
|
Assert.Equal("node-xyz", p.PeerNodeId);
|
|
Assert.True(p.NeedsSnapshot);
|
|
Assert.Equal(42, p.LastAppliedRemoteSeq);
|
|
Assert.Equal(987654, p.LastSeenHlc);
|
|
Assert.NotNull(p.LastSyncUtc);
|
|
|
|
await store.SetNeedsSnapshotAsync(false, default);
|
|
Assert.False((await store.GetPeerStateAsync(default)).NeedsSnapshot);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task GetMaxSeq_ReturnsHighestSeq_ZeroWhenEmpty()
|
|
{
|
|
using var db = await NewOrdersDb();
|
|
var store = new OplogStore(db, new ReplicationOptions());
|
|
Assert.Equal(0, await store.GetMaxSeqAsync(default));
|
|
|
|
await WriteRows(db, 4);
|
|
var batch = await store.ReadBatchAboveAsync(0, 100, default);
|
|
Assert.Equal(batch[^1].Seq, await store.GetMaxSeqAsync(default));
|
|
}
|
|
}
|