From 24f6d73d5c79b7cb6e61c9b68ad2306f7b447ca3 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sun, 13 Sep 2026 13:34:52 +1000 Subject: [PATCH] Serve fact queries live and drop the cache headers Fact answers must be as current as a backend's own, and X-Cache/Age are headers PuppetDB never sends. - serve /facts, /facts/[/] and /fact-names live on every request - keep the in-memory cache on merged /nodes only - drop X-Cache and Age everywhere; /healthz still reports cache state - answer successful queries with PuppetDB's application/json;charset=utf-8 --- README.md | 66 ++++---- cache_test.go | 369 ++++++++++++++++++++------------------------- config.go | 10 +- factroutes_test.go | 34 +++-- server.go | 88 +++++------ upstream_test.go | 24 +-- 6 files changed, 268 insertions(+), 323 deletions(-) diff --git a/README.md b/README.md index 11e2907..1ce2d23 100644 --- a/README.md +++ b/README.md @@ -60,9 +60,10 @@ exactly what backends disagree about), `408`, `429` and every `5xx` still return nothing about it is cached. Responses carry PuppetDB's `X-Records` when the query asked for a total, and on -the merged paths `X-Backends` (see [Backend health](#backend-health)). Cached -paths add two more headers `pdbmux` sets itself, `X-Cache` and `Age` — see -[Caching](#caching). +the merged paths `X-Backends` (see [Backend health](#backend-health)). Successful +query responses carry PuppetDB's own `Content-Type: application/json;charset=utf-8`. +Nothing tells a client whether a response came from the cache: PuppetDB sets no +`X-Cache` or `Age`, so neither does `pdbmux` — see [Caching](#caching). ## Merge semantics @@ -243,7 +244,7 @@ record of that name, so the route does not serve its own fan-out: it takes the records from the `/facts` merge that produces them, which makes the `certname` set, the owner and the `environment` identical to the ones an unfiltered `/facts` response reports, and lets the request's own `query` narrow the result upstream. -That costs one `/facts` fan-out per cache miss — the widest fan-out `pdbmux` +That costs one `/facts` fan-out per request — the widest fan-out `pdbmux` makes — on a rare, user-initiated path. `/facts/pdbmux_source/` pins the backend name, so it answers with the @@ -404,8 +405,8 @@ not answering. asked, never what a merged answer means: a response built from a subset is still served, as before. Every merged response carries `X-Backends: /` naming how many backends' records went into it, so - a client can tell a full answer from a partial one. On a cache hit the header - describes the stored body, not the current backend count. + a client can tell a full answer from a partial one. On a `/nodes` cache hit the + header describes the stored body, not the current backend count. - **`/healthz`** gives each backend a `state` (`healthy`, `unhealthy`, `probe_unsupported`, `unprobed`, or `unmonitored` when probing is off), `consecutive_failures`, @@ -425,21 +426,27 @@ not answering. ## Caching -`pdbmux` caches merged `/nodes`, `/facts`, `/facts/[/]` and -`/fact-names` record sets **in memory** so a busy Puppetboard does not re-fan-out -the same query every few seconds — its facts overview and fact drilldown are two -of the pages that hit hardest. Everything else runs uncached — including -`extract` aggregates on those 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. +`pdbmux` caches the merged `/nodes` record set **in memory** so a busy +Puppetboard does not re-fan-out the node list every few seconds. Its report +columns only move when a node finishes a run, so a `30s` answer is a `30s`-old +report timestamp and nothing else. + +**No fact-serving path is cached.** `/facts`, `/facts/[/]`, +`/fact-names`, `/factsets*`, `/nodes//facts`, `/fact-contents`, +`/fact-paths` and `/inventory` all fan out on every request, so a fact answer is +exactly as current as the backend's own. Cache fact data client-side if you want +it cached. Everything else runs uncached too — including `extract` aggregates on +`/nodes`, and the `/pdb/meta/v1/*` and `/metrics/*` endpoints. 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 +- **TTL** — `facts_ttl` (the key predates the cache narrowing to `/nodes`), + 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` @@ -462,29 +469,20 @@ backend later without further handler changes. 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 - stored (`0` on a `miss`). Uncached paths carry neither. +- **Invisible to clients** — a cached response is byte-for-byte a live one, with + the same headers. PuppetDB emits no `X-Cache` and no `Age`, so neither does + `pdbmux`; read the cache state off `/healthz` instead. - **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. -- **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. The one path that keys on less than - it is asked is the `pdbmux_source` drilldown, whose `` is dropped from - the key and applied to the shared entry instead. `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. +- **Merged bodies are stored, not re-merged** — an entry holds the fully merged + `/nodes` body, `pdbmux_source` already stamped on, and the + `X-Records`/`X-Backends` it was built with, so a hit reproduces all three. + Attribution is a property of that fetch, so it stays correct for as long as the + body does and ages out with it. Two requests share an entry only when they + share a key, and the key is path plus query. ## Config @@ -510,7 +508,7 @@ backends: # order is a tie-break only, not a ranking 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_ttl: 30s # merged /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 diff --git a/cache_test.go b/cache_test.go index 1f1eb25..5cc996d 100644 --- a/cache_test.go +++ b/cache_test.go @@ -126,13 +126,13 @@ func (cb *countingBackend) setBlock(ch chan struct{}) { cb.block = ch } -// newCachedServer builds a server whose facts cache runs on a fake clock. +// newCachedServer builds a server whose /nodes 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) + mc, ok := srv.nodeCache.(*memoryCache) if !ok { - t.Fatalf("expected a memory cache, got %T", srv.factsCache) + t.Fatalf("expected a memory cache, got %T", srv.nodeCache) } clk := newFakeClock() mc.now = clk.now @@ -267,14 +267,14 @@ func TestNoopCache_AlwaysMisses(t *testing.T) { } } -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: `[]`}) +func TestHandler_NodesCacheFreshHit(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)) - first := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) - second := doGet(t, srv.Handler(), factsPath, `["=","name","role"]`) + first := doGet(t, srv.Handler(), nodesPath, `["=","certname","h1"]`) + second := doGet(t, srv.Handler(), nodesPath, `["=","certname","h1"]`) if first.Code != http.StatusOK || second.Code != http.StatusOK { t.Fatalf("statuses %d/%d", first.Code, second.Code) @@ -282,40 +282,40 @@ func TestHandler_FactsCacheFreshHit(t *testing.T) { 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 { + if got := a.hitCount(nodesPath); got != 1 { t.Errorf("backend a saw %d requests, want 1 (second served from cache)", got) } - if got := b.hitCount(factsPath); got != 1 { + if got := b.hitCount(nodesPath); 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 { + doGet(t, srv.Handler(), nodesPath, `["=","certname","h2"]`) + if got := a.hitCount(nodesPath); 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: `[]`}) +func TestHandler_NodesCacheExpires(t *testing.T) { + a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) - doGet(t, srv.Handler(), factsPath, "") + doGet(t, srv.Handler(), nodesPath, "") clk.advance(31 * time.Second) - doGet(t, srv.Handler(), factsPath, "") + doGet(t, srv.Handler(), nodesPath, "") - if got := a.hitCount(factsPath); got != 2 { + if got := a.hitCount(nodesPath); 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: `[]`}) + a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("old", "2026-01-01T00:00:00.000Z") + `]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) - warm := doGet(t, srv.Handler(), factsPath, "") + warm := doGet(t, srv.Handler(), nodesPath, "") if warm.Code != http.StatusOK { t.Fatalf("warm-up status %d", warm.Code) } @@ -325,8 +325,8 @@ func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) { // 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, "") + a.setBody(nodesPath, `[`+node("new", "2026-01-01T00:00:00.000Z")+`]`) + refetch := doGet(t, srv.Handler(), nodesPath, "") if !strings.Contains(refetch.Body.String(), `"new"`) { t.Errorf("a healthy backend must not be shadowed by the stale entry: %s", refetch.Body.String()) } @@ -338,7 +338,7 @@ func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) { clk.advance(31 * time.Second) a.setFail(true) b.setFail(true) - stale := doGet(t, srv.Handler(), factsPath, "") + stale := doGet(t, srv.Handler(), nodesPath, "") if stale.Code != http.StatusOK { t.Fatalf("stale fallback status %d: %s", stale.Code, stale.Body.String()) } @@ -352,51 +352,14 @@ func TestHandler_ServesStaleOnlyWhenBackendsFail(t *testing.T) { } } -// The stored entry is the whole estate's record set and the pinned value narrows -// it per request, so the fallback has to keep narrowing: a client asking for one -// backend's records must not be handed every backend's because the entry expired. -func TestHandler_StaleSourceFactDrilldownStaysFilteredByOwner(t *testing.T) { - a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`}) - b := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h2", "osfamily", "Debian", "") + `]`}) - srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) - - drilldown := sourceFactURL + "/a" - want := map[string]string{"h1": "a"} - - warm := doGet(t, srv.Handler(), drilldown, "") - if warm.Code != http.StatusOK { - t.Fatalf("warm-up status %d: %s", warm.Code, warm.Body.String()) - } - if got, n := sourceValues(t, warm.Body.Bytes(), defaultSourceFact); n != len(want) || !reflect.DeepEqual(got, want) { - t.Fatalf("warm-up %s = %v (%d records), want %v", drilldown, got, n, want) - } - - // Every backend down and the entry expired: the stale copy is served. - clk.advance(31 * time.Second) - a.setFail(true) - b.setFail(true) - - rec := doGet(t, srv.Handler(), drilldown, "") - 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.Fatalf("%s = %q, want stale: the request did not take the fallback path", cacheStatusHeader, got) - } - got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact) - if n != len(want) || !reflect.DeepEqual(got, want) { - t.Errorf("stale %s = %v (%d records), want only backend a's %v", drilldown, got, n, want) - } -} - func TestHandler_NoCacheEntryMeansBackendFailureIs502(t *testing.T) { - a := newCountingBackend(t, map[string]string{factsPath: `[]`}) - b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + a := newCountingBackend(t, map[string]string{nodesPath: `[]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) a.setFail(true) b.setFail(true) srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) - rec := doGet(t, srv.Handler(), factsPath, "") + rec := doGet(t, srv.Handler(), nodesPath, "") if rec.Code != http.StatusBadGateway { t.Fatalf("status %d, want 502", rec.Code) } @@ -406,8 +369,8 @@ func TestHandler_NoCacheEntryMeansBackendFailureIs502(t *testing.T) { } func TestHandler_SingleFlightCollapsesConcurrentRequests(t *testing.T) { - a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) - b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) h := srv.Handler() @@ -422,13 +385,13 @@ func TestHandler_SingleFlightCollapsesConcurrentRequests(t *testing.T) { wg.Add(1) go func(i int) { defer wg.Done() - codes[i] = doGet(t, h, factsPath, "").Code + codes[i] = doGet(t, h, nodesPath, "").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 }) + waitFor(t, func() bool { return a.hitCount(nodesPath) >= 1 }) time.Sleep(250 * time.Millisecond) close(release) wg.Wait() @@ -438,10 +401,10 @@ func TestHandler_SingleFlightCollapsesConcurrentRequests(t *testing.T) { t.Fatalf("caller %d got %d", i, code) } } - if got := a.hitCount(factsPath); got != 1 { + if got := a.hitCount(nodesPath); got != 1 { t.Errorf("backend a saw %d requests, want 1 for %d concurrent callers", got, callers) } - if got := b.hitCount(factsPath); got != 1 { + if got := b.hitCount(nodesPath); got != 1 { t.Errorf("backend b saw %d requests, want 1 for %d concurrent callers", got, callers) } } @@ -557,15 +520,15 @@ func TestFlightGroup_LeaderPanicFailsLeaderAndWaiters(t *testing.T) { } func TestServeCached_BuildPanicIs502ThenStale(t *testing.T) { - a := newCountingBackend(t, map[string]string{factsPath: `[]`}) - b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + a := newCountingBackend(t, map[string]string{nodesPath: `[]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) 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) + req := httptest.NewRequest(http.MethodGet, nodesPath, nil) + srv.serveCached(rec, req, nodesPath, nil, build) return rec } @@ -575,7 +538,7 @@ func TestServeCached_BuildPanicIs502ThenStale(t *testing.T) { t.Fatalf("status %d (%s), want 502", rec.Code, rec.Body.String()) } - stored := `[` + fact("h1", "role", "web", "") + `]` + stored := `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]` rec = serve(func(context.Context) (cachedResponse, error) { return cachedResponse{Body: json.RawMessage(stored), Records: 1}, nil }) @@ -600,16 +563,16 @@ func TestServeCached_BuildPanicIs502ThenStale(t *testing.T) { } func TestServeCached_BuildPanicFailsConcurrentWaiters(t *testing.T) { - a := newCountingBackend(t, map[string]string{factsPath: `[]`}) - b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + a := newCountingBackend(t, map[string]string{nodesPath: `[]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) entered := make(chan struct{}) release := make(chan struct{}) 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) + req := httptest.NewRequest(http.MethodGet, nodesPath, nil) + srv.serveCached(rec, req, nodesPath, nil, build) return rec } @@ -710,8 +673,8 @@ func TestHandler_LeaderDisconnectDoesNotFailFollowers(t *testing.T) { // 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 := strings.TrimSpace(warmed.Body.String()); !sameJSON(t, got, want) { + t.Errorf("warmed body = %s, want %s: the abandoned leader's flight must still warm the cache", got, want) } if got := a.hitCount(nodesPath); got != 1 { t.Errorf("backend a saw %d requests after the cached read, want 1", got) @@ -750,7 +713,7 @@ func TestHandler_SoloDisconnectAbortsFanOutPromptly(t *testing.T) { }() start := time.Now() - req := httptest.NewRequest(http.MethodGet, factsPath, nil).WithContext(ctx) + req := httptest.NewRequest(http.MethodGet, nodesPath, 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) @@ -783,7 +746,7 @@ func TestHandler_DisconnectedRequestsDoNotPinBackends(t *testing.T) { time.Sleep(disconnectAfter) cancel() }() - target := factsPath + "?query=" + url.QueryEscape(strconv.Itoa(i)) + target := nodesPath + "?query=" + url.QueryEscape(strconv.Itoa(i)) req := httptest.NewRequest(http.MethodGet, target, nil).WithContext(ctx) h.ServeHTTP(httptest.NewRecorder(), req) }(i) @@ -1116,47 +1079,39 @@ func assertGoroutinesSettle(t *testing.T, baseline int, frames ...string) { } } -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: `[]`}) +// PuppetDB emits no X-Cache and no Age, so a client cannot tell a cached +// response from a live one. Covers the fresh hit, the miss and the stale +// fallback, which are the three ways a response can leave the cached path. +func TestHandler_CachedResponsesCarryNoCacheHeaders(t *testing.T) { + a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) 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) - } + assertNoCacheHeaders(t, doGet(t, h, nodesPath, ""), "miss") 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 got := rec.Header().Get(ageHeader); got != "7" { - t.Errorf("cached request %s = %q, want 7", ageHeader, got) - } + assertNoCacheHeaders(t, doGet(t, h, nodesPath, ""), "fresh hit") - // Past the TTL with every backend down, the stale fallback must say so. clk.advance(24 * time.Second) a.setFail(true) b.setFail(true) - rec = doGet(t, h, factsPath, "") + rec := doGet(t, h, nodesPath, "") 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 serving, _, _ := srv.stale.snapshot(); !serving { + t.Fatal("the third request did not take the stale fallback path") } - if got := rec.Header().Get(ageHeader); got != "31" { - t.Errorf("stale fallback %s = %q, want 31 seconds since the entry was stored", ageHeader, got) - } - 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) + assertNoCacheHeaders(t, rec, "stale fallback") +} + +func assertNoCacheHeaders(t *testing.T, rec *httptest.ResponseRecorder, what string) { + t.Helper() + for _, h := range []string{"X-Cache", "Age"} { + if got := rec.Header().Get(h); got != "" { + t.Errorf("%s response carries %s: %q, want no such header", what, h, got) + } } } @@ -1193,18 +1148,18 @@ func (c *recordingCache) Stats() CacheStats { return CacheStats{Backend: "record func TestServeCached_PutRunsOnDetachedContext(t *testing.T) { srv := newTestServer(cacheTestConfig("http://backend.invalid", "http://backend.invalid")) cache := newRecordingCache() - srv.factsCache = cache + srv.nodeCache = 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) + req := httptest.NewRequest(http.MethodGet, nodesPath, nil).WithContext(ctx) served := make(chan struct{}) go func() { defer close(served) - srv.serveCached(httptest.NewRecorder(), req, factsPath, nil, func(context.Context) (cachedResponse, error) { + srv.serveCached(httptest.NewRecorder(), req, nodesPath, nil, func(context.Context) (cachedResponse, error) { close(entered) <-release return cachedResponse{Body: json.RawMessage(`[]`), Records: -1}, nil @@ -1236,8 +1191,8 @@ func TestServeCached_PutRunsOnDetachedContext(t *testing.T) { } func TestHandler_HealthzReportsCacheState(t *testing.T) { - a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "role", "web", "") + `]`}) - b := newCountingBackend(t, map[string]string{factsPath: `[]`}) + a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) hr := health(t, srv) @@ -1248,7 +1203,7 @@ func TestHandler_HealthzReportsCacheState(t *testing.T) { t.Fatalf("a cold cache should be empty and not stale: %+v", hr.Cache) } - doGet(t, srv.Handler(), factsPath, "") + doGet(t, srv.Handler(), nodesPath, "") 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) @@ -1264,7 +1219,7 @@ func TestHandler_HealthzReportsCacheState(t *testing.T) { a.setFail(true) b.setFail(true) - doGet(t, srv.Handler(), factsPath, "") + doGet(t, srv.Handler(), nodesPath, "") 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) @@ -1444,9 +1399,9 @@ func TestNewServer_ClampsFactsTTL(t *testing.T) { cfg.CacheBytes = 1 << 20 srv := newTestServer(cfg) - mc, ok := srv.factsCache.(*memoryCache) + mc, ok := srv.nodeCache.(*memoryCache) if !ok { - t.Fatalf("expected a memory cache, got %T", srv.factsCache) + t.Fatalf("expected a memory cache, got %T", srv.nodeCache) } if mc.ttl != maxFactsTTL { t.Errorf("cache ttl = %s, want %s", mc.ttl, maxFactsTTL) @@ -1520,35 +1475,6 @@ func stamped(t *testing.T, raw, field, value string) string { // 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") + `]`}) @@ -1561,9 +1487,6 @@ func TestHandler_CachedNodesKeepSourceStamp(t *testing.T) { } 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) @@ -1577,77 +1500,115 @@ func TestHandler_CachedNodesKeepSourceStamp(t *testing.T) { // 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: `[]`}) + a := newCountingBackend(t, map[string]string{nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`}) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`}) srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) h := srv.Handler() - doGet(t, h, factsPath, "") + doGet(t, h, nodesPath, "") // The node moves to b while the entry is still fresh. - a.setBody(factsPath, `[]`) - b.setBody(factsPath, `[`+fact("h1", "role", "web", "")+`]`) + a.setBody(nodesPath, `[]`) + b.setBody(nodesPath, `[`+node("h1", "2026-01-01T00:00:00.000Z")+`]`) - cached := doGet(t, h, factsPath, "") - if got := cached.Header().Get(cacheStatusHeader); got != "hit" { - t.Fatalf("%s = %q, want hit", cacheStatusHeader, got) + cached := doGet(t, h, nodesPath, "") + if got := nodeSources(t, cached.Body.Bytes(), defaultSourceFact); got["h1"] != "a" { + t.Errorf("cached stamp = %v, want h1 -> a: the body and its attribution come from the same fetch", 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) + if got := a.hitCount(nodesPath); got != 1 { + t.Errorf("backend a saw %d requests, want 1 while the entry is fresh", got) } 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) + rebuilt := doGet(t, h, nodesPath, "") + if got := nodeSources(t, rebuilt.Body.Bytes(), defaultSourceFact); got["h1"] != "b" { + t.Errorf("rebuilt stamp = %v, want h1 -> b once the entry expired", got) } } -// 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) { +// Fact answers must be as current as a backend's own, so every fact-serving +// route refetches: the second request sees data that changed between the two. +func TestHandler_FactPathsAreNeverCached(t *testing.T) { + for _, path := range []string{ + factsPath, + factsPath + "/role", + factsPath + "/role/web", + factNamesPath, + factsetsPath, + nodesPath + "/h1/facts", + "/pdb/query/v4/fact-contents", + "/pdb/query/v4/fact-paths", + "/pdb/query/v4/inventory", + } { + t.Run(path, func(t *testing.T) { + before := `[{"certname":"h1","name":"role","value":"before"}]` + after := `[{"certname":"h1","name":"role","value":"after"}]` + a := newCountingBackend(t, map[string]string{path: before}) + b := newCountingBackend(t, map[string]string{path: `[]`}) + srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) + h := srv.Handler() + + if first := doGet(t, h, path, ""); !strings.Contains(first.Body.String(), "before") { + t.Fatalf("first %s = %s, want the backend's own records", path, first.Body.String()) + } + a.setBody(path, after) + + second := doGet(t, h, path, "") + if strings.Contains(second.Body.String(), "before") { + t.Errorf("second %s = %s, want the changed data rather than a cached copy", path, second.Body.String()) + } + if !strings.Contains(second.Body.String(), "after") { + t.Errorf("second %s = %s, want the changed data", path, second.Body.String()) + } + if got := a.hitCount(path); got != 2 { + t.Errorf("backend a saw %d requests for %s, want one per request", got, path) + } + }) + } +} + +// No fact response may carry the cache headers PuppetDB never sends, whatever +// else pdbmux is doing with the merged body. +func TestHandler_FactResponsesCarryNoCacheHeaders(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() + for _, path := range []string{factsPath, factsPath + "/role", factNamesPath, sourceFactURL} { + t.Run(path, func(t *testing.T) { + a := newCountingBackend(t, map[string]string{path: body, factsPath: body}) + b := newCountingBackend(t, map[string]string{path: `[]`, 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) + assertNoCacheHeaders(t, doGet(t, h, path, ""), path) + assertNoCacheHeaders(t, doGet(t, h, path, ""), path+" repeated") + }) } } -// 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: `[]`}) +// openvoxdb answers a successful query with application/json;charset=utf-8 +// (src/puppetlabs/puppetdb/query_eng.clj:555 through http.clj:80), so a merged +// response has to as well or a client can tell the proxy from the real thing. +func TestHandler_SuccessContentTypeMatchesPuppetDB(t *testing.T) { + bodies := map[string]string{ + nodesPath: `[` + node("h1", "2026-01-01T00:00:00.000Z") + `]`, + factsPath: `[` + fact("h1", "role", "web", "") + `]`, + factNamesPath: `["role"]`, + reportsPath: `[` + report("h1", "abc", "2026-01-01T00:00:00.000Z") + `]`, + } + a := newCountingBackend(t, bodies) + b := newCountingBackend(t, map[string]string{nodesPath: `[]`, factsPath: `[]`, factNamesPath: `[]`, reportsPath: `[]`}) 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) + for path := range bodies { + // Twice, so a cache hit is held to the same content type as a live build. + for _, pass := range []string{"live", "repeated"} { + rec := doGet(t, h, path, "") + if rec.Code != http.StatusOK { + t.Fatalf("%s %s status %d (%s)", pass, path, rec.Code, rec.Body.String()) + } + if got := rec.Header().Get("Content-Type"); got != "application/json;charset=utf-8" { + t.Errorf("%s %s Content-Type = %q, want application/json;charset=utf-8", pass, path, got) + } + } } } diff --git a/config.go b/config.go index 1666ffd..f7508c2 100644 --- a/config.go +++ b/config.go @@ -59,7 +59,7 @@ 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 + FactsTTL time.Duration `yaml:"facts_ttl"` // 0 disables the merged /nodes cache CacheBytes int64 `yaml:"facts_cache_bytes"` // byte budget for that cache SourceFact string `yaml:"source_fact"` @@ -344,8 +344,8 @@ func (c Config) Validate() error { return nil } -// cacheEnabled reports whether a facts/nodes cache should be built: both a TTL -// and a byte budget are required. +// cacheEnabled reports whether the merged /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 { @@ -366,8 +366,8 @@ func writeDefaultConfig(path string) error { "# PDBMUX_HEALTH_PROBE_ENABLED, PDBMUX_HEALTH_PROBE_PATH,\n" + "# PDBMUX_HEALTH_PROBE_INTERVAL, PDBMUX_HEALTH_PROBE_TIMEOUT,\n" + "# PDBMUX_HEALTH_PROBE_FAILURES, PDBMUX_HEALTH_PROBE_SUCCESSES.\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" + + "# facts_ttl caches merged /nodes in memory; it is capped at 30s (a larger\n" + + "# value is clamped) and 0 disables the cache. Fact answers are never cached.\n" + "# health_probe_* polls each backend's status endpoint so queries skip a\n" + "# backend that is down; when every backend is down all are queried anyway.\n\n") if err := os.WriteFile(path, append(header, data...), 0o644); err != nil { diff --git a/factroutes_test.go b/factroutes_test.go index 23de0bc..2648eb8 100644 --- a/factroutes_test.go +++ b/factroutes_test.go @@ -370,9 +370,9 @@ func TestHandler_FactsBySourceNameUnknownValueSkipsFanOut(t *testing.T) { } } -// The record set is the estate's, not the pinned value's, so every value shares -// one entry: two valid values must not cost two whole-estate fan-outs. -func TestHandler_FactsBySourceNameValuesShareOneFetch(t *testing.T) { +// The record set is the estate's, not the pinned value's: the segment is +// applied to the merged records rather than fanned out as a path of its own. +func TestHandler_FactsBySourceNameValuesFilterOneFetch(t *testing.T) { a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`}) b := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h2", "osfamily", "Debian", "") + `]`}) srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) @@ -392,8 +392,8 @@ func TestHandler_FactsBySourceNameValuesShareOneFetch(t *testing.T) { } } for _, cb := range []*countingBackend{a, b} { - if got := cb.hitCount(factsPath); got != 1 { - t.Errorf("%s fetched %d times, want 1 shared fetch", factsPath, got) + if got := cb.hitCount(factsPath); got != 3 { + t.Errorf("%s fetched %d times, want one uncached fetch per request", factsPath, got) } if got := cb.hitCount(sourceFactURL); got != 0 { t.Errorf("%s was fanned out %d times, want 0", sourceFactURL, got) @@ -678,21 +678,27 @@ func TestHandler_FactNamesSurvivesOneBackend(t *testing.T) { } } -// Both routes are merged record sets, so they use the same cache /facts does. -func TestHandler_FactRoutesAreCached(t *testing.T) { +// Every route that serves fact records fans out per request, even with the +// cache configured: a fact answer is only ever as old as the backend's own. +func TestHandler_FactRoutesAreNotCached(t *testing.T) { for _, path := range []string{roleFactPath, roleFactPath + "/web", sourceFactURL, factNamesPath} { t.Run(path, func(t *testing.T) { + // The source-fact drilldown is answered from the /facts merge, so + // that is the path its fan-out lands on. + fanOut := path + if path == sourceFactURL { + fanOut = factsPath + } a := newFakeBackend(t, `[]`, `[]`) - a.bodies[path] = `[]` + a.bodies[fanOut] = `[]` b := newFakeBackend(t, `[]`, `[]`) - b.bodies[path] = `[]` + b.bodies[fanOut] = `[]` srv := newTestServer(cacheTestConfig(a.srv.URL, b.srv.URL)) - if got := doGet(t, srv.Handler(), path, "").Header().Get(cacheStatusHeader); got != "miss" { - t.Errorf("first %s = %q, want miss", cacheStatusHeader, got) - } - if got := doGet(t, srv.Handler(), path, "").Header().Get(cacheStatusHeader); got != "hit" { - t.Errorf("second %s = %q, want hit", cacheStatusHeader, got) + doGet(t, srv.Handler(), path, "") + doGet(t, srv.Handler(), path, "") + if got := a.hits(fanOut); got != 2 { + t.Errorf("backend a saw %d requests for %s, want one per request", got, fanOut) } }) } diff --git a/server.go b/server.go index 6908d6d..e705996 100644 --- a/server.go +++ b/server.go @@ -32,13 +32,13 @@ 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" - // Set by pdbmux: "/" backends behind a merged response. backendsHeader = "X-Backends" + + // What openvoxdb answers a successful query with, byte for byte + // (src/puppetlabs/puppetdb/query_eng.clj:555 through http.clj:80, normalised + // by Jetty), so a merged response is indistinguishable from a backend's own. + jsonContentType = "application/json;charset=utf-8" ) type backendResult struct { @@ -55,16 +55,17 @@ type Server struct { 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 + // nodeCache is nil when caching is disabled; cacheFor hands out a noop then. + nodeCache Cache + flights flightGroup + stale staleTracker // health is nil when probing is disabled, which makes every backend healthy. health *prober partial partialTracker - // now is shared with the cache's clock so Age matches the stored timestamp. + // now is shared with the cache's clock so a stale fallback is reported + // against the same timestamps the cache stamps entries with. now func() time.Time // freshness cache (freshness merge only). @@ -82,7 +83,7 @@ func NewServer(cfg Config, logger *log.Logger) *Server { now: time.Now, } if cfg.cacheEnabled() { - s.factsCache = newMemoryCache(cfg.FactsTTL, cfg.CacheBytes) + s.nodeCache = newMemoryCache(cfg.FactsTTL, cfg.CacheBytes) } s.health = newProber(cfg, logger) return s @@ -95,20 +96,21 @@ func (s *Server) StartProbes(ctx context.Context) { s.health.Start(ctx) } // StopProbes stops the probing goroutines and waits for them to exit. func (s *Server) StopProbes() { s.health.Stop() } -// cacheFor picks the cache backing a request. The merged /nodes and fact record -// sets — /facts, /facts/[/] and /fact-names — 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. +// cacheFor picks the cache backing a request. Only the merged /nodes record set +// is cached: its report columns move once per node run, while fact answers must +// be as current as a backend's own, so no fact-serving path is cached here. +// 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 { - case path == factsPath, path == nodesPath, path == factNamesPath, isFactsSubPath(path): + switch path { + case nodesPath: // An aggregate row is a combined count, not the merged record set the // cache was built for, so it stays on the live path. if spec, err := parseAggregate(params.Get("query")); spec != nil || err != nil { return noopCache{}, false } - if s.factsCache != nil { - return s.factsCache, true + if s.nodeCache != nil { + return s.nodeCache, true } } return noopCache{}, false @@ -584,10 +586,9 @@ type cachedResponse struct { Configured int `json:"configured"` // backends configured at build time } -// recordFilter narrows a response's records after it has been built or read back -// from the cache, so requests differing only in the filter share one stored entry -// and one fan-out. It leaves Records alone, so it only suits responses that set -// no X-Records. +// recordFilter narrows a response's records after it has been built, so a route +// that answers from another path's merge can pin part of it per request. It +// leaves Records alone, so it only suits responses that set no X-Records. type recordFilter func([]json.RawMessage) []json.RawMessage func (f recordFilter) apply(resp cachedResponse) cachedResponse { @@ -633,7 +634,7 @@ func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path stri s.log.Printf("warning: cache lookup for %s failed: %v", key, err) case status == CacheFresh: s.stale.markFresh() - s.writeStored(w, ent, CacheFresh, filter) + s.writeStored(w, ent, filter) return case status == CacheStale: stale = &ent @@ -676,14 +677,13 @@ func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path stri 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, filter) + s.writeStored(w, *stale, filter) return } s.writeUpstreamError(w, err) return } s.stale.markFresh() - s.setCacheHeaders(w, CacheMiss, time.Time{}) writeCached(w, filter.apply(resp)) } @@ -696,41 +696,21 @@ func (s *Server) flightTimeout() time.Duration { return defaultTimeout } -func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status CacheStatus, filter recordFilter) { +// writeStored answers from a cache entry. Whether a response came from the cache +// is deliberately invisible to the client: PuppetDB sets no X-Cache or Age, so +// neither does pdbmux, and /healthz carries the cache state instead. +func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, filter recordFilter) { var resp cachedResponse 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 } - s.setCacheHeaders(w, status, ent.StoredAt) writeCached(w, filter.apply(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). 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: - label = "hit" - case CacheStale: - label = "stale" - } - age := 0 - if !storedAt.IsZero() { - if secs := int(s.now().Sub(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") + w.Header().Set("Content-Type", jsonContentType) if resp.Records >= 0 { w.Header().Set(recordsHeader, strconv.Itoa(resp.Records)) } @@ -1037,8 +1017,8 @@ type cacheHealth struct { func (s *Server) cacheHealth() cacheHealth { stats := CacheStats{Backend: "none"} ttl := time.Duration(0) - if s.factsCache != nil { - stats = s.factsCache.Stats() + if s.nodeCache != nil { + stats = s.nodeCache.Stats() ttl = s.cfg.FactsTTL } serving, served, last := s.stale.snapshot() @@ -1106,7 +1086,7 @@ func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) { } func writeJSON(w http.ResponseWriter, recs []json.RawMessage) { - w.Header().Set("Content-Type", "application/json") + w.Header().Set("Content-Type", jsonContentType) if recs == nil { recs = []json.RawMessage{} } diff --git a/upstream_test.go b/upstream_test.go index 58ecd74..a392b68 100644 --- a/upstream_test.go +++ b/upstream_test.go @@ -296,18 +296,18 @@ func TestHandler_RejectedQueryIsNotCached(t *testing.T) { b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) - rec := doGet(t, srv.Handler(), factsPath, "") + rec := doGet(t, srv.Handler(), nodesPath, "") if rec.Code != http.StatusBadRequest { t.Fatalf("status = %d, want the upstream 400 replayed: %s", rec.Code, rec.Body.String()) } - if n := srv.factsCache.Stats().Entries; n != 0 { + if n := srv.nodeCache.Stats().Entries; n != 0 { t.Fatalf("cache holds %d entries after a rejected query", n) } // The rejection left nothing behind, so the next good query is served fresh. a.reject, b.reject = 0, 0 - a.factsBody = `[` + fact("h1", "role", "web", "") + `]` - ok := doGet(t, srv.Handler(), factsPath, "") + a.nodesBody = `[` + node("web", "2026-01-01T00:00:00.000Z") + `]` + ok := doGet(t, srv.Handler(), nodesPath, "") if ok.Code != http.StatusOK || !strings.Contains(ok.Body.String(), "web") { t.Fatalf("follow-up = %d %s", ok.Code, ok.Body.String()) } @@ -319,10 +319,10 @@ func TestHandler_AllBackendsFailedIsNotCached(t *testing.T) { a.fail, b.fail = true, true srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) - if rec := doGet(t, srv.Handler(), factsPath, ""); rec.Code != http.StatusBadGateway { + if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway { t.Fatalf("status = %d, want 502", rec.Code) } - if n := srv.factsCache.Stats().Entries; n != 0 { + if n := srv.nodeCache.Stats().Entries; n != 0 { t.Fatalf("cache holds %d entries after a failed fan-out", n) } } @@ -330,18 +330,18 @@ func TestHandler_AllBackendsFailedIsNotCached(t *testing.T) { // Stale records answer an outage. They do not answer a query the estate refused: // the client has to be told why, not handed data for a question it did not ask. func TestHandler_RejectedQueryIsNotAnsweredFromStale(t *testing.T) { - a := newFakeBackend(t, `[]`, `[`+fact("h1", "role", "old", "")+`]`) + a := newFakeBackend(t, `[`+node("old", "2026-01-01T00:00:00.000Z")+`]`, `[]`) b := newFakeBackend(t, `[]`, `[]`) srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) - if warm := doGet(t, srv.Handler(), factsPath, ""); warm.Code != http.StatusOK { + if warm := doGet(t, srv.Handler(), nodesPath, ""); warm.Code != http.StatusOK { t.Fatalf("warm-up status %d", warm.Code) } clk.advance(31 * time.Second) a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy - rec := doGet(t, srv.Handler(), factsPath, "") + rec := doGet(t, srv.Handler(), nodesPath, "") if rec.Code != http.StatusBadRequest { t.Fatalf("status = %d, want the rejection rather than the stale entry: %s", rec.Code, rec.Body.String()) } @@ -355,16 +355,16 @@ func TestHandler_RejectedQueryIsNotAnsweredFromStale(t *testing.T) { // An outage still falls back to the stale copy, unchanged by the replay path. func TestHandler_OutageStillFallsBackToStale(t *testing.T) { - a := newFakeBackend(t, `[]`, `[`+fact("h1", "role", "old", "")+`]`) + a := newFakeBackend(t, `[`+node("old", "2026-01-01T00:00:00.000Z")+`]`, `[]`) b := newFakeBackend(t, `[]`, `[]`) srv, clk := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL)) - if warm := doGet(t, srv.Handler(), factsPath, ""); warm.Code != http.StatusOK { + if warm := doGet(t, srv.Handler(), nodesPath, ""); warm.Code != http.StatusOK { t.Fatalf("warm-up status %d", warm.Code) } clk.advance(31 * time.Second) a.fail, b.fail = true, true - rec := doGet(t, srv.Handler(), factsPath, "") + rec := doGet(t, srv.Handler(), nodesPath, "") if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "old") { t.Fatalf("stale fallback = %d %s", rec.Code, rec.Body.String()) } -- 2.47.3