From e90f8034f4cda5cae41fe2a9ad2a05b0641db42f Mon Sep 17 00:00:00 2001 From: dborup Date: Wed, 7 Oct 2026 21:35:41 +0200 Subject: [PATCH] perf(reach): bound concurrent cold-cache scans across distinct nodes /api/nodes/{pk}/reach already collapses a herd on one cache key with singleflight, but distinct node/window keys scan independently: a burst of cold keys runs that many windowed observation scans at once and can hold the whole 4-connection read pool, leaving /api/nodes and /api/packets queueing. Add a per-server counting semaphore for cold Reach builds (default 2, half the read pool) with a bounded 3s queue wait, taken inside the singleflight slot after the cache recheck and released with defer on every exit. A request still saturated at the end of the wait is shed with 429 + Retry-After and nothing is cached. Upstream's non-blocking variant refuses the third build instantly; we keep its cap but queue first, because a cold report normally completes in about a second and our front end only auto-retries 503, so an instant refusal would fail a page whose database is not under real pressure. Queue time costs a parked goroutine, not a database connection. Warm hits never reach the limiter, same-key coalescing is unchanged, and visibility filtering, rank snapshots and response/cache semantics are untouched. Relates to #341 --- cmd/server/node_reach.go | 23 ++ cmd/server/node_reach_cold_limit.go | 196 +++++++++++ cmd/server/node_reach_cold_limit_test.go | 425 +++++++++++++++++++++++ docs/api-spec.md | 19 + 4 files changed, 663 insertions(+) create mode 100644 cmd/server/node_reach_cold_limit.go create mode 100644 cmd/server/node_reach_cold_limit_test.go diff --git a/cmd/server/node_reach.go b/cmd/server/node_reach.go index 4fbcdb437..9b4061a0b 100644 --- a/cmd/server/node_reach.go +++ b/cmd/server/node_reach.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "encoding/json" + "errors" "log" "net/http" "sort" @@ -308,6 +309,12 @@ type reachState struct { // simultaneous callers run the scan + attribution once, not N times. sf singleflight.Group + // cold bounds how many *distinct* cold keys may scan the database at the + // same time, so a burst of different nodes/windows cannot occupy the + // whole read pool and starve unrelated endpoints (#341). sf covers one + // key; this covers the set of them. Zero value is ready to use. + cold reachColdLimiter + // lastSeenBlacklistGen is the BlacklistGeneration() value that the cache // was last reconciled with. When the live generation moves past this // value, the cache is purged wholesale on the next request to prevent @@ -616,6 +623,16 @@ func (s *Server) handleNodeReach(w http.ResponseWriter, r *http.Request) { if e, ok := s.reachCacheGet(cacheKey); ok { return e, nil } + // Only now is the work known to be genuinely cold (cache rechecked + // inside the singleflight slot), so this is where the cross-key cold + // build limit is taken — never on a warm hit, and once per coalesced + // herd rather than once per waiter (#341). The deferred release + // covers every exit: success, compute error, marshal error, a + // cancelled request and a panic alike. + if aErr := s.reach.cold.acquire(r.Context()); aErr != nil { + return nil, aErr + } + defer s.reach.cold.release() resp, ok, cErr := s.computeNodeReach(r.Context(), pubkey, days) if cErr != nil { // Real backend failure (e.g. DB scan exploded) — propagate so the @@ -648,6 +665,12 @@ func (s *Server) handleNodeReach(w http.ResponseWriter, r *http.Request) { return e, nil }) if err != nil { + // Shed cold work rather than queue it behind a saturated limiter: + // 429 + Retry-After, nothing cached, next request recomputes (#341). + if errors.Is(err, errReachColdBusy) { + writeReachColdBusy429(w) + return + } writeError(w, 500, "reach computation failed") return } diff --git a/cmd/server/node_reach_cold_limit.go b/cmd/server/node_reach_cold_limit.go new file mode 100644 index 000000000..32e3cf10b --- /dev/null +++ b/cmd/server/node_reach_cold_limit.go @@ -0,0 +1,196 @@ +package main + +import ( + "context" + "errors" + "net/http" + "strconv" + "sync" + "sync/atomic" + "time" +) + +// Bounded concurrency for *cold* Reach reports (#341). +// +// /api/nodes/{pk}/reach already collapses a thundering herd on one cache key +// with singleflight, but distinct node/window keys are independent: N cold +// keys arriving together run N windowed observation scans at once. The +// server's read pool is 4 connections (OpenDB), so a handful of concurrent +// cold Reach scans can hold every connection and leave /api/nodes and +// /api/packets queueing behind them. +// +// Upstream (fork PR 2127) shields the pool with a *non-blocking* two-build +// semaphore: the third concurrent build is refused immediately with HTTP 429 + +// Retry-After. Evaluated against our deployment: +// +// - The cap of 2 fits our pool as is: two cold scans leave at least two of +// the four read connections for every other endpoint, which is what the +// issue asks for. We keep it. +// - Refusing instantly does not. Our Reach page is one request per view and +// a cold report normally completes in well under a second, so a plain +// 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. +// - 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. +// +// 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 +// pool-pressure evaluation in #221. +// +// Vars (not consts) so tests can tighten the cap and the queue wait without +// standing up dozens of slow scans — same pattern as reachScanRowLimit. +var ( + // reachColdBuildLimit is how many cold Reach reports may compute at once. + reachColdBuildLimit = 2 + // reachColdBuildWait is how long a cold request queues for a slot before + // it is shed with 429. Sized above a typical cold scan (≈1s on a + // 400k-observation fixture) so a handful of simultaneous cold pages are + // served in turn rather than refused, while sustained saturation still + // sheds promptly. Queue time costs a parked goroutine, not a database + // connection. + reachColdBuildWait = 3 * time.Second +) + +// reachColdRetryAfterSeconds is the Retry-After advertised on a shed cold +// Reach request. Deliberately short: the limiter frees up as soon as one of +// the in-flight scans finishes. +const reachColdRetryAfterSeconds = 2 + +// 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. +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 +// 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 + + inflight atomic.Int64 // slots currently held + peak atomic.Int64 // high-water mark of inflight + acquired atomic.Uint64 // slots handed out + rejected atomic.Uint64 // requests shed with errReachColdBusy + canceled atomic.Uint64 // waiters whose context ended while queued +} + +// configure replaces the limiter's capacity, queue wait and acquire hook. +// Call it before any request can acquire (test setup): in-flight holders +// release into the channel this installs, so swapping it mid-flight loses +// their slots. limit <= 0 and wait <= 0 fall back to the package defaults; +// hook is nil in production. +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 +} + +// slots returns the (lazily built) slot channel, the queue wait and the +// acquire hook. +func (l *reachColdLimiter) slots() (chan struct{}, time.Duration, func()) { + l.mu.Lock() + defer l.mu.Unlock() + if l.sem == nil { + n := l.lim + if n <= 0 { + n = reachColdBuildLimit + } + if n <= 0 { + n = 1 + } + l.sem = make(chan struct{}, n) + } + w := l.wait + if w <= 0 { + w = reachColdBuildWait + } + return l.sem, 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. +func (l *reachColdLimiter) acquire(ctx context.Context) error { + if err := ctx.Err(); err != nil { + return err + } + sem, wait, hook := l.slots() + // Fast path: a free slot is taken without arming a timer. + select { + case sem <- struct{}{}: + l.took(hook) + return nil + default: + } + t := time.NewTimer(wait) + defer t.Stop() + select { + case sem <- struct{}{}: + l.took(hook) + return nil + case <-ctx.Done(): + l.canceled.Add(1) + return ctx.Err() + case <-t.C: + l.rejected.Add(1) + return errReachColdBusy + } +} + +// took records a handed-out slot and runs the test hook (if any) outside the +// limiter's own mutex. +func (l *reachColdLimiter) took(hook func()) { + l.acquired.Add(1) + n := l.inflight.Add(1) + for { + p := l.peak.Load() + if n <= p || l.peak.CompareAndSwap(p, n) { + break + } + } + if hook != nil { + hook() + } +} + +// release hands a slot back. Must be called exactly once per successful +// acquire — on success, on error and on cancellation alike (the handler +// defers it). Never blocks, so an unbalanced release cannot wedge a request. +func (l *reachColdLimiter) release() { + l.inflight.Add(-1) + sem, _, _ := l.slots() + select { + case <-sem: + default: + } +} + +func (l *reachColdLimiter) peakInflight() int64 { return l.peak.Load() } +func (l *reachColdLimiter) acquiredCount() uint64 { return l.acquired.Load() } +func (l *reachColdLimiter) rejectedCount() uint64 { return l.rejected.Load() } +func (l *reachColdLimiter) canceledCount() uint64 { return l.canceled.Load() } + +// writeReachColdBusy429 sheds a cold Reach request that could not get a slot. +// 429 (not 503) because the server is healthy and the work is rate-limited; +// Retry-After tells a well-behaved client how long to back off. +func writeReachColdBusy429(w http.ResponseWriter) { + ra := strconv.Itoa(reachColdRetryAfterSeconds) + w.Header().Set("Retry-After", ra) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusTooManyRequests) + _, _ = w.Write([]byte(`{"error":"reach is busy computing other reports","retryAfter":` + ra + `}`)) +} diff --git a/cmd/server/node_reach_cold_limit_test.go b/cmd/server/node_reach_cold_limit_test.go new file mode 100644 index 000000000..a9c2b1172 --- /dev/null +++ b/cmd/server/node_reach_cold_limit_test.go @@ -0,0 +1,425 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sort" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/gorilla/mux" +) + +// Tests for issue #341: bound how many *cold* Reach reports may scan the +// database concurrently across distinct node/window keys, so a burst of +// distinct cold keys cannot occupy the whole read pool and starve unrelated +// endpoints. The limit must be taken only for genuinely cold work (after the +// in-flight cache recheck), always released, and must leave warm hits and +// same-key singleflight coalescing untouched. + +// newColdLimitServer builds the shared reach integration fixture with the +// cold limiter configured for the test (limit, queue wait, acquire hook). +func newColdLimitServer(t *testing.T, limit int, wait time.Duration, hook func()) (*Server, *DB, string) { + t.Helper() + db, n := newReachIntegrationDB(t, `["AABB","01FA","CCDD"]`) + t.Cleanup(func() { db.conn.Close() }) + // The fixture is an anonymous :memory: database, where every extra pool + // connection would see its own empty schema. Pin the pool to one + // connection so concurrent handler goroutines share the seeded data (the + // limiter is taken before any query, so slot concurrency is unaffected). + db.conn.SetMaxOpenConns(1) + cfg := &Config{} + srv := &Server{store: newTestStoreWithDB(t, db, cfg), db: db, cfg: cfg, perfStats: NewPerfStats()} + resetReachState(t, srv) + srv.reach.cold.configure(limit, wait, hook) + return srv, db, n +} + +// serveReachCtx is serveReach with a caller-supplied request context so a +// test can cancel mid-scan. +func serveReachCtx(srv *Server, ctx context.Context, path string) *httptest.ResponseRecorder { + router := mux.NewRouter() + router.HandleFunc("/api/nodes/{pubkey}/reach", srv.handleNodeReach).Methods("GET") + req := httptest.NewRequest("GET", path, nil).WithContext(ctx) + rr := httptest.NewRecorder() + router.ServeHTTP(rr, req) + return rr +} + +func clearReachCache(t *testing.T, srv *Server) { + t.Helper() + srv.reach.cacheMu.Lock() + srv.reach.cache = map[string]reachCacheEntry{} + srv.reach.cacheMu.Unlock() +} + +// --- 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. +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 + } + } + time.Sleep(2 * time.Millisecond) + live.Add(-1) + l.release() + }() + } + wg.Wait() + + if got := peak.Load(); got != 2 { + t.Fatalf("observed peak concurrency = %d, want exactly 2 (bound honoured and reached)", got) + } + if got := l.peakInflight(); got != peak.Load() { + t.Fatalf("limiter peakInflight = %d, want %d", got, peak.Load()) + } + if got := l.inflight.Load(); got != 0 { + t.Fatalf("inflight after all releases = %d, want 0", got) + } + if got := l.acquiredCount(); got != 32 { + t.Fatalf("acquiredCount = %d, want 32", got) + } + if got := l.rejectedCount(); got != 0 { + t.Fatalf("rejectedCount = %d, want 0 with a generous queue wait", got) + } +} + +// The shipped defaults are part of the contract: the cap must leave read-pool +// headroom for other endpoints (pool is 4 connections in OpenDB), and the +// queue wait must be a real, bounded wait rather than an instant refusal. +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) + } + // Zero value must come up on the defaults, with no constructor. + var l reachColdLimiter + for i := 0; i < reachColdBuildLimit; i++ { + if err := l.acquire(context.Background()); err != nil { + t.Fatalf("acquire %d on a zero-value limiter: %v", i, err) + } + } + l.configure(0, 5*time.Millisecond, nil) // default cap, short wait for the shed check + for i := 0; i < reachColdBuildLimit; i++ { + if err := l.acquire(context.Background()); err != nil { + t.Fatalf("acquire %d after configure: %v", i, err) + } + } + if err := l.acquire(context.Background()); err != errReachColdBusy { + t.Fatalf("acquire past the default cap = %v, want errReachColdBusy", err) + } +} + +// Cancellation while queued must not consume a slot and must surface the +// context error (not the busy sentinel), so the handler keeps its existing +// behaviour for a disconnecting client. +func TestReachColdLimiter_CancellationDoesNotConsumeSlot(t *testing.T) { + var l reachColdLimiter + l.configure(1, 5*time.Second, nil) + if err := l.acquire(context.Background()); err != nil { + t.Fatalf("first acquire: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- l.acquire(ctx) }() + time.Sleep(20 * time.Millisecond) + cancel() + select { + case err := <-done: + if err != context.Canceled { + t.Fatalf("queued acquire after cancel = %v, want context.Canceled", err) + } + case <-time.After(2 * time.Second): + t.Fatal("queued acquire did not return after cancellation") + } + if got := l.inflight.Load(); got != 1 { + t.Fatalf("inflight = %d, want 1 (cancelled waiter must not take a slot)", got) + } + if got := l.canceledCount(); got != 1 { + t.Fatalf("canceledCount = %d, want 1", got) + } + + // An already-cancelled context fails fast, without taking the free slot. + l.release() + dead, stop := context.WithCancel(context.Background()) + stop() + if err := l.acquire(dead); err != context.Canceled { + t.Fatalf("acquire with dead context = %v, want context.Canceled", err) + } + if got := l.inflight.Load(); got != 0 { + t.Fatalf("inflight = %d, want 0", got) + } +} + +// Saturation then recovery: a full limiter rejects with the busy sentinel +// after the bounded queue wait, and accepts again as soon as a slot frees. +func TestReachColdLimiter_SaturationRejectsThenRecovers(t *testing.T) { + var l reachColdLimiter + l.configure(1, 5*time.Millisecond, nil) + if err := l.acquire(context.Background()); err != nil { + t.Fatalf("first acquire: %v", err) + } + if err := l.acquire(context.Background()); err != errReachColdBusy { + t.Fatalf("saturated acquire = %v, want errReachColdBusy", err) + } + if got := l.rejectedCount(); got != 1 { + t.Fatalf("rejectedCount = %d, want 1", got) + } + l.release() + if err := l.acquire(context.Background()); err != nil { + t.Fatalf("acquire after release = %v, want nil (recovery)", err) + } + l.release() + if got := l.inflight.Load(); got != 0 { + t.Fatalf("inflight = %d, want 0", got) + } +} + +// --- 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. +func TestNodeReach_ColdScansBoundedAcrossDistinctKeys(t *testing.T) { + hold := func() { time.Sleep(10 * time.Millisecond) } + srv, _, n := newColdLimitServer(t, 2, 10*time.Second, hold) + keys := []string{n, pk64("aabb"), pk64("ccdd")} + + var wg sync.WaitGroup + for _, pk := range keys { + for days := 1; days <= 8; days++ { + wg.Add(1) + go func(pk string, days int) { + defer wg.Done() + rr := serveReach(srv, "/api/nodes/"+pk+"/reach?days="+strconv.Itoa(days)) + if rr.Code != http.StatusOK { + t.Errorf("%s days=%d: status=%d want 200 (body=%s)", pk[:8], days, rr.Code, rr.Body.String()) + } + }(pk, days) + } + } + wg.Wait() + + if got := srv.reach.cold.peakInflight(); got > 2 { + t.Fatalf("peak concurrent cold scans = %d, want <= 2", got) + } + if got := srv.reach.cold.peakInflight(); got < 1 { + t.Fatalf("peak concurrent cold scans = %d, want >= 1 (handler must take the limit)", got) + } + if got := srv.reach.cold.acquiredCount(); got != uint64(len(keys)*8) { + t.Fatalf("acquiredCount = %d, want %d (one per distinct cold key)", got, len(keys)*8) + } + if got := srv.reach.cold.inflight.Load(); got != 0 { + t.Fatalf("inflight after all requests = %d, want 0", got) + } +} + +// Identical keys stay coalesced by singleflight and take the limit once, so a +// thundering herd on one cold key costs one slot, not N. +func TestNodeReach_IdenticalKeysCoalesceIntoOneSlot(t *testing.T) { + hold := func() { time.Sleep(25 * time.Millisecond) } + srv, _, n := newColdLimitServer(t, 1, 10*time.Second, hold) + + var wg sync.WaitGroup + for i := 0; i < 12; i++ { + wg.Add(1) + go func() { + defer wg.Done() + rr := serveReach(srv, "/api/nodes/"+n+"/reach?days=30") + if rr.Code != http.StatusOK { + t.Errorf("status=%d want 200 (body=%s)", rr.Code, rr.Body.String()) + } + }() + } + wg.Wait() + + if got := srv.reach.cold.acquiredCount(); got != 1 { + t.Fatalf("acquiredCount = %d, want 1 (same-key coalescing must hold the limit once)", got) + } + if got := srv.reach.cold.rejectedCount(); got != 0 { + t.Fatalf("rejectedCount = %d, want 0", got) + } +} + +// Warm hits must not touch the limiter at all: with every slot held and a +// long queue wait, a cached key still answers immediately. +func TestNodeReach_WarmHitSkipsColdLimit(t *testing.T) { + srv, _, n := newColdLimitServer(t, 1, 10*time.Second, nil) + path := "/api/nodes/" + n + "/reach?days=30" + if rr := serveReach(srv, path); rr.Code != http.StatusOK { + t.Fatalf("warm-up: status=%d want 200 (body=%s)", rr.Code, rr.Body.String()) + } + if got := srv.reach.cold.acquiredCount(); got != 1 { + t.Fatalf("acquiredCount after cold warm-up = %d, want 1", got) + } + + // Occupy the only slot; a warm hit must not queue behind it. + if err := srv.reach.cold.acquire(context.Background()); err != nil { + t.Fatalf("test hold: %v", err) + } + defer srv.reach.cold.release() + + start := time.Now() + rr := serveReach(srv, path) + elapsed := time.Since(start) + if rr.Code != http.StatusOK { + t.Fatalf("warm hit under saturation: status=%d want 200 (body=%s)", rr.Code, rr.Body.String()) + } + if elapsed > 2*time.Second { + t.Fatalf("warm hit took %v under saturation, want immediate (no queue wait)", elapsed) + } + if got := srv.reach.cold.acquiredCount(); got != 2 { + t.Fatalf("acquiredCount = %d, want 2 (one cold warm-up + the test's own hold only)", got) + } +} + +// A failing cold scan must release the slot: the 500 path is the one that +// leaks a semaphore if release is not deferred. +func TestNodeReach_ComputeErrorReleasesColdSlot(t *testing.T) { + srv, db, n := newColdLimitServer(t, 1, 5*time.Millisecond, nil) + path := "/api/nodes/" + n + "/reach?days=30" + if rr := serveReach(srv, path); rr.Code != http.StatusOK { + t.Fatalf("warm-up: status=%d want 200 (body=%s)", rr.Code, rr.Body.String()) + } + clearReachCache(t, srv) + if _, err := db.conn.Exec("DROP TABLE observations"); err != nil { + t.Fatalf("drop observations: %v", err) + } + + rr := serveReach(srv, path) + if rr.Code != http.StatusInternalServerError { + t.Fatalf("status=%d want 500 on scan failure (body=%s)", rr.Code, rr.Body.String()) + } + if got := srv.reach.cold.inflight.Load(); got != 0 { + t.Fatalf("inflight after failed scan = %d, want 0 (slot must be released on error)", got) + } + // The limiter must still hand out its only slot. + if err := srv.reach.cold.acquire(context.Background()); err != nil { + t.Fatalf("acquire after failed scan = %v, want nil (slot leaked)", err) + } + srv.reach.cold.release() +} + +// A cancelled request must release the slot too (handler path, not just the +// limiter unit). +func TestNodeReach_CancelledColdRequestReleasesSlot(t *testing.T) { + ctx, cancel := context.WithCancel(context.Background()) + hold := func() { cancel(); time.Sleep(5 * time.Millisecond) } + srv, _, n := newColdLimitServer(t, 1, 5*time.Millisecond, hold) + + rr := serveReachCtx(srv, ctx, "/api/nodes/"+n+"/reach?days=30") + if rr.Code != http.StatusOK && rr.Code != http.StatusInternalServerError { + t.Fatalf("status=%d, want 200 or 500 for a cancelled scan (body=%s)", rr.Code, rr.Body.String()) + } + if got := srv.reach.cold.inflight.Load(); got != 0 { + t.Fatalf("inflight after cancelled request = %d, want 0", got) + } + if err := srv.reach.cold.acquire(context.Background()); err != nil { + t.Fatalf("acquire after cancelled request = %v, want nil (slot leaked)", err) + } + srv.reach.cold.release() +} + +// Saturation and recovery through the HTTP surface: a cold request that +// cannot get a slot answers 429 + Retry-After, caches nothing, and the very +// same request returns the identical report once a slot frees. +func TestNodeReach_SaturatedColdRequestReturns429AndRecovers(t *testing.T) { + srv, _, n := newColdLimitServer(t, 1, 5*time.Millisecond, nil) + path := "/api/nodes/" + n + "/reach?days=30" + + base := serveReach(srv, path) + if base.Code != http.StatusOK { + t.Fatalf("baseline: status=%d want 200 (body=%s)", base.Code, base.Body.String()) + } + want := normaliseReachBody(t, base.Body.Bytes()) + clearReachCache(t, srv) + + if err := srv.reach.cold.acquire(context.Background()); err != nil { + t.Fatalf("test hold: %v", err) + } + rr := serveReach(srv, path) + if rr.Code != http.StatusTooManyRequests { + t.Fatalf("saturated cold request: status=%d want 429 (body=%s)", rr.Code, rr.Body.String()) + } + if ra := rr.Header().Get("Retry-After"); ra != strconv.Itoa(reachColdRetryAfterSeconds) { + t.Fatalf("Retry-After = %q, want %q", ra, strconv.Itoa(reachColdRetryAfterSeconds)) + } + if ct := rr.Header().Get("Content-Type"); ct != "application/json" { + t.Fatalf("Content-Type = %q, want application/json", ct) + } + var body struct { + Error string `json:"error"` + RetryAfter int `json:"retryAfter"` + } + if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil { + t.Fatalf("429 body is not JSON: %v (%s)", err, rr.Body.String()) + } + if body.Error == "" || body.RetryAfter != reachColdRetryAfterSeconds { + t.Fatalf("429 body = %+v, want non-empty error and retryAfter=%d", body, reachColdRetryAfterSeconds) + } + if got := srv.reachCacheLen(); got != 0 { + t.Fatalf("reach cache len = %d after a shed request, want 0 (nothing cached)", got) + } + + srv.reach.cold.release() + rec := serveReach(srv, path) + if rec.Code != http.StatusOK { + t.Fatalf("after recovery: status=%d want 200 (body=%s)", rec.Code, rec.Body.String()) + } + if got := normaliseReachBody(t, rec.Body.Bytes()); got != want { + t.Fatalf("report changed after shed+recovery:\n got %s\nwant %s", got, want) + } + if got := srv.reachCacheLen(); got != 1 { + t.Fatalf("reach cache len = %d after recovery, want 1", got) + } +} + +// normaliseReachBody re-marshals a reach report with the wall-clock-dependent +// window start removed and links/observers put in pubkey order, so two +// reports computed seconds apart compare equal. The ordering step is needed +// because links that tie on every sort key (bidir, bottleneck, we+they) come +// out in map-iteration order — a pre-existing tie-break gap, not something +// this change introduces. +func normaliseReachBody(t *testing.T, raw []byte) string { + t.Helper() + var resp NodeReachResponse + if err := json.Unmarshal(raw, &resp); err != nil { + t.Fatalf("unmarshal reach body: %v (%s)", err, raw) + } + resp.Window.Since = "" + sort.Slice(resp.Links, func(i, j int) bool { return resp.Links[i].Pubkey < resp.Links[j].Pubkey }) + sort.Slice(resp.DirectObservers, func(i, j int) bool { + return resp.DirectObservers[i].Pubkey < resp.DirectObservers[j].Pubkey + }) + out, err := json.Marshal(resp) + if err != nil { + t.Fatalf("remarshal reach body: %v", err) + } + return string(out) +} diff --git a/docs/api-spec.md b/docs/api-spec.md index 021f4aa1a..05c61f8e3 100644 --- a/docs/api-spec.md +++ b/docs/api-spec.md @@ -1033,6 +1033,25 @@ Returned when the node is unknown or hidden (see Visibility). { "error": "Not found" } ``` +### 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. + +``` +Retry-After: 2 +``` + +```json +{ "error": "reach is busy computing other reports", "retryAfter": 2 } +``` + ### Response `500` Returned when the scan fails, or when the live name lookup for visibility