From c7910156e8efc35ee171130fb49528094d5301e9 Mon Sep 17 00:00:00 2001 From: unkin-agent Date: Sat, 5 Sep 2026 21:29:34 +1000 Subject: [PATCH] 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 {