diff --git a/cmd/server/node_reach_cold_limit.go b/cmd/server/node_reach_cold_limit.go index 32e3cf10b..5d0e78e1c 100644 --- a/cmd/server/node_reach_cold_limit.go +++ b/cmd/server/node_reach_cold_limit.go @@ -31,13 +31,18 @@ import ( // two-tab burst would 429 a user whose database is not under any real // pressure. Our own front end also only auto-retries 503 + Retry-After // (app.js api()), never 429, so an instant refusal surfaces as a failed -// page rather than a short wait. +// page rather than a short wait. That was the initial evaluation; the +// frontend now handles bounded 429 retries too (#530 / #547). // - We therefore queue for a bounded reachColdBuildWait first and only // answer 429 + Retry-After when the limiter is still saturated at the end // of it. Queueing costs an idle HTTP goroutine, not a database // connection, so the pool protection is identical while a transient burst -// is absorbed instead of shed. Load shedding still happens — it just -// requires sustained saturation rather than an instant of it. +// is absorbed instead of shed. At most reachColdWaiterLimit distinct +// leaders may wait; a full queue is shed immediately before allocating +// another timer. Same-key singleflight followers share their leader's +// admission, so this does not bound those pre-existing followers. +// +// Queue admission reuses the existing correction in commit 1d2cd99d (#557). // // This is a per-endpoint cap, not global database admission control: it bounds // Reach's own cold work and nothing else. Any broader policy belongs with the @@ -62,23 +67,28 @@ var ( // the in-flight scans finishes. const reachColdRetryAfterSeconds = 2 +// reachColdWaiterLimit is a fixed internal default, not operator/customizer +// configuration. It bounds distinct leaders waiting for a build slot. +const reachColdWaiterLimit = 8 + // errReachColdBusy is returned by reachColdLimiter.acquire when the limiter -// stayed saturated for the whole queue wait. The handler maps it to 429 and -// never caches it (singleflight does not cache errors either), so the next -// request recomputes normally. +// has a full queue or stayed saturated for the whole wait. The handler maps +// it to 429 and never caches it (singleflight does not cache errors either), +// so the next request recomputes normally. var errReachColdBusy = errors.New("reach cold-scan concurrency limit reached") -// reachColdLimiter is a counting semaphore with a bounded queue wait, plus the -// counters the tests and the perf report read. Usable as a zero value: the +// reachColdLimiter is a counting semaphore with a bounded queue count/wait, +// plus counters the tests and the perf report read. Usable as a zero value: the // slot channel is built on first use from reachColdBuildLimit, so every // Server (including the ones tests assemble by hand) gets the limit without a // constructor. type reachColdLimiter struct { - mu sync.Mutex - sem chan struct{} // buffered to the limit; one slot per in-flight scan - lim int // 0 → reachColdBuildLimit - wait time.Duration // <=0 → reachColdBuildWait - hook func() // test injection, run just after a slot is taken + mu sync.Mutex + sem chan struct{} // buffered to the limit; one slot per in-flight scan + queued chan struct{} // one admission per distinct leader waiting for a slot + lim int // 0 → reachColdBuildLimit + wait time.Duration // <=0 → reachColdBuildWait + hook func() // test injection, run just after a slot is taken inflight atomic.Int64 // slots currently held peak atomic.Int64 // high-water mark of inflight @@ -95,12 +105,12 @@ type reachColdLimiter struct { func (l *reachColdLimiter) configure(limit int, wait time.Duration, hook func()) { l.mu.Lock() defer l.mu.Unlock() - l.lim, l.wait, l.hook, l.sem = limit, wait, hook, nil + l.lim, l.wait, l.hook, l.sem, l.queued = limit, wait, hook, nil, nil } -// slots returns the (lazily built) slot channel, the queue wait and the +// slots returns the lazily built slot and queue channels, the wait and the // acquire hook. -func (l *reachColdLimiter) slots() (chan struct{}, time.Duration, func()) { +func (l *reachColdLimiter) slots() (chan struct{}, chan struct{}, time.Duration, func()) { l.mu.Lock() defer l.mu.Unlock() if l.sem == nil { @@ -112,23 +122,24 @@ func (l *reachColdLimiter) slots() (chan struct{}, time.Duration, func()) { n = 1 } l.sem = make(chan struct{}, n) + l.queued = make(chan struct{}, reachColdWaiterLimit) } w := l.wait if w <= 0 { w = reachColdBuildWait } - return l.sem, w, l.hook + return l.sem, l.queued, w, l.hook } // acquire takes a slot for one cold Reach build. It returns nil when a slot // was taken (the caller MUST release it), ctx.Err() when the request went -// away while queued, or errReachColdBusy when the limiter stayed saturated -// for the whole queue wait. +// away while queued, or errReachColdBusy when the queue is full or the limiter +// stayed saturated for the whole queue wait. func (l *reachColdLimiter) acquire(ctx context.Context) error { if err := ctx.Err(); err != nil { return err } - sem, wait, hook := l.slots() + sem, queued, wait, hook := l.slots() // Fast path: a free slot is taken without arming a timer. select { case sem <- struct{}{}: @@ -136,6 +147,15 @@ func (l *reachColdLimiter) acquire(ctx context.Context) error { return nil default: } + // Bound waiting leaders before allocating a timer. Every exit below gives + // back the admission, including promotion to an active scan. + select { + case queued <- struct{}{}: + defer func() { <-queued }() + default: + l.rejected.Add(1) + return errReachColdBusy + } t := time.NewTimer(wait) defer t.Stop() select { @@ -172,7 +192,7 @@ func (l *reachColdLimiter) took(hook func()) { // defers it). Never blocks, so an unbalanced release cannot wedge a request. func (l *reachColdLimiter) release() { l.inflight.Add(-1) - sem, _, _ := l.slots() + sem, _, _, _ := l.slots() select { case <-sem: default: diff --git a/cmd/server/node_reach_cold_limit_test.go b/cmd/server/node_reach_cold_limit_test.go index a9c2b1172..0ac9dae37 100644 --- a/cmd/server/node_reach_cold_limit_test.go +++ b/cmd/server/node_reach_cold_limit_test.go @@ -60,36 +60,38 @@ func clearReachCache(t *testing.T, srv *Server) { // --- limiter unit semantics --- -// Many distinct holders: the limiter must never let more than `limit` run at -// once, yet must let every holder through eventually. The test tracks peak -// occupancy itself so the limiter's own counters are verified, not trusted. +// Many distinct holders in bursts within the queue capacity: the limiter must +// never let more than `limit` run at once, yet serve every admitted holder. +// The test tracks peak occupancy itself, independently of the limiter counters. func TestReachColdLimiter_BoundsManyDistinctHolders(t *testing.T) { var l reachColdLimiter l.configure(2, 5*time.Second, nil) var live, peak atomic.Int64 var wg sync.WaitGroup - for i := 0; i < 32; i++ { - wg.Add(1) - go func() { - defer wg.Done() - if err := l.acquire(context.Background()); err != nil { - t.Errorf("acquire: %v", err) - return - } - n := live.Add(1) - for { - p := peak.Load() - if n <= p || peak.CompareAndSwap(p, n) { - break + for batch := 0; batch < 4; batch++ { + for i := 0; i < 8; i++ { + wg.Add(1) + go func() { + defer wg.Done() + if err := l.acquire(context.Background()); err != nil { + t.Errorf("acquire: %v", err) + return } - } - time.Sleep(2 * time.Millisecond) - live.Add(-1) - l.release() - }() + n := live.Add(1) + for { + p := peak.Load() + if n <= p || peak.CompareAndSwap(p, n) { + break + } + } + time.Sleep(2 * time.Millisecond) + live.Add(-1) + l.release() + }() + } + wg.Wait() } - wg.Wait() if got := peak.Load(); got != 2 { t.Fatalf("observed peak concurrency = %d, want exactly 2 (bound honoured and reached)", got) @@ -115,8 +117,11 @@ func TestReachColdLimiter_DefaultsBoundToTwoWithBoundedWait(t *testing.T) { if reachColdBuildLimit != 2 { t.Fatalf("reachColdBuildLimit = %d, want 2 (half of the 4-connection read pool)", reachColdBuildLimit) } - if reachColdBuildWait <= 0 || reachColdBuildWait > 5*time.Second { - t.Fatalf("reachColdBuildWait = %v, want a bounded positive wait", reachColdBuildWait) + if reachColdBuildWait != 3*time.Second { + t.Fatalf("reachColdBuildWait = %v, want the documented 3-second wait", reachColdBuildWait) + } + if reachColdWaiterLimit != 8 { + t.Fatalf("reachColdWaiterLimit = %d, want the documented eight queued leaders", reachColdWaiterLimit) } // Zero value must come up on the defaults, with no constructor. var l reachColdLimiter @@ -204,8 +209,8 @@ func TestReachColdLimiter_SaturationRejectsThenRecovers(t *testing.T) { // --- handler integration --- -// Many distinct cold keys (3 nodes x 8 windows) must all be served, but never -// more than `limit` of them may scan at the same time. +// Many distinct cold keys (3 bursts of 8 windows, within the queue capacity) +// must all be served, but never more than `limit` may scan at the same time. func TestNodeReach_ColdScansBoundedAcrossDistinctKeys(t *testing.T) { hold := func() { time.Sleep(10 * time.Millisecond) } srv, _, n := newColdLimitServer(t, 2, 10*time.Second, hold) @@ -223,8 +228,8 @@ func TestNodeReach_ColdScansBoundedAcrossDistinctKeys(t *testing.T) { } }(pk, days) } + wg.Wait() } - wg.Wait() if got := srv.reach.cold.peakInflight(); got > 2 { t.Fatalf("peak concurrent cold scans = %d, want <= 2", got) diff --git a/cmd/server/node_reach_waiter_cap_test.go b/cmd/server/node_reach_waiter_cap_test.go new file mode 100644 index 000000000..34658080f --- /dev/null +++ b/cmd/server/node_reach_waiter_cap_test.go @@ -0,0 +1,322 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "strconv" + "sync" + "testing" + "time" +) + +// Done is evaluated only when acquire reaches its queued select. This lets +// tests fill the queue without sleeps or inspecting its implementation. +type reachQueueContext struct { + context.Context + entered chan struct{} + once sync.Once +} + +func (c *reachQueueContext) Done() <-chan struct{} { + c.once.Do(func() { close(c.entered) }) + return c.Context.Done() +} + +// receiveReachResult bounds both the test's waits and cleanup, including red +// runs where an admission regression could otherwise strand a goroutine. +func receiveReachResult[T any](t *testing.T, ch <-chan T) T { + t.Helper() + select { + case result := <-ch: + return result + case <-time.After(5 * time.Second): + t.Fatal("Reach test worker did not finish") + var zero T + return zero + } +} + +func waitForReachQueue(t *testing.T, ctx *reachQueueContext) { + t.Helper() + select { + case <-ctx.entered: + case <-time.After(2 * time.Second): + t.Fatal("cold build did not enter the queue") + } +} + +func requireReachQueueFull(t *testing.T, l *reachColdLimiter) { + t.Helper() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan error, 1) + go func() { done <- l.acquire(ctx) }() + select { + case err := <-done: + if err == nil { + l.release() + } + if err != errReachColdBusy { + t.Fatalf("overflow acquire = %v, want errReachColdBusy", err) + } + case <-time.After(time.Second): + cancel() + receiveReachResult(t, done) + t.Fatal("overflow acquire waited with eight cold builds already queued; want immediate rejection") + } + + // A full queue must reject before creating a timer or retaining another + // queued build. With the channels initialized, rejection needs no allocation. + allocs := testing.AllocsPerRun(100, func() { + if err := l.acquire(context.Background()); err != errReachColdBusy { + t.Fatalf("overflow acquire = %v, want errReachColdBusy", err) + } + }) + if allocs != 0 { + t.Fatalf("full-queue rejection allocated %g times, want 0 (no wait timer)", allocs) + } +} + +// The ninth distinct queued leader must be shed, and every way out of the +// queue must return its admission so a later leader can wait in its place. +func TestReachColdLimiter_QueueCapAndAdmissionReuse(t *testing.T) { + for _, exit := range []string{"success", "cancellation", "timeout"} { + t.Run(exit, func(t *testing.T) { + var l reachColdLimiter + l.configure(2, 3*time.Second, nil) + held := 0 + for i := 0; i < 2; i++ { + if err := l.acquire(context.Background()); err != nil { + t.Fatalf("hold slot %d: %v", i, err) + } + held++ + } + t.Cleanup(func() { + for i := 0; i < held; i++ { + l.release() + } + }) + + type result struct { + index int + err error + } + results := make(chan result, 16) + queue := func(index int) context.CancelFunc { + ctx, cancel := context.WithCancel(context.Background()) + entered := &reachQueueContext{Context: ctx, entered: make(chan struct{})} + finished := make(chan struct{}) + go func() { + defer close(finished) + err := l.acquire(entered) + results <- result{index, err} + if err == nil { + <-ctx.Done() // keep a promoted build active until cleanup + l.release() + } + }() + t.Cleanup(func() { cancel(); receiveReachResult(t, finished) }) + waitForReachQueue(t, entered) + return cancel + } + cancels := make([]context.CancelFunc, 8) + for i := range cancels { + cancels[i] = queue(i) + } + requireReachQueueFull(t, &l) + + vacated := 1 + switch exit { + case "success": + l.release() + held-- + if got := receiveReachResult(t, results); got.err != nil { + t.Fatalf("promoted acquire = %v, want nil", got.err) + } + case "cancellation": + cancels[0]() + if got := receiveReachResult(t, results); got.index != 0 || got.err != context.Canceled { + t.Fatalf("cancelled acquire = %+v, want waiter 0 with context.Canceled", got) + } + case "timeout": + vacated = len(cancels) + for i := 0; i < vacated; i++ { + if got := receiveReachResult(t, results); got.err != errReachColdBusy { + t.Fatalf("timed-out acquire = %v, want errReachColdBusy", got.err) + } + } + } + for i := 0; i < vacated; i++ { + queue(8 + i) + } + requireReachQueueFull(t, &l) + if got := l.peakInflight(); got != 2 { + t.Fatalf("peak active builds = %d, want 2", got) + } + }) + } +} + +// A disconnect can race promotion from the queue. Either select result is +// legal, but the waiter admission must be released exactly once in both cases. +func TestReachColdLimiter_CancelRacesPromotion(t *testing.T) { + var l reachColdLimiter + l.configure(2, 5*time.Second, nil) + for i := 0; i < 2; i++ { + if err := l.acquire(context.Background()); err != nil { + t.Fatal(err) + } + } + for i := 0; i < 64; i++ { + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + queued := &reachQueueContext{Context: ctx, entered: make(chan struct{})} + result := make(chan error, 1) + go func() { result <- l.acquire(queued) }() + waitForReachQueue(t, queued) + + start := make(chan struct{}) + cancelled, released := make(chan struct{}), make(chan struct{}) + go func() { <-start; cancel(); close(cancelled) }() + go func() { <-start; l.release(); close(released) }() + close(start) + err := receiveReachResult(t, result) + receiveReachResult(t, cancelled) + receiveReachResult(t, released) + if err == nil { + l.release() + } else if err != context.Canceled { + t.Fatalf("iteration %d: acquire = %v, want success or cancellation", i, err) + } + if got := len(l.queued); got != 0 { + t.Fatalf("iteration %d: queued admissions = %d, want 0", i, got) + } + if got := l.inflight.Load(); got != 1 { + t.Fatalf("iteration %d: active builds = %d, want the untouched holder only", i, got) + } + if err := l.acquire(context.Background()); err != nil { + t.Fatalf("iteration %d: restore second holder: %v", i, err) + } + } + l.release() + l.release() +} + +// A full queue protects the read pool while warm hits and same-key followers +// still work. Only the eight distinct cold leaders may start scans afterward. +func TestNodeReach_FullColdQueuePreservesWarmHitsAndCoalescing(t *testing.T) { + srv, db, n := newColdLimitServer(t, 0, 10*time.Second, nil) + path := "/api/nodes/" + n + "/reach?days=" + warm := serveReach(srv, path+"30") + if warm.Code != http.StatusOK { + t.Fatalf("warm-up status = %d, want 200", warm.Code) + } + for i := 0; i < 2; i++ { + if err := srv.reach.cold.acquire(context.Background()); err != nil { + t.Fatalf("hold slot %d: %v", i, err) + } + } + held := true + t.Cleanup(func() { + if held { + srv.reach.cold.release() + srv.reach.cold.release() + } + }) + // newColdLimitServer uses Config{}: no live hidden-name query runs before + // admission. The warm-up above also primes our fork's separate degree/rank + // cache, so warm responses below do not need a cold snapshot query. This is + // a limiter/pool test, not a claim that visibility lookups never use SQL. + // Hold the fixture's sole connection: any premature query would queue on + // the pool and increment WaitCount, even if it later completed normally. + conn, err := db.conn.Conn(context.Background()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { conn.Close() }) + waits := db.conn.Stats().WaitCount + + ctx, cancel := context.WithCancel(context.Background()) + var wg sync.WaitGroup + results := make(chan *httptest.ResponseRecorder, 20) + t.Cleanup(func() { + cancel() + joined := make(chan struct{}) + go func() { wg.Wait(); close(joined) }() + receiveReachResult(t, joined) + }) + for days := 1; days <= 8; days++ { + queued := &reachQueueContext{Context: ctx, entered: make(chan struct{})} + wg.Add(1) + go func(days int) { + defer wg.Done() + results <- serveReachCtx(srv, queued, path+strconv.Itoa(days)) + }(days) + waitForReachQueue(t, queued) + } + + overflowCtx, cancelOverflow := context.WithCancel(ctx) + defer cancelOverflow() + overflow := make(chan *httptest.ResponseRecorder, 1) + go func() { overflow <- serveReachCtx(srv, overflowCtx, path+"9") }() + select { + case rr := <-overflow: + if rr.Code != http.StatusTooManyRequests || rr.Header().Get("Retry-After") != "2" { + t.Fatalf("full queue: status=%d Retry-After=%q, want 429 and 2", rr.Code, rr.Header().Get("Retry-After")) + } + var body struct { + Error string `json:"error"` + RetryAfter int `json:"retryAfter"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil || body.Error == "" || body.RetryAfter != 2 { + t.Fatalf("full queue body = %s, want JSON error and retryAfter=2 (decode=%v)", rr.Body.String(), err) + } + case <-time.After(time.Second): + cancelOverflow() + receiveReachResult(t, overflow) + t.Fatal("ninth distinct cold leader queued; want immediate HTTP 429") + } + if rr := serveReach(srv, path+"30"); rr.Code != http.StatusOK || rr.Body.String() != warm.Body.String() { + t.Fatalf("warm hit with full queue: status=%d body=%s", rr.Code, rr.Body.String()) + } + for i := 0; i < 12; i++ { + wg.Add(1) + go func() { + defer wg.Done() + results <- serveReachCtx(srv, ctx, path+"1") + }() + } + select { + case rr := <-results: + t.Fatalf("queued leader or same-key follower returned early: status=%d", rr.Code) + case <-time.After(100 * time.Millisecond): + } + if got := db.conn.Stats().WaitCount; got != waits { + t.Fatalf("pool wait count = %d, want %d (queued work must not request a connection)", got, waits) + } + if got := srv.reachCacheLen(); got != 1 { + t.Fatalf("cache entries = %d, want only the warm report", got) + } + conn.Close() + srv.reach.cold.release() + srv.reach.cold.release() + held = false + for i := 0; i < 20; i++ { + select { + case rr := <-results: + if rr.Code != http.StatusOK { + t.Fatalf("after freeing slots: status=%d, want 200 (body=%s)", rr.Code, rr.Body.String()) + } + case <-time.After(5 * time.Second): + t.Fatal("admitted request did not finish after freeing slots") + } + } + if got := srv.reach.cold.acquiredCount(); got != 11 { + t.Fatalf("acquired = %d, want 1 warm-up + 2 holds + 8 distinct cold leaders", got) + } + if got := srv.reach.cold.rejectedCount(); got != 1 { + t.Fatalf("rejected = %d, want only the excess distinct leader", got) + } +} diff --git a/docs/api-spec.md b/docs/api-spec.md index b72c071de..8bf51bda4 100644 --- a/docs/api-spec.md +++ b/docs/api-spec.md @@ -1239,14 +1239,22 @@ Returned when the node is unknown or hidden (see Visibility). ### Response `429` -Returned when the report is **cold** (not cached) and too many other cold -reports are already scanning the database. At most two cold reports compute at -once, so a burst of different nodes/windows cannot occupy the read pool and -slow unrelated endpoints; a cold request queues briefly for a slot and is only -shed if the limit is still saturated. Warm (cached) responses are never -affected, and concurrent requests for the *same* node and window still share -one computation. Nothing is cached for a shed request — retry after -`Retry-After` seconds and it recomputes normally. +Returned when the report is **cold** (not cached) and its build cannot be +admitted. At most two cold reports compute at once; up to eight distinct cold +build leaders may wait for a slot for up to three seconds without holding a +database connection while waiting for that slot. Visibility and ranking checks +can still query the database outside this limiter. A further distinct leader +gets 429 immediately when the queue is full, before allocating a wait timer. +An admitted leader gets 429 if +its wait expires before a slot becomes available. These are fixed internal +defaults, not operator/customizer settings. + +Warm (cached) responses bypass this limiter. Concurrent requests for the +*same* node and window still share one computation and one queue admission; +the eight-leader cap does **not** bound all existing singleflight followers. +Nothing is cached for a shed request — retry after `Retry-After` seconds and +it recomputes normally. Existing request-cancellation/error and visibility +behavior is unchanged. ``` Retry-After: 2