Merge pull request 'feat: cache merged /facts and /nodes in memory, stale on backend failure' (#12) from benvin/cache-facts into main

Reviewed-on: #12

README.md

@@ -38,13 +38,17 @@ not PQL) is forwarded verbatim.
 | `GET /metrics/v2/read/<mbean>` | 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[/<mbean>]` | 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
 **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
@@ -208,6 +212,64 @@ 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** — `<path>?<params>`, 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. 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. 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. 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.
+- **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. `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
 
 Precedence (lowest → highest): **defaults < config file < env vars (`PDBMUX_*`) < flags**.
@@ -229,9 +291,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 +310,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 +347,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.

cache.go

@@ -0,0 +1,328 @@
+package main
+
+import (
+	"container/list"
+	"context"
+	"errors"
+	"fmt"
+	"net/url"
+	"runtime/debug"
+	"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.
+//
+// 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)
+	Put(ctx context.Context, key string, body []byte) error
+	Stats() CacheStats
+}
+
+// cacheKey is the request path when there are no params, else "<path>?<params>"
+// 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 {
+	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.
+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.
+//
+// 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)
+	}
+	c, shared := g.calls[key]
+	if shared {
+		c.participants++
+		g.mu.Unlock()
+	} 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)
+	}
+	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()}
+		}
+		// 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.forget(key, c)
+		close(c.done)
+	}()
+
+	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
+// 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
+}

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)
 	}

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, "")
 	}
 }

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)

server.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"log"
@@ -26,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 {
@@ -35,11 +41,21 @@ 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
+
+	// 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
@@ -47,11 +63,36 @@ 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,
+		now:    time.Now,
 	}
+	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 +150,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(ctx context.Context) (cachedResponse, error) {
+		alive, err := s.aliveResults(ctx, 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 +169,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(ctx context.Context) (cachedResponse, error) {
+		alive, err := s.aliveResults(ctx, 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 +224,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(ctx context.Context) (cachedResponse, error) {
+		alive, err := s.aliveResults(ctx, 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 +264,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 +277,161 @@ 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(context.Context) (cachedResponse, error)) {
+	cache, enabled := s.cacheFor(path, params)
+	if !enabled {
+		resp, err := build(r.Context())
+		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, CacheFresh)
+		return
+	case status == CacheStale:
+		stale = &ent
+	}
+
+	// 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
+		}
+		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 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
+	})
+	if err != nil {
+		// 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(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)
+			return
+		}
+		http.Error(w, err.Error(), http.StatusBadGateway)
+		return
+	}
+	s.stale.markFresh()
+	s.setCacheHeaders(w, CacheMiss, time.Time{})
+	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, ent CacheEntry, status CacheStatus) {
+	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, 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")
+	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 +607,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 {