From 068db986b31de3ce23d3919427b84fc6455d2328 Mon Sep 17 00:00:00 2001 From: Jorge Manrubia Date: Fri, 18 Sep 2026 16:23:06 +0200 Subject: [PATCH] The feed test reads the walk's position once the walk has saved it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The end-to-end intake test served a live event on the socket, waited for it to reach the ledger, and read the durable position there. A fresh ledger has no position to resume, so the entry is present-class: the feed HOLDS the page's position through the walk and saves it only after the drain has accepted every event buffered during it. The live event is one of those. Its row in the ledger says the drain reached it, not that the save has run, and the read that followed landed between the two — require.True over an `ok` that was not yet due. Under -race everything runs several times slower and that gap widens. It is what failed the Race Detection job on main at 66b1a3a9 (run 35349470544) and on at least one pull request built from it. The race detector reported no data race, because there is none: the check read across an order it had not waited for. Instrumented to say what was true at the moment of that read, under twenty competing processes on two cores: the position was ABSENT and arrived 3.8ms, 33.2ms and 34.8ms later on the three iterations of forty that caught it — with one poll page walked, both pointer lines written and both ids handed over. Everything the events owed was done; only the walk's position was outstanding. The check now waits for the position to be saved before it reads it — the condition, not a duration, the same shape as #752. The counts below it are waited on too: the ledger row is written before the pointer line and the hand-off, so the row says nothing about either, which the restart test next door already knew and said. That wait is for "at least two", so the equalities still measure — a third pointer line or a third id fails them rather than satisfying them. --- internal/connector/intake_feed_test.go | 24 +++++++++++++++++++++--- 1 file changed, 21 insertions(+), 3 deletions(-) diff --git a/internal/connector/intake_feed_test.go b/internal/connector/intake_feed_test.go index eb0681d87..ea04fc839 100644 --- a/internal/connector/intake_feed_test.go +++ b/internal/connector/intake_feed_test.go @@ -169,7 +169,18 @@ func TestIntakeRunsTheFeedThroughCatchUpAndStreaming(t *testing.T) { return err == nil && ok }, 5*time.Second, 5*time.Millisecond, "a live event reaches the ledger") - // Only the poll lane advances the durable position. + // Only the poll lane advances the durable position — and it advances it + // when the walk is OVER, not when the walk's events land. A fresh ledger + // has no position to resume, so this entry is present-class: the feed + // holds the page's position and saves it only once the drain has accepted + // every event buffered during the walk. The live event above is one of + // those, so seeing it in the ledger says the drain reached it, not that + // the save has run. Wait for the save rather than read across it. + require.Eventually(t, func() bool { + _, ok, err := ledger.Load(ctx, intake.CheckpointKey()) + return err == nil && ok + }, 5*time.Second, 5*time.Millisecond, "the walk's held position reaches the ledger") + position, ok, err := ledger.Load(ctx, intake.CheckpointKey()) require.NoError(t, err) require.True(t, ok) @@ -182,8 +193,15 @@ func TestIntakeRunsTheFeedThroughCatchUpAndStreaming(t *testing.T) { require.NoError(t, err) assert.Equal(t, LanePoll, polled.Lane) - assert.Equal(t, 2, countLines(pointers.String())) - assert.Equal(t, 2, queue.Depth()) + // The ledger row is written before the pointer line and the hand-off, so + // the counts are waited on rather than read the moment the rows appear. + // The wait is for "at least", so the equalities below still measure: a + // third pointer line or a third id fails them rather than satisfying them. + require.Eventually(t, func() bool { + return countLines(pointers.String()) >= 2 && queue.Depth() >= 2 + }, 5*time.Second, 5*time.Millisecond, "both events are written out and handed over") + assert.Equal(t, 2, countLines(pointers.String()), "one pointer line per event, and no more") + assert.Equal(t, 2, queue.Depth(), "one id handed over per event, and no more") cancel() select {