From 3fe8beaa4fb0a47be06d2ea233e011123ef85a55 Mon Sep 17 00:00:00 2001 From: Coldwings Date: Mon, 31 Aug 2026 15:48:28 +0800 Subject: [PATCH] fetch: unpublish a flight before waking its waiters (fixes #69) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit group.finish published the result (close(done), waking the leader and every waiter) before removing the flight from the group map. If the worker lost its P in the handful of instructions between the close and the delete, the woken leader could return to its caller, whose next same-key call then found the completed-but-still-registered entry in acquire and joined it — reusing a result, including an error, for a strictly sequential call. That made an error effectively cached: this is what flaked TestSizeHiddenTotalIsProbeFailure in CI (origin probes = 1, want 2), where the engine's second size probe was answered from a finished flight. Delete the entry before closing done: the map now holds only uncompleted flights, so once a call has returned its flight is unreachable and a later caller always leads a fresh fetch. The result writes still happen-before close(done); the stale-leader guard (g.m[key] == c) is unchanged. Mirror cost, accepted and documented: a caller arriving between the delete and the close leads a duplicate fetch instead of sharing the just-finished result. Regression pin: TestSequentialFetchNeverJoinsCompletedFlight — sequential identical fetches over an always-failing inner fetcher must each reach it; time-boxed -race stress (the race manifests in scheduling bursts), 8/8 pre-fix runs fail, post-fix clean. docs/fetch.md states the invariant in §3.5/§4 and refreshes the drifted coverage figure. --- docs/fetch.md | 22 +++++++++--- internal/fetch/fetch.go | 21 +++++++---- internal/fetch/issue69_test.go | 65 ++++++++++++++++++++++++++++++++++ 3 files changed, 98 insertions(+), 10 deletions(-) create mode 100644 internal/fetch/issue69_test.go diff --git a/docs/fetch.md b/docs/fetch.md index 53bf778..3f3fd45 100644 --- a/docs/fetch.md +++ b/docs/fetch.md @@ -147,6 +147,16 @@ used to be retained until the flight bound). A corollary of inline waiting: a flight everyone abandoned is *not* replaced at its deadline — nobody is left wanting the bytes — the stale entry is evicted lazily by the next real caller. +Sharing stops at completion: the worker removes the flight from the group +**before** publishing its result, so once any call has returned, its flight is +unreachable and a later — even immediately sequential — call for the same key +leads a fresh fetch. An error is therefore never replayed to a caller that was +not still waiting for it (the engine's size probe relies on this: a failed +probe is re-issued against origin, never served twice — issue #69). The mirror +cost: a caller arriving in the window between removal and publication leads a +duplicate fetch instead of sharing the just-finished result — bounded extra +traffic, by design. + Each caller's own `ctx` only bounds how long that caller waits (a cancelled caller returns `ctx.Err()`). @@ -219,9 +229,12 @@ bucket), and that changes three things: `start` discarded, at most `end-start+1` read) — the full body is never buffered. 2. **Coalescing**: N concurrent `Coalescing.Fetch` for the same key ⇒ exactly 1 - call to the underlying `Fetcher`; all callers receive the same result. The one - exception is an authorization refusal (401/403), where each joiner re-fetches - once with its own credential (§3.6). + call to the underlying `Fetcher`; all callers receive the same result. + Coalescing never crosses completion: a call issued after its key's flight + has returned leads a fresh fetch, so a failure can never be replayed to a + later caller (issue #69, §3.5). The one exception is an authorization + refusal (401/403), where each joiner re-fetches once with its own + credential (§3.6). 3. **Cancellation isolation**: a caller cancelling its `ctx` neither aborts the shared fetch nor affects other callers. The shared flight is itself bounded by `MaxFlight` (§3.5), so a stalled origin fails all waiters within the @@ -253,7 +266,7 @@ bucket), and that changes three things: ## 7. Testing - **Results**: `go vet` clean; `go test` all pass; `go test -race` clean. -- **Coverage**: **93.1%** of statements. The remaining uncovered lines are rare +- **Coverage**: **91.8%** of statements. The remaining uncovered lines are rare I/O-error branches (e.g. a mid-body read error) that need fault injection. - **Reproduce**: @@ -307,6 +320,7 @@ go test ./internal/fetch/ -cover -count=1 | `TestStartFromContentRange` | Content-Range start parsing incl. rejects | | `TestCoalescedResultIsMarked` | joined callers see Coalesced=true; the leader does not | | `TestUncontendedFetchNotMarkedCoalesced` | an uncontended fetch is not marked coalesced | +| **`TestSequentialFetchNeverJoinsCompletedFlight`** | **strictly sequential identical fetches never join a completed flight — each reaches the inner fetcher, so an error can never be replayed to a later caller (issue #69; time-boxed `-race` stress)** | | `TestFetchBlockPastEndErrors` | a block wholly past the object (or an inverted range) errors without contacting origin | | `TestFetchBlockUnknownSizeTailBlock` | an RFC-clamped 206 on the tail block is accepted and reveals Total | | **`TestFetchBlockRejectsOverflowGeometry`** | **negative / int64-overflowing block indices (and non-positive block size) error before the fetcher is invoked, size known or unknown (issue #52)** | diff --git a/internal/fetch/fetch.go b/internal/fetch/fetch.go index d0d0cc8..e189fbd 100644 --- a/internal/fetch/fetch.go +++ b/internal/fetch/fetch.go @@ -594,16 +594,25 @@ func (g *group) acquire(key string, maxFlight time.Duration) (c *call, lead bool return c, true } -// finish publishes a led flight's result and wakes its waiters: the writes -// happen-before the close of done, hence before any waiter's read. A -// late-finishing stale leader deletes only its own entry, never the -// replacement's. +// finish publishes a led flight's result and wakes its waiters. The entry is +// deleted BEFORE done is closed: the map holds only uncompleted flights, so +// once a call has returned, its flight is unreachable and a strictly +// sequential caller for the same key always leads a fresh fetch — an error +// result can never be served twice (issue #69; the engine's size probe pins +// "failures are not cached" on this). Publishing first would open a window +// in which a woken leader returns and its caller's next call joins the +// completed flight. The cost is the mirror image: a caller that arrives in +// the window between the delete and the close leads a duplicate fetch rather +// than sharing the just-finished result — bounded extra traffic, by design. +// The result writes still happen-before the close of done, hence before any +// waiter's read. A late-finishing stale leader deletes only its own entry, +// never the replacement's. func (g *group) finish(key string, c *call, v Range, err error) { - c.val, c.err = v, err - close(c.done) g.mu.Lock() if g.m[key] == c { delete(g.m, key) } g.mu.Unlock() + c.val, c.err = v, err + close(c.done) } diff --git a/internal/fetch/issue69_test.go b/internal/fetch/issue69_test.go new file mode 100644 index 0000000..2c2ded6 --- /dev/null +++ b/internal/fetch/issue69_test.go @@ -0,0 +1,65 @@ +package fetch + +import ( + "context" + "errors" + "runtime" + "sync/atomic" + "testing" + "time" +) + +// alwaysFailFetcher counts invocations and never succeeds. +type alwaysFailFetcher struct{ calls atomic.Int64 } + +func (f *alwaysFailFetcher) Fetch(context.Context, string, int64, int64) (Range, error) { + f.calls.Add(1) + return Range{}, errors.New("origin boom") +} + +// TestSequentialFetchNeverJoinsCompletedFlight pins issue #69: group.finish +// used to publish the result (close(done)) before deleting the flight's map +// entry, so a strictly sequential same-key Fetch issued right after the +// leader returned could join the COMPLETED flight and reuse its result — an +// error effectively became cached, which is how the engine's "probe failures +// are never cached" pin (TestSizeHiddenTotalIsProbeFailure) came to flake in +// CI. Every sequential Fetch must reach the inner fetcher. +func TestSequentialFetchNeverJoinsCompletedFlight(t *testing.T) { + // The pre-fix window is a handful of instructions in the flight worker + // between publishing the result and removing the map entry; it bites only + // when the worker loses its P exactly there. Fewer P's than runnable + // goroutines plus race instrumentation widen it (CI's shape: 4 vCPUs with + // parallel package tests). Pre-fix the race shows in scheduling bursts — a + // starved worker leaves its completed entry discoverable across many + // pairs — so the loop is time-boxed, not count-boxed: a short fixed count + // can slip between bursts. Measured pre-fix: the first violation lands + // well inside the budget (~2.7% of pairs in a standalone 2M-pair run, + // first hit typically within a few thousand pairs); post-fix: zero + // violations anywhere. + restore := runtime.GOMAXPROCS(4) + defer runtime.GOMAXPROCS(restore) + var stop atomic.Bool + for i := 0; i < 12; i++ { + go func() { + for !stop.Load() { + runtime.Gosched() + } + }() + } + defer stop.Store(true) + + c := &Coalescing{F: &alwaysFailFetcher{}} + inner := c.F.(*alwaysFailFetcher) + deadline := time.Now().Add(2 * time.Second) + for pair := 0; time.Now().Before(deadline); pair++ { + before := inner.calls.Load() + for j := 0; j < 2; j++ { + if _, err := c.Fetch(context.Background(), "http://origin/x", 0, 0); err == nil { + t.Fatal("the inner fetcher always fails; every call must return its error") + } + } + if got := inner.calls.Load() - before; got != 2 { + t.Fatalf("pair %d: two sequential fetches reached the inner fetcher %d times — a completed flight was joined and its error served twice (issue #69)", pair, got) + } + } +}