Replay a unanimous upstream rejection instead of a 502

Every backend gets the same query, so one they all refuse is the client's
mistake; flattening it into "all backends failed" threw openvoxdb's own
explanation away and logged a typo as an outage.

- Carry status, content type and body on a typed upstreamError
- Replay the status and explanation when every backend refuses alike
- Redact backend addresses from replayed bodies
- Keep a refused query out of the partial counters and the cache

README.md

@@ -48,6 +48,16 @@ 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.
 
+Every backend is asked the same question, so a query all of them *refuse* with
+the same client-shaped status — a `400` naming an unknown field, say — is the
+query's fault rather than an outage: that status and openvoxdb's own explanation
+are replayed to the client instead of a `502`, with any backend address stripped
+out of the body first. Backends disagreeing on the status, a `403` (`pdbmux`'s
+own credentials, not the client's), a `404` (which records a backend holds is
+exactly what backends disagree about), `408`, `429` and every `5xx` still return
+`502`. A refused query is not counted as a partial round on `/healthz`, and
+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
@@ -435,7 +445,8 @@ backend later without further handler changes.
 - **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.
+  healthy backend is never shadowed by a stale entry, and a query every backend
+  refuses is answered with the refusal rather than the stale copy.
 - **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

e2e_backend_test.go

@@ -342,6 +342,14 @@ func (b *backend) query(ctx context.Context, t fatalf, path string, params url.V
 
 // queryStatus is query without the body, for asserting what a backend rejects.
 func (b *backend) queryStatus(ctx context.Context, t fatalf, path string, params url.Values) int {
+	t.Helper()
+	status, _ := b.queryRaw(ctx, t, path, params)
+	return status
+}
+
+// queryRaw returns a backend's own status and body, so a test can compare what
+// pdbmux served against what the backend actually said.
+func (b *backend) queryRaw(ctx context.Context, t fatalf, path string, params url.Values) (int, []byte) {
 	t.Helper()
 	target := b.url + path
 	if len(params) > 0 {
@@ -356,8 +364,11 @@ func (b *backend) queryStatus(ctx context.Context, t fatalf, path string, params
 		t.Fatalf("querying %s%s: %v", b.name, path, err)
 	}
 	defer func() { _ = resp.Body.Close() }()
-	_, _ = io.Copy(io.Discard, resp.Body)
-	return resp.StatusCode
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		t.Fatalf("reading %s%s: %v", b.name, path, err)
+	}
+	return resp.StatusCode, body
 }
 
 func newNetwork(ctx context.Context, t fatalf) (string, func()) {

e2e_upstream_test.go

@@ -0,0 +1,58 @@
+//go:build e2e
+
+package main
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+// Every backend gets the same query, so a query none of them can answer is the
+// client's mistake rather than an outage. openvoxdb explains what is wrong with
+// it; pdbmux has to hand that explanation and its status back instead of a 502
+// saying the estate is down.
+func TestRejectedQuerySurfacesUpstreamStatus(t *testing.T) {
+	ctx := context.Background()
+	for _, tc := range []struct{ name, path, q string }{
+		{"merged", nodesPath, `["=","bogus","x"]`},
+		{"combined", nodesPath, `["extract",[["function","count"]],["=","bogus","x"]]`},
+		{"union", reportsPath, `["=","bogus","x"]`},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			params := query(tc.q)
+			status, upstream := h.a.queryRaw(ctx, t, tc.path, params)
+			if !clientShaped(status) {
+				t.Fatalf("backend %s answered HTTP %d for %s, which is not the client error this test needs: %s",
+					h.a.name, status, tc.q, upstream)
+			}
+			if other, _ := h.b.queryRaw(ctx, t, tc.path, params); other != status {
+				t.Fatalf("backends answered %d and %d, so the rejection is not unanimous", status, other)
+			}
+
+			_, before := healthz(t)
+			resp := rawGet(t, tc.path, params)
+			if resp.status != status {
+				t.Fatalf("pdbmux answered HTTP %d, want the upstream %d: %s", resp.status, status, resp.body)
+			}
+			if got, want := strings.TrimSpace(string(resp.body)), strings.TrimSpace(string(upstream)); got != want {
+				t.Errorf("pdbmux body = %q, want openvoxdb's own explanation %q", got, want)
+			}
+			for _, b := range []*backend{h.a, h.b} {
+				if strings.Contains(string(resp.body), b.url) {
+					t.Errorf("replayed body names backend %s: %q", b.name, resp.body)
+				}
+			}
+
+			// A refused query is not degraded service, so it must leave the merged
+			// fan-out's own health counters where it found them.
+			_, after := healthz(t)
+			if after.Query.PartialRounds != before.Query.PartialRounds || after.Query.Partial {
+				t.Errorf("query health = %+v, want %+v unchanged by a refused query", after.Query, before.Query)
+			}
+			if after.Status != "ok" {
+				t.Errorf("/healthz = %q after a refused query, want ok", after.Status)
+			}
+		})
+	}
+}

server.go

@@ -521,6 +521,10 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) {
 		alive = append(alive, res)
 	}
 	if len(alive) == 0 {
+		if ue := unanimousClientError(results); ue != nil {
+			s.writeUpstreamError(w, ue)
+			return
+		}
 		http.Error(w, "no backend holds this report", http.StatusNotFound)
 		return
 	}
@@ -533,17 +537,29 @@ func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) {
 	writeJSON(w, nil)
 }
 
-// Returns errAllBackendsFailed only when every backend failed.
+// Returns an *upstreamError when every backend refused the query the same way,
+// and errAllBackendsFailed when every backend failed for any other reason.
 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 {
+		if res.err == nil {
+			alive = append(alive, res)
+		}
+	}
+	if len(alive) == 0 {
+		// A query every backend refuses identically is the client's mistake, not an
+		// outage, so it is neither logged as one nor counted as degraded service.
+		if ue := unanimousClientError(results); ue != nil {
+			s.log.Printf("info: every backend refused %s: %v", path, ue)
+			return nil, ue
+		}
+	}
 	for _, res := range results {
 		if res.err != nil {
 			s.log.Printf("warning: backend %q failed for %s: %v", res.name, path, res.err)
-			continue
 		}
-		alive = append(alive, res)
 	}
 	s.partial.record(len(alive), len(s.cfg.Backends), s.now())
 	if len(alive) == 0 {
@@ -596,7 +612,7 @@ func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path stri
 	if !enabled {
 		resp, err := build(r.Context())
 		if err != nil {
-			http.Error(w, err.Error(), http.StatusBadGateway)
+			s.writeUpstreamError(w, err)
 			return
 		}
 		writeCached(w, filter.apply(resp))
@@ -648,14 +664,16 @@ func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path stri
 		if errors.Is(err, errFlightAbandoned) {
 			return
 		}
-		if stale != nil {
+		// A refused query is answered, not degraded, so stale records are no reply
+		// to it: the client has to see why the query was rejected.
+		if stale != nil && !errors.As(err, new(*upstreamError)) {
 			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)
 			return
 		}
-		http.Error(w, err.Error(), http.StatusBadGateway)
+		s.writeUpstreamError(w, err)
 		return
 	}
 	s.stale.markFresh()
@@ -877,7 +895,7 @@ func (s *Server) queryBackend(ctx context.Context, b Backend, path string, param
 		return nil, -1, err
 	}
 	if resp.StatusCode != http.StatusOK {
-		return nil, -1, fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
+		return nil, -1, newUpstreamError(resp.StatusCode, resp.Header.Get("Content-Type"), body)
 	}
 	total := -1
 	if n, err := strconv.Atoi(resp.Header.Get(recordsHeader)); err == nil && n >= 0 {
@@ -919,7 +937,7 @@ func (s *Server) proxyUnmerged(w http.ResponseWriter, r *http.Request) {
 	}
 	setContentType(w, fallback.contentType)
 	w.WriteHeader(fallback.status)
-	_, _ = w.Write(fallback.body)
+	_, _ = w.Write(s.redactBackends(fallback.body))
 }
 
 type bufferedResponse struct {

server_test.go

@@ -31,6 +31,10 @@ type fakeBackend struct {
 	totals map[string]int
 	fail   bool          // return 500 for everything
 	delay  time.Duration // artificial latency
+	// reject answers every request with this status and rejectBody, standing in
+	// for a PuppetDB refusing a query it cannot answer.
+	reject     int
+	rejectBody string
 
 	mu        sync.Mutex
 	gotParams map[string]url.Values
@@ -60,6 +64,10 @@ func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend {
 			http.Error(w, "boom", http.StatusInternalServerError)
 			return
 		}
+		if fb.reject != 0 {
+			http.Error(w, fb.rejectBody, fb.reject)
+			return
+		}
 		if body, ok := fb.bodies[r.URL.Path]; ok {
 			if n, ok := fb.totals[r.URL.Path]; ok && r.URL.Query().Get("include_total") == "true" {
 				w.Header().Set(recordsHeader, strconv.Itoa(n))

upstream.go

@@ -0,0 +1,134 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"strings"
+)
+
+// upstreamBodyLimit caps how much of a backend's error body is kept for replay.
+const upstreamBodyLimit = 8 << 10
+
+// redactedBackend stands in for anything naming a backend in a replayed body.
+const redactedBackend = "<backend>"
+
+// upstreamError is a non-2xx reply from a backend kept whole — status, content
+// type and body — so a rejection the backend explained can be replayed to the
+// client instead of collapsed into a gateway error that throws the explanation
+// away.
+type upstreamError struct {
+	status      int
+	contentType string
+	body        []byte
+}
+
+func newUpstreamError(status int, contentType string, body []byte) *upstreamError {
+	if len(body) > upstreamBodyLimit {
+		body = body[:upstreamBodyLimit]
+	}
+	return &upstreamError{status: status, contentType: contentType, body: body}
+}
+
+func (e *upstreamError) Error() string {
+	return fmt.Sprintf("HTTP %d: %s", e.status, strings.TrimSpace(string(e.body)))
+}
+
+// clientShaped reports whether a status blames the request rather than the
+// backend or pdbmux itself. openvoxdb answers every bad query with 400 —
+// src/puppetlabs/puppetdb/middleware.clj:98-116 and http.clj:115-129 — and its
+// other 4xx say something a client of pdbmux neither caused nor can fix: 403 is
+// pdbmux's own certificate being refused (middleware.clj:44-58); 404 is either
+// an absent object or an unknown path (http.clj:238-242,
+// middleware.clj:381-398), and which objects a backend holds is the one thing
+// backends are meant to disagree about; 408 and 429 report a backend's timing
+// and capacity. A backend that times out mid-query answers 200 with a truncated
+// body rather than any 4xx (query_eng.clj:463-483), so no timeout reaches here.
+func clientShaped(status int) bool {
+	switch status {
+	case http.StatusForbidden, http.StatusNotFound,
+		http.StatusRequestTimeout, http.StatusTooManyRequests:
+		return false
+	}
+	return status >= 400 && status < 500
+}
+
+// unanimousClientError returns the rejection to replay when every backend in a
+// fan-out refused the same query with the same client-shaped status, and nil
+// otherwise. Every backend is asked the same question, so unanimity is what
+// distinguishes a bad query from a sick estate: a transport failure, a 5xx, or
+// two backends disagreeing on the status all leave at least one backend whose
+// answer is evidence about the backend rather than about the request.
+//
+// The reply returned is the first in configured order, so a client retrying a
+// rejected query is told the same thing every time.
+func unanimousClientError(results []backendResult) *upstreamError {
+	var first *upstreamError
+	for _, res := range results {
+		var ue *upstreamError
+		if !errors.As(res.err, &ue) || !clientShaped(ue.status) {
+			return nil
+		}
+		if first == nil {
+			first = ue
+		} else if ue.status != first.status {
+			return nil
+		}
+	}
+	return first
+}
+
+// writeUpstreamError answers a fan-out that produced no records. A unanimous
+// client-shaped rejection is replayed with the backend's own status and
+// explanation; anything else is reported as a gateway failure.
+func (s *Server) writeUpstreamError(w http.ResponseWriter, err error) {
+	var ue *upstreamError
+	if !errors.As(err, &ue) {
+		http.Error(w, err.Error(), http.StatusBadGateway)
+		return
+	}
+	body := s.redactBackends(ue.body)
+	if len(strings.TrimSpace(string(body))) == 0 {
+		setContentType(w, "text/plain; charset=utf-8")
+		w.WriteHeader(ue.status)
+		_, _ = fmt.Fprintf(w, "upstream rejected the query: %s\n", http.StatusText(ue.status))
+		return
+	}
+	setContentType(w, ue.contentType)
+	w.WriteHeader(ue.status)
+	_, _ = w.Write(body)
+}
+
+// redactBackends strips anything naming a configured backend out of text bound
+// for a client. Hiding the estate behind one endpoint is the point of pdbmux, so
+// a backend's own words may reach the client but its address may not.
+func (s *Server) redactBackends(body []byte) []byte {
+	out := string(body)
+	for _, b := range s.cfg.Backends {
+		for _, term := range backendTerms(b) {
+			out = strings.ReplaceAll(out, term, redactedBackend)
+		}
+	}
+	return []byte(out)
+}
+
+// backendTerms is what identifies a backend in text: its URL, that URL's host
+// and the bare hostname, longest first so replacing one cannot leave a fragment
+// of another behind. The configured name is deliberately absent — no backend
+// knows it, and names are short enough to match ordinary words.
+func backendTerms(b Backend) []string {
+	terms := []string{}
+	if b.URL != "" {
+		terms = append(terms, strings.TrimRight(b.URL, "/"))
+	}
+	u, err := url.Parse(b.URL)
+	if err != nil || u.Host == "" {
+		return terms
+	}
+	terms = append(terms, u.Host)
+	if hn := u.Hostname(); hn != u.Host {
+		terms = append(terms, hn)
+	}
+	return terms
+}

upstream_test.go

@@ -0,0 +1,425 @@
+package main
+
+import (
+	"errors"
+	"fmt"
+	"net/http"
+	"net/url"
+	"strings"
+	"testing"
+	"time"
+)
+
+const badOrderBy = `Unrecognized column 'bogus' specified in :order_by`
+
+func upErr(status int, body string) error {
+	return newUpstreamError(status, "text/plain; charset=utf-8", []byte(body))
+}
+
+func TestUnanimousClientError_AgreementRule(t *testing.T) {
+	refused := errors.New("dial tcp: connection refused")
+	for _, tc := range []struct {
+		name string
+		errs []error
+		want int // 0 means "no replay"
+	}{
+		{"all agree on 400", []error{upErr(400, "bad"), upErr(400, "bad")}, 400},
+		{"all agree on 415", []error{upErr(415, "bad media"), upErr(415, "bad media")}, 415},
+		{"differing bodies still agree", []error{upErr(400, "one"), upErr(400, "two")}, 400},
+		{"disagreeing 4xx", []error{upErr(400, "bad"), upErr(404, "gone")}, 0},
+		{"4xx with a 5xx", []error{upErr(400, "bad"), upErr(500, "boom")}, 0},
+		{"4xx with a transport failure", []error{upErr(400, "bad"), refused}, 0},
+		{"all 403", []error{upErr(403, "denied"), upErr(403, "denied")}, 0},
+		{"all 404", []error{upErr(404, "gone"), upErr(404, "gone")}, 0},
+		{"all 429", []error{upErr(429, "slow down"), upErr(429, "slow down")}, 0},
+		{"all 408", []error{upErr(408, "too slow"), upErr(408, "too slow")}, 0},
+		{"all 500", []error{upErr(500, "boom"), upErr(500, "boom")}, 0},
+		{"all 503", []error{upErr(503, "unavailable"), upErr(503, "unavailable")}, 0},
+		{"no backends", nil, 0},
+		{"a backend succeeded", []error{upErr(400, "bad"), nil}, 0},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			results := make([]backendResult, len(tc.errs))
+			for i, err := range tc.errs {
+				results[i] = backendResult{name: fmt.Sprintf("b%d", i), err: err}
+			}
+			got := unanimousClientError(results)
+			switch {
+			case tc.want == 0 && got != nil:
+				t.Fatalf("want no replay, got HTTP %d", got.status)
+			case tc.want != 0 && got == nil:
+				t.Fatalf("want HTTP %d replayed, got none", tc.want)
+			case tc.want != 0 && got.status != tc.want:
+				t.Fatalf("status = %d, want %d", got.status, tc.want)
+			}
+		})
+	}
+}
+
+// Two backends can explain the same rejection differently; the reply is the
+// first in configured order so a retried query is answered the same way twice.
+func TestUnanimousClientError_PicksFirstInConfiguredOrder(t *testing.T) {
+	results := []backendResult{
+		{name: "a", err: upErr(400, "from a")},
+		{name: "b", err: upErr(400, "from b")},
+	}
+	got := unanimousClientError(results)
+	if got == nil || !strings.Contains(string(got.body), "from a") {
+		t.Fatalf("body = %q, want the first backend's", got)
+	}
+}
+
+func TestUpstreamError_BodyIsCapped(t *testing.T) {
+	ue := newUpstreamError(400, "text/plain", []byte(strings.Repeat("x", upstreamBodyLimit*2)))
+	if len(ue.body) != upstreamBodyLimit {
+		t.Fatalf("kept %d bytes, want %d", len(ue.body), upstreamBodyLimit)
+	}
+}
+
+func TestHandler_MergedAllBackendsRejectReplaysUpstreamStatus(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), nodesPath, url.Values{"order_by": {`[{"field":"certname"}]`}})
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want the upstream 400 replayed: %s", rec.Code, rec.Body.String())
+	}
+	if !strings.Contains(rec.Body.String(), "bogus") {
+		t.Errorf("body = %q, want the upstream explanation", rec.Body.String())
+	}
+}
+
+// 400 is not the only client-shaped status openvoxdb can answer with, so the
+// replay carries whatever status the backends agreed on.
+func TestHandler_MergedReplaysNonBadRequestStatus(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusUnsupportedMediaType, "unsupported media type"
+	b.reject, b.rejectBody = http.StatusUnsupportedMediaType, "unsupported media type"
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusUnsupportedMediaType {
+		t.Fatalf("status = %d, want 415", rec.Code)
+	}
+}
+
+// A unanimous 403 is pdbmux's own credentials being refused, not the client's
+// query, so it must not be handed back as the client's fault.
+func TestHandler_MergedForbiddenStays502(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusForbidden, "certificate not allowed"
+	b.reject, b.rejectBody = http.StatusForbidden, "certificate not allowed"
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
+		t.Fatalf("status = %d, want 502", rec.Code)
+	}
+}
+
+// Every backend answering 404 is ambiguous between a bad path and a record
+// nobody holds, so it stays a gateway error on a merged route.
+func TestHandler_MergedNotFoundStays502(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusNotFound, "Not Found"
+	b.reject, b.rejectBody = http.StatusNotFound, "Not Found"
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
+		t.Fatalf("status = %d, want 502", rec.Code)
+	}
+}
+
+func TestHandler_MergedMixedRejectionAndServerErrorStays502(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.fail = true
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
+		t.Fatalf("status = %d, want 502 when only one backend blamed the query", rec.Code)
+	}
+}
+
+func TestHandler_MergedMixedRejectionAndUnreachableStays502(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	dead := newFakeBackend(t, `[]`, `[]`)
+	deadURL := dead.srv.URL
+	dead.srv.Close()
+	srv := newTestServer(testConfig(a.srv.URL, deadURL, mergeStatic))
+
+	if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
+		t.Fatalf("status = %d, want 502 when a backend never answered", rec.Code)
+	}
+}
+
+func TestHandler_MergedDisagreeingRejectionsStay502(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.reject, b.rejectBody = http.StatusNotFound, "no such entity"
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadGateway {
+		t.Fatalf("status = %d, want 502 when backends disagree on the rejection", rec.Code)
+	}
+}
+
+// A backend that cannot answer a query must not fail one another backend can.
+func TestHandler_MergedOneRejectionOneAnswerServesTheAnswer(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), nodesPath, "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status = %d, want the surviving backend's 200: %s", rec.Code, rec.Body.String())
+	}
+	if !strings.Contains(rec.Body.String(), "h1") {
+		t.Errorf("body = %s, want the survivor's record", rec.Body.String())
+	}
+}
+
+func TestHandler_CombinedAllBackendsRejectReplaysUpstreamStatus(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), resourcesPath, `["extract",[["function","count"]],["=","environment","production"]]`)
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want the aggregate path to replay 400: %s", rec.Code, rec.Body.String())
+	}
+	if !strings.Contains(rec.Body.String(), "bogus") {
+		t.Errorf("body = %q, want the upstream explanation", rec.Body.String())
+	}
+}
+
+func TestHandler_CombinedMixedRejectionAndServerErrorStays502(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.fail = true
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), resourcesPath, `["extract",[["function","count"]],["=","environment","production"]]`)
+	if rec.Code != http.StatusBadGateway {
+		t.Fatalf("status = %d, want 502", rec.Code)
+	}
+}
+
+func TestHandler_UnionAllBackendsRejectReplaysUpstreamStatus(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"order_by": {receiveDesc}})
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want the union path to replay 400: %s", rec.Code, rec.Body.String())
+	}
+}
+
+// /fact-names orders locally, so its fan-out is the third shape aliveResults serves.
+func TestHandler_FactNamesAllBackendsRejectReplaysUpstreamStatus(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	if rec := doGet(t, srv.Handler(), factNamesPath, ""); rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want 400", rec.Code)
+	}
+}
+
+// A backend answering 404 means it does not hold the report, which is the whole
+// premise of this route, so it must keep producing pdbmux's own 404.
+func TestHandler_ReportSubResourceRejectionReplayedButNotItsAbsence(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	if rec := doGet(t, newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)).Handler(),
+		reportsPath+"/nope/events", ""); rec.Code != http.StatusNotFound {
+		t.Fatalf("status = %d, want pdbmux's own 404 when nobody holds the report", rec.Code)
+	}
+
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
+	rec := doGet(t, newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic)).Handler(),
+		reportsPath+"/nope/events", "")
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want the upstream 400 replayed: %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestHandler_RejectedQueryIsNotADegradedRound(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	if rec := doGet(t, srv.Handler(), nodesPath, ""); rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want 400", rec.Code)
+	}
+
+	hr := health(t, srv)
+	if hr.Query.Partial || hr.Query.PartialRounds != 0 || hr.Query.LastPartial != "" {
+		t.Fatalf("a refused query was counted as degraded service: %+v", hr.Query)
+	}
+	// The report's own reachability probe is a separate query the backends also
+	// refuse, so they read as unreachable while the estate itself is fine.
+	if hr.Status != "down" {
+		t.Errorf("status = %q, want the probe's own verdict", hr.Status)
+	}
+}
+
+func TestHandler_RejectedQueryIsNotCached(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, badOrderBy
+	b.reject, b.rejectBody = http.StatusBadRequest, badOrderBy
+	srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
+
+	rec := doGet(t, srv.Handler(), factsPath, "")
+	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 {
+		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, "")
+	if ok.Code != http.StatusOK || !strings.Contains(ok.Body.String(), "web") {
+		t.Fatalf("follow-up = %d %s", ok.Code, ok.Body.String())
+	}
+}
+
+func TestHandler_AllBackendsFailedIsNotCached(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(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 {
+		t.Fatalf("status = %d, want 502", rec.Code)
+	}
+	if n := srv.factsCache.Stats().Entries; n != 0 {
+		t.Fatalf("cache holds %d entries after a failed fan-out", n)
+	}
+}
+
+// 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", "")+`]`)
+	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 {
+		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, "")
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want the rejection rather than the stale entry: %s", rec.Code, rec.Body.String())
+	}
+	if strings.Contains(rec.Body.String(), "old") {
+		t.Errorf("body = %s, want the rejection, not cached records", rec.Body.String())
+	}
+	if serving, _, _ := srv.stale.snapshot(); serving {
+		t.Error("a refused query must not mark the cache as serving stale")
+	}
+}
+
+// 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", "")+`]`)
+	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 {
+		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, "")
+	if rec.Code != http.StatusOK || !strings.Contains(rec.Body.String(), "old") {
+		t.Fatalf("stale fallback = %d %s", rec.Code, rec.Body.String())
+	}
+}
+
+func TestWriteUpstreamError_RedactsBackendAddresses(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	leak := "connection to " + a.srv.URL + "/pdb/query/v4/nodes refused"
+	a.reject, a.rejectBody = http.StatusBadRequest, leak
+	b.reject, b.rejectBody = http.StatusBadRequest, leak
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), nodesPath, "")
+	body := rec.Body.String()
+	if strings.Contains(body, a.srv.URL) || strings.Contains(body, hostOf(t, a.srv.URL)) {
+		t.Fatalf("replayed body names a backend: %q", body)
+	}
+	if !strings.Contains(body, redactedBackend) {
+		t.Errorf("body = %q, want the address redacted", body)
+	}
+}
+
+// The pass-through path replays upstream errors too, so it redacts the same way.
+func TestProxyUnmerged_RedactsBackendAddresses(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, "upstream "+a.srv.URL+" said no"
+	b.reject, b.rejectBody = http.StatusBadRequest, "upstream "+b.srv.URL+" said no"
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), resourcesPath, `["=","type","File"]`)
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want the upstream 400", rec.Code)
+	}
+	if strings.Contains(rec.Body.String(), hostOf(t, a.srv.URL)) {
+		t.Errorf("replayed body names a backend: %q", rec.Body.String())
+	}
+}
+
+func TestWriteUpstreamError_EmptyBodyGetsAMessage(t *testing.T) {
+	a := newFakeBackend(t, `[]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.reject, a.rejectBody = http.StatusBadRequest, ""
+	b.reject, b.rejectBody = http.StatusBadRequest, ""
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), nodesPath, "")
+	if rec.Code != http.StatusBadRequest {
+		t.Fatalf("status = %d, want 400", rec.Code)
+	}
+	if strings.TrimSpace(rec.Body.String()) == "" {
+		t.Error("a replayed rejection must carry some explanation")
+	}
+}
+
+func hostOf(t *testing.T, raw string) string {
+	t.Helper()
+	u, err := url.Parse(raw)
+	if err != nil {
+		t.Fatalf("parsing %q: %v", raw, err)
+	}
+	return u.Host
+}