Joseph Doherty
cad3bcb6bf
A converged pair prunes its oplog on ack, so "no oplog rows at all" is the steady state of a healthy pair — not an anomaly. ComputeSnapshotRequiredAsync measured the peer's gap against the oldest surviving oplog row and treated an empty oplog as "no gap possible", which made that steady state the one state from which a wiped peer could never be healed: it reconnected with an empty database and a zero watermark, was told it needed no snapshot, and stayed empty indefinitely. With an empty oplog the oldest seq we could still stream is one past last_acked_seq — ack-pruning is the only thing that empties the oplog without also flagging needs_snapshot, and it deletes exactly seq <= last_acked_seq. Measuring the gap against that heals the wiped peer and leaves every other case identical: a converged peer's watermark equals last_acked_seq (no snapshot), and a pair that has never written anything has both at zero. Found by the OtOpcUa LocalDb Phase 1 live gate (check 4), which recorded it as a documented limitation. Covered by a test that reproduces the live-gate scenario through the real session stack: RED before this fix (the wiped side never converges, 15 s timeout), green after. 148/148 pass. Version 0.1.2. Claude-Session: https://claude.ai/code/session_01GASWkNEi68FSCtvr6rLoEW
501 lines
22 KiB
C#
501 lines
22 KiB
C#
using System.Threading.Channels;
|
|
using Microsoft.Extensions.Logging;
|
|
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>
|
|
/// Full-stack snapshot-resync coverage: a node whose peer fell behind the pruned oplog horizon
|
|
/// streams a full snapshot of its registered tables; the receiver MERGES it through LWW (never
|
|
/// truncates), tombstones are carried, and incremental deltas resume afterwards.
|
|
/// </summary>
|
|
public sealed class SnapshotResyncTests : IDisposable
|
|
{
|
|
private const string OrdersSql = "CREATE TABLE orders (id INTEGER PRIMARY KEY, sku TEXT, qty INTEGER)";
|
|
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, LwwApplier Applier, SyncSession Session);
|
|
|
|
// Builds a full replication stack with the snapshot hooks wired exactly as SyncSessionFactory does.
|
|
private async Task<Side> NewSideAsync(ReplicationOptions? options = null)
|
|
{
|
|
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(OrdersSql);
|
|
db.RegisterReplicated("orders");
|
|
|
|
options ??= new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20) };
|
|
var store = new OplogStore(db, options);
|
|
var applier = new LwwApplier(db);
|
|
var session = new SyncSession(db, store, applier, options, NullLogger.Instance);
|
|
|
|
var streamer = new SnapshotStreamer(db, store, options, NullLogger.Instance);
|
|
var snapshotApplier = new SnapshotApplier(db, applier, store, NullLogger.Instance);
|
|
session.SnapshotSender = streamer.SendAsync;
|
|
session.SnapshotApplier = snapshotApplier;
|
|
|
|
return new Side(db, store, applier, session);
|
|
}
|
|
|
|
private static (SyncDuplex A, SyncDuplex B) DuplexPair()
|
|
{
|
|
var aToB = Channel.CreateUnbounded<SyncMessage>();
|
|
var bToA = Channel.CreateUnbounded<SyncMessage>();
|
|
var a = new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), Inbox = bToA.Reader };
|
|
var b = new SyncDuplex { Send = (m, ct) => bToA.Writer.WriteAsync(m, ct).AsTask(), Inbox = aToB.Reader };
|
|
return (a, b);
|
|
}
|
|
|
|
private static (SyncDuplex Duplex, ChannelWriter<SyncMessage> ToSession, ChannelReader<SyncMessage> FromSession) ScriptedPeer()
|
|
{
|
|
var toSession = Channel.CreateUnbounded<SyncMessage>();
|
|
var fromSession = Channel.CreateUnbounded<SyncMessage>();
|
|
var duplex = new SyncDuplex { Send = (m, ct) => fromSession.Writer.WriteAsync(m, ct).AsTask(), Inbox = toSession.Reader };
|
|
return (duplex, toSession.Writer, fromSession.Reader);
|
|
}
|
|
|
|
private static Handshake MatchingHandshake(SqliteLocalDb sessionDb, string peerNodeId, long lastAppliedRemoteSeq = 0)
|
|
{
|
|
var h = new Handshake { NodeId = peerNodeId, LibSchemaVersion = 1, LastAppliedRemoteSeq = lastAppliedRemoteSeq };
|
|
foreach (var t in sessionDb.ReplicatedTables.Values.OrderBy(t => t.Name, StringComparer.Ordinal))
|
|
h.Tables.Add(new TableDigest { TableName = t.Name, Digest = t.Digest });
|
|
return h;
|
|
}
|
|
|
|
private static async Task<SyncMessage> NextAsync(ChannelReader<SyncMessage> reader, CancellationToken ct)
|
|
{
|
|
await reader.WaitToReadAsync(ct);
|
|
reader.TryRead(out var msg);
|
|
return msg!;
|
|
}
|
|
|
|
private static async Task<SyncMessage> NextOfCaseAsync(
|
|
ChannelReader<SyncMessage> reader, SyncMessage.MsgOneofCase msgCase, CancellationToken ct)
|
|
{
|
|
SyncMessage msg;
|
|
do { msg = await NextAsync(reader, ct); } while (msg.MsgCase != msgCase);
|
|
return msg;
|
|
}
|
|
|
|
// Completes the two-phase handshake against a real session: the peer never requests a snapshot.
|
|
private static async Task DriveHandshakeAsync(
|
|
SqliteLocalDb sessionDb, ChannelWriter<SyncMessage> toSession, ChannelReader<SyncMessage> fromSession, CancellationToken ct)
|
|
{
|
|
await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.Handshake, ct);
|
|
await toSession.WriteAsync(new SyncMessage { Handshake = MatchingHandshake(sessionDb, "peer") }, ct);
|
|
await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.HandshakeAck, ct);
|
|
await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = false } }, ct);
|
|
}
|
|
|
|
private static async Task SwallowAsync(Task task)
|
|
{
|
|
try { await task; }
|
|
catch (OperationCanceledException) { }
|
|
}
|
|
|
|
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<List<(long Id, string? Sku, long? Qty)>> ReadOrders(SqliteLocalDb db)
|
|
{
|
|
var rows = await db.QueryAsync(
|
|
"SELECT id, sku, qty FROM orders ORDER BY id",
|
|
x => (x.GetInt64(0), x.IsDBNull(1) ? null : x.GetString(1), (long?)(x.IsDBNull(2) ? null : x.GetInt64(2))));
|
|
return rows.ToList();
|
|
}
|
|
|
|
private static async Task<long> LastAppliedRemoteSeq(SqliteLocalDb db)
|
|
{
|
|
var r = await db.QueryAsync("SELECT last_applied_remote_seq FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0));
|
|
return r[0];
|
|
}
|
|
|
|
private static async Task<long> NeedsSnapshot(SqliteLocalDb db)
|
|
{
|
|
var r = await db.QueryAsync("SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0));
|
|
return r[0];
|
|
}
|
|
|
|
private static async Task<long> OplogCount(SqliteLocalDb db)
|
|
{
|
|
var r = await db.QueryAsync("SELECT COUNT(*) FROM __localdb_oplog", x => x.GetInt64(0));
|
|
return r[0];
|
|
}
|
|
|
|
private static SnapshotRow SnapRow(long id, string sku, long hlc, string nodeId = "peer") =>
|
|
new()
|
|
{
|
|
PkJson = $"{{\"id\":{id}}}",
|
|
RowJson = $"{{\"id\":{id},\"sku\":\"{sku}\",\"qty\":1}}",
|
|
Hlc = hlc,
|
|
NodeId = nodeId,
|
|
IsTombstone = false,
|
|
};
|
|
|
|
[Fact]
|
|
public async Task WipedPeer_AfterEverythingWasAckedAndPruned_StillGetsSnapshot()
|
|
{
|
|
// The gap the OtOpcUa Phase 1 live gate found (check 4). A converged pair prunes its oplog
|
|
// on ack, so the steady state is "no oplog at all". If the peer is then wiped — a rebuilt
|
|
// node, a lost volume — it comes back with an empty database and a zero watermark, and the
|
|
// healthy side has no delta left to send it. Snapshot detection compared the peer's
|
|
// watermark against the oldest oplog row, which does not exist, so it concluded no snapshot
|
|
// was needed and the wiped node stayed empty indefinitely: the one case where replication
|
|
// most obviously should heal a node was the case it could not.
|
|
var a = await NewSideAsync();
|
|
for (var i = 1; i <= 3; i++)
|
|
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
|
|
|
|
await a.Store.RecordPeerAckAsync(await a.Store.GetMaxSeqAsync());
|
|
Assert.Equal(0, await OplogCount(a.Db));
|
|
// Not the already-covered cap-exceeded path: nothing here flagged a resync.
|
|
Assert.Equal(0, await NeedsSnapshot(a.Db));
|
|
|
|
// A brand-new database is exactly what a wiped node is: no rows, watermark zero.
|
|
var b = await NewSideAsync();
|
|
Assert.Empty(await ReadOrders(b.Db));
|
|
|
|
var (da, db) = DuplexPair();
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var runA = a.Session.RunAsync(da, cts.Token);
|
|
var runB = b.Session.RunAsync(db, cts.Token);
|
|
|
|
await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 3, RunTimeout);
|
|
Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
|
|
|
|
cts.Cancel();
|
|
await SwallowAsync(runA);
|
|
await SwallowAsync(runB);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task PrunedPeer_GetsSnapshot_ThenConverges()
|
|
{
|
|
var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 });
|
|
var b = await NewSideAsync();
|
|
for (var i = 1; i <= 6; i++)
|
|
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
|
|
|
|
// Prune A's oplog below B's (empty) watermark and flag a resync: incremental deltas alone
|
|
// can no longer carry rows 1..4 to B.
|
|
Assert.True(await a.Store.EnforceCapsAsync());
|
|
var asOfSeq = await a.Store.GetMaxSeqAsync();
|
|
|
|
var (da, db) = DuplexPair();
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var runA = a.Session.RunAsync(da, cts.Token);
|
|
var runB = b.Session.RunAsync(db, cts.Token);
|
|
|
|
await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 6, RunTimeout);
|
|
Assert.Equal(await ReadOrders(a.Db), await ReadOrders(b.Db));
|
|
// The snapshot covered everything through asOfSeq, so the pump starts above it (the surviving
|
|
// tail oplog rows are not re-sent as deltas).
|
|
Assert.True(a.Session.SentThruSeq >= asOfSeq, $"SentThruSeq {a.Session.SentThruSeq} < asOfSeq {asOfSeq}");
|
|
|
|
// Deltas resume after the snapshot: a post-snapshot write on A reaches B incrementally.
|
|
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (7, 'POST', 7)");
|
|
await WaitForAsync(async () => (await ReadOrders(b.Db)).Any(r => r.Id == 7), RunTimeout);
|
|
|
|
cts.Cancel();
|
|
await SwallowAsync(runA);
|
|
await SwallowAsync(runB);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Snapshot_MergesNotTruncates()
|
|
{
|
|
var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 });
|
|
var b = await NewSideAsync();
|
|
|
|
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A_OLD', 1)");
|
|
await Task.Delay(5);
|
|
// B's row for the same pk is written later -> strictly higher HLC, so it must survive the merge.
|
|
await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'B_NEW', 9)");
|
|
// A owes B a snapshot; a single-row oplog would not trip the cap, so flag it directly.
|
|
await a.Store.SetNeedsSnapshotAsync(true);
|
|
|
|
var (da, db) = DuplexPair();
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var runA = a.Session.RunAsync(da, cts.Token);
|
|
var runB = b.Session.RunAsync(db, cts.Token);
|
|
|
|
async Task<bool> Converged()
|
|
{
|
|
var ra = await ReadOrders(a.Db);
|
|
var rb = await ReadOrders(b.Db);
|
|
return ra.Count == 1 && rb.Count == 1 && ra[0].Sku == "B_NEW" && rb[0].Sku == "B_NEW";
|
|
}
|
|
|
|
await WaitForAsync(Converged, RunTimeout);
|
|
|
|
cts.Cancel();
|
|
await SwallowAsync(runA);
|
|
await SwallowAsync(runB);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Snapshot_CarriesTombstones()
|
|
{
|
|
var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 });
|
|
var b = await NewSideAsync();
|
|
|
|
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'A1', 1), (2, 'A2', 2)");
|
|
await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (1, 'B1', 1), (2, 'B2', 2)");
|
|
await Task.Delay(5);
|
|
// A deletes id=1 last (newest HLC), so the snapshot's tombstone must win over B's live id=1.
|
|
await a.Db.ExecuteAsync("DELETE FROM orders WHERE id = 1");
|
|
Assert.True(await a.Store.EnforceCapsAsync());
|
|
|
|
var (da, db) = DuplexPair();
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var runA = a.Session.RunAsync(da, cts.Token);
|
|
var runB = b.Session.RunAsync(db, cts.Token);
|
|
|
|
await WaitForAsync(async () =>
|
|
{
|
|
var rb = await ReadOrders(b.Db);
|
|
return rb.Count == 1 && rb[0].Id == 2;
|
|
}, RunTimeout);
|
|
|
|
Assert.DoesNotContain(await ReadOrders(b.Db), r => r.Id == 1);
|
|
Assert.DoesNotContain(await ReadOrders(a.Db), r => r.Id == 1);
|
|
|
|
cts.Cancel();
|
|
await SwallowAsync(runA);
|
|
await SwallowAsync(runB);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Snapshot_WatermarkNoOp_WhenSeqZero()
|
|
{
|
|
// Snapshot rows carry seq 0; applying them must not regress an already-advanced watermark.
|
|
var a = await NewSideAsync();
|
|
await a.Store.SetLastAppliedRemoteSeqAsync(50, 12345);
|
|
|
|
var entries = new List<OplogEntryRecord>
|
|
{
|
|
new(0, "orders", "{\"id\":1}", "{\"id\":1,\"sku\":\"S\",\"qty\":1}", 6_000_000, "peer", false),
|
|
};
|
|
await a.Applier.ApplyBatchAsync(entries, default);
|
|
|
|
Assert.Equal(50, await LastAppliedRemoteSeq(a.Db)); // MAX(50, 0) -> unchanged
|
|
Assert.Single(await ReadOrders(a.Db));
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SnapshotComplete_AdvancesWatermark_DeltasResume()
|
|
{
|
|
var a = await NewSideAsync();
|
|
var (duplex, toSession, fromSession) = ScriptedPeer();
|
|
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var run = a.Session.RunAsync(duplex, cts.Token);
|
|
|
|
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
|
|
|
|
await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 100 } }, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage
|
|
{
|
|
SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } },
|
|
}, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 100 } }, cts.Token);
|
|
|
|
await WaitForAsync(async () => await LastAppliedRemoteSeq(a.Db) == 100, RunTimeout);
|
|
Assert.Single(await ReadOrders(a.Db));
|
|
|
|
// A normal delta above the snapshot horizon resumes incrementally.
|
|
await toSession.WriteAsync(new SyncMessage
|
|
{
|
|
DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
|
|
{
|
|
Seq = 101, TableName = "orders", PkJson = "{\"id\":2}",
|
|
RowJson = "{\"id\":2,\"sku\":\"D\",\"qty\":1}", Hlc = 6_000_001, NodeId = "peer", IsTombstone = false,
|
|
} } },
|
|
}, cts.Token);
|
|
var ack = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
|
|
Assert.Equal(101, ack.DeltaAck.AppliedThruSeq);
|
|
Assert.Equal(101, await LastAppliedRemoteSeq(a.Db));
|
|
Assert.Equal(2, (await ReadOrders(a.Db)).Count);
|
|
|
|
cts.Cancel();
|
|
await SwallowAsync(run);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SnapshotBatch_WithoutBegin_IsProtocolError()
|
|
{
|
|
var a = await NewSideAsync();
|
|
var (duplex, toSession, fromSession) = ScriptedPeer();
|
|
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var run = a.Session.RunAsync(duplex, cts.Token);
|
|
|
|
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage
|
|
{
|
|
SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "ORPHAN", 6_000_000) } },
|
|
}, cts.Token);
|
|
|
|
await Assert.ThrowsAsync<InvalidOperationException>(() => run);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SnapshotComplete_WithoutBegin_IsProtocolError()
|
|
{
|
|
var a = await NewSideAsync();
|
|
var (duplex, toSession, fromSession) = ScriptedPeer();
|
|
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var run = a.Session.RunAsync(duplex, cts.Token);
|
|
|
|
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 9 } }, cts.Token);
|
|
|
|
await Assert.ThrowsAsync<InvalidOperationException>(() => run);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task SnapshotComplete_EmptyOplogAsOfZero_DoesNotRegress()
|
|
{
|
|
// A sender with an empty oplog streams as_of_seq 0; a receiver already at watermark 500 must
|
|
// keep 500 (MAX-guarded complete), or it would re-request everything on the next handshake.
|
|
var a = await NewSideAsync();
|
|
await a.Store.SetLastAppliedRemoteSeqAsync(500, 0);
|
|
var (duplex, toSession, fromSession) = ScriptedPeer();
|
|
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var run = a.Session.RunAsync(duplex, cts.Token);
|
|
|
|
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 0 } }, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage
|
|
{
|
|
SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP", 6_000_000) } },
|
|
}, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 0 } }, cts.Token);
|
|
|
|
// Barrier (seq 0 does not move the watermark): its ack proves Complete{0} was processed.
|
|
await toSession.WriteAsync(new SyncMessage
|
|
{
|
|
DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
|
|
{
|
|
Seq = 0, TableName = "orders", PkJson = "{\"id\":2}",
|
|
RowJson = "{\"id\":2,\"sku\":\"BARRIER\",\"qty\":1}", Hlc = 6_000_001, NodeId = "peer", IsTombstone = false,
|
|
} } },
|
|
}, cts.Token);
|
|
await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
|
|
|
|
Assert.Equal(500, await LastAppliedRemoteSeq(a.Db));
|
|
Assert.Equal(2, (await ReadOrders(a.Db)).Count); // the snapshot row still merged
|
|
|
|
cts.Cancel();
|
|
await SwallowAsync(run);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task DeltaDuringSnapshot_AppliesSafely()
|
|
{
|
|
var a = await NewSideAsync();
|
|
var (duplex, toSession, fromSession) = ScriptedPeer();
|
|
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var run = a.Session.RunAsync(duplex, cts.Token);
|
|
|
|
await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token);
|
|
|
|
await toSession.WriteAsync(new SyncMessage { SnapshotBegin = new SnapshotBegin { AsOfSeq = 200 } }, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage
|
|
{
|
|
SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(1, "SNAP1", 6_000_000) } },
|
|
}, cts.Token);
|
|
// A normal delta interleaved MID-snapshot must still apply (LWW makes interleaving safe).
|
|
await toSession.WriteAsync(new SyncMessage
|
|
{
|
|
DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
|
|
{
|
|
Seq = 205, TableName = "orders", PkJson = "{\"id\":2}",
|
|
RowJson = "{\"id\":2,\"sku\":\"DELTA\",\"qty\":1}", Hlc = 6_000_010, NodeId = "peer", IsTombstone = false,
|
|
} } },
|
|
}, cts.Token);
|
|
await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage
|
|
{
|
|
SnapshotBatch = new SnapshotBatch { TableName = "orders", Rows = { SnapRow(3, "SNAP3", 6_000_020) } },
|
|
}, cts.Token);
|
|
await toSession.WriteAsync(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 200 } }, cts.Token);
|
|
|
|
await WaitForAsync(async () => (await ReadOrders(a.Db)).Count == 3, RunTimeout);
|
|
var rows = await ReadOrders(a.Db);
|
|
Assert.Equal([1L, 2L, 3L], rows.Select(r => r.Id));
|
|
|
|
// Barrier (seq 0 does not move the watermark): its ack proves Complete{200} was processed;
|
|
// the MAX-guarded complete must not have regressed the delta's 205.
|
|
await toSession.WriteAsync(new SyncMessage
|
|
{
|
|
DeltaBatch = new DeltaBatch { Entries = { new OplogEntry
|
|
{
|
|
Seq = 0, TableName = "orders", PkJson = "{\"id\":4}",
|
|
RowJson = "{\"id\":4,\"sku\":\"BARRIER\",\"qty\":1}", Hlc = 6_000_030, NodeId = "peer", IsTombstone = false,
|
|
} } },
|
|
}, cts.Token);
|
|
await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
|
|
Assert.Equal(205, await LastAppliedRemoteSeq(a.Db));
|
|
|
|
cts.Cancel();
|
|
await SwallowAsync(run);
|
|
}
|
|
|
|
[Fact]
|
|
public async Task Sender_ClearsNeedsSnapshotFlag()
|
|
{
|
|
var a = await NewSideAsync(new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), MaxOplogRows = 2 });
|
|
var b = await NewSideAsync();
|
|
for (var i = 1; i <= 5; i++)
|
|
await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i });
|
|
Assert.True(await a.Store.EnforceCapsAsync());
|
|
Assert.Equal(1, await NeedsSnapshot(a.Db));
|
|
|
|
var (da, db) = DuplexPair();
|
|
using var cts = new CancellationTokenSource(RunTimeout);
|
|
var runA = a.Session.RunAsync(da, cts.Token);
|
|
var runB = b.Session.RunAsync(db, cts.Token);
|
|
|
|
await WaitForAsync(async () => (await ReadOrders(b.Db)).Count == 5, RunTimeout);
|
|
await WaitForAsync(async () => await NeedsSnapshot(a.Db) == 0, RunTimeout);
|
|
|
|
cts.Cancel();
|
|
await SwallowAsync(runA);
|
|
await SwallowAsync(runB);
|
|
}
|
|
}
|