fetch: unpublish a flight before waking its waiters (fixes #69)

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)** |

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)
 }

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)
+		}
+	}
+}