From 1f79d033e842384c2e075c3db02ded5878275b61 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 17:52:21 +0200 Subject: [PATCH] A paused feed's checkpoint is proved by order, not by waiting 50ms MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit TestInvariantF1APausedFeedDoesNotMoveTheCheckpoint scripted one page, waited for the queue to pause, slept 50ms and asserted the checkpoint was absent. That is a window, not an order: a save that happens but happens slowly is a pass, so the test could false-pass and could never false-fail. It is worse than a loose window. The pause the test waits for happens INSIDE the page — the feed blocks offering the page's second event — so PageDelivered has not run for any page at the moment of the assertion. The test never observes a completed page while paused at all, so it has no guard against a per-page save, fast or slow. The walk is two pages now. The entry is present-class, so the feed holds every page's position and saves once: after the walk reaches its frozen head and the drain has accepted every event. The second poll is the barrier — the feed issues it only after every event of the first page was accepted and PageDelivered ran for it — so the checkpoint is read at three points the feed's own progress orders: paused mid-page-one, after the walk followed `next` into page two, and paused mid-page-two with the second page's first event already in the ledger. Each read is sequenced after any save that moved with a page, rather than 50ms after a clock. No sleep is left in the test. A test that cannot fail cannot be proved by reverting it, so the evidence is the other way round. With intake's PageDelivered observer made to save the page's position durably there and then — a paused feed really moving the checkpoint — the old test passes 20/20 plain and 10/10 under -race, and the new one fails 20/20 and 10/10 on "a page delivered in full is still not a position saved". With that regression removed the new test passes 30/30 plain and 20/20 under -race. invariants_test.go had exactly one sleep, this one. The rest of the file asserts after something synchronous has returned or after a WaitGroup, so they are ordered already. --- internal/connector/invariants_test.go | 80 +++++++++++++++++++++++---- 1 file changed, 69 insertions(+), 11 deletions(-) diff --git a/internal/connector/invariants_test.go b/internal/connector/invariants_test.go index e39605883..3b77b8680 100644 --- a/internal/connector/invariants_test.go +++ b/internal/connector/invariants_test.go @@ -155,6 +155,20 @@ func TestInvariantF2WarningEdgesSettleOnTheTrueState(t *testing.T) { // F1: while intake is paused on a full queue, the page it is in the middle of // is not checkpointed. +// +// That is an ORDER, not a window, and it has to be tested as one. On a +// present-class entry the feed holds every page's position and saves nothing +// until the walk has reached its frozen head AND the drain has accepted every +// event — so the hold is visible only against a feed that demonstrably moved +// on. Reading the checkpoint once, a fixed 50ms after the pause, measures a +// duration instead: a save that happens but happens slowly is a pass, so the +// test could never fail and never did. +// +// The walk here is two pages, and the second poll is the barrier. The feed +// issues it only after every event of the first page was accepted and +// `PageDelivered` ran for that page, so a save carried along with a page is +// ordered BEFORE the barrier — caught however slowly it lands, rather than +// only if it lands inside somebody's guess at a window. func TestInvariantF1APausedFeedDoesNotMoveTheCheckpoint(t *testing.T) { ledger := newTestLedger(t) queue, err := NewQueue(1, 1) @@ -163,25 +177,69 @@ func TestInvariantF1APausedFeedDoesNotMoveTheCheckpoint(t *testing.T) { intake.queue = queue intake.opts.Queue = queue minter.ScriptTicket(ticket()) - polls.ScriptPage(eventfeed.PollPage{Events: []eventfeed.Event{testEvent(500), testEvent(501)}, Position: "after-both"}) + polls.ScriptPage(eventfeed.PollPage{ + Events: []eventfeed.Event{testEvent(500), testEvent(501)}, + Position: "after-the-first-page", + Next: "https://3.basecampapi.com/2914079/events.json?position=after-the-first-page", + }) + polls.ScriptPage(eventfeed.PollPage{ + Events: []eventfeed.Event{testEvent(502), testEvent(503)}, + Position: "after-both-pages", + }) ctx, cancel := context.WithCancel(context.Background()) defer cancel() + + // Nothing durable has moved, read at a point the feed's own progress puts + // after any save that moved with a page. + checkpointUnmoved := func(why string) { + t.Helper() + _, ok, err := ledger.Load(ctx, intake.CheckpointKey()) + require.NoError(t, err) + require.False(t, ok, why) + } + // One id out of the queue, which is the only thing that lets the blocked + // offer behind it through. + makeRoom := func() { + t.Helper() + _, err := queue.Take(ctx) + require.NoError(t, err) + } + done := runInBackground(ctx, t, intake) subscribedConn(t, transport) - require.Eventually(t, queue.Paused, 5*time.Second, time.Millisecond) - time.Sleep(50 * time.Millisecond) - _, ok, err := ledger.Load(ctx, intake.CheckpointKey()) - require.NoError(t, err) - assert.False(t, ok, "a crash now must resume from before the page, not after it") - - _, err = queue.Take(ctx) - require.NoError(t, err) + // The queue holds one: 500 fills it and 501 waits for room, so the feed + // is paused in the middle of the first page. + require.Eventually(t, queue.Paused, 5*time.Second, time.Millisecond, + "the first page's second event must find the queue full") + checkpointUnmoved("a crash mid-page must resume from before the page, not after it") + + // Room for 501. The first page then completes and the walk follows its + // `next` — and that second poll is the barrier: the feed has advanced a + // whole page past the position it is holding. + makeRoom() + require.Eventually(t, func() bool { return polls.CallCount() >= 2 }, + 5*time.Second, 5*time.Millisecond, "the walk follows next into the second page") + checkpointUnmoved("a page delivered in full is still not a position saved") + + // And paused again, now inside the second page, with the first page's + // events long since handed over: a feed that demonstrably advanced over a + // checkpoint that demonstrably did not. + require.Eventually(t, func() bool { + _, ok, err := ledger.Get(ctx, 502) + return err == nil && ok && queue.Paused() + }, 5*time.Second, 5*time.Millisecond, "the second page pauses the feed the same way") + checkpointUnmoved("two pages in, paused, and nothing durable has moved") + + // Only the end of the walk moves it, and it moves to the last page's + // position — the whole walk, or none of it. + makeRoom() + makeRoom() require.Eventually(t, func() bool { position, ok, err := ledger.Load(ctx, intake.CheckpointKey()) - return err == nil && ok && position == "after-both" - }, 5*time.Second, 5*time.Millisecond) + return err == nil && ok && position == "after-both-pages" + }, 5*time.Second, 5*time.Millisecond, "a drained queue lets the walk finish and save") cancel() awaitReturn(t, done, "Run should return on shutdown")