From cab1d7ade0f23cca78883b076e1e6358ea986810 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 20:41:59 +1000 Subject: [PATCH 1/7] feat: cache merged /facts and /nodes in memory, stale on backend failure A busy Puppetboard re-fans-out the same /facts query every few seconds, and a 502 is worse than 30-second-old facts when every PuppetDB is unreachable. - Add a `Cache` interface (get reports fresh/stale/miss, put, stats) keyed on `?` with keys and repeated values sorted, plus a no-op default so uncached paths behave exactly as before. - Route serveMerged/serveUnion/serveSummed through `serveCached`, so the reports cache drops in at `cacheFor` without touching a handler. - Back /facts and /nodes with a byte-bounded LRU: `facts_ttl` (default 30s, clamped to a 30s cap) and `facts_cache_bytes` (default 64 MiB); expired entries are kept and served only when every backend fails. - Single-flight identical keys so N concurrent requests cause one fan-out. - Surface `cache` state and `serving_stale` in /healthz and the cache settings in `config show`. --- README.md | 52 +++- cache.go | 248 +++++++++++++++++++ cache_test.go | 657 +++++++++++++++++++++++++++++++++++++++++++++++++ config.go | 51 +++- config_test.go | 2 +- main.go | 14 ++ server.go | 252 ++++++++++++++++--- 7 files changed, 1230 insertions(+), 46 deletions(-) create mode 100644 cache.go create mode 100644 cache_test.go diff --git a/README.md b/README.md index 07fdba0..f080019 100644 --- a/README.md +++ b/README.md @@ -38,7 +38,7 @@ not PQL) is forwarded verbatim. | `GET /metrics/v2/read/` | Fan out to all and merge the Jolokia response; numeric attributes are **summed** by default (see merge semantics). | | `GET /metrics/v2/list` | Fan out to all and serve the **union** of the backends' MBean trees. | | `GET /metrics/v1/mbeans[/]` | Same merge, applied to the legacy envelope-less body. | -| `GET /healthz` | Per-backend reachability. `200 {"status":"ok"}` if all reachable, `200 degraded` if some fail, `503 down` if all fail. | +| `GET /healthz` | Per-backend reachability plus cache state. `200 {"status":"ok"}` if all reachable, `200 degraded` if some fail, `503 down` if all fail. | Fan-out is concurrent. If one backend errors or times out, `pdbmux` serves the surviving backends' results and logs a warning; a merged endpoint only returns `502` when @@ -208,6 +208,41 @@ Each backend applies `order_by`/`limit`/`offset` to its own slice only, so - A malformed `limit`, `offset` or `order_by` gets a `400` rather than being forwarded. +## Caching + +`pdbmux` caches merged `/nodes` and `/facts` record sets **in memory** so a busy +Puppetboard does not re-fan-out the same query every few seconds. Everything else +runs uncached — including `extract`/`count` aggregates on those two paths, and +the `/pdb/meta/v1/*` and `/metrics/*` endpoints, which are served live on every +request. The cache is an interface, and `/reports` gets its own (S3-backed) +backend later without further handler changes. + +- **Key** — `?`, where the params are the ones that actually + determine the response, URL-encoded with keys sorted ascending and a repeated + param's values sorted ascending. Param order in the request is therefore + irrelevant: one canonical key per distinct request. A request with no params + keys on the bare path. +- **TTL** — `facts_ttl`, default `30s`, **hard cap `30s`**. A larger configured + value is **clamped** down to the cap, not rejected, so a stray env var cannot + crash-loop a container; `pdbmux config show` prints + `facts_ttl : 30s (clamped from 600s, cap 30s)` when that happens. `facts_ttl: 0` + disables the cache entirely and the merged endpoints behave exactly as before. +- **Stale on failure only** — an expired entry is kept, not dropped. When the TTL + has passed `pdbmux` always re-queries the backends; the expired copy is served + **only** if every backend fails, which turns a `502` into slightly-old data. A + healthy backend is never shadowed by a stale entry. +- **Bounded** — `facts_cache_bytes` (default 64 MiB) is a byte budget, evicted + least-recently-used; reads count as use, so a stale entry that is still being + asked for survives. A single response larger than the whole budget is not + cached at all. +- **Single-flight** — concurrent requests for the same key collapse into one + upstream fan-out; the rest wait for it and share the result. +- **Visibility** — `/healthz` carries a `cache` object: `backend` + (`memory`/`none`), `ttl`, `entries`, `stale_entries`, `bytes`, `serving_stale`, + `stale_served` and `last_stale_served`. `serving_stale` is `true` from the + moment a stale fallback is served until the next response comes from a live + fan-out or a fresh entry. + ## Config Precedence (lowest → highest): **defaults < config file < env vars (`PDBMUX_*`) < flags**. @@ -229,9 +264,11 @@ backends: # order is a tie-break only, not a ranking url: http://puppetdb1.example.com:8080 - name: pdb-b url: https://puppetdb2.example.com -merge: freshness # freshness | static -timeout: 10s # per-upstream request timeout -freshness_ttl: 30s # freshness-map cache TTL (freshness merge only) +merge: freshness # freshness | static +timeout: 10s # per-upstream request timeout +freshness_ttl: 30s # freshness-map cache TTL (freshness merge only) +facts_ttl: 30s # /facts + /nodes response cache TTL; 0 disables, capped at 30s +facts_cache_bytes: 67108864 # byte budget for that cache (64 MiB), LRU-evicted source_fact: pdbmux_source # name of the synthetic provenance fact source_fact_enabled: true # false serves backends' records untouched ``` @@ -246,6 +283,8 @@ the `/pdb/query/v4/...` path per request. | `PDBMUX_MERGE` | `merge` | | `PDBMUX_TIMEOUT` | `timeout` (Go duration, e.g. `10s`) | | `PDBMUX_FRESHNESS_TTL` | `freshness_ttl` | +| `PDBMUX_FACTS_TTL` | `facts_ttl` (clamped to 30s) | +| `PDBMUX_FACTS_CACHE_BYTES` | `facts_cache_bytes` (plain integer bytes) | | `PDBMUX_BACKENDS` | whole backend list, as `name=url,name=url` | | `PDBMUX_SOURCE_FACT` | `source_fact` (default `pdbmux_source`) | | `PDBMUX_SOURCE_FACT_ENABLED` | `source_fact_enabled` (default `true`); `false` disables injection | @@ -281,5 +320,6 @@ A static (`CGO_ENABLED=0`) binary on a distroless base. Configure it with `PDBMUX_*` env vars (at minimum `PDBMUX_BACKENDS`), or mount a config file — a configmap at `/etc/pdbmux/config.yaml` is picked up with no env var at all, and any other mount path works via `PDBMUX_CONFIG`. Env vars still override file -values, so the two mix. Stateless, so run as many replicas as you like; use -`/healthz` for liveness/readiness probes. +values, so the two mix. Run as many replicas as you like — the only state is the +in-memory cache, which is per-replica and bounded by `facts_cache_bytes`, so size +the memory limit above it. Use `/healthz` for liveness/readiness probes. diff --git a/cache.go b/cache.go new file mode 100644 index 0000000..962dcc4 --- /dev/null +++ b/cache.go @@ -0,0 +1,248 @@ +package main + +import ( + "container/list" + "context" + "net/url" + "sort" + "sync" + "time" +) + +// CacheStatus distinguishes the three outcomes of a lookup: nothing stored, a +// stored entry within its TTL, and a stored entry past it. +type CacheStatus int + +const ( + CacheMiss CacheStatus = iota + CacheFresh + CacheStale +) + +func (s CacheStatus) String() string { + switch s { + case CacheFresh: + return "fresh" + case CacheStale: + return "stale" + default: + return "miss" + } +} + +// CacheEntry is a stored response body and the time it was stored. +type CacheEntry struct { + Body []byte + StoredAt time.Time +} + +// CacheStats is the cache state reported by /healthz. +type CacheStats struct { + Backend string `json:"backend"` + Entries int `json:"entries"` + StaleEntries int `json:"stale_entries"` + Bytes int64 `json:"bytes"` +} + +// Cache stores merged responses keyed by cacheKey. Get reports freshness rather +// than hiding expired entries so a caller can fall back to a stale body when the +// upstream fetch fails. The context and error exist for out-of-process backends +// (the reports cache lands on S3); an in-process backend ignores both. +// +// A Body handed back by Get aliases the cache's copy and must not be mutated. +type Cache interface { + Get(ctx context.Context, key string) (CacheEntry, CacheStatus, error) + Put(ctx context.Context, key string, body []byte) error + Stats() CacheStats +} + +// cacheKey is the request path when there are no params, else "?" +// where params is url.Values.Encode() over a copy whose repeated values have +// been sorted. Encode() already sorts keys ascending, so both the order params +// arrive in and the order of a repeated param's values are irrelevant to the +// key: one canonical string per distinct request. +func cacheKey(path string, params url.Values) string { + if len(params) == 0 { + return path + } + norm := make(url.Values, len(params)) + for k, vs := range params { + sorted := append([]string(nil), vs...) + sort.Strings(sorted) + norm[k] = sorted + } + encoded := norm.Encode() + if encoded == "" { + return path + } + return path + "?" + encoded +} + +// noopCache is the default for every path with no cache configured, so wiring a +// handler through the cache leaves its behaviour unchanged. +type noopCache struct{} + +func (noopCache) Get(context.Context, string) (CacheEntry, CacheStatus, error) { + return CacheEntry{}, CacheMiss, nil +} + +func (noopCache) Put(context.Context, string, []byte) error { return nil } + +func (noopCache) Stats() CacheStats { return CacheStats{Backend: "none"} } + +type memoryEntry struct { + key string + body []byte + storedAt time.Time +} + +// memoryCache is a byte-bounded LRU. Expired entries are kept, not dropped, so +// they remain available as a stale fallback; only the byte budget evicts. +type memoryCache struct { + ttl time.Duration + maxBytes int64 + now func() time.Time + + mu sync.Mutex + ll *list.List // front = most recently used + items map[string]*list.Element + bytes int64 +} + +func newMemoryCache(ttl time.Duration, maxBytes int64) *memoryCache { + return &memoryCache{ + ttl: ttl, + maxBytes: maxBytes, + now: time.Now, + ll: list.New(), + items: make(map[string]*list.Element), + } +} + +func (c *memoryCache) Get(_ context.Context, key string) (CacheEntry, CacheStatus, error) { + c.mu.Lock() + defer c.mu.Unlock() + el, ok := c.items[key] + if !ok { + return CacheEntry{}, CacheMiss, nil + } + c.ll.MoveToFront(el) + e := el.Value.(*memoryEntry) + status := CacheFresh + if c.now().Sub(e.storedAt) >= c.ttl { + status = CacheStale + } + return CacheEntry{Body: e.body, StoredAt: e.storedAt}, status, nil +} + +func (c *memoryCache) Put(_ context.Context, key string, body []byte) error { + // A response larger than the whole budget would evict everything else. + if int64(len(body)) > c.maxBytes { + return nil + } + stored := append([]byte(nil), body...) + + c.mu.Lock() + defer c.mu.Unlock() + if el, ok := c.items[key]; ok { + e := el.Value.(*memoryEntry) + c.bytes += int64(len(stored)) - int64(len(e.body)) + e.body, e.storedAt = stored, c.now() + c.ll.MoveToFront(el) + } else { + c.items[key] = c.ll.PushFront(&memoryEntry{key: key, body: stored, storedAt: c.now()}) + c.bytes += int64(len(stored)) + } + for c.bytes > c.maxBytes { + back := c.ll.Back() + if back == nil { + break + } + e := c.ll.Remove(back).(*memoryEntry) + delete(c.items, e.key) + c.bytes -= int64(len(e.body)) + } + return nil +} + +func (c *memoryCache) Stats() CacheStats { + c.mu.Lock() + defer c.mu.Unlock() + st := CacheStats{Backend: "memory", Entries: len(c.items), Bytes: c.bytes} + now := c.now() + for el := c.ll.Front(); el != nil; el = el.Next() { + if now.Sub(el.Value.(*memoryEntry).storedAt) >= c.ttl { + st.StaleEntries++ + } + } + return st +} + +// flightGroup collapses concurrent identical builds so N simultaneous requests +// for one key cause one upstream fan-out. +type flightGroup struct { + mu sync.Mutex + calls map[string]*flightCall +} + +type flightCall struct { + wg sync.WaitGroup + resp cachedResponse + err error +} + +// Do returns fn's result and whether this caller shared another's in-flight run. +func (g *flightGroup) Do(key string, fn func() (cachedResponse, error)) (cachedResponse, error, bool) { + g.mu.Lock() + if g.calls == nil { + g.calls = make(map[string]*flightCall) + } + if c, ok := g.calls[key]; ok { + g.mu.Unlock() + c.wg.Wait() + return c.resp, c.err, true + } + c := &flightCall{} + c.wg.Add(1) + g.calls[key] = c + g.mu.Unlock() + + defer func() { + c.wg.Done() + g.mu.Lock() + delete(g.calls, key) + g.mu.Unlock() + }() + + c.resp, c.err = fn() + return c.resp, c.err, false +} + +// staleTracker records stale fallbacks for /healthz. serving flips back to false +// as soon as a response is served from a live fan-out or a fresh entry. +type staleTracker struct { + mu sync.Mutex + serving bool + served uint64 + last time.Time +} + +func (t *staleTracker) markStale(now time.Time) { + t.mu.Lock() + defer t.mu.Unlock() + t.serving = true + t.served++ + t.last = now +} + +func (t *staleTracker) markFresh() { + t.mu.Lock() + defer t.mu.Unlock() + t.serving = false +} + +func (t *staleTracker) snapshot() (serving bool, served uint64, last time.Time) { + t.mu.Lock() + defer t.mu.Unlock() + return t.serving, t.served, t.last +} diff --git a/cache_test.go b/cache_test.go new file mode 100644 index 0000000..c9231a9 --- /dev/null +++ b/cache_test.go @@ -0,0 +1,657 @@ +package main + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "net/url" + "os" + "path/filepath" + "slices" + "strings" + "sync" + "sync/atomic" + "testing" + "time" +) + +// fakeClock drives the cache's TTL without sleeping. +type fakeClock struct { + mu sync.Mutex + t time.Time +} + +func newFakeClock() *fakeClock { return &fakeClock{t: time.Unix(1_800_000_000, 0)} } + +func (c *fakeClock) now() time.Time { + c.mu.Lock() + defer c.mu.Unlock() + return c.t +} + +func (c *fakeClock) advance(d time.Duration) { + c.mu.Lock() + defer c.mu.Unlock() + c.t = c.t.Add(d) +} + +// countingBackend is a PuppetDB stand-in that counts requests per path and can +// be made to fail or block, so cache hits and single-flight are observable. +type countingBackend struct { + srv *httptest.Server + + mu sync.Mutex + hits map[string]int + bodies map[string]string + fail bool + block chan struct{} +} + +func newCountingBackend(t *testing.T, bodies map[string]string) *countingBackend { + t.Helper() + cb := &countingBackend{hits: map[string]int{}, bodies: bodies} + cb.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cb.mu.Lock() + cb.hits[r.URL.Path]++ + fail, block, body := cb.fail, cb.block, cb.bodies[r.URL.Path] + cb.mu.Unlock() + + if block != nil { + <-block + } + if fail { + http.Error(w, "boom", http.StatusInternalServerError) + return + } + if body == "" { + body = "[]" + } + w.Header().Set("Content-Type", "application/json") + _, _ = io.WriteString(w, body) + })) + t.Cleanup(cb.srv.Close) + return cb +} + +func (cb *countingBackend) hitCount(path string) int { + cb.mu.Lock() + defer cb.mu.Unlock() + return cb.hits[path] +} + +func (cb *countingBackend) setFail(v bool) { + cb.mu.Lock() + defer cb.mu.Unlock() + cb.fail = v +} + +func (cb *countingBackend) setBody(path, body string) { + cb.mu.Lock() + defer cb.mu.Unlock() + cb.bodies[path] = body +} + +func (cb *countingBackend) setBlock(ch chan struct{}) { + cb.mu.Lock() + defer cb.mu.Unlock() + cb.block = ch +} + +// newCachedServer builds a server whose facts cache runs on a fake clock. +func newCachedServer(t *testing.T, cfg Config) (*Server, *fakeClock) { + t.Helper() + srv := newTestServer(cfg) + mc, ok := srv.factsCache.(*memoryCache) + if !ok { + t.Fatalf("expected a memory cache, got %T", srv.factsCache) + } + clk := newFakeClock() + mc.now = clk.now + return srv, clk +} + +func cacheTestConfig(aURL, bURL string) Config { + cfg := testConfig(aURL, bURL, mergeStatic) + cfg.FactsTTL = 30 * time.Second + cfg.CacheBytes = defaultCacheSize + return cfg +} + +func TestCacheKey_CanonicalOrdering(t *testing.T) { + a := cacheKey(factsPath, url.Values{ + "query": {`["=","name","role"]`}, + "limit": {"10"}, + "expand": {"z", "a"}, + }) + b := cacheKey(factsPath, url.Values{ + "expand": {"a", "z"}, + "limit": {"10"}, + "query": {`["=","name","role"]`}, + }) + if a != b { + t.Errorf("param order must not change the key:\n %s\n %s", a, b) + } + if !strings.HasPrefix(a, factsPath+"?") { + t.Errorf("key must start with the path, got %q", a) + } + if got := cacheKey(nodesPath, nil); got != nodesPath { + t.Errorf("no params should give the bare path, got %q", got) + } + if cacheKey(factsPath, url.Values{"query": {"x"}}) == cacheKey(factsPath, url.Values{"query": {"y"}}) { + t.Error("different queries must not share a key") + } + if cacheKey(factsPath, url.Values{"query": {"x"}}) == cacheKey(nodesPath, url.Values{"query": {"x"}}) { + t.Error("different paths must not share a key") + } +} + +func TestMemoryCache_FreshThenStale(t *testing.T) { + c := newMemoryCache(30*time.Second, 1<<20) + clk := newFakeClock() + c.now = clk.now + + if _, status, _ := c.Get(context.Background(), "k"); status != CacheMiss { + t.Fatalf("empty cache should miss, got %s", status) + } + if err := c.Put(context.Background(), "k", []byte(`{"body":[]}`)); err != nil { + t.Fatal(err) + } + + ent, status, err := c.Get(context.Background(), "k") + if err != nil || status != CacheFresh || string(ent.Body) != `{"body":[]}` { + t.Fatalf("want a fresh hit, got %s %q (%v)", status, ent.Body, err) + } + + clk.advance(29 * time.Second) + if _, status, _ := c.Get(context.Background(), "k"); status != CacheFresh { + t.Fatalf("inside the TTL should still be fresh, got %s", status) + } + + clk.advance(2 * time.Second) + ent, status, _ = c.Get(context.Background(), "k") + if status != CacheStale { + t.Fatalf("past the TTL should be stale, got %s", status) + } + if string(ent.Body) != `{"body":[]}` { + t.Errorf("a stale entry must still carry its body, got %q", ent.Body) + } + if st := c.Stats(); st.Entries != 1 || st.StaleEntries != 1 { + t.Errorf("stats should report 1 entry, 1 stale, got %+v", st) + } +} + +func TestMemoryCache_EvictsLeastRecentlyUsed(t *testing.T) { + body := []byte("0123456789") // 10 bytes + c := newMemoryCache(time.Minute, 25) + ctx := context.Background() + + for _, k := range []string{"a", "b"} { + if err := c.Put(ctx, k, body); err != nil { + t.Fatal(err) + } + } + // Reading "a" makes "b" the eviction candidate. + if _, status, _ := c.Get(ctx, "a"); status != CacheFresh { + t.Fatalf("a should be cached, got %s", status) + } + if err := c.Put(ctx, "c", body); err != nil { + t.Fatal(err) + } + + if _, status, _ := c.Get(ctx, "b"); status != CacheMiss { + t.Errorf("b was least recently used and should have been evicted, got %s", status) + } + for _, k := range []string{"a", "c"} { + if _, status, _ := c.Get(ctx, k); status != CacheFresh { + t.Errorf("%s should have survived eviction, got %s", k, status) + } + } + if st := c.Stats(); st.Entries != 2 || st.Bytes != 20 { + t.Errorf("stats after eviction = %+v, want 2 entries / 20 bytes", st) + } +} + +func TestMemoryCache_SkipsOversizedEntry(t *testing.T) { + c := newMemoryCache(time.Minute, 8) + if err := c.Put(context.Background(), "big", []byte("123456789")); err != nil { + t.Fatal(err) + } + if _, status, _ := c.Get(context.Background(), "big"); status != CacheMiss { + t.Error("an entry larger than the whole budget must not be stored") + } + if st := c.Stats(); st.Bytes != 0 { + t.Errorf("bytes = %d, want 0", st.Bytes) + } +} + +func TestNoopCache_AlwaysMisses(t *testing.T) { + var c Cache = noopCache{} + if err := c.Put(context.Background(), "k", []byte("x")); err != nil { + t.Fatal(err) + } + if _, status, _ := c.Get(context.Background(), "k"); status != CacheMiss { + t.Errorf("noop cache must always miss, got %s", status) + } + if st := c.Stats(); st.Backend != "none" { + t.Errorf("backend = %q, want none", st.Backend) + } +} + +func TestHandler_FactsCacheFreshHit(t *testing.T) { + body := `[` + fact("h1", "role", "web", "") + `]` + a := newCountingBackend(t, map[string]string{factsPath: body}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + first := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) + second := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) + + if first.Code != http.StatusOK || second.Code != http.StatusOK { + t.Fatalf("statuses %d/%d", first.Code, second.Code) + } + if first.Body.String() != second.Body.String() { + t.Errorf("cache hit changed the body:\n %s\n %s", first.Body.String(), second.Body.String()) + } + if got := a.hitCount(factsPath); got != 1 { + t.Errorf("backend a saw %d requests, want 1 (second served from cache)", got) + } + if got := b.hitCount(factsPath); got != 1 { + t.Errorf("backend b saw %d requests, want 1", got) + } + + // A different query is a different key and must go upstream. + doGet(t, srv.Handler(), factsPath, `["=","name","osfamily"]`) + if got := a.hitCount(factsPath); got != 2 { + t.Errorf("a different query should refetch: %d requests, want 2", got) + } +} + +func TestHandler_FactsCacheExpires(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + doGet(t, srv.Handler(), factsPath, "") + clk.advance(31 * time.Second) + doGet(t, srv.Handler(), factsPath, "") + + if got := a.hitCount(factsPath); got != 2 { + t.Errorf("an expired entry should refetch: %d requests, want 2", got) + } +} + +func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "old", "") + `]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + warm := doGet(t, srv.Handler(), factsPath, "") + if warm.Code != http.StatusOK { + t.Fatalf("warm-up status %d", warm.Code) + } + if !strings.Contains(warm.Body.String(), `"old"`) { + t.Fatalf("warm-up body = %s", warm.Body.String()) + } + + // Backends healthy but the entry expired: fresh data wins, never the stale copy. + clk.advance(31 * time.Second) + a.setBody(factsPath, `[`+fact("h1", "role", "new", "")+`]`) + refetch := doGet(t, srv.Handler(), factsPath, "") + if !strings.Contains(refetch.Body.String(), `"new"`) { + t.Errorf("a healthy backend must not be shadowed by the stale entry: %s", refetch.Body.String()) + } + if serving, _, _ := srv.stale.snapshot(); serving { + t.Error("serving_stale must stay false while backends are healthy") + } + + // Every backend down and the entry expired: the stale copy is served. + clk.advance(31 * time.Second) + a.setFail(true) + b.setFail(true) + stale := doGet(t, srv.Handler(), factsPath, "") + if stale.Code != http.StatusOK { + t.Fatalf("stale fallback status %d: %s", stale.Code, stale.Body.String()) + } + if !strings.Contains(stale.Body.String(), `"new"`) { + t.Errorf("stale body = %s, want the last cached copy", stale.Body.String()) + } + + serving, served, last := srv.stale.snapshot() + if !serving || served != 1 || last.IsZero() { + t.Errorf("stale tracker = %v/%d/%v, want serving=true served=1", serving, served, last) + } +} + +func TestHandler_NoCacheEntryMeansBackendFailureIs502(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + a.setFail(true) + b.setFail(true) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + rec := doGet(t, srv.Handler(), factsPath, "") + if rec.Code != http.StatusBadGateway { + t.Fatalf("status %d, want 502", rec.Code) + } + if !strings.Contains(rec.Body.String(), "all backends failed") { + t.Errorf("body = %q", rec.Body.String()) + } +} + +func TestHandler_SingleFlightCollapsesConcurrentRequests(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + h := srv.Handler() + + release := make(chan struct{}) + a.setBlock(release) + b.setBlock(release) + + const callers = 16 + var wg sync.WaitGroup + codes := make([]int, callers) + for i := range callers { + wg.Add(1) + go func(i int) { + defer wg.Done() + codes[i] = doGet(t, h, factsPath, "").Code + }(i) + } + + // The leader is parked inside the blocked backend, so every caller that + // reaches the handler before the release joins its flight. + waitFor(t, func() bool { return a.hitCount(factsPath) >= 1 }) + time.Sleep(250 * time.Millisecond) + close(release) + wg.Wait() + + for i, code := range codes { + if code != http.StatusOK { + t.Fatalf("caller %d got %d", i, code) + } + } + if got := a.hitCount(factsPath); got != 1 { + t.Errorf("backend a saw %d requests, want 1 for %d concurrent callers", got, callers) + } + if got := b.hitCount(factsPath); got != 1 { + t.Errorf("backend b saw %d requests, want 1 for %d concurrent callers", got, callers) + } +} + +func TestFlightGroup_LeaderRunsOnce(t *testing.T) { + var g flightGroup + var calls, shared atomic.Int64 + entered := make(chan struct{}) + release := make(chan struct{}) + + run := func() (cachedResponse, error) { + if calls.Add(1) == 1 { + close(entered) + } + <-release + return cachedResponse{Body: json.RawMessage(`[]`), Records: -1}, nil + } + + var wg sync.WaitGroup + for range 16 { + wg.Add(1) + go func() { + defer wg.Done() + if _, _, s := g.Do("k", run); s { + shared.Add(1) + } + }() + } + <-entered + time.Sleep(250 * time.Millisecond) + close(release) + wg.Wait() + + if calls.Load() != 1 { + t.Errorf("fn ran %d times, want 1", calls.Load()) + } + if shared.Load() != 15 { + t.Errorf("%d callers shared the flight, want 15", shared.Load()) + } + // The key is released once the flight finishes. + if _, _, s := g.Do("k", func() (cachedResponse, error) { return cachedResponse{Records: -1}, nil }); s { + t.Error("a later call must start its own flight") + } +} + +func TestHandler_HealthzReportsCacheState(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + hr := health(t, srv) + if hr.Cache.Backend != "memory" || hr.Cache.TTL != "30s" { + t.Fatalf("cache health = %+v, want memory/30s", hr.Cache) + } + if hr.Cache.Entries != 0 || hr.Cache.ServingStale { + t.Fatalf("a cold cache should be empty and not stale: %+v", hr.Cache) + } + + doGet(t, srv.Handler(), factsPath, "") + hr = health(t, srv) + if hr.Cache.Entries != 1 || hr.Cache.StaleEntries != 0 || hr.Cache.Bytes == 0 { + t.Fatalf("after one request: %+v", hr.Cache) + } + + clk.advance(31 * time.Second) + if hr = health(t, srv); hr.Cache.StaleEntries != 1 { + t.Fatalf("expired entry should count as stale: %+v", hr.Cache) + } + if hr.Cache.ServingStale { + t.Error("an expired entry alone is not serving_stale") + } + + a.setFail(true) + b.setFail(true) + doGet(t, srv.Handler(), factsPath, "") + hr = health(t, srv) + if !hr.Cache.ServingStale || hr.Cache.StaleServed != 1 || hr.Cache.LastStale == "" { + t.Errorf("staleness not surfaced in /healthz: %+v", hr.Cache) + } + if hr.Status != "down" { + t.Errorf("status = %q, want down", hr.Status) + } +} + +func TestHandler_HealthzReportsDisabledCache(t *testing.T) { + a := newCountingBackend(t, nil) + b := newCountingBackend(t, nil) + srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)) // FactsTTL zero + + hr := health(t, srv) + if hr.Cache.Backend != "none" || hr.Cache.TTL != "0" { + t.Errorf("disabled cache health = %+v, want none/0", hr.Cache) + } +} + +func TestHandler_UncachedPathsStillRefetch(t *testing.T) { + a := newCountingBackend(t, map[string]string{reportsPath: `[{"hash":"h"}]`}) + b := newCountingBackend(t, map[string]string{reportsPath: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + doGet(t, srv.Handler(), reportsPath, "") + doGet(t, srv.Handler(), reportsPath, "") + if got := a.hitCount(reportsPath); got != 2 { + t.Errorf("/reports is uncached: %d requests, want 2", got) + } +} + +func TestHandler_NodesRecordSetCached(t *testing.T) { + a := newCountingBackend(t, map[string]string{nodesPath: `[{"certname":"h1"}]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + doGet(t, srv.Handler(), nodesPath, "") + doGet(t, srv.Handler(), nodesPath, "") + if got := a.hitCount(nodesPath); got != 1 { + t.Errorf("merged /nodes is cached: %d requests, want 1", got) + } +} + +func TestHandler_NodesAggregateNotCached(t *testing.T) { + a := newCountingBackend(t, map[string]string{nodesPath: `[{"count":90}]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[{"count":53}]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + first := doGet(t, srv.Handler(), nodesPath, nodeCountQuery) + second := doGet(t, srv.Handler(), nodesPath, nodeCountQuery) + if first.Code != http.StatusOK || second.Code != http.StatusOK { + t.Fatalf("statuses %d/%d", first.Code, second.Code) + } + if got := a.hitCount(nodesPath); got != 2 { + t.Errorf("/nodes aggregates are uncached: %d requests, want 2", got) + } + if got := counts(t, second.Body.Bytes(), "count"); !slices.Equal(got, []float64{143}) { + t.Errorf("count = %v, want [143]", got) + } +} + +func TestHandler_ResourcesAggregateNotCached(t *testing.T) { + a := newCountingBackend(t, map[string]string{resourcesPath: `[{"count":7}]`}) + b := newCountingBackend(t, map[string]string{resourcesPath: `[{"count":5}]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + doGet(t, srv.Handler(), resourcesPath, nodeCountQuery) + doGet(t, srv.Handler(), resourcesPath, nodeCountQuery) + if got := a.hitCount(resourcesPath); got != 2 { + t.Errorf("/resources aggregates are uncached: %d requests, want 2", got) + } +} + +func TestHandler_MetaAndMetricsNotCached(t *testing.T) { + bodies := map[string]string{ + metaVersionPath: `{"version":"8.4.0"}`, + metaServerTimePath: `{"server_time":"2026-01-01T00:00:00.000Z"}`, + "/metrics/v2/read/x": `{"request":{},"value":{"Value":1},"status":200}`, + } + a := newCountingBackend(t, bodies) + b := newCountingBackend(t, bodies) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + for path := range bodies { + doGet(t, srv.Handler(), path, "") + doGet(t, srv.Handler(), path, "") + if got := a.hitCount(path); got != 2 { + t.Errorf("%s is uncached: %d requests, want 2", path, got) + } + } +} + +func TestConfig_FactsTTLClampedAtCap(t *testing.T) { + dir := t.TempDir() + t.Setenv("XDG_CONFIG_HOME", dir) + clearEnv(t) + + path := filepath.Join(dir, appName, configFileName) + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(path, []byte("facts_ttl: 5m\n"), 0o644); err != nil { + t.Fatal(err) + } + + cfg, err := Load("") + if err != nil { + t.Fatal(err) + } + if cfg.FactsTTL != maxFactsTTL { + t.Errorf("file facts_ttl = %s, want it clamped to %s", cfg.FactsTTL, maxFactsTTL) + } + + t.Setenv(envPrefix+"FACTS_TTL", "10m") + cfg, err = Load("") + if err != nil { + t.Fatal(err) + } + if cfg.FactsTTL != maxFactsTTL { + t.Errorf("env facts_ttl = %s, want it clamped to %s", cfg.FactsTTL, maxFactsTTL) + } + if cfg.factsTTLClamped != 10*time.Minute { + t.Errorf("pre-clamp value = %s, want 10m0s", cfg.factsTTLClamped) + } + + out := captureStdout(t, func() { printConfig(cfg) }) + if !strings.Contains(out, "facts_ttl : 30s (clamped from 600s, cap 30s)") { + t.Errorf("config show must report the clamp, got:\n%s", out) + } + + t.Setenv(envPrefix+"FACTS_TTL", "5s") + cfg, err = Load("") + if err != nil { + t.Fatal(err) + } + if cfg.FactsTTL != 5*time.Second || cfg.factsTTLClamped != 0 { + t.Errorf("a value under the cap must pass through: %s / %s", cfg.FactsTTL, cfg.factsTTLClamped) + } + + t.Setenv(envPrefix+"FACTS_TTL", "0s") + cfg, err = Load("") + if err != nil { + t.Fatal(err) + } + if cfg.cacheEnabled() { + t.Error("facts_ttl 0 must disable the cache") + } + out = captureStdout(t, func() { printConfig(cfg) }) + if !strings.Contains(out, "(cache disabled)") { + t.Errorf("config show must say the cache is off, got:\n%s", out) + } +} + +func TestNewServer_ClampsFactsTTL(t *testing.T) { + cfg := testConfig("http://a.invalid", "http://b.invalid", mergeStatic) + cfg.FactsTTL = time.Hour + cfg.CacheBytes = 1 << 20 + srv := newTestServer(cfg) + + mc, ok := srv.factsCache.(*memoryCache) + if !ok { + t.Fatalf("expected a memory cache, got %T", srv.factsCache) + } + if mc.ttl != maxFactsTTL { + t.Errorf("cache ttl = %s, want %s", mc.ttl, maxFactsTTL) + } +} + +func TestConfig_ValidateRejectsNegativeCacheSettings(t *testing.T) { + base := testConfig("http://a.invalid", "http://b.invalid", mergeStatic) + for name, mutate := range map[string]func(*Config){ + "negative facts_ttl": func(c *Config) { c.FactsTTL = -time.Second }, + "negative facts_cache_bytes": func(c *Config) { c.CacheBytes = -1 }, + } { + cfg := base + mutate(&cfg) + if err := cfg.Validate(); err == nil { + t.Errorf("%s should not validate", name) + } + } +} + +func health(t *testing.T, srv *Server) healthReport { + t.Helper() + rec := doGet(t, srv.Handler(), "/healthz", "") + var hr healthReport + if err := json.Unmarshal(rec.Body.Bytes(), &hr); err != nil { + t.Fatalf("decode healthz: %v (%s)", err, rec.Body.String()) + } + return hr +} + +func waitFor(t *testing.T, cond func() bool) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for !cond() { + if time.Now().After(deadline) { + t.Fatal("condition not met within 5s") + } + time.Sleep(time.Millisecond) + } +} diff --git a/config.go b/config.go index 6b5cddc..ae12542 100644 --- a/config.go +++ b/config.go @@ -27,6 +27,13 @@ const ( defaultFreshnessTTL = 30 * time.Second defaultSourceFact = "pdbmux_source" + + // maxFactsTTL is a hard cap, not a default: a larger configured value is + // clamped down to it rather than rejected, so a stray env var cannot make a + // container crash-loop. + maxFactsTTL = 30 * time.Second + defaultFactsTTL = 30 * time.Second + defaultCacheSize = int64(64 << 20) ) var exampleBackends = []Backend{ @@ -45,11 +52,14 @@ type Config struct { Merge string `yaml:"merge"` Timeout time.Duration `yaml:"timeout"` FreshnessTTL time.Duration `yaml:"freshness_ttl"` + FactsTTL time.Duration `yaml:"facts_ttl"` // 0 disables the /facts+/nodes cache + CacheBytes int64 `yaml:"facts_cache_bytes"` // byte budget for that cache SourceFact string `yaml:"source_fact"` SourceFactEnabled bool `yaml:"source_fact_enabled"` - sourcePath string // file this config was read from, empty if none was found + sourcePath string // file this config was read from, empty if none was found + factsTTLClamped time.Duration // pre-clamp facts_ttl, zero when nothing was clamped } // SourcePath returns the config file Load read, or "" when none was loaded. @@ -66,6 +76,8 @@ func DefaultConfig() Config { Merge: mergeFreshness, Timeout: defaultTimeout, FreshnessTTL: defaultFreshnessTTL, + FactsTTL: defaultFactsTTL, + CacheBytes: defaultCacheSize, SourceFact: defaultSourceFact, SourceFactEnabled: true, } @@ -146,9 +158,19 @@ func Load(flagPath string) (Config, error) { } applyEnv(&cfg, os.Getenv) + cfg.clampFactsTTL() return cfg, nil } +// clampFactsTTL pins facts_ttl to maxFactsTTL, remembering the configured value +// so `config show` can say the cap was applied. +func (c *Config) clampFactsTTL() { + if c.FactsTTL > maxFactsTTL { + c.factsTTLClamped = c.FactsTTL + c.FactsTTL = maxFactsTTL + } +} + func applyEnv(cfg *Config, getenv func(string) string) { if v := getenv(envPrefix + "LISTEN"); v != "" { cfg.Listen = v @@ -174,6 +196,16 @@ func applyEnv(cfg *Config, getenv func(string) string) { cfg.SourceFactEnabled = b } } + if v := getenv(envPrefix + "FACTS_TTL"); v != "" { + if d, err := time.ParseDuration(v); err == nil { + cfg.FactsTTL = d + } + } + if v := getenv(envPrefix + "FACTS_CACHE_BYTES"); v != "" { + if n, err := strconv.ParseInt(v, 10, 64); err == nil { + cfg.CacheBytes = n + } + } if v := getenv(envPrefix + "BACKENDS"); v != "" { if bs := parseBackends(v); len(bs) > 0 { cfg.Backends = bs @@ -234,9 +266,19 @@ func (c Config) Validate() error { if c.SourceFactEnabled && c.SourceFact == "" { return fmt.Errorf("source_fact must be non-empty, or set source_fact_enabled to false") } + if c.FactsTTL < 0 { + return fmt.Errorf("facts_ttl must not be negative (0 disables the cache)") + } + if c.CacheBytes < 0 { + return fmt.Errorf("facts_cache_bytes must not be negative") + } return nil } +// cacheEnabled reports whether a facts/nodes cache should be built: both a TTL +// and a byte budget are required. +func (c Config) cacheEnabled() bool { return c.FactsTTL > 0 && c.CacheBytes > 0 } + func writeDefaultConfig(path string) error { if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { return fmt.Errorf("creating config dir: %w", err) @@ -249,8 +291,11 @@ func writeDefaultConfig(path string) error { "# A merging proxy presenting one PuppetDB v4 query surface over several\n" + "# PuppetDB backends. The backend URLs below are placeholders — edit them.\n" + "# Env overrides: PDBMUX_LISTEN, PDBMUX_MERGE, PDBMUX_TIMEOUT,\n" + - "# PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url),\n" + - "# PDBMUX_SOURCE_FACT, PDBMUX_SOURCE_FACT_ENABLED.\n\n") + "# PDBMUX_FRESHNESS_TTL, PDBMUX_FACTS_TTL, PDBMUX_FACTS_CACHE_BYTES,\n" + + "# PDBMUX_BACKENDS (name=url,name=url),\n" + + "# PDBMUX_SOURCE_FACT, PDBMUX_SOURCE_FACT_ENABLED.\n" + + "# facts_ttl caches merged /facts and /nodes in memory; it is capped at 30s\n" + + "# (a larger value is clamped) and 0 disables the cache.\n\n") if err := os.WriteFile(path, append(header, data...), 0o644); err != nil { return fmt.Errorf("writing config: %w", err) } diff --git a/config_test.go b/config_test.go index bfdbf36..ef89ea4 100644 --- a/config_test.go +++ b/config_test.go @@ -456,7 +456,7 @@ func captureStdout(t *testing.T, f func()) string { func clearEnv(t *testing.T) { t.Helper() - for _, k := range []string{"CONFIG", "LISTEN", "MERGE", "TIMEOUT", "FRESHNESS_TTL", "BACKENDS"} { + for _, k := range []string{"CONFIG", "LISTEN", "MERGE", "TIMEOUT", "FRESHNESS_TTL", "FACTS_TTL", "FACTS_CACHE_BYTES", "BACKENDS"} { t.Setenv(envPrefix+k, "") } } diff --git a/main.go b/main.go index de1cd43..ebe0521 100644 --- a/main.go +++ b/main.go @@ -151,6 +151,18 @@ func runServer(cfg Config) error { } } +func factsTTLString(cfg Config) string { + s := durationString(cfg.FactsTTL) + switch { + case cfg.factsTTLClamped > 0: + return fmt.Sprintf("%s (clamped from %s, cap %s)", + s, durationString(cfg.factsTTLClamped), durationString(maxFactsTTL)) + case !cfg.cacheEnabled(): + return s + " (cache disabled)" + } + return s +} + func printConfig(cfg Config) { if p := cfg.SourcePath(); p != "" { fmt.Printf("config file : %s (loaded)\n", p) @@ -166,6 +178,8 @@ func printConfig(cfg Config) { } else { fmt.Printf("source_fact : disabled\n") } + fmt.Printf("facts_ttl : %s\n", factsTTLString(cfg)) + fmt.Printf("facts_cache : %d bytes\n", cfg.CacheBytes) fmt.Println("backends:") for _, b := range cfg.Backends { fmt.Printf(" - %-8s %s\n", b.Name, b.URL) diff --git a/server.go b/server.go index 4e046e6..05f6c40 100644 --- a/server.go +++ b/server.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "fmt" "io" "log" @@ -35,11 +36,18 @@ type backendResult struct { err error } +var errAllBackendsFailed = errors.New("all backends failed") + type Server struct { cfg Config client *http.Client log *log.Logger + // factsCache is nil when caching is disabled; cacheFor hands out a noop then. + factsCache Cache + flights flightGroup + stale staleTracker + // freshness cache (freshness merge only). mu sync.Mutex freshData freshness @@ -47,11 +55,35 @@ type Server struct { } func NewServer(cfg Config, logger *log.Logger) *Server { - return &Server{ + cfg.clampFactsTTL() + s := &Server{ cfg: cfg, client: &http.Client{Timeout: cfg.Timeout}, log: logger, } + if cfg.cacheEnabled() { + s.factsCache = newMemoryCache(cfg.FactsTTL, cfg.CacheBytes) + } + return s +} + +// cacheFor picks the cache backing a request. Merged /facts and /nodes record +// sets share the in-memory cache; every other path is uncached until the reports +// cache lands, and a new backend is a case here rather than a change to any +// handler. +func (s *Server) cacheFor(path string, params url.Values) (Cache, bool) { + switch path { + case factsPath, nodesPath: + // An aggregate row is a summed count, not the merged record set the + // cache was built for, so it stays on the live path. + if parseAggregate(params.Get("query")) != nil { + return noopCache{}, false + } + if s.factsCache != nil { + return s.factsCache, true + } + } + return noopCache{}, false } func (s *Server) Handler() http.Handler { @@ -109,11 +141,14 @@ func isReportSubResource(path string) bool { } func (s *Server) serveMerged(w http.ResponseWriter, r *http.Request, path string, merge func([]backendResult) []json.RawMessage) { - alive, ok := s.aliveResults(w, r, path, queryParams(r.URL.Query().Get("query"))) - if !ok { - return - } - writeJSON(w, merge(alive)) + params := queryParams(r.URL.Query().Get("query")) + s.serveCached(w, r, path, params, func() (cachedResponse, error) { + alive, err := s.aliveResults(r.Context(), path, params) + if err != nil { + return cachedResponse{}, err + } + return cachedResponse{Body: encodeRecords(merge(alive)), Records: -1}, nil + }) } // Reports and events are immutable history, so both backends' records belong in the merged view. @@ -125,19 +160,23 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, return } - alive, ok := s.aliveResults(w, r, path, page.upstreamParams(in)) - if !ok { - return - } - - merged := mergeUnion(alive, key) - sortRecords(merged, page.order) - if page.wantTotal { - if total := sumTotals(alive); total >= 0 { - w.Header().Set(recordsHeader, strconv.Itoa(total)) + // Keyed on the request's own params, not the upstream ones: upstreamParams + // folds offset into limit, so different windows would collide. + s.serveCached(w, r, path, in, func() (cachedResponse, error) { + alive, err := s.aliveResults(r.Context(), path, page.upstreamParams(in)) + if err != nil { + return cachedResponse{}, err } - } - writeJSON(w, page.apply(merged)) + merged := mergeUnion(alive, key) + sortRecords(merged, page.order) + resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1} + if page.wantTotal { + if total := sumTotals(alive); total >= 0 { + resp.Records = total + } + } + return resp, nil + }) } // A count row carries no certname, so the certname-keyed merge would collapse every backend's count into one backend's; aggregates take the summing path instead. @@ -176,17 +215,19 @@ func (s *Server) serveSummed(w http.ResponseWriter, r *http.Request, path string return } - alive, ok := s.aliveResults(w, r, path, page.upstreamParams(in)) - if !ok { - return - } - - merged := sumRows(alive, columns) - sortRecords(merged, page.order) - if page.wantTotal { - w.Header().Set(recordsHeader, strconv.Itoa(len(merged))) - } - writeJSON(w, page.apply(merged)) + s.serveCached(w, r, path, in, func() (cachedResponse, error) { + alive, err := s.aliveResults(r.Context(), path, page.upstreamParams(in)) + if err != nil { + return cachedResponse{}, err + } + merged := sumRows(alive, columns) + sortRecords(merged, page.order) + resp := cachedResponse{Body: encodeRecords(page.apply(merged)), Records: -1} + if page.wantTotal { + resp.Records = len(merged) + } + return resp, nil + }) } // A backend without the report answers 404, indistinguishable from a failure, so every backend is consulted before serving empty. @@ -214,9 +255,9 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) { writeJSON(w, nil) } -// Writes a 502 and returns ok=false only when every backend failed. -func (s *Server) aliveResults(w http.ResponseWriter, r *http.Request, path string, params url.Values) ([]backendResult, bool) { - results := s.fanOut(r.Context(), path, params) +// Returns errAllBackendsFailed only when every backend failed. +func (s *Server) aliveResults(ctx context.Context, path string, params url.Values) ([]backendResult, error) { + results := s.fanOut(ctx, path, params) var alive []backendResult for _, res := range results { @@ -227,10 +268,114 @@ func (s *Server) aliveResults(w http.ResponseWriter, r *http.Request, path strin alive = append(alive, res) } if len(alive) == 0 { - http.Error(w, "all backends failed", http.StatusBadGateway) - return nil, false + return nil, errAllBackendsFailed } - return alive, true + return alive, nil +} + +// cachedResponse is the stored form of a merged response: the JSON body plus the +// X-Records value it carried, so a cache hit reproduces both. +type cachedResponse struct { + Body json.RawMessage `json:"body"` + Records int `json:"records"` // -1 when the response sets no X-Records +} + +// serveCached answers from the cache when the entry is fresh, otherwise runs +// build — single-flighted, so N concurrent identical requests cause one upstream +// fan-out — and stores the result. A build failure falls back to a stale entry +// when one exists; that is the only path on which stale data is served. Paths +// with no cache configured run build directly, unchanged. +func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string, params url.Values, build func() (cachedResponse, error)) { + cache, enabled := s.cacheFor(path, params) + if !enabled { + resp, err := build() + if err != nil { + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + writeCached(w, resp) + return + } + + key := cacheKey(path, params) + var stale *CacheEntry + ent, status, err := cache.Get(r.Context(), key) + switch { + case err != nil: + s.log.Printf("warning: cache lookup for %s failed: %v", key, err) + case status == CacheFresh: + s.stale.markFresh() + s.writeStored(w, ent.Body) + return + case status == CacheStale: + stale = &ent + } + + resp, err, _ := s.flights.Do(key, func() (cachedResponse, error) { + built, buildErr := build() + if buildErr != nil { + return cachedResponse{}, buildErr + } + body, marshalErr := json.Marshal(built) + if marshalErr != nil { + s.log.Printf("warning: encoding cache entry for %s failed: %v", key, marshalErr) + return built, nil + } + // The leader's request may be cancelled while followers still wait. + if putErr := cache.Put(context.WithoutCancel(r.Context()), key, body); putErr != nil { + s.log.Printf("warning: cache store for %s failed: %v", key, putErr) + } + return built, nil + }) + if err != nil { + if stale != nil { + s.stale.markStale(time.Now()) + s.log.Printf("warning: serving stale %s from cache (stored %s): %v", + path, stale.StoredAt.UTC().Format(time.RFC3339), err) + s.writeStored(w, stale.Body) + return + } + http.Error(w, err.Error(), http.StatusBadGateway) + return + } + s.stale.markFresh() + writeCached(w, resp) +} + +func (s *Server) writeStored(w http.ResponseWriter, body []byte) { + var resp cachedResponse + if err := json.Unmarshal(body, &resp); err != nil { + s.log.Printf("warning: unreadable cache entry: %v", err) + http.Error(w, "unreadable cache entry", http.StatusBadGateway) + return + } + writeCached(w, resp) +} + +func writeCached(w http.ResponseWriter, resp cachedResponse) { + w.Header().Set("Content-Type", "application/json") + if resp.Records >= 0 { + w.Header().Set(recordsHeader, strconv.Itoa(resp.Records)) + } + // resp.Body is shared with the cache and with every caller of a single + // flight, so it is written, never appended to. + body := []byte(resp.Body) + if len(body) == 0 { + body = []byte("[]") + } + _, _ = w.Write(body) + _, _ = w.Write([]byte("\n")) +} + +func encodeRecords(recs []json.RawMessage) json.RawMessage { + if recs == nil { + recs = []json.RawMessage{} + } + b, err := json.Marshal(recs) + if err != nil { + return json.RawMessage("[]") + } + return b } func queryParams(query string) url.Values { @@ -406,13 +551,48 @@ func setContentType(w http.ResponseWriter, contentType string) { type healthReport struct { Status string `json:"status"` Backends map[string]string `json:"backends"` // name -> "ok" | error text + Cache cacheHealth `json:"cache"` +} + +type cacheHealth struct { + Backend string `json:"backend"` // "memory" | "none" + TTL string `json:"ttl"` + Entries int `json:"entries"` + StaleEntries int `json:"stale_entries"` // cached entries past their TTL + Bytes int64 `json:"bytes"` + ServingStale bool `json:"serving_stale"` // last cached response came from a stale entry + StaleServed uint64 `json:"stale_served"` + LastStale string `json:"last_stale_served,omitempty"` +} + +func (s *Server) cacheHealth() cacheHealth { + stats := CacheStats{Backend: "none"} + ttl := time.Duration(0) + if s.factsCache != nil { + stats = s.factsCache.Stats() + ttl = s.cfg.FactsTTL + } + serving, served, last := s.stale.snapshot() + h := cacheHealth{ + Backend: stats.Backend, + TTL: durationString(ttl), + Entries: stats.Entries, + StaleEntries: stats.StaleEntries, + Bytes: stats.Bytes, + ServingStale: serving, + StaleServed: served, + } + if !last.IsZero() { + h.LastStale = last.UTC().Format(time.RFC3339) + } + return h } func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { probe := `["=","certname","pdbmux-healthz-probe"]` results := s.fanOut(r.Context(), nodesPath, queryParams(probe)) - report := healthReport{Backends: map[string]string{}} + report := healthReport{Backends: map[string]string{}, Cache: s.cacheHealth()} healthy := 0 for _, res := range results { if res.err != nil { -- 2.47.3 From 0c1fe7f1dd0804a1431bbc839640703bbe3f23ec Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 20:55:09 +1000 Subject: [PATCH 2/7] Convert a single-flight panic into an error - flightGroup.Do recovers a panicking fn so the leader and every waiter get a non-nil error instead of a zero-value success served as 200 [] - Note in the README that facts_cache_bytes budgets body bytes only --- README.md | 4 +- cache.go | 21 ++++++- cache_test.go | 164 ++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 187 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index f080019..751cd3d 100644 --- a/README.md +++ b/README.md @@ -234,7 +234,9 @@ backend later without further handler changes. - **Bounded** — `facts_cache_bytes` (default 64 MiB) is a byte budget, evicted least-recently-used; reads count as use, so a stale entry that is still being asked for survives. A single response larger than the whole budget is not - cached at all. + cached at all. The budget counts stored response bodies only — cache keys and + the list/map bookkeeping are not accounted for, so it is a target for body + bytes rather than a hard cap on process memory. - **Single-flight** — concurrent requests for the same key collapse into one upstream fan-out; the rest wait for it and share the result. - **Visibility** — `/healthz` carries a `cache` object: `backend` diff --git a/cache.go b/cache.go index 962dcc4..97b8811 100644 --- a/cache.go +++ b/cache.go @@ -3,7 +3,9 @@ package main import ( "container/list" "context" + "fmt" "net/url" + "runtime/debug" "sort" "sync" "time" @@ -191,8 +193,20 @@ type flightCall struct { err error } +// flightPanic is a panic from a flight's fn, reported to the leader and to every +// waiter as an error so callers keep their error handling (stale fallback, 502) +// instead of seeing a zero-value success. +type flightPanic struct { + value any + stack []byte +} + +func (p *flightPanic) Error() string { + return fmt.Sprintf("panic building response: %v\n%s", p.value, p.stack) +} + // Do returns fn's result and whether this caller shared another's in-flight run. -func (g *flightGroup) Do(key string, fn func() (cachedResponse, error)) (cachedResponse, error, bool) { +func (g *flightGroup) Do(key string, fn func() (cachedResponse, error)) (resp cachedResponse, err error, shared bool) { g.mu.Lock() if g.calls == nil { g.calls = make(map[string]*flightCall) @@ -208,6 +222,11 @@ func (g *flightGroup) Do(key string, fn func() (cachedResponse, error)) (cachedR g.mu.Unlock() defer func() { + if r := recover(); r != nil { + c.resp, c.err = cachedResponse{}, &flightPanic{value: r, stack: debug.Stack()} + resp, err = c.resp, c.err + } + // Done only after the results are stored, so waiters read them. c.wg.Done() g.mu.Lock() delete(g.calls, key) diff --git a/cache_test.go b/cache_test.go index c9231a9..e52d519 100644 --- a/cache_test.go +++ b/cache_test.go @@ -422,6 +422,170 @@ func TestFlightGroup_LeaderRunsOnce(t *testing.T) { } } +func TestFlightGroup_LeaderPanicFailsLeaderAndWaiters(t *testing.T) { + var g flightGroup + entered := make(chan struct{}) + release := make(chan struct{}) + + type result struct { + resp cachedResponse + err error + shared bool + } + waiters := make([]result, 8) + var wg sync.WaitGroup + + var leader result + wg.Add(1) + go func() { + defer wg.Done() + leader.resp, leader.err, leader.shared = g.Do("k", func() (cachedResponse, error) { + close(entered) + <-release + panic("build exploded") + }) + }() + + <-entered + for i := range waiters { + wg.Add(1) + go func(i int) { + defer wg.Done() + waiters[i].resp, waiters[i].err, waiters[i].shared = g.Do("k", func() (cachedResponse, error) { + t.Error("a waiter must not run its own fn") + return cachedResponse{}, nil + }) + }(i) + } + // Park the waiters in Wait() before the leader panics. + time.Sleep(250 * time.Millisecond) + close(release) + wg.Wait() + + if leader.err == nil { + t.Errorf("leader err = nil, want a panic error (resp %+v)", leader.resp) + } + for i, w := range waiters { + if !w.shared { + t.Errorf("waiter %d did not share the flight", i) + } + if w.err == nil { + t.Fatalf("waiter %d unblocked with err = nil and resp %+v, want an error", i, w.resp) + } + if !strings.Contains(w.err.Error(), "build exploded") { + t.Errorf("waiter %d err = %v, want the panic value", i, w.err) + } + } + + // The key is released on the panic path, so a later call leads its own flight. + resp, err, shared := g.Do("k", func() (cachedResponse, error) { + return cachedResponse{Body: json.RawMessage(`[]`), Records: 0}, nil + }) + if shared { + t.Error("a call after a panicking flight must start its own flight") + } + if err != nil || resp.Records != 0 { + t.Errorf("later call = (%+v, %v), want a clean success", resp, err) + } +} + +func TestServeCached_BuildPanicIs502ThenStale(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + panicBuild := func() (cachedResponse, error) { panic("build exploded") } + serve := func(build func() (cachedResponse, error)) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, factsPath, nil) + srv.serveCached(rec, req, factsPath, nil, build) + return rec + } + + // No entry to fall back to: a panicking build must not look like a success. + rec := serve(panicBuild) + if rec.Code != http.StatusBadGateway { + t.Fatalf("status %d (%s), want 502", rec.Code, rec.Body.String()) + } + + stored := `[` + fact("h1", "role", "web", "") + `]` + rec = serve(func() (cachedResponse, error) { + return cachedResponse{Body: json.RawMessage(stored), Records: 1}, nil + }) + if rec.Code != http.StatusOK { + t.Fatalf("priming status %d (%s)", rec.Code, rec.Body.String()) + } + + clk.advance(31 * time.Second) + rec = serve(panicBuild) + if rec.Code != http.StatusOK { + t.Fatalf("stale fallback status %d (%s), want 200", rec.Code, rec.Body.String()) + } + if got := strings.TrimSpace(rec.Body.String()); got != stored { + t.Errorf("stale body = %s, want %s", got, stored) + } + if got := rec.Header().Get(recordsHeader); got != "1" { + t.Errorf("%s = %q, want 1", recordsHeader, got) + } + if hr := health(t, srv); !hr.Cache.ServingStale || hr.Cache.StaleServed != 1 { + t.Errorf("panic fallback not counted as stale: %+v", hr.Cache) + } +} + +func TestServeCached_BuildPanicFailsConcurrentWaiters(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + + entered := make(chan struct{}) + release := make(chan struct{}) + serve := func(build func() (cachedResponse, error)) *httptest.ResponseRecorder { + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, factsPath, nil) + srv.serveCached(rec, req, factsPath, nil, build) + return rec + } + + var wg sync.WaitGroup + wg.Add(1) + var leaderRec *httptest.ResponseRecorder + go func() { + defer wg.Done() + leaderRec = serve(func() (cachedResponse, error) { + close(entered) + <-release + panic("build exploded") + }) + }() + + <-entered + const waiters = 8 + recs := make([]*httptest.ResponseRecorder, waiters) + for i := range waiters { + wg.Add(1) + go func(i int) { + defer wg.Done() + recs[i] = serve(func() (cachedResponse, error) { + t.Error("a waiter must not run its own build") + return cachedResponse{}, nil + }) + }(i) + } + time.Sleep(250 * time.Millisecond) + close(release) + wg.Wait() + + if leaderRec.Code != http.StatusBadGateway { + t.Errorf("leader status %d, want 502", leaderRec.Code) + } + for i, rec := range recs { + if rec.Code != http.StatusBadGateway { + t.Fatalf("waiter %d got %d (%s), want 502 rather than an empty success", + i, rec.Code, rec.Body.String()) + } + } +} + func TestHandler_HealthzReportsCacheState(t *testing.T) { a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) b := newCountingBackend(t, map[string]string{factsPath: `[]`}) -- 2.47.3 From 743cd9a6abd384078d09c5c5917195fe4257a1c8 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 21:29:20 +1000 Subject: [PATCH 3/7] Detach the shared fan-out from its leader's request context A single flight is built by whichever request arrived first, but every request on that key waits for it. Running the fan-out on the leader's cancelable request context hands the leader's disconnect to followers whose own connections are healthy: they get 502 all backends failed. Waiters also parked on a WaitGroup, so a follower whose own client went away stayed blocked until the leader finished. Run the flight on a context detached from the leader's request and bounded by the configured timeout, and pass that context into build so the fan-out uses it. Give Do a context so a waiter can abandon a flight it no longer needs; the leader ignores it and always runs fn to completion, keeping the cache warm for the others. A caller that abandons on its own cancellation writes no response. --- cache.go | 25 ++++-- cache_test.go | 212 +++++++++++++++++++++++++++++++++++++++++++++++--- server.go | 44 ++++++++--- 3 files changed, 250 insertions(+), 31 deletions(-) diff --git a/cache.go b/cache.go index 97b8811..65307cd 100644 --- a/cache.go +++ b/cache.go @@ -188,7 +188,7 @@ type flightGroup struct { } type flightCall struct { - wg sync.WaitGroup + done chan struct{} resp cachedResponse err error } @@ -206,18 +206,26 @@ func (p *flightPanic) Error() string { } // Do returns fn's result and whether this caller shared another's in-flight run. -func (g *flightGroup) Do(key string, fn func() (cachedResponse, error)) (resp cachedResponse, err error, shared bool) { +// +// ctx belongs to the caller alone. A waiter that gives up returns ctx.Err() and +// leaves the flight running for everyone else; the leader ignores ctx entirely +// and always runs fn to completion, so one participant walking away can neither +// cancel nor fail the others. fn is therefore responsible for its own deadline. +func (g *flightGroup) Do(ctx context.Context, key string, fn func() (cachedResponse, error)) (resp cachedResponse, err error, shared bool) { g.mu.Lock() if g.calls == nil { g.calls = make(map[string]*flightCall) } if c, ok := g.calls[key]; ok { g.mu.Unlock() - c.wg.Wait() - return c.resp, c.err, true + select { + case <-c.done: + return c.resp, c.err, true + case <-ctx.Done(): + return cachedResponse{}, ctx.Err(), true + } } - c := &flightCall{} - c.wg.Add(1) + c := &flightCall{done: make(chan struct{})} g.calls[key] = c g.mu.Unlock() @@ -226,11 +234,12 @@ func (g *flightGroup) Do(key string, fn func() (cachedResponse, error)) (resp ca c.resp, c.err = cachedResponse{}, &flightPanic{value: r, stack: debug.Stack()} resp, err = c.resp, c.err } - // Done only after the results are stored, so waiters read them. - c.wg.Done() + // The key is released before the results are published, so a caller that + // arrives late leads a new flight instead of joining a finished one. g.mu.Lock() delete(g.calls, key) g.mu.Unlock() + close(c.done) }() c.resp, c.err = fn() diff --git a/cache_test.go b/cache_test.go index e52d519..98b0ba0 100644 --- a/cache_test.go +++ b/cache_test.go @@ -3,6 +3,7 @@ package main import ( "context" "encoding/json" + "errors" "io" "net/http" "net/http/httptest" @@ -400,7 +401,7 @@ func TestFlightGroup_LeaderRunsOnce(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - if _, _, s := g.Do("k", run); s { + if _, _, s := g.Do(context.Background(), "k", run); s { shared.Add(1) } }() @@ -417,7 +418,7 @@ func TestFlightGroup_LeaderRunsOnce(t *testing.T) { t.Errorf("%d callers shared the flight, want 15", shared.Load()) } // The key is released once the flight finishes. - if _, _, s := g.Do("k", func() (cachedResponse, error) { return cachedResponse{Records: -1}, nil }); s { + if _, _, s := g.Do(context.Background(), "k", func() (cachedResponse, error) { return cachedResponse{Records: -1}, nil }); s { t.Error("a later call must start its own flight") } } @@ -439,7 +440,7 @@ func TestFlightGroup_LeaderPanicFailsLeaderAndWaiters(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - leader.resp, leader.err, leader.shared = g.Do("k", func() (cachedResponse, error) { + leader.resp, leader.err, leader.shared = g.Do(context.Background(), "k", func() (cachedResponse, error) { close(entered) <-release panic("build exploded") @@ -451,7 +452,7 @@ func TestFlightGroup_LeaderPanicFailsLeaderAndWaiters(t *testing.T) { wg.Add(1) go func(i int) { defer wg.Done() - waiters[i].resp, waiters[i].err, waiters[i].shared = g.Do("k", func() (cachedResponse, error) { + waiters[i].resp, waiters[i].err, waiters[i].shared = g.Do(context.Background(), "k", func() (cachedResponse, error) { t.Error("a waiter must not run its own fn") return cachedResponse{}, nil }) @@ -478,7 +479,7 @@ func TestFlightGroup_LeaderPanicFailsLeaderAndWaiters(t *testing.T) { } // The key is released on the panic path, so a later call leads its own flight. - resp, err, shared := g.Do("k", func() (cachedResponse, error) { + resp, err, shared := g.Do(context.Background(), "k", func() (cachedResponse, error) { return cachedResponse{Body: json.RawMessage(`[]`), Records: 0}, nil }) if shared { @@ -494,8 +495,8 @@ func TestServeCached_BuildPanicIs502ThenStale(t *testing.T) { b := newCountingBackend(t, map[string]string{factsPath: `[]`}) srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) - panicBuild := func() (cachedResponse, error) { panic("build exploded") } - serve := func(build func() (cachedResponse, error)) *httptest.ResponseRecorder { + panicBuild := func(context.Context) (cachedResponse, error) { panic("build exploded") } + serve := func(build func(context.Context) (cachedResponse, error)) *httptest.ResponseRecorder { rec := httptest.NewRecorder() req := httptest.NewRequest(http.MethodGet, factsPath, nil) srv.serveCached(rec, req, factsPath, nil, build) @@ -509,7 +510,7 @@ func TestServeCached_BuildPanicIs502ThenStale(t *testing.T) { } stored := `[` + fact("h1", "role", "web", "") + `]` - rec = serve(func() (cachedResponse, error) { + rec = serve(func(context.Context) (cachedResponse, error) { return cachedResponse{Body: json.RawMessage(stored), Records: 1}, nil }) if rec.Code != http.StatusOK { @@ -539,7 +540,7 @@ func TestServeCached_BuildPanicFailsConcurrentWaiters(t *testing.T) { entered := make(chan struct{}) release := make(chan struct{}) - serve := func(build func() (cachedResponse, error)) *httptest.ResponseRecorder { + serve := func(build func(context.Context) (cachedResponse, error)) *httptest.ResponseRecorder { rec := httptest.NewRecorder() req := httptest.NewRequest(http.MethodGet, factsPath, nil) srv.serveCached(rec, req, factsPath, nil, build) @@ -551,7 +552,7 @@ func TestServeCached_BuildPanicFailsConcurrentWaiters(t *testing.T) { var leaderRec *httptest.ResponseRecorder go func() { defer wg.Done() - leaderRec = serve(func() (cachedResponse, error) { + leaderRec = serve(func(context.Context) (cachedResponse, error) { close(entered) <-release panic("build exploded") @@ -565,7 +566,7 @@ func TestServeCached_BuildPanicFailsConcurrentWaiters(t *testing.T) { wg.Add(1) go func(i int) { defer wg.Done() - recs[i] = serve(func() (cachedResponse, error) { + recs[i] = serve(func(context.Context) (cachedResponse, error) { t.Error("a waiter must not run its own build") return cachedResponse{}, nil }) @@ -586,6 +587,195 @@ func TestServeCached_BuildPanicFailsConcurrentWaiters(t *testing.T) { } } +// The leader's client disconnecting mid-fetch must not fail the followers whose +// own connections are healthy. +func TestHandler_LeaderDisconnectDoesNotFailFollowers(t *testing.T) { + body := `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]` + a := newCountingBackend(t, map[string]string{nodesPath: body}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + h := srv.Handler() + + release := make(chan struct{}) + a.setBlock(release) + b.setBlock(release) + + leaderCtx, cancelLeader := context.WithCancel(context.Background()) + defer cancelLeader() + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + req := httptest.NewRequest(http.MethodGet, nodesPath, nil).WithContext(leaderCtx) + h.ServeHTTP(httptest.NewRecorder(), req) + }() + + // Only once the leader is inside the fan-out does a second caller join its + // flight rather than starting one of its own. + waitFor(t, func() bool { return a.hitCount(nodesPath) >= 1 }) + + follower := httptest.NewRecorder() + wg.Add(1) + go func() { + defer wg.Done() + h.ServeHTTP(follower, httptest.NewRequest(http.MethodGet, nodesPath, nil)) + }() + + time.Sleep(250 * time.Millisecond) + cancelLeader() + time.Sleep(100 * time.Millisecond) + close(release) + wg.Wait() + + if follower.Code != http.StatusOK { + t.Fatalf("follower status %d (%s), want 200: a healthy client must not inherit the leader's cancellation", + follower.Code, follower.Body.String()) + } + if got := strings.TrimSpace(follower.Body.String()); got != body { + t.Errorf("follower body = %s, want %s", got, body) + } + if got := a.hitCount(nodesPath); got != 1 { + t.Errorf("backend a saw %d requests, want 1", got) + } +} + +// A follower whose own client goes away must unpark rather than wait out the +// leader, and must not disturb the flight the others are sharing. +func TestFlightGroup_WaiterAbandonsOnContextCancel(t *testing.T) { + var g flightGroup + entered := make(chan struct{}) + release := make(chan struct{}) + + var leaderResp cachedResponse + var leaderErr error + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + leaderResp, leaderErr, _ = g.Do(context.Background(), "k", func() (cachedResponse, error) { + close(entered) + <-release + return cachedResponse{Body: json.RawMessage(`[1]`), Records: 1}, nil + }) + }() + <-entered + + // A patient waiter proves the flight survives the abandoning one. + var patientResp cachedResponse + var patientErr error + wg.Add(1) + go func() { + defer wg.Done() + patientResp, patientErr, _ = g.Do(context.Background(), "k", func() (cachedResponse, error) { + t.Error("a waiter must not run its own fn") + return cachedResponse{}, nil + }) + }() + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + var abandoned error + var abandonedShared bool + go func() { + defer close(done) + _, abandoned, abandonedShared = g.Do(ctx, "k", func() (cachedResponse, error) { + t.Error("a waiter must not run its own fn") + return cachedResponse{}, nil + }) + }() + + time.Sleep(100 * time.Millisecond) + cancel() + select { + case <-done: + case <-time.After(2 * time.Second): + t.Fatal("a waiter whose context was cancelled stayed parked on the leader") + } + if !errors.Is(abandoned, context.Canceled) { + t.Errorf("abandoning waiter err = %v, want context.Canceled", abandoned) + } + if !abandonedShared { + t.Error("the abandoning waiter did share the flight") + } + + close(release) + wg.Wait() + + if leaderErr != nil || leaderResp.Records != 1 { + t.Errorf("leader = (%+v, %v), want a clean success", leaderResp, leaderErr) + } + if patientErr != nil || patientResp.Records != 1 { + t.Errorf("patient waiter = (%+v, %v), want the leader's result", patientResp, patientErr) + } + + g.mu.Lock() + remaining := len(g.calls) + g.mu.Unlock() + if remaining != 0 { + t.Errorf("%d flights left registered, want 0", remaining) + } +} + +// Every participant walking away must still leave the flight bounded and the +// group empty: nothing parked, nothing registered. +func TestFlightGroup_AllCallersAbandon(t *testing.T) { + var g flightGroup + entered := make(chan struct{}) + release := make(chan struct{}) + + leaderCtx, cancelLeader := context.WithCancel(context.Background()) + leaderDone := make(chan struct{}) + go func() { + defer close(leaderDone) + _, _, _ = g.Do(leaderCtx, "k", func() (cachedResponse, error) { + close(entered) + <-release + return cachedResponse{Records: -1}, nil + }) + }() + <-entered + + const waiters = 8 + ctx, cancel := context.WithCancel(context.Background()) + var wg sync.WaitGroup + for range waiters { + wg.Add(1) + go func() { + defer wg.Done() + _, _, _ = g.Do(ctx, "k", func() (cachedResponse, error) { + t.Error("a waiter must not run its own fn") + return cachedResponse{}, nil + }) + }() + } + + time.Sleep(100 * time.Millisecond) + cancel() + cancelLeader() + waitDone := make(chan struct{}) + go func() { wg.Wait(); close(waitDone) }() + select { + case <-waitDone: + case <-time.After(2 * time.Second): + t.Fatal("waiters stayed parked after their contexts were cancelled") + } + + close(release) + select { + case <-leaderDone: + case <-time.After(2 * time.Second): + t.Fatal("the leader goroutine leaked") + } + + g.mu.Lock() + remaining := len(g.calls) + g.mu.Unlock() + if remaining != 0 { + t.Errorf("%d flights left registered, want 0", remaining) + } +} + func TestHandler_HealthzReportsCacheState(t *testing.T) { a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) b := newCountingBackend(t, map[string]string{factsPath: `[]`}) diff --git a/server.go b/server.go index 05f6c40..c0b4cc0 100644 --- a/server.go +++ b/server.go @@ -142,8 +142,8 @@ func isReportSubResource(path string) bool { func (s *Server) serveMerged(w http.ResponseWriter, r *http.Request, path string, merge func([]backendResult) []json.RawMessage) { params := queryParams(r.URL.Query().Get("query")) - s.serveCached(w, r, path, params, func() (cachedResponse, error) { - alive, err := s.aliveResults(r.Context(), path, params) + s.serveCached(w, r, path, params, func(ctx context.Context) (cachedResponse, error) { + alive, err := s.aliveResults(ctx, path, params) if err != nil { return cachedResponse{}, err } @@ -162,8 +162,8 @@ func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, // Keyed on the request's own params, not the upstream ones: upstreamParams // folds offset into limit, so different windows would collide. - s.serveCached(w, r, path, in, func() (cachedResponse, error) { - alive, err := s.aliveResults(r.Context(), path, page.upstreamParams(in)) + s.serveCached(w, r, path, in, func(ctx context.Context) (cachedResponse, error) { + alive, err := s.aliveResults(ctx, path, page.upstreamParams(in)) if err != nil { return cachedResponse{}, err } @@ -215,8 +215,8 @@ func (s *Server) serveSummed(w http.ResponseWriter, r *http.Request, path string return } - s.serveCached(w, r, path, in, func() (cachedResponse, error) { - alive, err := s.aliveResults(r.Context(), path, page.upstreamParams(in)) + s.serveCached(w, r, path, in, func(ctx context.Context) (cachedResponse, error) { + alive, err := s.aliveResults(ctx, path, page.upstreamParams(in)) if err != nil { return cachedResponse{}, err } @@ -285,10 +285,10 @@ type cachedResponse struct { // fan-out — and stores the result. A build failure falls back to a stale entry // when one exists; that is the only path on which stale data is served. Paths // with no cache configured run build directly, unchanged. -func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string, params url.Values, build func() (cachedResponse, error)) { +func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string, params url.Values, build func(context.Context) (cachedResponse, error)) { cache, enabled := s.cacheFor(path, params) if !enabled { - resp, err := build() + resp, err := build(r.Context()) if err != nil { http.Error(w, err.Error(), http.StatusBadGateway) return @@ -311,8 +311,14 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string stale = &ent } - resp, err, _ := s.flights.Do(key, func() (cachedResponse, error) { - built, buildErr := build() + // The flight is shared, so it runs on a context detached from whichever + // request happened to lead it: one client disconnecting must not cancel the + // fan-out its followers are waiting on. cfg.Timeout keeps it bounded. + resp, err, _ := s.flights.Do(r.Context(), key, func() (cachedResponse, error) { + ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), s.flightTimeout()) + defer cancel() + + built, buildErr := build(ctx) if buildErr != nil { return cachedResponse{}, buildErr } @@ -321,13 +327,18 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string s.log.Printf("warning: encoding cache entry for %s failed: %v", key, marshalErr) return built, nil } - // The leader's request may be cancelled while followers still wait. - if putErr := cache.Put(context.WithoutCancel(r.Context()), key, body); putErr != nil { + if putErr := cache.Put(ctx, key, body); putErr != nil { s.log.Printf("warning: cache store for %s failed: %v", key, putErr) } return built, nil }) if err != nil { + // This caller abandoned the flight because its own client went away; the + // flight itself is still running for everyone else and there is nobody + // left to write to. + if rerr := r.Context().Err(); rerr != nil && errors.Is(err, rerr) { + return + } if stale != nil { s.stale.markStale(time.Now()) s.log.Printf("warning: serving stale %s from cache (stored %s): %v", @@ -342,6 +353,15 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string writeCached(w, resp) } +// http.Client reads a zero Timeout as "no deadline", but it would expire a +// context immediately, so an unset value falls back to the default. +func (s *Server) flightTimeout() time.Duration { + if s.cfg.Timeout > 0 { + return s.cfg.Timeout + } + return defaultTimeout +} + func (s *Server) writeStored(w http.ResponseWriter, body []byte) { var resp cachedResponse if err := json.Unmarshal(body, &resp); err != nil { -- 2.47.3 From c7910156e8efc35ee171130fb49528094d5301e9 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 21:29:34 +1000 Subject: [PATCH 4/7] Mark cached responses with X-Cache and Age A stale fallback is byte-identical to a fresh response, so a client has no way to tell it is holding data pdbmux served only because every backend was down; the sole signal is a log line and a /healthz counter. Set X-Cache to hit, miss or stale and Age to whole seconds since the served copy was stored on every response from a cached path. Neither header is emitted by OpenVoxDB, so nothing upstream is shadowed. --- README.md | 13 ++++++++++++- cache_test.go | 43 +++++++++++++++++++++++++++++++++++++++++++ server.go | 36 ++++++++++++++++++++++++++++++++---- 3 files changed, 87 insertions(+), 5 deletions(-) diff --git a/README.md b/README.md index 751cd3d..d27df08 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,10 @@ surviving backends' results and logs a warning; a merged endpoint only returns ` **every** backend fails. Response records are passed through as raw JSON so unknown fields survive untouched. +Responses carry PuppetDB's `X-Records` when the query asked for a total, and on +the cached paths two headers `pdbmux` adds itself, `X-Cache` and `Age` — see +[Caching](#caching). + ## Merge semantics - **`/nodes`** — dedupe by `certname`; the record with the strictly-newer @@ -238,7 +242,14 @@ backend later without further handler changes. the list/map bookkeeping are not accounted for, so it is a target for body bytes rather than a hard cap on process memory. - **Single-flight** — concurrent requests for the same key collapse into one - upstream fan-out; the rest wait for it and share the result. + upstream fan-out; the rest wait for it and share the result. That fan-out runs + on its own context, bounded by `timeout`, so a client that disconnects can + neither cancel nor fail the requests sharing its flight; a waiter whose own + client goes away leaves the flight running for the others. +- **Response headers** — every response on a cached path carries `X-Cache` + (`hit` served from a fresh entry, `miss` built by this request, `stale` the + expired-entry fallback) and `Age` in whole seconds since the served copy was + stored (`0` on a `miss`). Uncached paths carry neither. - **Visibility** — `/healthz` carries a `cache` object: `backend` (`memory`/`none`), `ttl`, `entries`, `stale_entries`, `bytes`, `serving_stale`, `stale_served` and `last_stale_served`. `serving_stale` is `true` from the diff --git a/cache_test.go b/cache_test.go index 98b0ba0..bb23c89 100644 --- a/cache_test.go +++ b/cache_test.go @@ -11,6 +11,7 @@ import ( "os" "path/filepath" "slices" + "strconv" "strings" "sync" "sync/atomic" @@ -776,6 +777,48 @@ func TestFlightGroup_AllCallersAbandon(t *testing.T) { } } +func TestServeCached_CacheStatusHeaders(t *testing.T) { + stored := `[` + fact("h1", "role", "web", "") + `]` + a := newCountingBackend(t, map[string]string{factsPath: stored}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + h := srv.Handler() + + rec := doGet(t, h, factsPath, "") + if got := rec.Header().Get(cacheStatusHeader); got != "miss" { + t.Errorf("first request %s = %q, want miss", cacheStatusHeader, got) + } + if got := rec.Header().Get(ageHeader); got != "0" { + t.Errorf("first request %s = %q, want 0", ageHeader, got) + } + + rec = doGet(t, h, factsPath, "") + if got := rec.Header().Get(cacheStatusHeader); got != "hit" { + t.Errorf("cached request %s = %q, want hit", cacheStatusHeader, got) + } + if _, err := strconv.Atoi(rec.Header().Get(ageHeader)); err != nil { + t.Errorf("cached request %s = %q, want whole seconds", ageHeader, rec.Header().Get(ageHeader)) + } + + // Past the TTL with every backend down, the stale fallback must say so. + clk.advance(31 * time.Second) + a.setFail(true) + b.setFail(true) + rec = doGet(t, h, factsPath, "") + if rec.Code != http.StatusOK { + t.Fatalf("stale fallback status %d (%s)", rec.Code, rec.Body.String()) + } + if got := rec.Header().Get(cacheStatusHeader); got != "stale" { + t.Errorf("stale fallback %s = %q, want stale", cacheStatusHeader, got) + } + if got := rec.Header().Get(ageHeader); got == "" { + t.Errorf("stale fallback must carry an %s header", ageHeader) + } + if got := strings.TrimSpace(rec.Body.String()); got != stored { + t.Errorf("stale body = %s, want %s", got, stored) + } +} + func TestHandler_HealthzReportsCacheState(t *testing.T) { a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) b := newCountingBackend(t, map[string]string{factsPath: `[]`}) diff --git a/server.go b/server.go index c0b4cc0..b94db43 100644 --- a/server.go +++ b/server.go @@ -27,6 +27,11 @@ const ( // PuppetDB only sends this when the request carries include_total=true. recordsHeader = "X-Records" + + // Set by pdbmux, not by PuppetDB: how a cache-backed response was answered + // and how old the served copy is. + cacheStatusHeader = "X-Cache" + ageHeader = "Age" ) type backendResult struct { @@ -305,7 +310,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string s.log.Printf("warning: cache lookup for %s failed: %v", key, err) case status == CacheFresh: s.stale.markFresh() - s.writeStored(w, ent.Body) + s.writeStored(w, ent, CacheFresh) return case status == CacheStale: stale = &ent @@ -343,13 +348,14 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string s.stale.markStale(time.Now()) s.log.Printf("warning: serving stale %s from cache (stored %s): %v", path, stale.StoredAt.UTC().Format(time.RFC3339), err) - s.writeStored(w, stale.Body) + s.writeStored(w, *stale, CacheStale) return } http.Error(w, err.Error(), http.StatusBadGateway) return } s.stale.markFresh() + setCacheHeaders(w, CacheMiss, time.Time{}) writeCached(w, resp) } @@ -362,16 +368,38 @@ func (s *Server) flightTimeout() time.Duration { return defaultTimeout } -func (s *Server) writeStored(w http.ResponseWriter, body []byte) { +func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status CacheStatus) { var resp cachedResponse - if err := json.Unmarshal(body, &resp); err != nil { + if err := json.Unmarshal(ent.Body, &resp); err != nil { s.log.Printf("warning: unreadable cache entry: %v", err) http.Error(w, "unreadable cache entry", http.StatusBadGateway) return } + setCacheHeaders(w, status, ent.StoredAt) writeCached(w, resp) } +// setCacheHeaders labels a response from a cache-backed path: X-Cache is +// hit/stale/miss and Age is whole seconds since the served copy was stored (0 +// for a response built by this request). +func setCacheHeaders(w http.ResponseWriter, status CacheStatus, storedAt time.Time) { + label := "miss" + switch status { + case CacheFresh: + label = "hit" + case CacheStale: + label = "stale" + } + age := 0 + if !storedAt.IsZero() { + if secs := int(time.Since(storedAt).Seconds()); secs > 0 { + age = secs + } + } + w.Header().Set(cacheStatusHeader, label) + w.Header().Set(ageHeader, strconv.Itoa(age)) +} + func writeCached(w http.ResponseWriter, resp cachedResponse) { w.Header().Set("Content-Type", "application/json") if resp.Records >= 0 { -- 2.47.3 From fc811d4ecac093cccc1a2ee96f48ef5027bcefab Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 22:05:16 +1000 Subject: [PATCH 5/7] Cancel a shared flight when its last participant leaves Reference-count flightCall so the fan-out context ends with the last caller waiting on it, keeping cfg.Timeout as the upper bound. A leader leaving with a follower still parked no longer disturbs the flight, and a solo requester disconnecting releases the upstream sockets at once instead of holding them for the whole timeout. Return errFlightAbandoned from Do rather than inferring the abandon path from the request context's sentinel, and read Age off the server's clock so it matches the timestamp the cache stored. --- README.md | 4 +- cache.go | 90 ++++++++++---- cache_test.go | 316 ++++++++++++++++++++++++++++++++++++++++++++------ server.go | 36 +++--- 4 files changed, 374 insertions(+), 72 deletions(-) diff --git a/README.md b/README.md index d27df08..3ca1b9e 100644 --- a/README.md +++ b/README.md @@ -245,7 +245,9 @@ backend later without further handler changes. upstream fan-out; the rest wait for it and share the result. That fan-out runs on its own context, bounded by `timeout`, so a client that disconnects can neither cancel nor fail the requests sharing its flight; a waiter whose own - client goes away leaves the flight running for the others. + client goes away leaves the flight running for the others. The flight is + cancelled once its last participant leaves, so a lone client disconnecting + releases the upstream connections straight away. - **Response headers** — every response on a cached path carries `X-Cache` (`hit` served from a fresh entry, `miss` built by this request, `stale` the expired-entry fallback) and `Age` in whole seconds since the served copy was diff --git a/cache.go b/cache.go index 65307cd..50dea96 100644 --- a/cache.go +++ b/cache.go @@ -3,6 +3,7 @@ package main import ( "container/list" "context" + "errors" "fmt" "net/url" "runtime/debug" @@ -191,8 +192,18 @@ type flightCall struct { done chan struct{} resp cachedResponse err error + + cancel context.CancelFunc + // participants is the number of callers still waiting on this flight, + // guarded by flightGroup.mu. + participants int } +// errFlightAbandoned reports that this caller stopped waiting because its own +// context ended. It says nothing about the flight, which may still be running +// for other participants. +var errFlightAbandoned = errors.New("abandoned the shared flight") + // flightPanic is a panic from a flight's fn, reported to the leader and to every // waiter as an error so callers keep their error handling (stale fallback, 502) // instead of seeing a zero-value success. @@ -207,43 +218,80 @@ func (p *flightPanic) Error() string { // Do returns fn's result and whether this caller shared another's in-flight run. // -// ctx belongs to the caller alone. A waiter that gives up returns ctx.Err() and -// leaves the flight running for everyone else; the leader ignores ctx entirely -// and always runs fn to completion, so one participant walking away can neither -// cancel nor fail the others. fn is therefore responsible for its own deadline. -func (g *flightGroup) Do(ctx context.Context, key string, fn func() (cachedResponse, error)) (resp cachedResponse, err error, shared bool) { +// fn runs on a context of the flight's own: detached from every caller's, +// bounded by timeout, and cancelled once the last participant leaves. A caller +// that gives up returns errFlightAbandoned and leaves the flight running for +// whoever is still waiting, so one participant walking away can neither cancel +// nor fail the others, while a flight nobody waits on any more is dropped at +// once rather than holding upstream sockets until the timeout. +func (g *flightGroup) Do(ctx context.Context, key string, timeout time.Duration, fn func(context.Context) (cachedResponse, error)) (resp cachedResponse, err error, shared bool) { g.mu.Lock() if g.calls == nil { g.calls = make(map[string]*flightCall) } - if c, ok := g.calls[key]; ok { + c, shared := g.calls[key] + if shared { + c.participants++ g.mu.Unlock() - select { - case <-c.done: - return c.resp, c.err, true - case <-ctx.Done(): - return cachedResponse{}, ctx.Err(), true - } + } else { + flightCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), timeout) + c = &flightCall{done: make(chan struct{}), cancel: cancel, participants: 1} + g.calls[key] = c + g.mu.Unlock() + go g.run(flightCtx, key, c, fn) } - c := &flightCall{done: make(chan struct{})} - g.calls[key] = c - g.mu.Unlock() + defer g.leave(key, c) + select { + case <-c.done: + return c.resp, c.err, shared + case <-ctx.Done(): + return cachedResponse{}, fmt.Errorf("%w: %w", errFlightAbandoned, ctx.Err()), shared + } +} + +func (g *flightGroup) run(ctx context.Context, key string, c *flightCall, fn func(context.Context) (cachedResponse, error)) { defer func() { if r := recover(); r != nil { c.resp, c.err = cachedResponse{}, &flightPanic{value: r, stack: debug.Stack()} - resp, err = c.resp, c.err } // The key is released before the results are published, so a caller that // arrives late leads a new flight instead of joining a finished one. - g.mu.Lock() - delete(g.calls, key) - g.mu.Unlock() + g.forget(key, c) close(c.done) }() - c.resp, c.err = fn() - return c.resp, c.err, false + c.resp, c.err = fn(ctx) +} + +// leave drops one participant and, when it was the last, unregisters the key and +// cancels the flight so a lone requester disconnecting aborts the fan-out +// instead of pinning a socket per backend for the whole timeout. Unregistering +// under the same lock that admits joiners keeps anyone from joining a flight +// that is about to be cancelled. +func (g *flightGroup) leave(key string, c *flightCall) { + g.mu.Lock() + c.participants-- + last := c.participants == 0 + if last { + g.unregister(key, c) + } + g.mu.Unlock() + if last { + c.cancel() + } +} + +func (g *flightGroup) forget(key string, c *flightCall) { + g.mu.Lock() + g.unregister(key, c) + g.mu.Unlock() +} + +func (g *flightGroup) unregister(key string, c *flightCall) { + if cur, ok := g.calls[key]; ok && cur == c { + delete(g.calls, key) + } } // staleTracker records stale fallbacks for /healthz. serving flips back to false diff --git a/cache_test.go b/cache_test.go index bb23c89..5ba0d5b 100644 --- a/cache_test.go +++ b/cache_test.go @@ -5,11 +5,13 @@ import ( "encoding/json" "errors" "io" + "math/rand/v2" "net/http" "net/http/httptest" "net/url" "os" "path/filepath" + "runtime" "slices" "strconv" "strings" @@ -19,6 +21,16 @@ import ( "time" ) +const ( + // testFlightTimeout is long enough that a flight ending early can only be the + // refcount, never the deadline. + testFlightTimeout = 30 * time.Second + + // disconnectAfter is long enough for the fan-out to reach the backends and + // short enough to leave the whole abort well inside cfg.Timeout. + disconnectAfter = 25 * time.Millisecond +) + // fakeClock drives the cache's TTL without sleeping. type fakeClock struct { mu sync.Mutex @@ -111,6 +123,7 @@ func newCachedServer(t *testing.T, cfg Config) (*Server, *fakeClock) { } clk := newFakeClock() mc.now = clk.now + srv.now = clk.now return srv, clk } @@ -389,7 +402,7 @@ func TestFlightGroup_LeaderRunsOnce(t *testing.T) { entered := make(chan struct{}) release := make(chan struct{}) - run := func() (cachedResponse, error) { + run := func(context.Context) (cachedResponse, error) { if calls.Add(1) == 1 { close(entered) } @@ -402,7 +415,7 @@ func TestFlightGroup_LeaderRunsOnce(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - if _, _, s := g.Do(context.Background(), "k", run); s { + if _, _, s := g.Do(context.Background(), "k", testFlightTimeout, run); s { shared.Add(1) } }() @@ -419,7 +432,9 @@ func TestFlightGroup_LeaderRunsOnce(t *testing.T) { t.Errorf("%d callers shared the flight, want 15", shared.Load()) } // The key is released once the flight finishes. - if _, _, s := g.Do(context.Background(), "k", func() (cachedResponse, error) { return cachedResponse{Records: -1}, nil }); s { + if _, _, s := g.Do(context.Background(), "k", testFlightTimeout, func(context.Context) (cachedResponse, error) { + return cachedResponse{Records: -1}, nil + }); s { t.Error("a later call must start its own flight") } } @@ -441,7 +456,7 @@ func TestFlightGroup_LeaderPanicFailsLeaderAndWaiters(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - leader.resp, leader.err, leader.shared = g.Do(context.Background(), "k", func() (cachedResponse, error) { + leader.resp, leader.err, leader.shared = g.Do(context.Background(), "k", testFlightTimeout, func(context.Context) (cachedResponse, error) { close(entered) <-release panic("build exploded") @@ -453,7 +468,7 @@ func TestFlightGroup_LeaderPanicFailsLeaderAndWaiters(t *testing.T) { wg.Add(1) go func(i int) { defer wg.Done() - waiters[i].resp, waiters[i].err, waiters[i].shared = g.Do(context.Background(), "k", func() (cachedResponse, error) { + waiters[i].resp, waiters[i].err, waiters[i].shared = g.Do(context.Background(), "k", testFlightTimeout, func(context.Context) (cachedResponse, error) { t.Error("a waiter must not run its own fn") return cachedResponse{}, nil }) @@ -480,7 +495,7 @@ func TestFlightGroup_LeaderPanicFailsLeaderAndWaiters(t *testing.T) { } // The key is released on the panic path, so a later call leads its own flight. - resp, err, shared := g.Do(context.Background(), "k", func() (cachedResponse, error) { + resp, err, shared := g.Do(context.Background(), "k", testFlightTimeout, func(context.Context) (cachedResponse, error) { return cachedResponse{Body: json.RawMessage(`[]`), Records: 0}, nil }) if shared { @@ -639,6 +654,117 @@ func TestHandler_LeaderDisconnectDoesNotFailFollowers(t *testing.T) { if got := a.hitCount(nodesPath); got != 1 { t.Errorf("backend a saw %d requests, want 1", got) } + + // The flight the leader started still populated the cache. + warmed := doGet(t, h, nodesPath, "") + if got := warmed.Header().Get(cacheStatusHeader); got != "hit" { + t.Errorf("%s = %q, want hit: the abandoned leader's flight must still warm the cache", cacheStatusHeader, got) + } + if got := a.hitCount(nodesPath); got != 1 { + t.Errorf("backend a saw %d requests after the cached read, want 1", got) + } + assertNoFlights(t, &srv.flights) +} + +// blockingBackend parks every request until its own context ends, reporting when +// that happened, so an abandoned fan-out is observable from upstream. +func blockingBackend(t *testing.T, aborted chan<- time.Time) string { + t.Helper() + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case <-r.Context().Done(): + aborted <- time.Now() + case <-time.After(10 * time.Second): + } + })) + t.Cleanup(srv.Close) + return srv.URL +} + +// A requester with nobody else on its flight must take the fan-out down with it +// rather than leave a socket per backend held open until cfg.Timeout. +func TestHandler_SoloDisconnectAbortsFanOutPromptly(t *testing.T) { + aborted := make(chan time.Time, 4) + cfg := cacheTestConfig(blockingBackend(t, aborted), blockingBackend(t, aborted)) + cfg.Timeout = 400 * time.Millisecond + srv, _ := newCachedServer(t, cfg) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + time.Sleep(disconnectAfter) + cancel() + }() + + start := time.Now() + req := httptest.NewRequest(http.MethodGet, factsPath, nil).WithContext(ctx) + srv.Handler().ServeHTTP(httptest.NewRecorder(), req) + if elapsed := time.Since(start); elapsed >= cfg.Timeout/2 { + t.Errorf("handler returned after %s, want well under cfg.Timeout %s", elapsed, cfg.Timeout) + } + + waitForReleases(t, aborted, len(cfg.Backends), start.Add(cfg.Timeout/2)) + assertNoFlights(t, &srv.flights) +} + +// Distinct cache keys do not collapse into one flight, so disconnecting clients +// must not each hold len(backends) sockets for the whole timeout. +func TestHandler_DisconnectedRequestsDoNotPinBackends(t *testing.T) { + aborted := make(chan time.Time, 128) + cfg := cacheTestConfig(blockingBackend(t, aborted), blockingBackend(t, aborted)) + cfg.Timeout = 400 * time.Millisecond + srv, _ := newCachedServer(t, cfg) + h := srv.Handler() + + // Keep-alive plumbing outlives the requests, so the transport is ours to shut + // down before counting goroutines. + transport := &http.Transport{} + srv.client.Transport = transport + + baseline := runtime.NumGoroutine() + const callers = 25 + var wg sync.WaitGroup + start := time.Now() + for i := range callers { + wg.Add(1) + go func(i int) { + defer wg.Done() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go func() { + time.Sleep(disconnectAfter) + cancel() + }() + target := factsPath + "?query=" + url.QueryEscape(strconv.Itoa(i)) + req := httptest.NewRequest(http.MethodGet, target, nil).WithContext(ctx) + h.ServeHTTP(httptest.NewRecorder(), req) + }(i) + } + wg.Wait() + if elapsed := time.Since(start); elapsed >= cfg.Timeout/2 { + t.Errorf("%d disconnecting callers took %s, want well under cfg.Timeout %s", callers, elapsed, cfg.Timeout) + } + + // A caller cancelled before its fan-out was dispatched leaves the backend + // nothing to release, so one release per caller is the floor. + waitForReleases(t, aborted, callers, start.Add(cfg.Timeout/2)) + assertNoFlights(t, &srv.flights) + transport.CloseIdleConnections() + assertGoroutinesSettle(t, baseline, 2) +} + +// waitForReleases fails unless at least want backend requests were released by +// cutoff, which is set well inside cfg.Timeout so only the flight going away can +// have freed them. +func waitForReleases(t *testing.T, aborted <-chan time.Time, want int, cutoff time.Time) { + t.Helper() + for got := 0; got < want; got++ { + select { + case <-aborted: + case <-time.After(time.Until(cutoff)): + t.Fatalf("%d of %d backend requests released before the cutoff, want all of them", got, want) + } + } } // A follower whose own client goes away must unpark rather than wait out the @@ -654,7 +780,7 @@ func TestFlightGroup_WaiterAbandonsOnContextCancel(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - leaderResp, leaderErr, _ = g.Do(context.Background(), "k", func() (cachedResponse, error) { + leaderResp, leaderErr, _ = g.Do(context.Background(), "k", testFlightTimeout, func(context.Context) (cachedResponse, error) { close(entered) <-release return cachedResponse{Body: json.RawMessage(`[1]`), Records: 1}, nil @@ -668,10 +794,7 @@ func TestFlightGroup_WaiterAbandonsOnContextCancel(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - patientResp, patientErr, _ = g.Do(context.Background(), "k", func() (cachedResponse, error) { - t.Error("a waiter must not run its own fn") - return cachedResponse{}, nil - }) + patientResp, patientErr, _ = g.Do(context.Background(), "k", testFlightTimeout, waiterMustNotBuild(t)) }() ctx, cancel := context.WithCancel(context.Background()) @@ -680,10 +803,7 @@ func TestFlightGroup_WaiterAbandonsOnContextCancel(t *testing.T) { var abandonedShared bool go func() { defer close(done) - _, abandoned, abandonedShared = g.Do(ctx, "k", func() (cachedResponse, error) { - t.Error("a waiter must not run its own fn") - return cachedResponse{}, nil - }) + _, abandoned, abandonedShared = g.Do(ctx, "k", testFlightTimeout, waiterMustNotBuild(t)) }() time.Sleep(100 * time.Millisecond) @@ -693,8 +813,11 @@ func TestFlightGroup_WaiterAbandonsOnContextCancel(t *testing.T) { case <-time.After(2 * time.Second): t.Fatal("a waiter whose context was cancelled stayed parked on the leader") } + if !errors.Is(abandoned, errFlightAbandoned) { + t.Errorf("abandoning waiter err = %v, want errFlightAbandoned", abandoned) + } if !errors.Is(abandoned, context.Canceled) { - t.Errorf("abandoning waiter err = %v, want context.Canceled", abandoned) + t.Errorf("abandoning waiter err = %v, want it to carry context.Canceled", abandoned) } if !abandonedShared { t.Error("the abandoning waiter did share the flight") @@ -718,21 +841,22 @@ func TestFlightGroup_WaiterAbandonsOnContextCancel(t *testing.T) { } } -// Every participant walking away must still leave the flight bounded and the -// group empty: nothing parked, nothing registered. -func TestFlightGroup_AllCallersAbandon(t *testing.T) { +// The last participant leaving must cancel the flight's context rather than let +// it burn the whole timeout, and must leave the group empty. +func TestFlightGroup_LastParticipantLeavingCancelsFlight(t *testing.T) { var g flightGroup entered := make(chan struct{}) - release := make(chan struct{}) + flightCancelled := make(chan struct{}) leaderCtx, cancelLeader := context.WithCancel(context.Background()) leaderDone := make(chan struct{}) go func() { defer close(leaderDone) - _, _, _ = g.Do(leaderCtx, "k", func() (cachedResponse, error) { + _, _, _ = g.Do(leaderCtx, "k", testFlightTimeout, func(ctx context.Context) (cachedResponse, error) { close(entered) - <-release - return cachedResponse{Records: -1}, nil + <-ctx.Done() + close(flightCancelled) + return cachedResponse{}, ctx.Err() }) }() <-entered @@ -744,10 +868,7 @@ func TestFlightGroup_AllCallersAbandon(t *testing.T) { wg.Add(1) go func() { defer wg.Done() - _, _, _ = g.Do(ctx, "k", func() (cachedResponse, error) { - t.Error("a waiter must not run its own fn") - return cachedResponse{}, nil - }) + _, _, _ = g.Do(ctx, "k", testFlightTimeout, waiterMustNotBuild(t)) }() } @@ -762,13 +883,123 @@ func TestFlightGroup_AllCallersAbandon(t *testing.T) { t.Fatal("waiters stayed parked after their contexts were cancelled") } - close(release) + // testFlightTimeout is far longer, so only the refcount can have cancelled it. + select { + case <-flightCancelled: + case <-time.After(2 * time.Second): + t.Fatal("the flight ran on after its last participant left") + } select { case <-leaderDone: case <-time.After(2 * time.Second): t.Fatal("the leader goroutine leaked") } + assertNoFlights(t, &g) +} + +// A leader walking away while a waiter is still parked must leave the flight's +// context untouched, so the waiter gets a real result. +func TestFlightGroup_LeaderLeavingKeepsFlightAliveForWaiter(t *testing.T) { + var g flightGroup + entered := make(chan struct{}) + release := make(chan struct{}) + + leaderCtx, cancelLeader := context.WithCancel(context.Background()) + leaderDone := make(chan struct{}) + var leaderErr error + go func() { + defer close(leaderDone) + _, leaderErr, _ = g.Do(leaderCtx, "k", testFlightTimeout, func(ctx context.Context) (cachedResponse, error) { + close(entered) + <-release + if err := ctx.Err(); err != nil { + return cachedResponse{}, err + } + return cachedResponse{Body: json.RawMessage(`[1]`), Records: 1}, nil + }) + }() + <-entered + + waiterDone := make(chan struct{}) + var waiterResp cachedResponse + var waiterErr error + go func() { + defer close(waiterDone) + waiterResp, waiterErr, _ = g.Do(context.Background(), "k", testFlightTimeout, waiterMustNotBuild(t)) + }() + + time.Sleep(100 * time.Millisecond) + cancelLeader() + select { + case <-leaderDone: + case <-time.After(2 * time.Second): + t.Fatal("the leader stayed parked after its context was cancelled") + } + if !errors.Is(leaderErr, errFlightAbandoned) { + t.Errorf("leader err = %v, want errFlightAbandoned", leaderErr) + } + + close(release) + select { + case <-waiterDone: + case <-time.After(2 * time.Second): + t.Fatal("the waiter never got a result") + } + if waiterErr != nil || waiterResp.Records != 1 { + t.Errorf("waiter = (%+v, %v), want the flight's result", waiterResp, waiterErr) + } + + assertNoFlights(t, &g) +} + +// Random cancellations across many keys must leave nothing registered and no +// goroutines behind. +func TestFlightGroup_HammerRandomCancellations(t *testing.T) { + var g flightGroup + baseline := runtime.NumGoroutine() + + const callers = 400 + var wg sync.WaitGroup + for i := range callers { + wg.Add(1) + go func(i int) { + defer wg.Done() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + if i%3 == 0 { + go func() { + time.Sleep(time.Duration(rand.IntN(3000)) * time.Microsecond) + cancel() + }() + } + key := strconv.Itoa(i % 17) + _, _, _ = g.Do(ctx, key, testFlightTimeout, func(ctx context.Context) (cachedResponse, error) { + select { + case <-ctx.Done(): + return cachedResponse{}, ctx.Err() + case <-time.After(time.Duration(rand.IntN(3000)) * time.Microsecond): + return cachedResponse{Records: -1}, nil + } + }) + }(i) + } + wg.Wait() + + assertNoFlights(t, &g) + assertGoroutinesSettle(t, baseline, 2) +} + +func waiterMustNotBuild(t *testing.T) func(context.Context) (cachedResponse, error) { + t.Helper() + return func(context.Context) (cachedResponse, error) { + t.Error("a waiter must not run its own fn") + return cachedResponse{}, nil + } +} + +func assertNoFlights(t *testing.T, g *flightGroup) { + t.Helper() g.mu.Lock() remaining := len(g.calls) g.mu.Unlock() @@ -777,6 +1008,24 @@ func TestFlightGroup_AllCallersAbandon(t *testing.T) { } } +// Goroutines unwind after their caller returns, so settling is polled rather +// than sampled once. +func assertGoroutinesSettle(t *testing.T, baseline, slack int) { + t.Helper() + deadline := time.Now().Add(5 * time.Second) + for { + got := runtime.NumGoroutine() + if got <= baseline+slack { + return + } + if time.Now().After(deadline) { + t.Errorf("goroutines = %d, want back near the baseline of %d", got, baseline) + return + } + time.Sleep(10 * time.Millisecond) + } +} + func TestServeCached_CacheStatusHeaders(t *testing.T) { stored := `[` + fact("h1", "role", "web", "") + `]` a := newCountingBackend(t, map[string]string{factsPath: stored}) @@ -792,16 +1041,17 @@ func TestServeCached_CacheStatusHeaders(t *testing.T) { t.Errorf("first request %s = %q, want 0", ageHeader, got) } + clk.advance(7 * time.Second) rec = doGet(t, h, factsPath, "") if got := rec.Header().Get(cacheStatusHeader); got != "hit" { t.Errorf("cached request %s = %q, want hit", cacheStatusHeader, got) } - if _, err := strconv.Atoi(rec.Header().Get(ageHeader)); err != nil { - t.Errorf("cached request %s = %q, want whole seconds", ageHeader, rec.Header().Get(ageHeader)) + if got := rec.Header().Get(ageHeader); got != "7" { + t.Errorf("cached request %s = %q, want 7", ageHeader, got) } // Past the TTL with every backend down, the stale fallback must say so. - clk.advance(31 * time.Second) + clk.advance(24 * time.Second) a.setFail(true) b.setFail(true) rec = doGet(t, h, factsPath, "") @@ -811,8 +1061,8 @@ func TestServeCached_CacheStatusHeaders(t *testing.T) { if got := rec.Header().Get(cacheStatusHeader); got != "stale" { t.Errorf("stale fallback %s = %q, want stale", cacheStatusHeader, got) } - if got := rec.Header().Get(ageHeader); got == "" { - t.Errorf("stale fallback must carry an %s header", ageHeader) + if got := rec.Header().Get(ageHeader); got != "31" { + t.Errorf("stale fallback %s = %q, want 31 seconds since the entry was stored", ageHeader, got) } if got := strings.TrimSpace(rec.Body.String()); got != stored { t.Errorf("stale body = %s, want %s", got, stored) diff --git a/server.go b/server.go index b94db43..c670b8f 100644 --- a/server.go +++ b/server.go @@ -53,6 +53,9 @@ type Server struct { flights flightGroup stale staleTracker + // now is shared with the cache's clock so Age matches the stored timestamp. + now func() time.Time + // freshness cache (freshness merge only). mu sync.Mutex freshData freshness @@ -65,6 +68,7 @@ func NewServer(cfg Config, logger *log.Logger) *Server { cfg: cfg, client: &http.Client{Timeout: cfg.Timeout}, log: logger, + now: time.Now, } if cfg.cacheEnabled() { s.factsCache = newMemoryCache(cfg.FactsTTL, cfg.CacheBytes) @@ -316,13 +320,11 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string stale = &ent } - // The flight is shared, so it runs on a context detached from whichever - // request happened to lead it: one client disconnecting must not cancel the - // fan-out its followers are waiting on. cfg.Timeout keeps it bounded. - resp, err, _ := s.flights.Do(r.Context(), key, func() (cachedResponse, error) { - ctx, cancel := context.WithTimeout(context.WithoutCancel(r.Context()), s.flightTimeout()) - defer cancel() - + // The flight is shared, so it runs on its own context rather than the leading + // request's: one client disconnecting must not cancel the fan-out its + // followers are waiting on, and the flight ends as soon as the last of them + // goes. cfg.Timeout keeps it bounded. + resp, err, _ := s.flights.Do(r.Context(), key, s.flightTimeout(), func(ctx context.Context) (cachedResponse, error) { built, buildErr := build(ctx) if buildErr != nil { return cachedResponse{}, buildErr @@ -338,14 +340,13 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string return built, nil }) if err != nil { - // This caller abandoned the flight because its own client went away; the - // flight itself is still running for everyone else and there is nobody - // left to write to. - if rerr := r.Context().Err(); rerr != nil && errors.Is(err, rerr) { + // This caller left the flight because its own client went away, so there + // is nobody to write to. + if errors.Is(err, errFlightAbandoned) { return } if stale != nil { - s.stale.markStale(time.Now()) + s.stale.markStale(s.now()) s.log.Printf("warning: serving stale %s from cache (stored %s): %v", path, stale.StoredAt.UTC().Format(time.RFC3339), err) s.writeStored(w, *stale, CacheStale) @@ -355,7 +356,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string return } s.stale.markFresh() - setCacheHeaders(w, CacheMiss, time.Time{}) + s.setCacheHeaders(w, CacheMiss, time.Time{}) writeCached(w, resp) } @@ -375,14 +376,15 @@ func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status Cache http.Error(w, "unreadable cache entry", http.StatusBadGateway) return } - setCacheHeaders(w, status, ent.StoredAt) + s.setCacheHeaders(w, status, ent.StoredAt) writeCached(w, resp) } // setCacheHeaders labels a response from a cache-backed path: X-Cache is // hit/stale/miss and Age is whole seconds since the served copy was stored (0 -// for a response built by this request). -func setCacheHeaders(w http.ResponseWriter, status CacheStatus, storedAt time.Time) { +// for a response built by this request). It reads the same clock the cache +// stamps entries with, so the two never disagree. +func (s *Server) setCacheHeaders(w http.ResponseWriter, status CacheStatus, storedAt time.Time) { label := "miss" switch status { case CacheFresh: @@ -392,7 +394,7 @@ func setCacheHeaders(w http.ResponseWriter, status CacheStatus, storedAt time.Ti } age := 0 if !storedAt.IsZero() { - if secs := int(time.Since(storedAt).Seconds()); secs > 0 { + if secs := int(s.now().Sub(storedAt).Seconds()); secs > 0 { age = secs } } -- 2.47.3 From 45ac52df65cbd5de2074a6aed706ef53d720e448 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 22:20:53 +1000 Subject: [PATCH 6/7] Store a completed build on a context detached from the flight --- cache.go | 4 +++ cache_test.go | 75 +++++++++++++++++++++++++++++++++++++++++++++++++++ server.go | 8 +++++- 3 files changed, 86 insertions(+), 1 deletion(-) diff --git a/cache.go b/cache.go index 50dea96..e2b24e6 100644 --- a/cache.go +++ b/cache.go @@ -52,6 +52,10 @@ type CacheStats struct { // upstream fetch fails. The context and error exist for out-of-process backends // (the reports cache lands on S3); an in-process backend ignores both. // +// Put's context is detached from the request and flight that produced the body, +// so a store still runs when the last caller has walked away; it carries its own +// timeout. +// // A Body handed back by Get aliases the cache's copy and must not be mutated. type Cache interface { Get(ctx context.Context, key string) (CacheEntry, CacheStatus, error) diff --git a/cache_test.go b/cache_test.go index 5ba0d5b..7ace493 100644 --- a/cache_test.go +++ b/cache_test.go @@ -1069,6 +1069,81 @@ func TestServeCached_CacheStatusHeaders(t *testing.T) { } } +// recordingCache captures what Put was handed, so a test can assert the store +// does not run on an already-cancelled context. memoryCache ignores its context +// and so cannot show the difference. +type recordingCache struct { + putCalled chan struct{} + putCtxErr error + putBody []byte +} + +func newRecordingCache() *recordingCache { + return &recordingCache{putCalled: make(chan struct{})} +} + +func (c *recordingCache) Get(context.Context, string) (CacheEntry, CacheStatus, error) { + return CacheEntry{}, CacheMiss, nil +} + +// The fields are read only after putCalled closes, which orders the two. +func (c *recordingCache) Put(ctx context.Context, _ string, body []byte) error { + c.putCtxErr = ctx.Err() + c.putBody = append([]byte(nil), body...) + close(c.putCalled) + return nil +} + +func (c *recordingCache) Stats() CacheStats { return CacheStats{Backend: "recording"} } + +// A build that succeeded must still reach the cache once the last participant +// has left and cancelled the flight, or an out-of-process backend would drop the +// write and lose the entry the next caller would have hit warm. +func TestServeCached_PutRunsOnDetachedContext(t *testing.T) { + srv := newTestServer(cacheTestConfig("http://backend.invalid", "http://backend.invalid")) + cache := newRecordingCache() + srv.factsCache = cache + + entered := make(chan struct{}) + release := make(chan struct{}) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + req := httptest.NewRequest(http.MethodGet, factsPath, nil).WithContext(ctx) + + served := make(chan struct{}) + go func() { + defer close(served) + srv.serveCached(httptest.NewRecorder(), req, factsPath, nil, func(context.Context) (cachedResponse, error) { + close(entered) + <-release + return cachedResponse{Body: json.RawMessage(`[]`), Records: -1}, nil + }) + }() + + <-entered + cancel() + select { + case <-served: + case <-time.After(2 * time.Second): + t.Fatal("the abandoning request stayed parked") + } + + // The flight's own context is cancelled by now, so only a detached one can + // carry the store. + close(release) + select { + case <-cache.putCalled: + case <-time.After(2 * time.Second): + t.Fatal("a completed build never reached the cache") + } + if cache.putCtxErr != nil { + t.Errorf("Put ran on a cancelled context: %v", cache.putCtxErr) + } + if len(cache.putBody) == 0 { + t.Error("Put stored an empty body") + } +} + func TestHandler_HealthzReportsCacheState(t *testing.T) { a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) b := newCountingBackend(t, map[string]string{factsPath: `[]`}) diff --git a/server.go b/server.go index c670b8f..5e9fbb6 100644 --- a/server.go +++ b/server.go @@ -334,7 +334,13 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string s.log.Printf("warning: encoding cache entry for %s failed: %v", key, marshalErr) return built, nil } - if putErr := cache.Put(ctx, key, body); putErr != nil { + // The build succeeded, so the entry is worth storing even if the last + // participant has already left and cancelled ctx: warming the cache for + // the next caller is the whole point. Same bound as the flight so an + // out-of-process cache cannot hang the store forever. + putCtx, cancelPut := context.WithTimeout(context.WithoutCancel(ctx), s.flightTimeout()) + defer cancelPut() + if putErr := cache.Put(putCtx, key, body); putErr != nil { s.log.Printf("warning: cache store for %s failed: %v", key, putErr) } return built, nil -- 2.47.3 From de61ec5081c41b47ba9540e49ea62040dec15102 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 23:05:54 +1000 Subject: [PATCH 7/7] Cache the merged body with its provenance already injected MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Rebasing onto main brought in the per-request sourceInjector, which now runs inside the cached build: what a cache entry holds is the merged body with pdbmux_source already stamped and upstream records of that name already dropped. Injecting on the way out instead would mean storing the un-injected records plus a per-certname backend map and re-marshalling every record on every hit, which is the work the cache exists to avoid. Baking it in stays correct because the value names the backend that supplied the data — a property of that fetch, not of the caller reading it — so it ages out with the body it labels, and because the injection gate is a pure function of path and query, both of which are already in the cache key. Update the two cache tests whose byte-exact bodies predate the fact, and add tests for the composition: attribution survives a cache hit on /facts and /nodes, it ages with its entry rather than tracking a node that moved, gated and ungated queries cache separately, and suppression of an upstream fact of that name survives into the entry. --- README.md | 12 ++++ cache_test.go | 177 ++++++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 185 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 3ca1b9e..d60e7b0 100644 --- a/README.md +++ b/README.md @@ -257,6 +257,18 @@ backend later without further handler changes. `stale_served` and `last_stale_served`. `serving_stale` is `true` from the moment a stale fallback is served until the next response comes from a live fan-out or a fresh entry. +- **Provenance is stored, not re-applied** — what a cache entry holds is the + fully merged body, `pdbmux_source` already injected and upstream records of + that name already dropped. Attribution names the backend that supplied the + data, which is a property of that fetch, so it stays correct for as long as the + body does and ages out with it — `X-Cache` and `Age` say how old both are. Two + requests can only share an entry when they share a key, and the key is path + plus query, which is exactly what decides whether injection applies; a + name-filtered `/facts` query and a plain one therefore cache separately and + neither is ever served the other's shape. `source_fact` and + `source_fact_enabled` are read once at startup, and the cache lives for the + same process, so changing either cannot leave differently-shaped entries + behind. ## Config diff --git a/cache_test.go b/cache_test.go index 7ace493..02cf087 100644 --- a/cache_test.go +++ b/cache_test.go @@ -11,6 +11,7 @@ import ( "net/url" "os" "path/filepath" + "reflect" "runtime" "slices" "strconv" @@ -648,8 +649,11 @@ func TestHandler_LeaderDisconnectDoesNotFailFollowers(t *testing.T) { t.Fatalf("follower status %d (%s), want 200: a healthy client must not inherit the leader's cancellation", follower.Code, follower.Body.String()) } - if got := strings.TrimSpace(follower.Body.String()); got != body { - t.Errorf("follower body = %s, want %s", got, body) + // The leader built this body, so its provenance names the backend that + // answered the leader's fan-out. + want := `[` + stamped(t, node("h1", "2026-01-01T00:00:00.000Z"), defaultSourceFact, "a") + `]` + if got := strings.TrimSpace(follower.Body.String()); !sameJSON(t, got, want) { + t.Errorf("follower body = %s, want %s", got, want) } if got := a.hitCount(nodesPath); got != 1 { t.Errorf("backend a saw %d requests, want 1", got) @@ -1064,8 +1068,9 @@ func TestServeCached_CacheStatusHeaders(t *testing.T) { if got := rec.Header().Get(ageHeader); got != "31" { t.Errorf("stale fallback %s = %q, want 31 seconds since the entry was stored", ageHeader, got) } - if got := strings.TrimSpace(rec.Body.String()); got != stored { - t.Errorf("stale body = %s, want %s", got, stored) + want := `[` + fact("h1", "role", "web", "") + `,` + factEnv("h1", defaultSourceFact, "a", "") + `]` + if got := strings.TrimSpace(rec.Body.String()); !sameJSON(t, got, want) { + t.Errorf("stale body = %s, want %s", got, want) } } @@ -1377,3 +1382,167 @@ func waitFor(t *testing.T, cond func() bool) { time.Sleep(time.Millisecond) } } + +// sameJSON compares two JSON documents by value, so a test need not track the +// key order json.Marshal produces for a stamped record. +func sameJSON(t *testing.T, got, want string) bool { + t.Helper() + var g, w any + if err := json.Unmarshal([]byte(got), &g); err != nil { + t.Fatalf("unmarshal got %s: %v", got, err) + } + if err := json.Unmarshal([]byte(want), &w); err != nil { + t.Fatalf("unmarshal want %s: %v", want, err) + } + return reflect.DeepEqual(g, w) +} + +// stamped is a /nodes record with one extra string key, as the merge serves it. +func stamped(t *testing.T, raw, field, value string) string { + t.Helper() + var obj map[string]json.RawMessage + if err := json.Unmarshal([]byte(raw), &obj); err != nil { + t.Fatalf("unmarshal %s: %v", raw, err) + } + obj[field] = json.RawMessage(strconv.Quote(value)) + out, err := json.Marshal(obj) + if err != nil { + t.Fatalf("marshal stamped record: %v", err) + } + return string(out) +} + +// The injector is per-request but a cache entry is shared, so a second caller is +// served a body built for the first. Provenance names the backend that supplied +// the data, which is a property of that fetch, so the shared body stays correct. +func TestHandler_CachedFactsKeepSourceAttribution(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + h := srv.Handler() + + first := doGet(t, h, factsPath, "") + if got := first.Header().Get(cacheStatusHeader); got != "miss" { + t.Fatalf("first %s = %q, want miss", cacheStatusHeader, got) + } + srcs, n := sourceValues(t, first.Body.Bytes(), defaultSourceFact) + if n != 1 || srcs["h1"] != "a" { + t.Fatalf("first request sources = %v (%d records), want h1 -> a", srcs, n) + } + + second := doGet(t, h, factsPath, "") + if got := second.Header().Get(cacheStatusHeader); got != "hit" { + t.Fatalf("second %s = %q, want hit", cacheStatusHeader, got) + } + srcs, n = sourceValues(t, second.Body.Bytes(), defaultSourceFact) + if n != 1 || srcs["h1"] != "a" { + t.Errorf("cached request sources = %v (%d records), want exactly one h1 -> a", srcs, n) + } + if got := a.hitCount(factsPath); got != 1 { + t.Errorf("backend a saw %d requests, want 1: the second read must come from the cache", got) + } +} + +// The same, for the /nodes stamp rather than the synthetic /facts record. +func TestHandler_CachedNodesKeepSourceStamp(t *testing.T) { + a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-02T00:00:00.000Z") + `]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + h := srv.Handler() + + first := doGet(t, h, nodesPath, "") + if got := nodeSources(t, first.Body.Bytes(), defaultSourceFact); got["h1"] != "a" { + t.Fatalf("first request stamp = %v, want h1 -> a", got) + } + + second := doGet(t, h, nodesPath, "") + if got := second.Header().Get(cacheStatusHeader); got != "hit" { + t.Fatalf("second %s = %q, want hit", cacheStatusHeader, got) + } + want := `[` + stamped(t, node("h1", "2026-01-02T00:00:00.000Z"), defaultSourceFact, "a") + `]` + if got := strings.TrimSpace(second.Body.String()); !sameJSON(t, got, want) { + t.Errorf("cached body = %s, want %s", got, want) + } + if got := a.hitCount(nodesPath); got != 1 { + t.Errorf("backend a saw %d requests, want 1", got) + } +} + +// Provenance is baked into the cached body, so it ages with the data it labels: +// while the entry is served the attribution is the one that fetch had, and the +// rebuild after the TTL picks up the move. +func TestHandler_CachedSourceAgesWithItsData(t *testing.T) { + a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + h := srv.Handler() + + doGet(t, h, factsPath, "") + + // The node moves to b while the entry is still fresh. + a.setBody(factsPath, `[]`) + b.setBody(factsPath, `[`+fact("h1", "role", "web", "")+`]`) + + cached := doGet(t, h, factsPath, "") + if got := cached.Header().Get(cacheStatusHeader); got != "hit" { + t.Fatalf("%s = %q, want hit", cacheStatusHeader, got) + } + if srcs, _ := sourceValues(t, cached.Body.Bytes(), defaultSourceFact); srcs["h1"] != "a" { + t.Errorf("cached sources = %v, want h1 -> a: the body and its attribution come from the same fetch", srcs) + } + + clk.advance(31 * time.Second) + rebuilt := doGet(t, h, factsPath, "") + if srcs, _ := sourceValues(t, rebuilt.Body.Bytes(), defaultSourceFact); srcs["h1"] != "b" { + t.Errorf("rebuilt sources = %v, want h1 -> b once the entry expired", srcs) + } +} + +// The injection gate is a pure function of path and query, both of which are in +// the cache key, so a gated request can never be served an injected body cached +// for an ungated one. +func TestHandler_SourceGateIsPerCacheKey(t *testing.T) { + body := `[` + fact("h1", "role", "web", "") + `]` + a := newCountingBackend(t, map[string]string{factsPath: body}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + h := srv.Handler() + + const nameFiltered = `["=","name","role"]` + for _, pass := range []string{"first", "cached"} { + open := doGet(t, h, factsPath, "") + if srcs, n := sourceValues(t, open.Body.Bytes(), defaultSourceFact); n != 1 || srcs["h1"] != "a" { + t.Errorf("%s unfiltered request sources = %v (%d records), want h1 -> a", pass, srcs, n) + } + gated := doGet(t, h, factsPath, nameFiltered) + if _, n := sourceValues(t, gated.Body.Bytes(), defaultSourceFact); n != 0 { + t.Errorf("%s name-filtered request carried %d synthetic record(s), want none", pass, n) + } + if got := strings.TrimSpace(gated.Body.String()); !sameJSON(t, got, body) { + t.Errorf("%s name-filtered body = %s, want %s", pass, got, body) + } + } + if got := a.hitCount(factsPath); got != 2 { + t.Errorf("backend a saw %d requests, want 2: one per distinct cache key", got) + } +} + +// Suppression of an upstream fact of the configured name is part of the merged +// body, so it survives into the cache rather than being re-applied per request. +func TestHandler_SuppressionSurvivesCacheHit(t *testing.T) { + upstream := `[` + fact("h1", defaultSourceFact, "somewhere-else", "") + `,` + fact("h1", "role", "web", "") + `]` + a := newCountingBackend(t, map[string]string{factsPath: upstream}) + b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + h := srv.Handler() + + doGet(t, h, factsPath, "") + cached := doGet(t, h, factsPath, "") + if got := cached.Header().Get(cacheStatusHeader); got != "hit" { + t.Fatalf("%s = %q, want hit", cacheStatusHeader, got) + } + srcs, n := sourceValues(t, cached.Body.Bytes(), defaultSourceFact) + if n != 1 || srcs["h1"] != "a" { + t.Errorf("cached sources = %v (%d records), want exactly one h1 -> a, the upstream value dropped", srcs, n) + } +} -- 2.47.3