amadan

carlos / platform Public

console: sweep metrics without holding the cache lock, on a detached context

LoadMetrics held metricsCache.mu across the whole bucket sweep — a List,
48 Gets per box, and the alerts pass — and swept on the request's own
context with no deadline. handleApp calls it for every tenant's app
page, so a cold sweep made the mutex a queue the whole console stood in;
and because a failed sweep is never cached, a request cancelled
mid-sweep aborted the shared work and left the next request to start
over, so an impatiently reloaded page could starve the cache
indefinitely.

The lock now guards only the cache fields. One in-flight sweep is
tracked by a flag plus a done channel: concurrent cold callers coalesce
onto it instead of duplicating or blocking, and a waiter whose own
context is cancelled gives up without touching the sweep. The sweep
itself runs on context.WithoutCancel of the caller's, bounded by a new
15s metricsSweepTimeout, so no single request can abort or poison it.

Contract is unchanged: same signature, one shared sweep, a failed sweep
still never cached.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 1f72d1f0c91f4cc807ddb06ab36e7ca178b08199 parent 3ef9cba
2 files changed, +212 −17
  • internal/console/metricsview.go +96 −17
  • internal/console/metricsview_test.go +116 −0
diff --git a/internal/console/metricsview.go b/internal/console/metricsview.go
index aab226c37..4f01c8ec6 100644
--- a/internal/console/metricsview.go
+++ b/internal/console/metricsview.go
@@ -80,53 +80,132 @@ const metricsFreshWindow = 5 * time.Minute
// prefix and reads the path segment right after "boxes/".
const metricsBoxesPrefix = "metrics/boxes/"
+// metricsSweepTimeout bounds one whole bucket sweep: a List plus 48 Gets per
+// box plus the alerts pass. It exists because the sweep no longer runs on the
+// requesting page's context (see LoadMetrics) and so has no deadline of its
+// own to inherit — a store that accepts connections and then hangs would
+// otherwise pin the in-flight flag forever and every later caller with it.
+// Fifteen seconds is far beyond a healthy sweep of a fleet this size and far
+// short of any browser's patience.
+const metricsSweepTimeout = 15 * time.Second
+
// metricsCache holds the result of the last successful bucket sweep. A
// failed sweep never reaches here — LoadMetrics returns the error to its
// caller and leaves the previous (possibly stale-but-valid, possibly zero)
// cache exactly as it was.
+//
+// mu guards the fields and is deliberately NOT held across the sweep's bucket
+// IO; sweeping/done are what serialise the IO instead, so that a cold sweep
+// delays only the callers that actually need its result rather than every
+// caller that touches the cache. sweeping is true while one goroutine is out
+// at the bucket, and done is the channel it closes when it comes back.
type metricsCache struct {
- mu sync.Mutex
- at time.Time
- boxes []BoxMetricsView
- alerts []AlertRow
+ mu sync.Mutex
+ at time.Time
+ boxes []BoxMetricsView
+ alerts []AlertRow
+ sweeping bool
+ done chan struct{}
}
// LoadMetrics reads every box under metrics/boxes/, the raw chunks of the
// last 48 hours (missing hours skipped silently), and alerts/boxes/*.json,
// caching the result in-process for 60s so concurrent callers — the
// dashboard page and every owner's app-metrics page — share one sweep.
+//
+// Two properties matter beyond the cache itself, both learned from the shape
+// this replaced (mutex held across the whole sweep, on the request's own
+// context):
+//
+// - The cache mutex is never held across bucket IO. Only the goroutine that
+// actually sweeps is slow; everyone else either takes the cached answer
+// immediately or waits on the in-flight sweep's channel, and neither
+// blocks a caller whose cache is fresh. handleApp calls this on every
+// tenant's app page, so a cold sweep must not become a queue every page
+// on the console stands in.
+// - The sweep runs on its own context — context.WithoutCancel of the
+// caller's, plus metricsSweepTimeout — never the request's. A request
+// cancelled mid-sweep (a reload, a navigate-away) would otherwise abort
+// the shared sweep AND, since a failed sweep is never cached, leave the
+// next request to start from scratch: reload impatiently enough and the
+// cache never populates at all. The caller's own context still governs
+// WAITING (a cancelled waiter gives up rather than hanging), it just no
+// longer governs the sweep.
func (s *Server) LoadMetrics(ctx context.Context) ([]BoxMetricsView, []AlertRow, error) {
- s.metricsCache.mu.Lock()
- defer s.metricsCache.mu.Unlock()
+ c := &s.metricsCache
+ for {
+ c.mu.Lock()
+ if !c.at.IsZero() && time.Since(c.at) < metricsCacheTTL {
+ boxes, alerts := c.boxes, c.alerts
+ c.mu.Unlock()
+ return boxes, alerts, nil
+ }
+ if c.sweeping {
+ // Someone is already at the bucket: wait for their result
+ // rather than duplicating the sweep, then re-check (their
+ // sweep may have failed, in which case this iteration takes
+ // the work over).
+ wait := c.done
+ c.mu.Unlock()
+ select {
+ case <-wait:
+ continue
+ case <-ctx.Done():
+ return nil, nil, ctx.Err()
+ }
+ }
+ c.sweeping = true
+ c.done = make(chan struct{})
+ done := c.done
+ c.mu.Unlock()
+
+ boxes, alerts, err := s.sweepAll(ctx)
+
+ c.mu.Lock()
+ if err == nil {
+ // Only a fully successful sweep is cached; a failed one
+ // leaves the previous cache exactly as it was.
+ c.at = time.Now()
+ c.boxes = boxes
+ c.alerts = alerts
+ }
+ c.sweeping = false
+ close(done)
+ c.mu.Unlock()
- if !s.metricsCache.at.IsZero() && time.Since(s.metricsCache.at) < metricsCacheTTL {
- return s.metricsCache.boxes, s.metricsCache.alerts, nil
+ if err != nil {
+ return nil, nil, err
+ }
+ return boxes, alerts, nil
}
+}
+
+// sweepAll is the bucket half of LoadMetrics: discovery, every box's 48h raw
+// sweep, then the alerts pass, on a context detached from the caller's
+// (WithoutCancel) and bounded by metricsSweepTimeout. It holds no lock — see
+// LoadMetrics's doc comment for why that is the point.
+func (s *Server) sweepAll(ctx context.Context) ([]BoxMetricsView, []AlertRow, error) {
+ sweepCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), metricsSweepTimeout)
+ defer cancel()
- boxNames, err := s.discoverBoxes(ctx)
+ boxNames, err := s.discoverBoxes(sweepCtx)
if err != nil {
return nil, nil, err
}
views := make([]BoxMetricsView, 0, len(boxNames))
for _, box := range boxNames {
- view, err := s.sweepBox(ctx, box)
+ view, err := s.sweepBox(sweepCtx, box)
if err != nil {
return nil, nil, err
}
views = append(views, view)
}
- alerts, err := s.sweepAlerts(ctx)
+ alerts, err := s.sweepAlerts(sweepCtx)
if err != nil {
return nil, nil, err
}
-
- // Only a fully successful sweep is cached — a partial or errored one
- // falls through to the returns above without ever reaching this line.
- s.metricsCache.at = time.Now()
- s.metricsCache.boxes = views
- s.metricsCache.alerts = alerts
return views, alerts, nil
}
diff --git a/internal/console/metricsview_test.go b/internal/console/metricsview_test.go
index 11fc60613..d54d4213c 100644
--- a/internal/console/metricsview_test.go
+++ b/internal/console/metricsview_test.go
@@ -6,6 +6,7 @@ import (
"path/filepath"
"strconv"
"strings"
+ "sync"
"testing"
"time"
@@ -186,6 +187,121 @@ func TestLoadMetricsEmptyBucket(t *testing.T) {
}
}
+// ctxStore is a release.Store that honours its context: every read whose
+// context is already done fails with that context's error. DirStore ignores
+// its ctx entirely (see internal/release/store.go), so a cancelled-caller
+// test built on a bare DirStore would pass vacuously no matter what
+// LoadMetrics does with the context it is handed.
+type ctxStore struct{ release.Store }
+
+func (c ctxStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ if err := ctx.Err(); err != nil {
+ return nil, "", err
+ }
+ return c.Store.Get(ctx, key)
+}
+
+func (c ctxStore) List(ctx context.Context, prefix string) ([]string, error) {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ return c.Store.List(ctx, prefix)
+}
+
+// newCtxMetricsFixture is newMetricsFixture with the server reading through a
+// context-honouring store. The returned release.Store is the raw one, for
+// seeding.
+func newCtxMetricsFixture(t *testing.T) (*Server, release.Store) {
+ t.Helper()
+ st, err := Open(filepath.Join(t.TempDir(), "console.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { st.Close() })
+ rel := release.NewDirStore(filepath.Join(t.TempDir(), "bucket"))
+ srv, err := NewServer(Config{Releases: ctxStore{rel}}, st)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return srv, rel
+}
+
+// seedOneBox writes a single box's chunk with one sample, the minimum a sweep
+// needs to produce a non-empty result.
+func seedOneBox(t *testing.T, rel release.Store, box string) {
+ t.Helper()
+ at := time.Now().UTC().Add(-time.Minute)
+ if err := metrics.WriteRaw(context.Background(), rel, metrics.RawChunk{
+ V: 1, Box: box, Hour: metrics.HourOf(at),
+ Samples: []metrics.Sample{{At: at, Host: metrics.HostSample{CPUFrac: p(0.5)}}},
+ }); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// TestLoadMetricsSweepSurvivesCancelledCaller pins the review fix: the shared
+// sweep runs on its own context (WithoutCancel + its own timeout), never the
+// request's. Otherwise the first caller to navigate away mid-sweep aborts the
+// sweep every other caller is waiting on, and — since a failed sweep is never
+// cached — the next request starts over, so a page reloaded impatiently can
+// starve the cache indefinitely. A caller whose context is ALREADY cancelled
+// is the extreme of that case and must still get real data.
+func TestLoadMetricsSweepSurvivesCancelledCaller(t *testing.T) {
+ srv, rel := newCtxMetricsFixture(t)
+ seedOneBox(t, rel, "box-a")
+
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+
+ boxes, _, err := srv.LoadMetrics(ctx)
+ if err != nil {
+ t.Fatalf("LoadMetrics with a cancelled caller context: %v — the sweep must not inherit the request's cancellation", err)
+ }
+ if len(boxes) != 1 || boxes[0].Box != "box-a" || boxes[0].CPU == nil || *boxes[0].CPU != 0.5 {
+ t.Fatalf("boxes = %+v, want box-a at 0.5", boxes)
+ }
+}
+
+// TestLoadMetricsConcurrentColdCallers: several callers arriving on a cold
+// cache at once — every tenant's app page plus the dashboard on a console
+// that just restarted — all get the same correct data, whether they coalesce
+// onto one sweep or duplicate it. Run under -race, this is also the guard on
+// the cache fields being written while the lock is released across the IO.
+func TestLoadMetricsConcurrentColdCallers(t *testing.T) {
+ srv, rel := newMetricsFixture(t)
+ seedOneBox(t, rel, "box-a")
+
+ const callers = 4
+ type result struct {
+ boxes []BoxMetricsView
+ err error
+ }
+ results := make([]result, callers)
+ var start sync.WaitGroup
+ var done sync.WaitGroup
+ start.Add(1)
+ for i := 0; i < callers; i++ {
+ done.Add(1)
+ go func(i int) {
+ defer done.Done()
+ start.Wait()
+ boxes, _, err := srv.LoadMetrics(context.Background())
+ results[i] = result{boxes: boxes, err: err}
+ }(i)
+ }
+ start.Done()
+ done.Wait()
+
+ for i, r := range results {
+ if r.err != nil {
+ t.Fatalf("caller %d: %v", i, r.err)
+ }
+ if len(r.boxes) != 1 || r.boxes[0].Box != "box-a" || r.boxes[0].CPU == nil || *r.boxes[0].CPU != 0.5 {
+ t.Fatalf("caller %d got %+v, want box-a at 0.5", i, r.boxes)
+ }
+ }
+}
+
// TestOwnerAppMetricsEmptyHosts: an app with no routes asks for nothing and
// gets nothing — empty, no error, and (proven by the poisoned-free fixture
// having no boxes at all) never an error path the app page has to handle.