diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncDuplex.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncDuplex.cs index a2a9043..1341542 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncDuplex.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncDuplex.cs @@ -6,9 +6,9 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal; /// /// The transport seam a runs over: an outbound send delegate plus an /// inbound message reader. Task 11 adapts a gRPC duplex stream to this; tests wire two in-memory -/// pairs crosswise. may be called concurrently by the -/// session's pump and receive loops — the session serializes those calls, so an adapter's writer -/// only ever sees one send at a time. +/// pairs crosswise. The session's single writer task is the only caller +/// of , so an adapter's stream writer never sees concurrent sends. Send MUST +/// honor its CancellationToken even when blocked on transport flow control (teardown relies on it). /// internal sealed class SyncDuplex { diff --git a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs index 0aa796d..f31c8e7 100644 --- a/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs +++ b/ZB.MOM.WW.LocalDb/src/ZB.MOM.WW.LocalDb.Replication/Internal/SyncSession.cs @@ -1,3 +1,5 @@ +using System.Globalization; +using System.Threading.Channels; using Microsoft.Extensions.Logging; using ZB.MOM.WW.LocalDb.Contracts; using ZB.MOM.WW.LocalDb.Hlc; @@ -9,13 +11,20 @@ namespace ZB.MOM.WW.LocalDb.Replication.Internal; /// The symmetric 2-node sync protocol as one state machine: both the gRPC service and the gRPC /// client run this identical logic over a . It performs the fail-closed /// handshake, then runs a bidirectional steady state — an outbound change pump and an inbound -/// apply/ack loop — until the token is cancelled or the stream ends. Local writes never block on -/// the peer; every failure is a typed exception that tears down both loops and propagates. +/// apply/ack loop — until the token is cancelled or the stream ends. All outbound traffic flows +/// through a single writer task fed by two lanes: an unbounded CONTROL lane (handshake messages, +/// acks) and a bounded DATA lane (delta/snapshot batches). Receiving therefore never blocks on the +/// ability to write — acks always fit — while the bounded data lane is the pump's backpressure. +/// Local writes never block on the peer; the first fault tears down all loops and propagates. /// internal sealed class SyncSession { private const uint LibSchemaVersion = 1; + // Data-lane depth bounds outbound batch buffering; control-lane volume is bounded by the + // inbound batch rate (the peer's flow control), so its lane can safely be unbounded. + private const int DataLaneCapacity = 16; + private readonly SqliteLocalDb _db; private readonly OplogStore _store; private readonly LwwApplier _applier; @@ -23,14 +32,12 @@ internal sealed class SyncSession private readonly ILogger _logger; private readonly Func _utcNow; - // gRPC stream writers are not safe for concurrent writes; the pump and receive loops both send, - // so every outbound message goes through this gate to serialize them into one writer. - private readonly SemaphoreSlim _sendGate = new(1, 1); - // Highest oplog seq handed to the wire this session; the pump reads strictly above it so a batch // is never re-sent before its ack (the reliable ordered stream delivers exactly once in-session). private long _sentThruSeq; private long _peerAckedSeq; + private string? _peerNodeId; + private bool _driftWarned; public SyncSession( SqliteLocalDb db, OplogStore store, LwwApplier applier, ReplicationOptions options, ILogger logger, @@ -44,24 +51,74 @@ internal sealed class SyncSession _utcNow = utcNow ?? (() => DateTimeOffset.UtcNow); } - /// Invoked when the peer's HandshakeAck requests a snapshot before deltas. Task 12 fills this in. - public Func? SnapshotSender { get; set; } + /// + /// Invoked when the peer's HandshakeAck requests a snapshot before deltas. Receives a gated + /// send delegate that enqueues on the bounded data lane (never the raw transport). Task 12 fills this in. + /// + public Func, CancellationToken, Task>? SnapshotSender { get; set; } - /// Invoked on an inbound SnapshotBegin. Task 12 fills this in. - public Func? SnapshotReceiver { get; set; } + /// + /// Invoked on an inbound SnapshotBegin, with the triggering message and a gated data-lane send + /// delegate. Task 12 fills this in. + /// + public Func, CancellationToken, Task>? SnapshotReceiver { get; set; } /// Highest seq the peer has acknowledged applying this session. public long PeerAckedSeq => Interlocked.Read(ref _peerAckedSeq); - /// The peer node's identity, once the handshake has completed. - public string? PeerNodeId { get; private set; } + /// The peer node's identity, once the handshake has completed. Volatile: written on the session task, read from test/metric threads. + public string? PeerNodeId + { + get => Volatile.Read(ref _peerNodeId); + private set => Volatile.Write(ref _peerNodeId, value); + } public async Task RunAsync(SyncDuplex duplex, CancellationToken ct) { - var peerState = await _store.GetPeerStateAsync(ct); + var control = Channel.CreateUnbounded(new UnboundedChannelOptions { SingleReader = true }); + var data = Channel.CreateBounded(new BoundedChannelOptions(DataLaneCapacity) { SingleReader = true }); + Task GatedDataSend(SyncMessage message, CancellationToken token) => data.Writer.WriteAsync(message, token).AsTask(); + using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct); + var writer = WriterLoopAsync(duplex, control.Reader, data.Reader, linked.Token); + try + { + await HandshakeAsync(duplex, control.Writer, GatedDataSend, linked.Token); + + var pump = PumpLoopAsync(data.Writer, linked.Token); + var receive = ReceiveLoopAsync(duplex, control.Writer, GatedDataSend, linked.Token); + var loops = new[] { pump, receive, writer }; + + // The first COMPLETED task is the primary outcome. Cancel the siblings, observe every + // one of them under a swallow (so no secondary exception can replace the primary and + // nothing is left running/unobserved), then surface the primary by awaiting it. + var finished = await Task.WhenAny(loops); + linked.Cancel(); + foreach (var loop in loops) + if (!ReferenceEquals(loop, finished)) + await DrainSecondaryAsync(loop); + await finished; + } + catch + { + // Handshake-phase faults (and the primary rethrow above) land here: the writer may + // still be running, so stop and observe it before the primary exception surfaces. + linked.Cancel(); + await DrainSecondaryAsync(writer); + throw; + } + } + + private async Task HandshakeAsync( + SyncDuplex duplex, ChannelWriter controlLane, + Func dataSend, CancellationToken ct) + { + var peerState = await _store.GetPeerStateAsync(ct); var localHandshake = BuildHandshake(peerState); - await SendAsync(duplex, new SyncMessage { Handshake = localHandshake }, ct); + + // Both peers send before reading, so the transport must buffer at least two in-flight + // messages (Handshake + HandshakeAck); gRPC's initial flow-control window covers that. + await controlLane.WriteAsync(new SyncMessage { Handshake = localHandshake }, ct); var firstInbound = await ReceiveOneAsync(duplex, ct); if (firstInbound.MsgCase != SyncMessage.MsgOneofCase.Handshake) @@ -77,7 +134,8 @@ internal sealed class SyncSession Interlocked.Exchange(ref _peerAckedSeq, peerHandshake.LastAppliedRemoteSeq); var snapshotRequired = await ComputeSnapshotRequiredAsync(peerHandshake, peerState, ct); - await SendAsync(duplex, new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotRequired } }, ct); + await controlLane.WriteAsync( + new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = _db.NodeId, SnapshotRequired = snapshotRequired } }, ct); var peerAckMsg = await ReceiveOneAsync(duplex, ct); if (peerAckMsg.MsgCase != SyncMessage.MsgOneofCase.HandshakeAck) @@ -88,39 +146,49 @@ internal sealed class SyncSession { if (SnapshotSender is null) throw new NotSupportedException("snapshot required but no snapshot sender configured"); - await SnapshotSender(duplex, ct); + await SnapshotSender(dataSend, ct); } - - await RunSteadyStateAsync(duplex, ct); } - private async Task RunSteadyStateAsync(SyncDuplex duplex, CancellationToken ct) + // The ONLY caller of duplex.Send. Control (handshake + acks) always preempts data so the peer's + // inbound processing is never starved of its acks behind a backlog of DeltaBatches. + private static async Task WriterLoopAsync( + SyncDuplex duplex, ChannelReader control, ChannelReader data, CancellationToken ct) { - using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct); - var pump = PumpLoopAsync(duplex, linked.Token); - var receive = ReceiveLoopAsync(duplex, linked.Token); - - // Whichever loop completes first (fault or stream-end) drives teardown: cancel the sibling, - // drain it (swallowing only the cancellation we caused), then surface the primary outcome. - var finished = await Task.WhenAny(pump, receive); - linked.Cancel(); - var other = finished == pump ? receive : pump; - try { await other; } - catch (OperationCanceledException) { } - await finished; - } - - private async Task PumpLoopAsync(SyncDuplex duplex, CancellationToken ct) - { - while (!ct.IsCancellationRequested) + Task? controlWait = null; + Task? dataWait = null; + while (true) { + ct.ThrowIfCancellationRequested(); + if (control.TryRead(out var message) || data.TryRead(out message)) + { + await duplex.Send(message, ct); + continue; + } + // At most one outstanding waiter per lane (SingleReader); a stale already-completed + // waiter only costs one extra TryRead pass. + controlWait ??= control.WaitToReadAsync(ct).AsTask(); + dataWait ??= data.WaitToReadAsync(ct).AsTask(); + var completed = await Task.WhenAny(controlWait, dataWait); + await completed; + if (controlWait.IsCompleted) controlWait = null; + if (dataWait.IsCompleted) dataWait = null; + } + } + + private async Task PumpLoopAsync(ChannelWriter dataLane, CancellationToken ct) + { + while (true) + { + ct.ThrowIfCancellationRequested(); var batch = await _store.ReadBatchAboveAsync(_sentThruSeq, _options.MaxBatchSize, ct); if (batch.Count > 0) { var delta = new DeltaBatch(); foreach (var entry in batch) delta.Entries.Add(ToProto(entry)); - await SendAsync(duplex, new SyncMessage { DeltaBatch = delta }, ct); + // Blocking here on a full data lane IS the pump's backpressure. + await dataLane.WriteAsync(new SyncMessage { DeltaBatch = delta }, ct); _sentThruSeq = batch[^1].Seq; continue; } @@ -129,7 +197,9 @@ internal sealed class SyncSession } } - private async Task ReceiveLoopAsync(SyncDuplex duplex, CancellationToken ct) + private async Task ReceiveLoopAsync( + SyncDuplex duplex, ChannelWriter controlLane, + Func dataSend, CancellationToken ct) { while (await duplex.Inbox.WaitToReadAsync(ct)) { @@ -138,7 +208,7 @@ internal sealed class SyncSession switch (msg.MsgCase) { case SyncMessage.MsgOneofCase.DeltaBatch: - await HandleDeltaBatchAsync(duplex, msg.DeltaBatch, ct); + await HandleDeltaBatchAsync(controlLane, msg.DeltaBatch, ct); break; case SyncMessage.MsgOneofCase.DeltaAck: var acked = msg.DeltaAck.AppliedThruSeq; @@ -148,7 +218,7 @@ internal sealed class SyncSession case SyncMessage.MsgOneofCase.SnapshotBegin: if (SnapshotReceiver is null) throw new NotSupportedException("snapshot received but no snapshot receiver configured"); - await SnapshotReceiver(msg.SnapshotBegin, duplex, ct); + await SnapshotReceiver(msg.SnapshotBegin, dataSend, ct); break; default: throw new InvalidOperationException( @@ -158,33 +228,82 @@ internal sealed class SyncSession } } - private async Task HandleDeltaBatchAsync(SyncDuplex duplex, DeltaBatch batch, CancellationToken ct) + private async Task HandleDeltaBatchAsync(ChannelWriter controlLane, DeltaBatch batch, CancellationToken ct) { - var entries = new List(batch.Entries.Count); - var maxHlc = 0L; - foreach (var e in batch.Entries) - { - entries.Add(FromProto(e)); - if (e.Hlc > maxHlc) maxHlc = e.Hlc; - } + if (batch.Entries.Count == 0) + return; - if (entries.Count > 0) + var now = _utcNow(); + var nowMs = now.ToUnixTimeMilliseconds(); + var driftLimitMs = nowMs + (long)_options.MaxHlcDriftAhead.TotalMilliseconds; + var toApply = new List(batch.Entries.Count); + var maxSeq = 0L; + + foreach (var proto in batch.Entries) { - var maxHlcMs = HybridLogicalClock.PhysicalMs(maxHlc); - var driftLimitMs = _utcNow().Add(_options.MaxHlcDriftAhead).ToUnixTimeMilliseconds(); - if (maxHlcMs > driftLimitMs) + if (proto.Seq > maxSeq) maxSeq = proto.Seq; + var entry = FromProto(proto); + var physicalMs = HybridLogicalClock.PhysicalMs(entry.Hlc); + if (physicalMs > driftLimitMs) { - _logger.LogWarning( - "Inbound delta HLC physical time {HlcMs} ms exceeds local clock + MaxHlcDriftAhead ({LimitMs} ms).", - maxHlcMs, driftLimitMs); + if (!_driftWarned) + { + // Rate-limited to one warning per session: drift is usually a sustained clock + // condition and would otherwise flood the log at the delta rate. + _driftWarned = true; + _logger.LogWarning( + "Inbound oplog entry HLC physical time {EntryMs} ms is ahead of local UTC {NowMs} ms by more than MaxHlcDriftAhead ({Drift}). Further drift this session is not re-logged.", + physicalMs, nowMs, _options.MaxHlcDriftAhead); + } + // Dead-letter is the fail-closed disposition: faulting the session instead would + // livelock on reconnect (the peer resends the same drifted entries forever). if (_options.FailClosedOnDrift) - throw new InvalidOperationException( - $"Inbound HLC physical time {maxHlcMs} ms exceeds MaxHlcDriftAhead limit {driftLimitMs} ms; rejecting batch (fail-closed)."); + { + await DeadLetterDriftedAsync(entry, physicalMs, nowMs, now, ct); + continue; + } } + toApply.Add(entry); } - var result = await _applier.ApplyBatchAsync(entries, ct); - await SendAsync(duplex, new SyncMessage { DeltaAck = new DeltaAck { AppliedThruSeq = result.AppliedThruSeq } }, ct); + var result = await _applier.ApplyBatchAsync(toApply, ct); + var ackThru = Math.Max(result.AppliedThruSeq, maxSeq); + if (ackThru > result.AppliedThruSeq) + // Dead-lettered entries carried the batch's top seqs; advance the durable remote + // watermark over them (MAX-guarded like the applier's) so a reconnect handshake does + // not make the peer resend what this node has already discarded. + await _db.ExecuteAsync( + "UPDATE __localdb_peer_state SET last_applied_remote_seq = MAX(last_applied_remote_seq, @seq) WHERE id = 1", + new { seq = ackThru }, ct); + await controlLane.WriteAsync(new SyncMessage { DeltaAck = new DeltaAck { AppliedThruSeq = ackThru } }, ct); + } + + 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); + + private async Task DrainSecondaryAsync(Task task) + { + try + { + await task; + } + catch (Exception ex) + { + _logger.LogDebug(ex, "Secondary sync-session task ended during teardown."); + } } private Handshake BuildHandshake(PeerState peerState) @@ -237,13 +356,6 @@ internal sealed class SyncSession return minSeq > 0 && peerHandshake.LastAppliedRemoteSeq < minSeq - 1; } - private async Task SendAsync(SyncDuplex duplex, SyncMessage message, CancellationToken ct) - { - await _sendGate.WaitAsync(ct); - try { await duplex.Send(message, ct); } - finally { _sendGate.Release(); } - } - private static async Task ReceiveOneAsync(SyncDuplex duplex, CancellationToken ct) { if (await duplex.Inbox.WaitToReadAsync(ct) && duplex.Inbox.TryRead(out var message)) diff --git a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SyncSessionTests.cs b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SyncSessionTests.cs index 5ed8dd1..208f73d 100644 --- a/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SyncSessionTests.cs +++ b/ZB.MOM.WW.LocalDb/tests/ZB.MOM.WW.LocalDb.Tests/SyncSessionTests.cs @@ -20,7 +20,7 @@ public sealed class SyncSessionTests : IDisposable { foreach (var d in _disposables) d.Dispose(); - SqliteConnection_ClearAllPools(); + Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); foreach (var p in _paths) { if (File.Exists(p)) File.Delete(p); @@ -29,8 +29,6 @@ public sealed class SyncSessionTests : IDisposable } } - private static void SqliteConnection_ClearAllPools() => Microsoft.Data.Sqlite.SqliteConnection.ClearAllPools(); - private sealed record Side( SqliteLocalDb Db, OplogStore Store, LwwApplier Applier, SyncSession Session); @@ -55,10 +53,13 @@ public sealed class SyncSessionTests : IDisposable return new Side(db, store, applier, session); } - private static (SyncDuplex A, SyncDuplex B) DuplexPair() + private static (SyncDuplex A, SyncDuplex B) DuplexPair(int? capacity = null) { - var aToB = Channel.CreateUnbounded(); - var bToA = Channel.CreateUnbounded(); + Channel New() => capacity is int c + ? Channel.CreateBounded(c) + : Channel.CreateUnbounded(); + var aToB = New(); + var bToA = New(); var a = new SyncDuplex { Send = (m, ct) => aToB.Writer.WriteAsync(m, ct).AsTask(), @@ -99,6 +100,14 @@ public sealed class SyncSessionTests : IDisposable return msg!; } + private static async Task NextOfCaseAsync( + ChannelReader reader, SyncMessage.MsgOneofCase msgCase, CancellationToken ct) + { + SyncMessage msg; + do { msg = await NextAsync(reader, ct); } while (msg.MsgCase != msgCase); + return msg; + } + private static async Task SwallowAsync(Task task) { try { await task; } @@ -136,6 +145,21 @@ public sealed class SyncSessionTests : IDisposable return r[0]; } + private static OplogEntry ProtoEntry(long seq, long id, string sku, long hlc, string nodeId = "peer") => + new() + { + Seq = seq, + TableName = "orders", + PkJson = $"{{\"id\":{id}}}", + RowJson = $"{{\"id\":{id},\"sku\":\"{sku}\",\"qty\":1}}", + Hlc = hlc, + NodeId = nodeId, + IsTombstone = false, + }; + + private static long AheadHlc(TimeSpan ahead) => + DateTimeOffset.UtcNow.Add(ahead).ToUnixTimeMilliseconds() << 16; + [Fact] public async Task Handshake_Exchanged_BothProceed() { @@ -255,6 +279,101 @@ public sealed class SyncSessionTests : IDisposable await SwallowAsync(runB); } + [Fact] + public async Task Backpressure_BoundedDuplex_NoDeadlock() + { + // Capacity-1 crosswise channels with backlog in BOTH directions: under the old shared + // send-gate design each side's pump held the gate while blocked on the full channel, and + // each side's receive loop then blocked on the gate trying to send its ack — a circular + // wait. The two-lane writer breaks it: receiving enqueues acks without ever blocking. + var options = new ReplicationOptions { MaxBatchSize = 2, FlushInterval = TimeSpan.FromMilliseconds(20) }; + var a = await NewSideAsync(options: options); + var b = await NewSideAsync(options: options); + + const int rowsPerSide = 40; + for (var i = 1; i <= rowsPerSide; i++) + { + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'A', @id)", new { id = i }); + await b.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'B', @id)", new { id = 100 + i }); + } + + var (da, db) = DuplexPair(capacity: 1); + 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(a.Db)).Count == 2 * rowsPerSide && (await ReadOrders(b.Db)).Count == 2 * rowsPerSide, + RunTimeout); + + cts.Cancel(); + await SwallowAsync(runA); + await SwallowAsync(runB); + } + + [Fact] + public async Task OutOfOrderAck_DoesNotRegress() + { + var a = await NewSideAsync(); + for (var i = 1; i <= 5; i++) + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'R', @id)", new { id = i }); + + 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 { DeltaAck = new DeltaAck { AppliedThruSeq = 5 } }, cts.Token); + await toSession.WriteAsync(new SyncMessage { DeltaAck = new DeltaAck { AppliedThruSeq = 2 } }, cts.Token); + // Barrier: the session applies inbound messages in order, so its DeltaAck for this batch + // proves both out-of-order acks above were already processed. + await toSession.WriteAsync(new SyncMessage + { + DeltaBatch = new DeltaBatch { Entries = { ProtoEntry(1, 100, "BARRIER", 5_000_000) } }, + }, cts.Token); + await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token); + + Assert.Equal(5, a.Session.PeerAckedSeq); + Assert.Equal(5, await LastAckedSeq(a.Db)); + var snap = await a.Db.QueryAsync("SELECT needs_snapshot FROM __localdb_peer_state WHERE id = 1", x => x.GetInt64(0)); + Assert.Equal(0, snap[0]); + + cts.Cancel(); + await SwallowAsync(run); + } + + [Fact] + public async Task Teardown_PrimaryFaultSurfaces() + { + var a = await NewSideAsync(); + // Backlog keeps the pump loop mid-work while the receive loop faults. + for (var i = 1; i <= 10; i++) + await a.Db.ExecuteAsync("INSERT INTO orders (id, sku, qty) VALUES (@id, 'R', @id)", new { id = i }); + + 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 + { + DeltaBatch = new DeltaBatch + { + Entries = { new OplogEntry + { + Seq = 1, TableName = "widgets", PkJson = "{\"id\":9}", + RowJson = "{\"id\":9}", Hlc = 5_000_000, NodeId = "peer", IsTombstone = false, + } }, + }, + }, cts.Token); + + // The receive loop's schema fault must surface — not an OperationCanceledException from + // the torn-down pump/writer siblings. + var ex = await Assert.ThrowsAsync(() => run); + Assert.Contains("widgets", ex.Message); + } + [Fact] public async Task HlcDriftAhead_Warns() { @@ -269,33 +388,30 @@ public sealed class SyncSessionTests : IDisposable await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token); - var aheadHlc = (DateTimeOffset.UtcNow.AddMinutes(10).ToUnixTimeMilliseconds() << 16); + // Two separate drifted batches: both apply, but only ONE warning is logged per session. await toSession.WriteAsync(new SyncMessage { - DeltaBatch = new DeltaBatch - { - Entries = { new OplogEntry - { - Seq = 1, TableName = "orders", PkJson = "{\"id\":1}", - RowJson = "{\"id\":1,\"sku\":\"AHEAD\",\"qty\":9}", Hlc = aheadHlc, NodeId = "peer", IsTombstone = false, - } }, - }, + DeltaBatch = new DeltaBatch { Entries = { ProtoEntry(1, 1, "AHEAD1", AheadHlc(TimeSpan.FromMinutes(10))) } }, }, cts.Token); + var ack1 = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token); + Assert.Equal(1, ack1.DeltaAck.AppliedThruSeq); - // Drain the DeltaAck the session emits after applying. - SyncMessage ack; - do { ack = await NextAsync(fromSession, cts.Token); } while (ack.MsgCase != SyncMessage.MsgOneofCase.DeltaAck); - Assert.Equal(1, ack.DeltaAck.AppliedThruSeq); + await toSession.WriteAsync(new SyncMessage + { + DeltaBatch = new DeltaBatch { Entries = { ProtoEntry(2, 2, "AHEAD2", AheadHlc(TimeSpan.FromMinutes(11))) } }, + }, cts.Token); + var ack2 = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token); + Assert.Equal(2, ack2.DeltaAck.AppliedThruSeq); - Assert.Single(await ReadOrders(a.Db)); // applied despite drift - Assert.Contains(logger.Entries, e => e.Level == LogLevel.Warning); + Assert.Equal(2, (await ReadOrders(a.Db)).Count); // applied despite drift + Assert.Equal(1, logger.Entries.Count(e => e.Level == LogLevel.Warning)); cts.Cancel(); await SwallowAsync(run); } [Fact] - public async Task HlcDriftAhead_FailClosed_Throws() + public async Task HlcDriftAhead_FailClosed_DeadLetters() { var a = await NewSideAsync( options: new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), FailClosedOnDrift = true }); @@ -306,32 +422,59 @@ public sealed class SyncSessionTests : IDisposable await DriveHandshakeAsync(a.Db, toSession, fromSession, cts.Token); - var aheadHlc = (DateTimeOffset.UtcNow.AddMinutes(10).ToUnixTimeMilliseconds() << 16); + // One batch: seq 1 drifted (dead-lettered), seq 2 clean (applied). Watermark covers both. await toSession.WriteAsync(new SyncMessage { DeltaBatch = new DeltaBatch { - Entries = { new OplogEntry + Entries = { - Seq = 1, TableName = "orders", PkJson = "{\"id\":1}", - RowJson = "{\"id\":1,\"sku\":\"AHEAD\",\"qty\":9}", Hlc = aheadHlc, NodeId = "peer", IsTombstone = false, - } }, + ProtoEntry(1, 1, "DRIFTED", AheadHlc(TimeSpan.FromMinutes(10))), + ProtoEntry(2, 2, "CLEAN", 5_000_000), + }, }, }, cts.Token); + var ack = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token); + Assert.Equal(2, ack.DeltaAck.AppliedThruSeq); - await Assert.ThrowsAsync(() => run); - Assert.Empty(await ReadOrders(a.Db)); // rejected, not applied + var rows = await ReadOrders(a.Db); + Assert.Single(rows); + Assert.Equal(2, rows[0].Id); + + var dead = await a.Db.QueryAsync( + "SELECT pk_json, error FROM __localdb_dead_letter", + x => (Pk: x.GetString(0), Err: x.GetString(1))); + Assert.Single(dead); + Assert.Equal("{\"id\":1}", dead[0].Pk); + Assert.Contains("drift", dead[0].Err, StringComparison.OrdinalIgnoreCase); + + // The session stays alive after dead-lettering: a follow-up clean batch still applies. + await toSession.WriteAsync(new SyncMessage + { + DeltaBatch = new DeltaBatch { Entries = { ProtoEntry(3, 3, "AFTER", 5_000_001) } }, + }, cts.Token); + var ack3 = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token); + Assert.Equal(3, ack3.DeltaAck.AppliedThruSeq); + Assert.Equal(2, (await ReadOrders(a.Db)).Count); + + cts.Cancel(); + await SwallowAsync(run); } [Fact] public async Task PeerNeedsSnapshot_InvokesSnapshotHooks() { - // (a) peer's HandshakeAck.snapshot_required = true -> our SnapshotSender fires. + // (a) peer's HandshakeAck.snapshot_required = true -> our SnapshotSender fires, and its + // gated send delegate routes messages through the writer onto the wire. { var a = await NewSideAsync(); var (duplex, toSession, fromSession) = ScriptedPeer(); var sent = new TaskCompletionSource(); - a.Session.SnapshotSender = (_, _) => { sent.TrySetResult(); return Task.CompletedTask; }; + a.Session.SnapshotSender = (send, token) => + { + sent.TrySetResult(); + return send(new SyncMessage { SnapshotComplete = new SnapshotComplete { AsOfSeq = 7 } }, token); + }; using var cts = new CancellationTokenSource(RunTimeout); var run = a.Session.RunAsync(duplex, cts.Token); @@ -342,24 +485,31 @@ public sealed class SyncSessionTests : IDisposable await toSession.WriteAsync(new SyncMessage { HandshakeAck = new HandshakeAck { NodeId = "peer", SnapshotRequired = true } }, cts.Token); await sent.Task.WaitAsync(RunTimeout); + var complete = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.SnapshotComplete, cts.Token); + Assert.Equal(7, complete.SnapshotComplete.AsOfSeq); + cts.Cancel(); await SwallowAsync(run); } - // (b) inbound SnapshotBegin -> our SnapshotReceiver fires. + // (b) inbound SnapshotBegin -> our SnapshotReceiver fires with the triggering message. { var a = await NewSideAsync(); var (duplex, toSession, fromSession) = ScriptedPeer(); - var received = new TaskCompletionSource(); - a.Session.SnapshotReceiver = (_, _, _) => { received.TrySetResult(); return Task.CompletedTask; }; + var received = new TaskCompletionSource(); + a.Session.SnapshotReceiver = (begin, _, _) => + { + received.TrySetResult(begin.AsOfSeq); + return Task.CompletedTask; + }; 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 { SnapshotBegin = new SnapshotBegin { AsOfSeq = 42 } }, cts.Token); - await received.Task.WaitAsync(RunTimeout); + Assert.Equal(42, await received.Task.WaitAsync(RunTimeout)); cts.Cancel(); await SwallowAsync(run); }