Merge pull request 'feat: merge /reports and /events across both PuppetDBs' (#5) from benvin/reports-merge into main

Reviewed-on: #5

README.md

@@ -2,8 +2,8 @@
 
 `pdbmux` is a small HTTP daemon that fronts **two** PuppetDB backends and serves
 a single, merged PuppetDB v4 query surface on one address. Point `node-lookup`,
-`pblastreport`, or anything else at `pdbmux` instead of a raw PuppetDB and it
-sees one consistent view spanning both.
+`pblastreport`, Puppetboard, or anything else at `pdbmux` instead of a raw
+PuppetDB and it sees one consistent view spanning both.
 
 ## Why
 
@@ -26,6 +26,9 @@ not PQL) is forwarded verbatim.
 |---|---|
 | `GET /pdb/query/v4/nodes` | Fan out to both backends, dedupe by `certname`, keep the record with the newer `report_timestamp`. |
 | `GET /pdb/query/v4/facts` | Fan out to both, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics). |
+| `GET /pdb/query/v4/reports` | Fan out to both and serve the **union**, deduped by report `hash`, re-ordered and re-paged across the two backends. |
+| `GET /pdb/query/v4/events` | Fan out to both and serve the **union**, deduped by record identity, re-ordered and re-paged. |
+| `GET /pdb/query/v4/reports/<hash>/{events,logs,metrics}` | Ask both; serve the answer from whichever backend actually holds that report. `404` when neither does. |
 | `GET /pdb/query/v4/*` (any other) | Transparently proxied to the **primary** backend, unmerged, streamed verbatim. |
 | `GET /healthz` | Per-backend reachability. `200 {"status":"ok"}` if all reachable, `200 degraded` if some fail, `503 down` if all fail. |
 
@@ -50,6 +53,29 @@ unknown fields survive untouched.
     No extra `/nodes` query.
   - A node present in only one backend always appears (falls back to whichever
     backend actually returned facts for it).
+- **`/reports`, `/events`** — **union**, not a per-node winner. Reports are
+  immutable history, so a node that migrated legitimately has reports in the old
+  PuppetDB *and* the new one and both belong in the merged view. Reports dedupe
+  on `hash`; events, which carry no id of their own, dedupe on the verbatim
+  record (a node briefly reporting to both PuppetDBs stores identical records in
+  each). Records the merge cannot key — `extract`/`group_by` aggregate rows — are
+  never deduped, so every backend's rows pass through even when byte-identical;
+  summing those aggregates across backends is not implemented yet.
+
+### Paging and ordering on the merged endpoints
+
+Each backend applies `order_by`/`limit`/`offset` to its own slice only, so
+`pdbmux` re-does all three over the union:
+
+- `order_by` is parsed and the merged set re-sorted by those fields (ties keep
+  backend precedence). A record missing an ordered field sorts first.
+- Backends are asked for the first `offset + limit` records — never an `offset`
+  — and the requested window is then cut from the merged, re-sorted set.
+- `include_total=true` makes `pdbmux` sum each backend's `X-Records` header into
+  one merged header. Deduped records are counted once per backend, so the total
+  is an upper bound.
+- A malformed `limit`, `offset` or `order_by` gets a `400` rather than being
+  forwarded.
 
 ## Config
 

main.go

@@ -3,19 +3,26 @@
 // During the VM -> k8s Puppet migration there are two PuppetDBs — the legacy
 // Consul-registered one and the new k8s one — and nodes move between them as
 // they migrate. pdbmux presents a single merged PuppetDB v4 query surface so
-// node-lookup and pblastreport (and anything else) see one consistent view:
+// node-lookup, pblastreport and Puppetboard (and anything else) see one
+// consistent view:
 //
 //   - GET /pdb/query/v4/nodes  — fan out to both backends, dedupe by certname,
 //     keep the record with the newer report_timestamp.
 //   - GET /pdb/query/v4/facts  — fan out to both, and for a certname present in
 //     both keep ALL facts from the backend holding that node's newer report
 //     (freshness merge) or a static preferred backend (static merge).
+//   - GET /pdb/query/v4/reports and /events — fan out to both and serve the
+//     deduped union, re-ordered and re-paged across the two backends, because
+//     reports are immutable history and a migrated node has some in each.
+//   - GET /pdb/query/v4/reports/<hash>/{events,logs,metrics} — served by
+//     whichever backend actually holds that report.
 //   - any other GET /pdb/query/v4/* — transparently proxied to the primary.
 //   - GET /healthz — per-backend reachability.
 //
-// The query param is forwarded verbatim (PuppetDB AST JSON). If one backend
-// errors/times out, the other's results are served; only if both fail does a
-// merged endpoint return 502.
+// The query param is forwarded verbatim (PuppetDB AST JSON); order_by, limit,
+// offset and include_total are re-applied over the merged result set. If one
+// backend errors/times out, the other's results are served; only if both fail
+// does a merged endpoint return 502.
 package main
 
 import (
@@ -67,9 +74,9 @@ func main() {
 		Use:   appName,
 		Short: "Merging HTTP proxy over two PuppetDB backends.",
 		Long: "pdbmux presents a single merged PuppetDB v4 query surface over the old\n" +
-			"(Consul) and new (k8s) PuppetDBs during the migration, so node-lookup and\n" +
-			"pblastreport see one consistent view. Running pdbmux with no subcommand\n" +
-			"(or `pdbmux serve`) starts the proxy.",
+			"(Consul) and new (k8s) PuppetDBs during the migration, so node-lookup,\n" +
+			"pblastreport and Puppetboard see one consistent view. Running pdbmux with\n" +
+			"no subcommand (or `pdbmux serve`) starts the proxy.",
 		SilenceUsage: true,
 		RunE:         func(cmd *cobra.Command, args []string) error { return serve(cmd) },
 	}

merge.go

@@ -6,19 +6,21 @@ import (
 )
 
 // record is a single PuppetDB result element kept as raw JSON so unknown fields
-// survive the merge untouched. certname/report_timestamp are decoded only for
-// merge decisions.
+// survive the merge untouched. certname/report_timestamp/hash are decoded only
+// for merge decisions.
 type record struct {
 	Raw             json.RawMessage
 	Certname        string
 	ReportTimestamp string // only populated for /nodes records
+	Hash            string // only populated for /reports records
 }
 
-// recordMeta is the subset we decode from any /nodes or /facts element to drive
-// merge decisions.
+// recordMeta is the subset we decode from any /nodes, /facts or /reports element
+// to drive merge decisions.
 type recordMeta struct {
 	Certname        string `json:"certname"`
 	ReportTimestamp string `json:"report_timestamp"`
+	Hash            string `json:"hash"`
 }
 
 // decodeRecords turns a raw PuppetDB JSON array into records, preserving each
@@ -36,6 +38,7 @@ func decodeRecords(body []byte) ([]record, error) {
 			Raw:             raw,
 			Certname:        m.Certname,
 			ReportTimestamp: m.ReportTimestamp,
+			Hash:            m.Hash,
 		})
 	}
 	return out, nil

merge_test.go

@@ -74,6 +74,20 @@ func fact(cn, name, val, ts string) string {
 	return `{"certname":"` + cn + `","name":"` + name + `","value":"` + val + `","report_timestamp":"` + ts + `"}`
 }
 
+// report builds a /reports record with the fields the merge and ordering paths
+// care about.
+func report(cn, hash, receive string) string {
+	return `{"certname":"` + cn + `","hash":"` + hash + `","receive_time":"` + receive +
+		`","end_time":"` + receive + `","status":"changed","environment":"production"}`
+}
+
+// event builds an /events record, which carries its report's hash but no id of
+// its own.
+func event(cn, reportHash, resource string) string {
+	return `{"certname":"` + cn + `","report":"` + reportHash + `","resource_title":"` + resource +
+		`","status":"success","timestamp":"2026-07-01T00:00:00Z"}`
+}
+
 func TestMergeNodes_NewerWins(t *testing.T) {
 	old := recs(t, "old", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-10T00:00:00Z"))
 	nw := recs(t, "new", node("h1", "2026-07-20T00:00:00Z"), node("h3", "2026-07-05T00:00:00Z"))

reports.go

@@ -0,0 +1,247 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/url"
+	"sort"
+	"strconv"
+	"strings"
+)
+
+// mergeUnion concatenates every backend's records and drops duplicates by key.
+// Reports and events are immutable history, so a certname that migrated between
+// PuppetDBs legitimately has records in both and the union — not a per-node
+// winner — is the correct merged view. results must be ordered by precedence;
+// the first backend holding a key supplies the record. A key func returning
+// ok=false means the record has no dedupe identity and is always kept.
+func mergeUnion(results []backendResult, key func(record) (string, bool)) []json.RawMessage {
+	seen := make(map[string]bool)
+	out := []json.RawMessage{}
+	for _, res := range results {
+		for _, rec := range res.records {
+			if k, ok := key(rec); ok {
+				if seen[k] {
+					continue
+				}
+				seen[k] = true
+			}
+			out = append(out, rec.Raw)
+		}
+	}
+	return out
+}
+
+// reportKey identifies a report by its content hash, which PuppetDB guarantees
+// is unique per report. An `extract`/`group_by` query returns synthetic rows
+// with no hash and no identity — two backends can emit byte-identical aggregate
+// rows that both count — so those are never deduped.
+func reportKey(rec record) (string, bool) {
+	if rec.Hash == "" {
+		return "", false
+	}
+	return "hash\x00" + rec.Hash, true
+}
+
+// rawKey identifies a record by its verbatim JSON. Events carry no unique id,
+// but two byte-identical events from the same PuppetDB serialiser describe the
+// same resource change, so raw equality is a safe dedupe key.
+func rawKey(rec record) (string, bool) { return "raw\x00" + string(rec.Raw), true }
+
+// orderField is one entry of PuppetDB's order_by param.
+type orderField struct {
+	Field string
+	Desc  bool
+}
+
+// parseOrderBy decodes PuppetDB's order_by param, a JSON array of
+// {"field":..., "order":"asc"|"desc"} objects. An empty param yields no fields.
+func parseOrderBy(s string) ([]orderField, error) {
+	if strings.TrimSpace(s) == "" {
+		return nil, nil
+	}
+	var raw []struct {
+		Field string `json:"field"`
+		Order string `json:"order"`
+	}
+	if err := json.Unmarshal([]byte(s), &raw); err != nil {
+		return nil, fmt.Errorf("order_by is not a JSON array: %w", err)
+	}
+	out := make([]orderField, 0, len(raw))
+	for _, r := range raw {
+		if r.Field == "" {
+			return nil, fmt.Errorf("order_by entry is missing a field")
+		}
+		out = append(out, orderField{Field: r.Field, Desc: strings.EqualFold(r.Order, "desc")})
+	}
+	return out, nil
+}
+
+// sortRecords re-sorts a merged record set by order. Each backend only ordered
+// its own slice, so the union has to be ordered again here. The sort is stable,
+// so ties keep backend precedence order.
+func sortRecords(recs []json.RawMessage, order []orderField) {
+	if len(order) == 0 || len(recs) < 2 {
+		return
+	}
+	objs := make([]map[string]any, len(recs))
+	for i, raw := range recs {
+		_ = json.Unmarshal(raw, &objs[i]) // non-objects sort as all-missing fields
+	}
+	idx := make([]int, len(recs))
+	for i := range idx {
+		idx[i] = i
+	}
+	sort.SliceStable(idx, func(a, b int) bool {
+		oa, ob := objs[idx[a]], objs[idx[b]]
+		for _, f := range order {
+			c := compareValues(oa[f.Field], ob[f.Field])
+			if c == 0 {
+				continue
+			}
+			if f.Desc {
+				return c > 0
+			}
+			return c < 0
+		}
+		return false
+	})
+	sorted := make([]json.RawMessage, len(recs))
+	for i, j := range idx {
+		sorted[i] = recs[j]
+	}
+	copy(recs, sorted)
+}
+
+// compareValues orders two decoded JSON values. Unlike types are ordered by
+// kind (null < bool < number < string) so a missing field always sorts first.
+func compareValues(a, b any) int {
+	ra, rb := valueRank(a), valueRank(b)
+	if ra != rb {
+		if ra < rb {
+			return -1
+		}
+		return 1
+	}
+	switch av := a.(type) {
+	case bool:
+		bv := b.(bool)
+		switch {
+		case av == bv:
+			return 0
+		case bv:
+			return -1
+		default:
+			return 1
+		}
+	case float64:
+		bv := b.(float64)
+		switch {
+		case av < bv:
+			return -1
+		case av > bv:
+			return 1
+		default:
+			return 0
+		}
+	case string:
+		return strings.Compare(av, b.(string))
+	}
+	return 0
+}
+
+func valueRank(v any) int {
+	switch v.(type) {
+	case nil:
+		return 0
+	case bool:
+		return 1
+	case float64:
+		return 2
+	case string:
+		return 3
+	default:
+		return 4
+	}
+}
+
+// paging holds the PuppetDB paging params a merged endpoint has to re-apply
+// itself: each backend applies limit/offset to its own result set only, so the
+// proxy must page the union instead.
+type paging struct {
+	limit     int // -1 when unset
+	offset    int
+	order     []orderField
+	wantTotal bool
+}
+
+// parsePaging reads limit, offset, order_by and include_total from a request's
+// query params.
+func parsePaging(v url.Values) (paging, error) {
+	p := paging{limit: -1}
+	if s := v.Get("limit"); s != "" {
+		n, err := strconv.Atoi(s)
+		if err != nil || n < 0 {
+			return p, fmt.Errorf("limit must be a non-negative integer, got %q", s)
+		}
+		p.limit = n
+	}
+	if s := v.Get("offset"); s != "" {
+		n, err := strconv.Atoi(s)
+		if err != nil || n < 0 {
+			return p, fmt.Errorf("offset must be a non-negative integer, got %q", s)
+		}
+		p.offset = n
+	}
+	order, err := parseOrderBy(v.Get("order_by"))
+	if err != nil {
+		return p, err
+	}
+	p.order = order
+	p.wantTotal = v.Get("include_total") == "true"
+	return p, nil
+}
+
+// upstreamParams rewrites the client's params for the fan-out. A backend must
+// return everything that could land in the merged page, so it is asked for the
+// first offset+limit records and the offset is applied locally instead.
+func (p paging) upstreamParams(in url.Values) url.Values {
+	out := url.Values{}
+	for k, vs := range in {
+		out[k] = append([]string(nil), vs...)
+	}
+	out.Del("offset")
+	if p.limit >= 0 {
+		out.Set("limit", strconv.Itoa(p.limit+p.offset))
+	}
+	return out
+}
+
+// apply slices the merged, ordered record set down to the requested page.
+func (p paging) apply(recs []json.RawMessage) []json.RawMessage {
+	if p.offset >= len(recs) {
+		return []json.RawMessage{}
+	}
+	recs = recs[p.offset:]
+	if p.limit >= 0 && p.limit < len(recs) {
+		recs = recs[:p.limit]
+	}
+	return recs
+}
+
+// sumTotals adds up the X-Records counts the backends reported, ignoring any
+// backend that did not send one. It returns -1 when no backend reported a count.
+// Deduped records are counted once per backend, so the total is an upper bound.
+func sumTotals(results []backendResult) int {
+	total := -1
+	for _, res := range results {
+		if res.total < 0 {
+			continue
+		}
+		if total < 0 {
+			total = 0
+		}
+		total += res.total
+	}
+	return total
+}

reports_test.go

@@ -0,0 +1,254 @@
+package main
+
+import (
+	"encoding/json"
+	"net/url"
+	"slices"
+	"testing"
+)
+
+func TestMergeUnion_KeepsBothBackendsHistory(t *testing.T) {
+	old := recs(t, "old", report("h1", "r1", "2026-07-01T00:00:00Z"))
+	nw := recs(t, "new", report("h1", "r2", "2026-07-02T00:00:00Z"))
+	merged := mergeUnion([]backendResult{nw, old}, reportKey)
+	if got := hashesOf(t, merged); !slices.Equal(got, []string{"r2", "r1"}) {
+		t.Errorf("union = %v, want both reports in precedence order", got)
+	}
+}
+
+func TestMergeUnion_DedupesSharedHash(t *testing.T) {
+	dup := report("h1", "r1", "2026-07-01T00:00:00Z")
+	merged := mergeUnion([]backendResult{recs(t, "new", dup), recs(t, "old", dup)}, reportKey)
+	if got := hashesOf(t, merged); !slices.Equal(got, []string{"r1"}) {
+		t.Errorf("union = %v, want a single r1", got)
+	}
+}
+
+func TestMergeUnion_HashlessRowsAreAllKept(t *testing.T) {
+	// extract/group_by queries return synthetic rows with no hash; dropping the
+	// second backend's rows as "duplicates" would silently lose half the data.
+	old := recs(t, "old", `{"status":"changed","count":3}`)
+	nw := recs(t, "new", `{"status":"changed","count":5}`)
+	merged := mergeUnion([]backendResult{old, nw}, reportKey)
+	if len(merged) != 2 {
+		t.Errorf("expected both aggregate rows, got %d: %v", len(merged), merged)
+	}
+}
+
+func TestMergeUnion_IdenticalHashlessRowsAreNotCollapsed(t *testing.T) {
+	// Two backends can legitimately produce the same aggregate row; collapsing
+	// them as duplicates undercounts the merged result.
+	same := `{"status":"changed","count":1}`
+	merged := mergeUnion([]backendResult{recs(t, "old", same), recs(t, "new", same)}, reportKey)
+	if len(merged) != 2 {
+		t.Errorf("expected both backends' aggregate rows, got %d: %v", len(merged), merged)
+	}
+}
+
+func TestMergeUnion_EventsDedupeOnRawIdentity(t *testing.T) {
+	same := event("h1", "r1", "Package[nginx]")
+	other := event("h1", "r1", "Service[nginx]")
+	merged := mergeUnion([]backendResult{recs(t, "new", same, other), recs(t, "old", same)}, rawKey)
+	if len(merged) != 2 {
+		t.Errorf("expected 2 distinct events, got %d: %v", len(merged), merged)
+	}
+}
+
+func TestParseOrderBy(t *testing.T) {
+	got, err := parseOrderBy(`[{"field":"receive_time","order":"desc"},{"field":"certname"}]`)
+	if err != nil {
+		t.Fatal(err)
+	}
+	want := []orderField{{Field: "receive_time", Desc: true}, {Field: "certname"}}
+	if !slices.Equal(got, want) {
+		t.Errorf("parseOrderBy = %v, want %v", got, want)
+	}
+
+	if got, err := parseOrderBy("  "); err != nil || got != nil {
+		t.Errorf("empty order_by = %v, %v; want nil, nil", got, err)
+	}
+	for _, bad := range []string{`receive_time`, `[{"order":"desc"}]`} {
+		if _, err := parseOrderBy(bad); err == nil {
+			t.Errorf("parseOrderBy(%q) should have failed", bad)
+		}
+	}
+}
+
+func TestSortRecords_MultipleFieldsAndStability(t *testing.T) {
+	raws := rawsOf(t,
+		`{"certname":"b","status":"failed","hash":"r1"}`,
+		`{"certname":"a","status":"changed","hash":"r2"}`,
+		`{"certname":"a","status":"changed","hash":"r3"}`,
+		`{"certname":"a","status":"failed","hash":"r4"}`,
+	)
+	sortRecords(raws, []orderField{{Field: "certname"}, {Field: "status", Desc: true}})
+	// certname asc, then status desc; r2/r3 tie fully and keep input order.
+	if got := hashesOf(t, raws); !slices.Equal(got, []string{"r4", "r2", "r3", "r1"}) {
+		t.Errorf("sorted = %v, want [r4 r2 r3 r1]", got)
+	}
+}
+
+func TestSortRecords_MissingFieldSortsFirst(t *testing.T) {
+	raws := rawsOf(t,
+		`{"hash":"r1","receive_time":"2026-07-01T00:00:00Z"}`,
+		`{"hash":"r2"}`,
+	)
+	sortRecords(raws, []orderField{{Field: "receive_time"}})
+	if got := hashesOf(t, raws); !slices.Equal(got, []string{"r2", "r1"}) {
+		t.Errorf("sorted = %v, want the record missing the field first", got)
+	}
+}
+
+func TestSortRecords_NoOrderLeavesInputOrder(t *testing.T) {
+	raws := rawsOf(t, `{"hash":"r1"}`, `{"hash":"r2"}`)
+	sortRecords(raws, nil)
+	if got := hashesOf(t, raws); !slices.Equal(got, []string{"r1", "r2"}) {
+		t.Errorf("sorted = %v, want unchanged", got)
+	}
+}
+
+func TestCompareValues_AcrossKinds(t *testing.T) {
+	cases := []struct {
+		a, b any
+		want int
+	}{
+		{nil, false, -1},
+		{false, true, -1},
+		{true, 1.0, -1},
+		{1.0, 2.0, -1},
+		{2.0, 2.0, 0},
+		{2.0, "x", -1},
+		{"a", "b", -1},
+		{"b", "a", 1},
+	}
+	for _, c := range cases {
+		if got := compareValues(c.a, c.b); got != c.want {
+			t.Errorf("compareValues(%v, %v) = %d, want %d", c.a, c.b, got, c.want)
+		}
+		if got := compareValues(c.b, c.a); got != -c.want {
+			t.Errorf("compareValues(%v, %v) = %d, want %d (antisymmetry)", c.b, c.a, got, -c.want)
+		}
+	}
+}
+
+func TestParsePaging(t *testing.T) {
+	p, err := parsePaging(url.Values{
+		"limit":         {"25"},
+		"offset":        {"50"},
+		"include_total": {"true"},
+		"order_by":      {`[{"field":"receive_time","order":"desc"}]`},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if p.limit != 25 || p.offset != 50 || !p.wantTotal || len(p.order) != 1 {
+		t.Fatalf("parsePaging = %+v", p)
+	}
+
+	if p, err := parsePaging(nil); err != nil || p.limit != -1 || p.offset != 0 || p.wantTotal {
+		t.Errorf("empty params = %+v, %v; want limit=-1 and no paging", p, err)
+	}
+	for _, bad := range []url.Values{{"limit": {"-1"}}, {"limit": {"x"}}, {"offset": {"x"}}} {
+		if _, err := parsePaging(bad); err == nil {
+			t.Errorf("parsePaging(%v) should have failed", bad)
+		}
+	}
+}
+
+func TestPagingUpstreamParams(t *testing.T) {
+	in := url.Values{
+		"query":  {`["=","certname","h1"]`},
+		"limit":  {"25"},
+		"offset": {"50"},
+	}
+	p, err := parsePaging(in)
+	if err != nil {
+		t.Fatal(err)
+	}
+	out := p.upstreamParams(in)
+	if out.Get("limit") != "75" {
+		t.Errorf("upstream limit = %q, want 75 (offset+limit)", out.Get("limit"))
+	}
+	if out.Has("offset") {
+		t.Errorf("upstream offset = %q, want it dropped", out.Get("offset"))
+	}
+	if out.Get("query") != in.Get("query") {
+		t.Errorf("query should pass through verbatim, got %q", out.Get("query"))
+	}
+	if in.Get("limit") != "25" {
+		t.Errorf("upstreamParams must not mutate the caller's params, limit is now %q", in.Get("limit"))
+	}
+}
+
+func TestPagingUpstreamParams_NoLimitLeavesQueryUnbounded(t *testing.T) {
+	in := url.Values{"offset": {"5"}}
+	p, err := parsePaging(in)
+	if err != nil {
+		t.Fatal(err)
+	}
+	out := p.upstreamParams(in)
+	if out.Has("limit") || out.Has("offset") {
+		t.Errorf("upstream params = %v, want neither limit nor offset", out)
+	}
+}
+
+func TestPagingApply(t *testing.T) {
+	raws := rawsOf(t, `{"hash":"r1"}`, `{"hash":"r2"}`, `{"hash":"r3"}`)
+	cases := []struct {
+		name string
+		page paging
+		want []string
+	}{
+		{name: "window", page: paging{limit: 1, offset: 1}, want: []string{"r2"}},
+		{name: "limit past end", page: paging{limit: 10}, want: []string{"r1", "r2", "r3"}},
+		{name: "offset past end", page: paging{limit: 2, offset: 9}, want: nil},
+		{name: "unset limit", page: paging{limit: -1, offset: 2}, want: []string{"r3"}},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			got := hashesOf(t, c.page.apply(raws))
+			if len(got) == 0 && len(c.want) == 0 {
+				return
+			}
+			if !slices.Equal(got, c.want) {
+				t.Errorf("apply = %v, want %v", got, c.want)
+			}
+		})
+	}
+}
+
+func TestSumTotals(t *testing.T) {
+	if got := sumTotals([]backendResult{{total: 40}, {total: 60}}); got != 100 {
+		t.Errorf("sumTotals = %d, want 100", got)
+	}
+	if got := sumTotals([]backendResult{{total: -1}, {total: 7}}); got != 7 {
+		t.Errorf("sumTotals should skip backends without a count, got %d", got)
+	}
+	if got := sumTotals([]backendResult{{total: -1}, {total: -1}}); got != -1 {
+		t.Errorf("sumTotals with no counts = %d, want -1", got)
+	}
+}
+
+// rawsOf builds a raw record slice from literal JSON elements.
+func rawsOf(t *testing.T, elems ...string) []json.RawMessage {
+	t.Helper()
+	out := make([]json.RawMessage, 0, len(elems))
+	for _, e := range elems {
+		out = append(out, json.RawMessage(e))
+	}
+	return out
+}
+
+// hashesOf extracts the hash field from raw records, in order.
+func hashesOf(t *testing.T, raws []json.RawMessage) []string {
+	t.Helper()
+	out := make([]string, 0, len(raws))
+	for _, r := range raws {
+		var m recordMeta
+		if err := json.Unmarshal(r, &m); err != nil {
+			t.Fatalf("unmarshal %s: %v", r, err)
+		}
+		out = append(out, m.Hash)
+	}
+	return out
+}

server.go

@@ -9,15 +9,22 @@ import (
 	"net/http"
 	"net/url"
 	"sort"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
 )
 
 const (
-	factsPath = "/pdb/query/v4/facts"
-	nodesPath = "/pdb/query/v4/nodes"
-	queryV4   = "/pdb/query/v4/"
+	factsPath   = "/pdb/query/v4/facts"
+	nodesPath   = "/pdb/query/v4/nodes"
+	reportsPath = "/pdb/query/v4/reports"
+	eventsPath  = "/pdb/query/v4/events"
+	queryV4     = "/pdb/query/v4/"
+
+	// recordsHeader is PuppetDB's total-result-count header, returned when a
+	// request carries include_total=true.
+	recordsHeader = "X-Records"
 )
 
 // backendResult is one backend's decoded response for a query. err is non-nil
@@ -26,6 +33,7 @@ const (
 type backendResult struct {
 	name    string
 	records []record
+	total   int // upstream X-Records count, or -1 when the backend sent none
 	err     error
 }
 
@@ -59,7 +67,9 @@ func (s *Server) Handler() http.Handler {
 }
 
 // handleQuery dispatches /pdb/query/v4/* requests: /facts and /nodes are merged
-// across backends; every other v4 path is transparently proxied to the primary.
+// per node, /reports and /events are unioned across backends, a report's
+// sub-resources resolve to whichever backend stores that report, and every other
+// v4 path is transparently proxied to the primary.
 func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
 	if r.Method != http.MethodGet {
 		http.Error(w, "only GET is supported", http.StatusMethodNotAllowed)
@@ -70,17 +80,109 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
 		s.serveMerged(w, r, nodesPath, s.mergeNodesResponse)
 	case factsPath:
 		s.serveMerged(w, r, factsPath, s.mergeFactsResponse)
+	case reportsPath:
+		s.serveUnion(w, r, reportsPath, reportKey)
+	case eventsPath:
+		s.serveUnion(w, r, eventsPath, rawKey)
 	default:
+		if isReportSubResource(r.URL.Path) {
+			s.serveFirstHolder(w, r)
+			return
+		}
 		s.proxyPrimary(w, r)
 	}
 }
 
+// isReportSubResource reports whether path is a per-report child endpoint —
+// /pdb/query/v4/reports/<hash>/{events,logs,metrics} — whose data lives in
+// exactly one backend.
+func isReportSubResource(path string) bool {
+	rest, ok := strings.CutPrefix(path, reportsPath+"/")
+	if !ok {
+		return false
+	}
+	hash, sub, ok := strings.Cut(rest, "/")
+	if !ok || hash == "" {
+		return false
+	}
+	switch sub {
+	case "events", "logs", "metrics":
+		return true
+	}
+	return false
+}
+
 // serveMerged fans out the request to all backends, then hands the per-backend
 // results to merge to produce the response body. If every backend fails it
 // returns 502; if some fail it serves the survivors and logs a warning.
 func (s *Server) serveMerged(w http.ResponseWriter, r *http.Request, path string, merge func([]backendResult) []json.RawMessage) {
-	query := r.URL.Query().Get("query")
-	results := s.fanOut(r.Context(), path, query)
+	alive, ok := s.aliveResults(w, r, path, queryParams(r.URL.Query().Get("query")))
+	if !ok {
+		return
+	}
+	writeJSON(w, merge(alive))
+}
+
+// serveUnion fans out a request whose records are immutable history — reports
+// and events — and serves the deduped union of every backend. Because each
+// backend ordered and paged only its own slice, the union is re-ordered and
+// re-paged here from the client's order_by/limit/offset.
+func (s *Server) serveUnion(w http.ResponseWriter, r *http.Request, path string, key func(record) (string, bool)) {
+	in := r.URL.Query()
+	page, err := parsePaging(in)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+
+	alive, ok := s.aliveResults(w, r, path, page.upstreamParams(in))
+	if !ok {
+		return
+	}
+
+	merged := mergeUnion(s.byPrecedence(alive), key)
+	sortRecords(merged, page.order)
+	if page.wantTotal {
+		if total := sumTotals(alive); total >= 0 {
+			w.Header().Set(recordsHeader, strconv.Itoa(total))
+		}
+	}
+	writeJSON(w, page.apply(merged))
+}
+
+// serveFirstHolder answers a per-report sub-resource request. The report lives
+// in exactly one backend, so all are asked concurrently and the first one (in
+// precedence order) that actually holds it wins. Backends that do not have the
+// report answer 404, which is indistinguishable here from any other failure, so
+// an empty result is only served once every backend has been consulted.
+func (s *Server) serveFirstHolder(w http.ResponseWriter, r *http.Request) {
+	results := s.fanOut(r.Context(), r.URL.Path, r.URL.Query())
+
+	var alive []backendResult
+	for _, res := range results {
+		if res.err != nil {
+			s.log.Printf("info: backend %q has no %s: %v", res.name, r.URL.Path, res.err)
+			continue
+		}
+		alive = append(alive, res)
+	}
+	if len(alive) == 0 {
+		http.Error(w, "no backend holds this report", http.StatusNotFound)
+		return
+	}
+	for _, res := range s.byPrecedence(alive) {
+		if len(res.records) > 0 {
+			writeJSON(w, rawRecords(res.records))
+			return
+		}
+	}
+	writeJSON(w, nil)
+}
+
+// aliveResults fans out to every backend and returns the successful results.
+// It writes a 502 and returns ok=false 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)
 
 	var alive []backendResult
 	for _, res := range results {
@@ -92,11 +194,27 @@ func (s *Server) serveMerged(w http.ResponseWriter, r *http.Request, path string
 	}
 	if len(alive) == 0 {
 		http.Error(w, "all backends failed", http.StatusBadGateway)
-		return
+		return nil, false
 	}
+	return alive, true
+}
 
-	merged := merge(alive)
-	writeJSON(w, merged)
+// queryParams builds the upstream param set for a merged endpoint that only
+// forwards the PuppetDB query.
+func queryParams(query string) url.Values {
+	if query == "" {
+		return nil
+	}
+	return url.Values{"query": []string{query}}
+}
+
+// rawRecords strips decoded metadata back down to the verbatim JSON elements.
+func rawRecords(recs []record) []json.RawMessage {
+	out := make([]json.RawMessage, 0, len(recs))
+	for _, rec := range recs {
+		out = append(out, rec.Raw)
+	}
+	return out
 }
 
 // mergeNodesResponse merges /nodes results (dedupe by certname, newer wins).
@@ -154,7 +272,7 @@ func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness
 	s.mu.Unlock()
 
 	// Empty query = all nodes; cheap enough for a short-TTL cache.
-	nodeResults := s.fanOut(ctx, nodesPath, "")
+	nodeResults := s.fanOut(ctx, nodesPath, nil)
 	var alive []backendResult
 	for _, res := range nodeResults {
 		if res.err != nil {
@@ -172,48 +290,52 @@ func (s *Server) freshnessMap(ctx context.Context, _ []backendResult) freshness
 	return f
 }
 
-// fanOut queries every backend concurrently for path?query=... and returns one
-// backendResult per backend, in config order.
-func (s *Server) fanOut(ctx context.Context, path, query string) []backendResult {
+// fanOut queries every backend concurrently for path with the given params and
+// returns one backendResult per backend, in config order.
+func (s *Server) fanOut(ctx context.Context, path string, params url.Values) []backendResult {
 	results := make([]backendResult, len(s.cfg.Backends))
 	var wg sync.WaitGroup
 	for i, b := range s.cfg.Backends {
 		wg.Add(1)
 		go func(i int, b Backend) {
 			defer wg.Done()
-			recs, err := s.queryBackend(ctx, b, path, query)
-			results[i] = backendResult{name: b.Name, records: recs, err: err}
+			recs, total, err := s.queryBackend(ctx, b, path, params)
+			results[i] = backendResult{name: b.Name, records: recs, total: total, err: err}
 		}(i, b)
 	}
 	wg.Wait()
 	return results
 }
 
-// queryBackend performs one GET b.URL+path?query=... and decodes the JSON array.
-func (s *Server) queryBackend(ctx context.Context, b Backend, path, query string) ([]record, error) {
+// queryBackend performs one GET b.URL+path?params and decodes the JSON array.
+// It also returns the upstream X-Records count, or -1 when the backend sent none.
+func (s *Server) queryBackend(ctx context.Context, b Backend, path string, params url.Values) ([]record, int, error) {
 	target := strings.TrimRight(b.URL, "/") + path
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, target, nil)
 	if err != nil {
-		return nil, err
+		return nil, -1, err
 	}
-	if query != "" {
-		q := url.Values{}
-		q.Set("query", query)
-		req.URL.RawQuery = q.Encode()
+	if len(params) > 0 {
+		req.URL.RawQuery = params.Encode()
 	}
 	resp, err := s.client.Do(req)
 	if err != nil {
-		return nil, err
+		return nil, -1, err
 	}
 	defer func() { _ = resp.Body.Close() }()
 	body, err := io.ReadAll(resp.Body)
 	if err != nil {
-		return nil, err
+		return nil, -1, err
 	}
 	if resp.StatusCode != http.StatusOK {
-		return nil, fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
+		return nil, -1, fmt.Errorf("HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(body)))
 	}
-	return decodeRecords(body)
+	total := -1
+	if n, err := strconv.Atoi(resp.Header.Get(recordsHeader)); err == nil && n >= 0 {
+		total = n
+	}
+	recs, err := decodeRecords(body)
+	return recs, total, err
 }
 
 // proxyPrimary transparently forwards a non-merged /pdb/query/v4/* request to
@@ -254,7 +376,7 @@ type healthReport struct {
 // reachable, "degraded" if some fail, "down" if all fail (503 in that case).
 func (s *Server) handleHealth(w http.ResponseWriter, r *http.Request) {
 	probe := `["=","certname","pdbmux-healthz-probe"]`
-	results := s.fanOut(r.Context(), nodesPath, probe)
+	results := s.fanOut(r.Context(), nodesPath, queryParams(probe))
 
 	report := healthReport{Backends: map[string]string{}}
 	healthy := 0

server_test.go

@@ -7,7 +7,10 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"slices"
+	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 )
@@ -15,26 +18,55 @@ import (
 // fakeBackend is an httptest PuppetDB that returns canned bodies per path and
 // records the query params it received.
 type fakeBackend struct {
-	srv        *httptest.Server
-	nodesBody  string
-	factsBody  string
-	fail       bool          // return 500 for everything
-	delay      time.Duration // artificial latency
-	gotQueries map[string]string
+	srv       *httptest.Server
+	nodesBody string
+	factsBody string
+	// bodies holds extra canned responses keyed by path (reports, events, a
+	// report's sub-resources). A path under /reports/ that is absent from bodies
+	// answers 404, like a PuppetDB that does not hold that report.
+	bodies map[string]string
+	// totals is the X-Records count advertised per path when the request asks
+	// for include_total.
+	totals map[string]int
+	fail   bool          // return 500 for everything
+	delay  time.Duration // artificial latency
+
+	mu        sync.Mutex
+	gotParams map[string]url.Values
 }
 
 func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend {
 	t.Helper()
-	fb := &fakeBackend{nodesBody: nodesBody, factsBody: factsBody, gotQueries: map[string]string{}}
+	fb := &fakeBackend{
+		nodesBody: nodesBody,
+		factsBody: factsBody,
+		bodies:    map[string]string{},
+		totals:    map[string]int{},
+		gotParams: map[string]url.Values{},
+	}
 	fb.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		if fb.delay > 0 {
 			time.Sleep(fb.delay)
 		}
-		fb.gotQueries[r.URL.Path] = r.URL.Query().Get("query")
+		fb.mu.Lock()
+		fb.gotParams[r.URL.Path] = r.URL.Query()
+		fb.mu.Unlock()
 		if fb.fail {
 			http.Error(w, "boom", http.StatusInternalServerError)
 			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))
+			}
+			w.Header().Set("Content-Type", "application/json")
+			_, _ = io.WriteString(w, truncate(t, body, r.URL.Query().Get("limit")))
+			return
+		}
+		if strings.HasPrefix(r.URL.Path, reportsPath+"/") {
+			http.Error(w, "no report with that hash", http.StatusNotFound)
+			return
+		}
 		w.Header().Set("Content-Type", "application/json")
 		switch r.URL.Path {
 		case nodesPath:
@@ -49,6 +81,43 @@ func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend {
 	return fb
 }
 
+// params returns the query params the backend saw for a path, and whether it was
+// asked for that path at all.
+func (fb *fakeBackend) params(path string) (url.Values, bool) {
+	fb.mu.Lock()
+	defer fb.mu.Unlock()
+	v, ok := fb.gotParams[path]
+	return v, ok
+}
+
+// gotQuery returns the PuppetDB query param the backend saw for a path.
+func (fb *fakeBackend) gotQuery(path string) string {
+	v, _ := fb.params(path)
+	return v.Get("query")
+}
+
+// truncate applies an upstream limit param to a canned JSON array body, the way
+// a real PuppetDB would, so paging tests exercise the proxy's re-paging.
+func truncate(t *testing.T, body, limit string) string {
+	t.Helper()
+	n, err := strconv.Atoi(limit)
+	if err != nil {
+		return body
+	}
+	var raws []json.RawMessage
+	if err := json.Unmarshal([]byte(body), &raws); err != nil {
+		return body
+	}
+	if n < len(raws) {
+		raws = raws[:n]
+	}
+	out, err := json.Marshal(raws)
+	if err != nil {
+		t.Fatalf("re-marshal truncated body: %v", err)
+	}
+	return string(out)
+}
+
 func testConfig(oldURL, newURL, merge string) Config {
 	return Config{
 		Listen:       ":0",
@@ -109,11 +178,11 @@ func TestHandler_QueryPassthrough(t *testing.T) {
 
 	q := `["=","certname","abc.example.net"]`
 	doGet(t, srv.Handler(), factsPath, q)
-	if old.gotQueries[factsPath] != q {
-		t.Errorf("old backend got query %q, want %q", old.gotQueries[factsPath], q)
+	if old.gotQuery(factsPath) != q {
+		t.Errorf("old backend got query %q, want %q", old.gotQuery(factsPath), q)
 	}
-	if nw.gotQueries[factsPath] != q {
-		t.Errorf("new backend got query %q, want %q", nw.gotQueries[factsPath], q)
+	if nw.gotQuery(factsPath) != q {
+		t.Errorf("new backend got query %q, want %q", nw.gotQuery(factsPath), q)
 	}
 }
 
@@ -190,23 +259,24 @@ func TestHandler_BothBackendsDown(t *testing.T) {
 }
 
 func TestHandler_PassThroughToPrimary(t *testing.T) {
-	// A non-merged v4 path (e.g. /reports) goes only to the primary (new).
+	// A non-merged v4 path (e.g. /resources) goes only to the primary (new).
 	old := newFakeBackend(t, `[]`, `[]`)
 	nw := newFakeBackend(t, `[]`, `[]`)
 	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
 
-	rec := doGet(t, srv.Handler(), "/pdb/query/v4/reports", `["=","certname","h1"]`)
+	const path = "/pdb/query/v4/resources"
+	rec := doGet(t, srv.Handler(), path, `["=","certname","h1"]`)
 	if rec.Code != http.StatusOK {
 		t.Fatalf("status %d", rec.Code)
 	}
-	if !strings.Contains(rec.Body.String(), "/pdb/query/v4/reports") {
+	if !strings.Contains(rec.Body.String(), path) {
 		t.Errorf("expected pass-through body, got %s", rec.Body.String())
 	}
 	// Only primary (new) should have been queried.
-	if _, hit := old.gotQueries["/pdb/query/v4/reports"]; hit {
+	if _, hit := old.params(path); hit {
 		t.Errorf("non-primary backend should not be queried for pass-through")
 	}
-	if _, hit := nw.gotQueries["/pdb/query/v4/reports"]; !hit {
+	if _, hit := nw.params(path); !hit {
 		t.Errorf("primary backend should be queried for pass-through")
 	}
 }
@@ -284,3 +354,241 @@ func TestFreshnessCache_Reused(t *testing.T) {
 		}
 	}
 }
+
+// doGetParams issues a GET with an arbitrary param set, for the paging/ordering
+// params the reports endpoints accept.
+func doGetParams(t *testing.T, h http.Handler, path string, params url.Values) *httptest.ResponseRecorder {
+	t.Helper()
+	target := path
+	if len(params) > 0 {
+		target += "?" + params.Encode()
+	}
+	req := httptest.NewRequest(http.MethodGet, target, nil)
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, req)
+	return rec
+}
+
+// hashes extracts report hashes from a merged response body, in order.
+func hashes(t *testing.T, body []byte) []string {
+	t.Helper()
+	var raws []json.RawMessage
+	if err := json.Unmarshal(body, &raws); err != nil {
+		t.Fatalf("unmarshal %s: %v", body, err)
+	}
+	return hashesOf(t, raws)
+}
+
+const receiveDesc = `[{"field":"receive_time","order":"desc"}]`
+
+func TestHandler_ReportsUnioned(t *testing.T) {
+	// h1 migrated: its pre-migration reports are in old, later ones in new.
+	// Both must show up, unlike /facts where one backend wins the node.
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-10T00:00:00Z") + `,` +
+		report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + report("h1", "r4", "2026-07-30T00:00:00Z") + `,` +
+		report("h1", "r3", "2026-07-20T00:00:00Z") + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
+		"query":    {`["=","certname","h1"]`},
+		"order_by": {receiveDesc},
+	})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	got := hashes(t, rec.Body.Bytes())
+	want := []string{"r4", "r3", "r2", "r1"}
+	if !slices.Equal(got, want) {
+		t.Errorf("merged reports = %v, want %v (union re-sorted by receive_time desc)", got, want)
+	}
+}
+
+func TestHandler_ReportsDedupedByHash(t *testing.T) {
+	// A node reporting to both PuppetDBs mid-migration stores the same report
+	// hash in each; the merged view must show it once.
+	dup := report("h1", "r1", "2026-07-01T00:00:00Z")
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[` + dup + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + dup + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, nil)
+	if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r1"}) {
+		t.Errorf("merged reports = %v, want one r1", got)
+	}
+}
+
+func TestHandler_ReportsPagedAcrossBackends(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[` + report("h1", "r5", "2026-07-05T00:00:00Z") + `,` +
+		report("h1", "r3", "2026-07-03T00:00:00Z") + `,` +
+		report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + report("h1", "r6", "2026-07-06T00:00:00Z") + `,` +
+		report("h1", "r4", "2026-07-04T00:00:00Z") + `,` +
+		report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
+		"order_by": {receiveDesc},
+		"limit":    {"2"},
+		"offset":   {"2"},
+	})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	// Globally-ordered page 2 of the union, not each backend's own page 2.
+	if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r4", "r3"}) {
+		t.Errorf("page = %v, want [r4 r3]", got)
+	}
+	// Each backend must be asked for the first offset+limit records so the
+	// merged window is fully covered.
+	for name, fb := range map[string]*fakeBackend{"old": old, "new": nw} {
+		p, ok := fb.params(reportsPath)
+		if !ok {
+			t.Fatalf("%s backend was not queried", name)
+		}
+		if p.Get("limit") != "4" {
+			t.Errorf("%s backend got limit=%q, want 4 (offset+limit)", name, p.Get("limit"))
+		}
+		if p.Has("offset") {
+			t.Errorf("%s backend got offset=%q, want it applied locally instead", name, p.Get("offset"))
+		}
+	}
+}
+
+func TestHandler_ReportsIncludeTotalSummed(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
+	old.totals[reportsPath] = 40
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + report("h1", "r2", "2026-07-02T00:00:00Z") + `]`
+	nw.totals[reportsPath] = 60
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{
+		"include_total": {"true"},
+		"limit":         {"1"},
+	})
+	if got := rec.Header().Get(recordsHeader); got != "100" {
+		t.Errorf("%s = %q, want 100 (sum of both backends)", recordsHeader, got)
+	}
+}
+
+func TestHandler_ReportsNoTotalWhenNotRequested(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[reportsPath] = `[]`
+	old.totals[reportsPath] = 40
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[]`
+	nw.totals[reportsPath] = 60
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, nil)
+	if got := rec.Header().Get(recordsHeader); got != "" {
+		t.Errorf("%s = %q, want it unset without include_total", recordsHeader, got)
+	}
+}
+
+func TestHandler_ReportsBadPagingParam(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	nw := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	for _, params := range []url.Values{
+		{"limit": {"lots"}},
+		{"offset": {"-1"}},
+		{"order_by": {"receive_time"}},
+	} {
+		rec := doGetParams(t, srv.Handler(), reportsPath, params)
+		if rec.Code != http.StatusBadRequest {
+			t.Errorf("%v: expected 400, got %d", params, rec.Code)
+		}
+	}
+}
+
+func TestHandler_EventsUnioned(t *testing.T) {
+	// Puppetboard fetches a report's events as /events?query=["=","report",hash],
+	// and the report may live in either backend.
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[eventsPath] = `[` + event("h1", "r1", "Package[nginx]") + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[eventsPath] = `[` + event("h1", "r2", "Service[nginx]") + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), eventsPath, `["=","certname","h1"]`)
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	body := rec.Body.String()
+	if !strings.Contains(body, "Package[nginx]") || !strings.Contains(body, "Service[nginx]") {
+		t.Errorf("expected both backends' events: %s", body)
+	}
+}
+
+func TestHandler_EventsDedupedByIdentity(t *testing.T) {
+	dup := event("h1", "r1", "Package[nginx]")
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[eventsPath] = `[` + dup + `]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[eventsPath] = `[` + dup + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), eventsPath, "")
+	var got []json.RawMessage
+	if err := json.Unmarshal(rec.Body.Bytes(), &got); err != nil {
+		t.Fatal(err)
+	}
+	if len(got) != 1 {
+		t.Errorf("expected the duplicate event once, got %d: %s", len(got), rec.Body.String())
+	}
+}
+
+func TestHandler_ReportSubResourceFromHoldingBackend(t *testing.T) {
+	// Only old holds report r1, so its logs must come from old even though new
+	// is the primary — a pass-through would have 404'd.
+	const path = reportsPath + "/r1/logs"
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.bodies[path] = `[{"level":"notice","message":"from-old"}]`
+	nw := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), path, "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	if !strings.Contains(rec.Body.String(), "from-old") {
+		t.Errorf("expected old's logs, got %s", rec.Body.String())
+	}
+}
+
+func TestHandler_ReportSubResourceMissingEverywhere(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	nw := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), reportsPath+"/nope/events", "")
+	if rec.Code != http.StatusNotFound {
+		t.Fatalf("expected 404 when no backend holds the report, got %d", rec.Code)
+	}
+}
+
+func TestHandler_ReportsOneBackendDown(t *testing.T) {
+	old := newFakeBackend(t, `[]`, `[]`)
+	old.fail = true
+	nw := newFakeBackend(t, `[]`, `[]`)
+	nw.bodies[reportsPath] = `[` + report("h1", "r1", "2026-07-01T00:00:00Z") + `]`
+	srv := newTestServer(testConfig(old.srv.URL, nw.srv.URL, mergeStatic))
+
+	rec := doGetParams(t, srv.Handler(), reportsPath, url.Values{"order_by": {receiveDesc}})
+	if rec.Code != http.StatusOK {
+		t.Fatalf("expected 200 serving the survivor, got %d", rec.Code)
+	}
+	if got := hashes(t, rec.Body.Bytes()); !slices.Equal(got, []string{"r1"}) {
+		t.Errorf("merged reports = %v, want [r1]", got)
+	}
+}