diff --git a/docs/fetch.md b/docs/fetch.md index f4e4967..86f3473 100644 --- a/docs/fetch.md +++ b/docs/fetch.md @@ -115,13 +115,24 @@ identical fetches share one call to `F`. The **shared origin request runs on a bounded background context** (`MaxFlight`, default 10m): one caller's cancellation does not abort it — the block still completes for the other waiters (desirable for a cache) — but a stalled origin cannot pin it forever -either. A waiter blocked past the flight's deadline re-checks, evicts the -stale entry, and leads a replacement (so abandoned waiters are released at -the deadline even when no later caller ever arrives), and a late-finishing -stale leader never deletes the replacement's entry; without the bound, one -half-dead origin connection would poison the cache key for the process -lifetime and leak one goroutine per abandoned waiter. Each caller's own `ctx` only bounds how long that caller -waits (a cancelled caller returns `ctx.Err()`). +either. The flight's own context expires at `MaxFlight`; a short grace later +(`evictGrace`, so a ctx-respecting worker's deadline delivery always lands +first) the flight becomes **stale**: a caller still waiting re-checks, evicts +the stale entry, and leads a replacement; a late-finishing stale leader never +deletes the replacement's entry. Without the bound, one half-dead origin +connection would poison the cache key for the process lifetime. + +A flight keeps **exactly one worker goroutine** — started by whoever leads it — +no matter how many callers join or abandon it: joiners wait inline on the +flight's done channel, so a caller that stops waiting retains no goroutine of +its own. A cancellation storm on one stalled key therefore costs one worker, +not one goroutine per abandoned caller (the issue #53 gap: waiter goroutines +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. + +Each caller's own `ctx` only bounds how long that caller waits (a cancelled +caller returns `ctx.Err()`). A joiner whose flight was refused (401/403) retries once with its own credential — but only if the joiner itself is still alive: the retry serves @@ -273,7 +284,8 @@ go test ./internal/fetch/ -cover -count=1 | `TestLateStaleLeaderKeepsReplacement` | a late-finishing stale leader never deletes the replacement flight's entry | | `TestFlightContextBoundsStalledOrigin` | a ctx-respecting but stalled flight fails all waiters within `MaxFlight` | | `TestJoinerRetrySkippedWhenCallerGone` | a cancelled joiner does not fire a refusal retry | -| `TestAbandonedWaitersReleasedAtDeadline` | a waiter abandoned with a stuck leader is released at the flight deadline even with no later caller | +| **`TestCancelledWaitersRetainNoGoroutines`** | **64 cancelled waiters on a stalled flight retain no goroutines — one worker per flight, released promptly (issue #53)** | +| **`TestStaleFlightReplacedOnlyByALiveCaller`** | **an abandoned flight is not respawned at its deadline; the next real caller evicts and replaces it** | | `TestHTTPFetcher206ShortBody` | a 206 shorter than the window is an error (no silent cache poisoning) | | `TestHTTPFetcher206WrongOffset` | a 206 at the wrong offset is an error | | `TestStartFromContentRange` | Content-Range start parsing incl. rejects | diff --git a/internal/fetch/fetch.go b/internal/fetch/fetch.go index df709bb..27c1011 100644 --- a/internal/fetch/fetch.go +++ b/internal/fetch/fetch.go @@ -386,6 +386,12 @@ func redactTransportErr(err error) error { // cancellation does not abort it — the block still completes for the other // waiters (desirable for a cache). Each caller's own ctx only bounds how long // that caller waits. +// +// A flight keeps exactly one worker goroutine — started by whoever leads it — +// regardless of how many callers join or abandon it: joiners wait inline on +// the flight's done channel, so a caller that stops waiting retains no +// goroutine of its own (issue #53). A cancellation storm on one stalled key +// therefore costs one worker, not one goroutine per abandoned caller. type Coalescing struct { F Fetcher // Key maps an origin URL to the identity used for deduplication. Nil uses the @@ -400,8 +406,8 @@ type Coalescing struct { // (chunk.ObjectID). Key func(url string) string // MaxFlight bounds how long one shared flight may run before a later call - // for the same key abandons it and leads a replacement (see group.Do). The - // flight's own context also expires at this bound. Zero means + // for the same key abandons it and leads a replacement (see group.acquire). + // The flight's own context also expires at this bound. Zero means // DefaultMaxFlight. MaxFlight time.Duration g group @@ -423,8 +429,9 @@ func (c *Coalescing) identity(url string) string { // a degenerate flight can carry a whole object (a Range-ignoring origin), not // just a block. The bound exists so a stalled origin connection — accepted, // then silent — cannot poison a cache key forever: the flight's context -// expires, and the key's map entry is evicted once the deadline passes even if -// the leader's fetcher ignores context cancellation. +// expires at the bound, and the key's map entry becomes evictable a short +// grace later (evictGrace) even if the leader's fetcher ignores context +// cancellation. const DefaultMaxFlight = 10 * time.Minute func (c *Coalescing) maxFlight() time.Duration { @@ -435,7 +442,9 @@ func (c *Coalescing) maxFlight() time.Duration { } // Fetch coalesces duplicate concurrent fetches. It returns ctx.Err() if the -// caller's context is cancelled before the shared fetch completes. +// caller's context is cancelled before the shared fetch completes — and, once +// it has returned, retains no goroutine: only the flight's single worker +// outlives its callers. // // If the shared fetch was refused on authorization grounds (401/403) and this // caller merely joined it, the caller retries alone with its *own* URL. That @@ -447,37 +456,53 @@ func (c *Coalescing) maxFlight() time.Duration { // refusal keeps the thundering-herd protection intact on every normal path. func (c *Coalescing) Fetch(ctx context.Context, url string, start, end int64) (Range, error) { key := c.identity(url) + "\x00" + strconv.FormatInt(start, 10) + ":" + strconv.FormatInt(end, 10) - type result struct { - r Range - err error - } - ch := make(chan result, 1) - go func() { - r, err, shared := c.g.Do(key, c.maxFlight(), func() (Range, error) { - // The shared flight runs on a bounded background context: caller - // cancellation must not abort it (a cancelled caller's peers may - // still want the bytes), but a stalled origin must not pin it - // forever either. - flightCtx, cancel := context.WithTimeout(context.Background(), c.maxFlight()) - defer cancel() - return c.F.Fetch(flightCtx, url, start, end) - }) - r.Coalesced = shared + for { + flight, lead := c.g.acquire(key, c.maxFlight()) + if lead { + // Exactly one worker goroutine per flight, started by its leader. + // It deliberately outlives the leading caller: caller cancellation + // must not abort a flight its peers may still be waiting on, so + // the worker runs on a background context — bounded by the + // flight's own deadline, so a stalled origin (accepted, then + // silent) cannot pin it forever. + go func() { + flightCtx, cancel := context.WithDeadline(context.Background(), flight.end) + defer cancel() + r, err := c.F.Fetch(flightCtx, url, start, end) + c.g.finish(key, flight, r, err) + }() + } + // Wait inline rather than on a per-caller goroutine: a caller that + // stops waiting — cancelled, or still waiting when the flight goes + // stale — leaves nothing behind. The wait runs to flight.stale, a + // grace margin past the flight's own deadline, so a ctx-respecting + // worker's deadline delivery always wins over eviction; the stale + // path exists only for a worker that ignores its context. + remaining := time.Until(flight.stale) + if remaining <= 0 { + continue // the flight went stale between acquire and the wait + } + timer := time.NewTimer(remaining) + select { + case <-ctx.Done(): + timer.Stop() + return Range{}, ctx.Err() + case <-timer.C: + continue // deadline passed: evict the stale flight or join its replacement + case <-flight.done: + timer.Stop() + } + r, err := flight.val, flight.err + r.Coalesced = !lead // Only a joiner retries: if we led the flight, the credential that was // refused was our own and asking again would change nothing. The retry // serves this caller alone, so it uses the caller's context and is // skipped entirely when that caller is already gone. - if shared && refused(err) && ctx.Err() == nil { + if !lead && refused(err) && ctx.Err() == nil { r, err = c.F.Fetch(ctx, url, start, end) r.Coalesced = false // these bytes did cross the network for us } - ch <- result{r, err} - }() - select { - case <-ctx.Done(): - return Range{}, ctx.Err() - case res := <-ch: - return res.r, res.err + return r, err } } @@ -498,67 +523,72 @@ func refused(err error) bool { return errors.As(err, &se) && se.Refused() } -// group is a minimal singleflight: concurrent Do calls with the same key share -// one execution of fn; all callers receive its result. +// group is a minimal singleflight keyed by string. One leader at a time owns +// a key's flight and runs the shared work on the flight's single worker +// goroutine; every other caller waits inline on the flight's done channel, so +// waiters hold no per-caller state beyond their own stack. type group struct { mu sync.Mutex m map[string]*call } type call struct { - done chan struct{} // closed when the flight completes - val Range - err error - end time.Time // deadline; zero when unbounded + done chan struct{} // closed when the flight completes + val Range + err error + end time.Time // the flight context's deadline + stale time.Time // after this, a caller evicts the flight and leads a replacement } -// Do executes fn once per in-flight key. shared reports whether the result was -// shared with a concurrent caller. -// -// maxFlight bounds how long a flight may occupy its key. A joiner waits only -// until the flight's deadline, then re-checks: if the leader overran (a stalled -// origin, or a fetcher ignoring context), the stale entry is evicted and a -// replacement flight led — so a dead flight releases its waiters at the -// deadline even when nobody else ever calls. A late-finishing stale leader -// deletes only its own entry, never the replacement's. -func (g *group) Do(key string, maxFlight time.Duration, fn func() (Range, error)) (v Range, err error, shared bool) { - for { - g.mu.Lock() - if g.m == nil { - g.m = make(map[string]*call) - } - c, ok := g.m[key] - if ok && (maxFlight <= 0 || time.Now().Before(c.end)) { - g.mu.Unlock() - if maxFlight <= 0 { - <-c.done - return c.val, c.err, true - } - select { - case <-c.done: - return c.val, c.err, true - case <-time.After(time.Until(c.end)): - continue // deadline passed: evict the stale flight or join its replacement - } - } - if ok { - delete(g.m, key) // stale flight: abandon it and lead a replacement - } - c = &call{done: make(chan struct{})} - if maxFlight > 0 { - c.end = time.Now().Add(maxFlight) - } - g.m[key] = c - g.mu.Unlock() - - c.val, c.err = fn() - close(c.done) +// evictGrace is the margin between a flight's own deadline and the moment it +// becomes evictable. A ctx-respecting worker delivers its result right at the +// deadline; the margin guarantees that delivery lands before any waiter +// declares the flight dead, so eviction only ever fires on a worker that +// ignores its context. +func evictGrace(maxFlight time.Duration) time.Duration { + g := maxFlight / 8 + if g < 25*time.Millisecond { + g = 25 * time.Millisecond + } + if g > 30*time.Second { + g = 30 * time.Second + } + return g +} - g.mu.Lock() - if g.m[key] == c { - delete(g.m, key) +// acquire returns the in-flight call for key, or — when none is flying or the +// existing one has gone stale — creates a new one and reports lead=true. A +// caller that leads must run the shared work for the flight and then call +// finish with the returned call, exactly once. maxFlight must be positive +// (Coalescing.maxFlight guarantees this). +func (g *group) acquire(key string, maxFlight time.Duration) (c *call, lead bool) { + g.mu.Lock() + defer g.mu.Unlock() + if g.m == nil { + g.m = make(map[string]*call) + } + if c, ok := g.m[key]; ok { + if time.Now().Before(c.stale) { + return c, false } - g.mu.Unlock() - return c.val, c.err, false + delete(g.m, key) // stale flight: abandon it and lead a replacement } + end := time.Now().Add(maxFlight) + c = &call{done: make(chan struct{}), end: end, stale: end.Add(evictGrace(maxFlight))} + g.m[key] = c + 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. +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() } diff --git a/internal/fetch/fetch_test.go b/internal/fetch/fetch_test.go index b3bb600..943ca3f 100644 --- a/internal/fetch/fetch_test.go +++ b/internal/fetch/fetch_test.go @@ -8,6 +8,7 @@ import ( "net/http" "net/http/httptest" neturl "net/url" + "runtime" "strings" "sync" "sync/atomic" @@ -803,37 +804,92 @@ func TestJoinerRetrySkippedWhenCallerGone(t *testing.T) { } } -// TestAbandonedWaitersReleasedAtDeadline pins the review-found residual of -// issue #4: a joiner whose caller gave up must not block in the flight wait -// forever when the leader ignores cancellation — even when no later caller -// ever arrives for the key. At the flight deadline the waiter re-checks, -// evicts the stale flight, and leads a replacement. -func TestAbandonedWaitersReleasedAtDeadline(t *testing.T) { - var calls int64 +// TestCancelledWaitersRetainNoGoroutines pins issue #53: every Fetch used to +// start a per-caller goroutine that waited inside the singleflight group, so a +// caller whose context was cancelled returned while its goroutine stayed +// blocked until the flight finished or hit MaxFlight — a cancellation storm on +// one stalled key retained one goroutine per abandoned caller. Now joiners +// wait inline: a flight keeps exactly one worker goroutine no matter how many +// callers come and go. +func TestCancelledWaitersRetainNoGoroutines(t *testing.T) { stall := make(chan struct{}) - t.Cleanup(func() { close(stall) }) // release all stuck fn calls at test end + t.Cleanup(func() { close(stall) }) // release the flight worker at test end + var calls int64 f := fetcherFunc(func(ctx context.Context, url string, start, end int64) (Range, error) { atomic.AddInt64(&calls, 1) - <-stall // ignores ctx: simulates a permanently stuck origin + <-stall // ignores ctx: simulates an origin that accepted then went silent return Range{Data: blob(8), Total: 8}, nil }) - c := &Coalescing{F: f, MaxFlight: 40 * time.Millisecond} + c := &Coalescing{F: f, MaxFlight: time.Minute} + + time.Sleep(50 * time.Millisecond) // let the runtime settle + base := runtime.NumGoroutine() + + const callers = 64 + var wg sync.WaitGroup + for i := 0; i < callers; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, err := c.Fetch(ctx, "u", 0, 7); !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("caller: want its own ctx deadline, got %v", err) + } + }() + } + wg.Wait() // every caller has already returned on its own deadline - // Two callers join the same stuck flight and give up; no later caller ever - // arrives for the key. - for i := 0; i < 2; i++ { - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) - if _, err := c.Fetch(ctx, "u", 0, 7); err == nil { - t.Fatal("caller: want its own ctx deadline, got nil") + // Prompt release is the contract: only the flight's single worker may + // outlive the callers (plus nothing else). On the old code the delta was + // one goroutine per caller. + time.Sleep(200 * time.Millisecond) + if held := runtime.NumGoroutine() - base; held > 4 { + t.Errorf("goroutines retained after all %d callers returned = %d, want <= 4 (one flight worker)", + callers, held) + } + if got := atomic.LoadInt64(&calls); got != 1 { + t.Errorf("underlying calls = %d, want 1 (all callers shared one flight)", got) + } +} + +// TestStaleFlightReplacedOnlyByALiveCaller pins the lifecycle contract that +// replaces the issue #4 residual: with the per-caller waiter goroutine gone +// (issue #53), a flight everyone abandoned is *not* replaced at its deadline — +// nobody is left wanting the bytes, so firing a fresh origin fetch would be +// pure waste. The stale entry is evicted lazily, by the next real caller. +func TestStaleFlightReplacedOnlyByALiveCaller(t *testing.T) { + var calls int64 + stall := make(chan struct{}) + t.Cleanup(func() { close(stall) }) // release any stuck fn call at test end + f := fetcherFunc(func(ctx context.Context, url string, start, end int64) (Range, error) { + if atomic.AddInt64(&calls, 1) == 1 { + <-stall // stuck leader: ignores ctx } - cancel() + return Range{Data: blob(8), Total: 8}, nil + }) + c := &Coalescing{F: f, MaxFlight: 40 * time.Millisecond} + + ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + if _, err := c.Fetch(ctx, "u", 0, 7); err == nil { + t.Fatal("caller: want its own ctx deadline, got nil") } - // After the deadline the abandoned waiter must have been released and led - // a replacement flight — not be blocked in the wait forever. + // Well past the flight deadline, with no live caller: no replacement + // flight may have been led. time.Sleep(200 * time.Millisecond) - if got := atomic.LoadInt64(&calls); got < 2 { - t.Fatalf("underlying calls = %d, want >= 2 (abandoned waiter must be released at the deadline)", got) + if got := atomic.LoadInt64(&calls); got != 1 { + t.Fatalf("underlying calls = %d, want 1 (an abandoned flight must not respawn without a caller)", got) + } + + // The next real caller evicts the stale entry and leads the replacement. + r, err := c.Fetch(context.Background(), "u", 0, 7) + if err != nil || len(r.Data) != 8 { + t.Fatalf("replacement flight: err=%v len=%d", err, len(r.Data)) + } + if got := atomic.LoadInt64(&calls); got != 2 { + t.Fatalf("underlying calls = %d, want 2 (stale flight evicted and replaced by the live caller)", got) } }