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

@@ -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
@@ -170,7 +174,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

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: `[]`})

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 {