test(localdb): pin drift-watermark branch; guard writer loop against completed-lane spin

2026-07-17 22:50:33 -04:00
parent 92ccbac8db
commit 183901e8ac
2 changed files with 105 additions and 6 deletions
@@ -157,6 +157,8 @@ internal sealed class SyncSession
    {
        Task<bool>? controlWait = null;
        Task<bool>? dataWait = null;
+        var controlDone = false;
+        var dataDone = false;
        while (true)
        {
            ct.ThrowIfCancellationRequested();
@@ -165,14 +167,27 @@ internal sealed class SyncSession
                await duplex.Send(message, ct);
                continue;
            }
+            // WaitToReadAsync => false means the lane was COMPLETED and is drained: stop waiting on
+            // it (re-creating its waiter would instantly re-complete false — a 100% CPU spin) and
+            // exit once both lanes are terminal. Nothing completes the lanes today, but Task 12 or
+            // a future refactor must not silently detonate this loop.
+            if (controlDone && dataDone)
+                return;
            // 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;
+            if (!controlDone) controlWait ??= control.WaitToReadAsync(ct).AsTask();
+            if (!dataDone) dataWait ??= data.WaitToReadAsync(ct).AsTask();
+            await Task.WhenAny(controlWait ?? dataWait!, dataWait ?? controlWait!);
+            if (controlWait is { IsCompleted: true })
+            {
+                controlDone = !await controlWait;
+                controlWait = null;
+            }
+            if (dataWait is { IsCompleted: true })
+            {
+                dataDone = !await dataWait;
+                dataWait = null;
+            }
        }
    }

@@ -278,6 +293,8 @@ internal sealed class SyncSession
        await controlLane.WriteAsync(new SyncMessage { DeltaAck = new DeltaAck { AppliedThruSeq = ackThru } }, ct);
    }

+    // Not transactional with the watermark update: a crash between them redelivers the batch and
+    // dead-letters the same entry again (at-least-once — a benign duplicate dead-letter row).
    private Task DeadLetterDriftedAsync(
        OplogEntryRecord entry, long physicalMs, long nowMs, DateTimeOffset now, CancellationToken ct) =>
        _db.ExecuteAsync(
@@ -145,6 +145,12 @@ public sealed class SyncSessionTests : IDisposable
        return r[0];
    }

+    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 OplogEntry ProtoEntry(long seq, long id, string sku, long hlc, string nodeId = "peer") =>
        new()
        {
@@ -461,6 +467,82 @@ public sealed class SyncSessionTests : IDisposable
        await SwallowAsync(run);
    }

+    [Fact]
+    public async Task TopSeqDrifted_WatermarkStillAdvances()
+    {
+        // The batch's TOP seq is dead-lettered, so the applier's watermark stops below it; the
+        // session's standalone MAX update must still advance the durable watermark and the ack.
+        var a = await NewSideAsync(
+            options: new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), FailClosedOnDrift = true });
+        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 =
+                {
+                    ProtoEntry(1, 1, "CLEAN", 5_000_000),
+                    ProtoEntry(2, 2, "DRIFTED", AheadHlc(TimeSpan.FromMinutes(10))),
+                },
+            },
+        }, cts.Token);
+        var ack = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
+
+        Assert.Equal(2, ack.DeltaAck.AppliedThruSeq);
+        Assert.Equal(2, await LastAppliedRemoteSeq(a.Db));
+        var rows = await ReadOrders(a.Db);
+        Assert.Single(rows);
+        Assert.Equal(1, rows[0].Id);
+        var dead = await a.Db.QueryAsync("SELECT pk_json FROM __localdb_dead_letter", x => x.GetString(0));
+        Assert.Equal(["{\"id\":2}"], dead);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
+    [Fact]
+    public async Task AllEntriesDrifted_WatermarkStillAdvances()
+    {
+        // Every entry dead-lettered: the applier sees an empty batch (AppliedThruSeq 0), yet the
+        // durable watermark and the ack must still cover the batch's max seq.
+        var a = await NewSideAsync(
+            options: new ReplicationOptions { FlushInterval = TimeSpan.FromMilliseconds(20), FailClosedOnDrift = true });
+        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 =
+                {
+                    ProtoEntry(1, 1, "DRIFTED1", AheadHlc(TimeSpan.FromMinutes(10))),
+                    ProtoEntry(2, 2, "DRIFTED2", AheadHlc(TimeSpan.FromMinutes(11))),
+                },
+            },
+        }, cts.Token);
+        var ack = await NextOfCaseAsync(fromSession, SyncMessage.MsgOneofCase.DeltaAck, cts.Token);
+
+        Assert.Equal(2, ack.DeltaAck.AppliedThruSeq);
+        Assert.Equal(2, await LastAppliedRemoteSeq(a.Db));
+        Assert.Empty(await ReadOrders(a.Db));
+        var dead = await a.Db.QueryAsync("SELECT pk_json FROM __localdb_dead_letter ORDER BY pk_json", x => x.GetString(0));
+        Assert.Equal(["{\"id\":1}", "{\"id\":2}"], dead);
+
+        cts.Cancel();
+        await SwallowAsync(run);
+    }
+
    [Fact]
    public async Task PeerNeedsSnapshot_InvokesSnapshotHooks()
    {