Inject a pdbmux_source provenance fact

## Why
With several PuppetDBs behind one endpoint, consumers cannot tell which backend a node's data came from.

## How
- Add a synthetic `pdbmux_source` fact per certname on `/facts`, valued with the backend that won the facts merge, and stamp the same key on merged `/nodes` records.
- Emit all four fact keys including `environment`, which clients index directly.
- Skip injection for top-level `extract` queries, so `count()` and other aggregates keep the backends' own numbers, and for `/facts` queries constraining `name`; a `name` filter inside an `in` subquery still injects.
- Replace, never duplicate, an upstream fact of the configured name.
- Configure via `source_fact` / `source_fact_enabled` (`PDBMUX_SOURCE_FACT`, `PDBMUX_SOURCE_FACT_ENABLED`), defaulting to `pdbmux_source` enabled.

README.md

@@ -24,8 +24,8 @@ not PQL) is forwarded verbatim.
 
 | Path | Behaviour |
 |---|---|
-| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`. |
-| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics). |
+| `GET /pdb/query/v4/nodes` | Fan out to all backends, dedupe by `certname`, keep the record with the newer `report_timestamp`. Stamped with the winning backend's name (see provenance). |
+| `GET /pdb/query/v4/facts` | Fan out to all, and per `certname` keep **all** facts from the backend that owns that node (see merge semantics), plus a synthetic `pdbmux_source` fact naming it. |
 | `GET /pdb/query/v4/reports` | Fan out to all and serve the **union**, deduped by report `hash`, re-ordered and re-paged across backends. |
 | `GET /pdb/query/v4/events` | Fan out to all and serve the **union**, deduped by record identity, re-ordered and re-paged. |
 | `GET /pdb/query/v4/event-counts` | Fan out to all and **sum** each subject's counts into one row per subject. |
@@ -77,6 +77,62 @@ unknown fields survive untouched.
   - `include_total=true` on a summed endpoint reports the **merged** row count,
     not the sum of the backends' `X-Records`, since shared keys collapse.
 
+### Provenance: the `pdbmux_source` fact
+
+Once several PuppetDBs sit behind one endpoint, a consumer can no longer tell
+which backend a node's data came from. `pdbmux` makes that visible in the
+response itself, so nothing has to query each backend to find out:
+
+- **`/facts`** gains one extra fact record per `certname`, alongside the node's
+  real facts, in the shape of a real fact record — `certname`, `name`, `value`,
+  `environment` — with `value` set to the **backend name** from `backends` /
+  `PDBMUX_BACKENDS`. `environment` is copied from that node's own facts (all
+  four keys are always present, since clients index them directly).
+- **`/nodes`** gains a `pdbmux_source` **key** on each merged node record. A
+  node record carries no facts, so this is a synthetic field, not a fact — the
+  one key outside PuppetDB's documented node schema. Clients read node fields by
+  name, so an extra key is ignored by anything that doesn't want it.
+
+The value always names the backend **whose data won that endpoint's merge**, not
+a backend that merely holds the node. The two endpoints resolve their winner
+separately, so under `merge: static` they can legitimately disagree: `/facts`
+attributes a shared node to the first backend in configured order, while
+`/nodes` always attributes it to the backend holding the newer
+`report_timestamp`. Each answer describes the record it is attached to.
+
+If a backend genuinely reports a fact of the configured name, `pdbmux`
+**overrides** it — the upstream record is dropped and replaced, never duplicated,
+so the fact means exactly one thing and a node never carries two of it. Rename
+the synthetic fact via `source_fact` if the real one matters more.
+
+**Injection is skipped**, and the response passes through untouched, when:
+
+- the query has a top-level `extract` — it projects a column subset, and with a
+  `["function", ...]` column it aggregates. Injecting there would break the row
+  shape or silently inflate a `count()`, so **aggregate results are never
+  changed**;
+- (`/facts` only) the query constrains `name` — `["=","name","osfamily"]` and
+  friends ask for specific facts, and the synthetic record is not one of them.
+  Only the outer query is inspected: a `name` filter inside an `in`/`select_facts`
+  subquery narrows which *nodes* match, not which facts come back, so injection
+  still happens;
+- injection is turned off (see `source_fact_enabled`).
+
+**Not supported in v1: server-side filtering on the fact.** A query that selects
+it — `["=","name","pdbmux_source"]`, or an `extract` naming it — is forwarded to
+the backends like any other, and they return nothing, because the fact does not
+exist upstream. `pdbmux` does not evaluate the AST itself, so it cannot answer
+such a query correctly for every operator (`not`, `or`, subqueries) and does not
+pretend to for some. Read the fact from an unfiltered (or `certname`-filtered)
+`/facts` response and filter client-side. The same applies to the
+`/pdb/query/v4/facts/<name>` route, which is served unmerged pass-through.
+
+**Not covered:** `/factsets` and `/inventory`. Both carry facts, but `pdbmux`
+does not merge either today — they take the unmerged pass-through path, where
+the answer comes from whichever backend replied first rather than from a merge
+winner, so there is no owner to attribute. Injecting there would state a
+provenance that isn't true.
+
 ### Paging and ordering on the merged endpoints
 
 Each backend applies `order_by`/`limit`/`offset` to its own slice only, so
@@ -117,6 +173,8 @@ backends:                          # order is a tie-break only, not a ranking
 merge: freshness      # freshness | static
 timeout: 10s          # per-upstream request timeout
 freshness_ttl: 30s    # freshness-map cache TTL (freshness merge only)
+source_fact: pdbmux_source   # name of the synthetic provenance fact
+source_fact_enabled: true    # false serves backends' records untouched
 ```
 
 `backends[*].url` is a **base** URL (`scheme://host[:port]`); `pdbmux` appends
@@ -130,6 +188,8 @@ the `/pdb/query/v4/...` path per request.
 | `PDBMUX_TIMEOUT` | `timeout` (Go duration, e.g. `10s`) |
 | `PDBMUX_FRESHNESS_TTL` | `freshness_ttl` |
 | `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 |
 
 Flags: `--config`, `--listen`, `--merge`.
 

config.go

@@ -25,6 +25,8 @@ const (
 
 	defaultTimeout      = 10 * time.Second
 	defaultFreshnessTTL = 30 * time.Second
+
+	defaultSourceFact = "pdbmux_source"
 )
 
 var exampleBackends = []Backend{
@@ -44,6 +46,9 @@ type Config struct {
 	Timeout      time.Duration `yaml:"timeout"`
 	FreshnessTTL time.Duration `yaml:"freshness_ttl"`
 
+	SourceFact        string `yaml:"source_fact"`
+	SourceFactEnabled bool   `yaml:"source_fact_enabled"`
+
 	sourcePath string // file this config was read from, empty if none was found
 }
 
@@ -57,10 +62,12 @@ const (
 
 func DefaultConfig() Config {
 	return Config{
-		Listen:       defaultListen,
-		Merge:        mergeFreshness,
-		Timeout:      defaultTimeout,
-		FreshnessTTL: defaultFreshnessTTL,
+		Listen:            defaultListen,
+		Merge:             mergeFreshness,
+		Timeout:           defaultTimeout,
+		FreshnessTTL:      defaultFreshnessTTL,
+		SourceFact:        defaultSourceFact,
+		SourceFactEnabled: true,
 	}
 }
 
@@ -159,6 +166,14 @@ func applyEnv(cfg *Config, getenv func(string) string) {
 			cfg.FreshnessTTL = d
 		}
 	}
+	if v := getenv(envPrefix + "SOURCE_FACT"); v != "" {
+		cfg.SourceFact = v
+	}
+	if v := getenv(envPrefix + "SOURCE_FACT_ENABLED"); v != "" {
+		if b, err := strconv.ParseBool(v); err == nil {
+			cfg.SourceFactEnabled = b
+		}
+	}
 	if v := getenv(envPrefix + "BACKENDS"); v != "" {
 		if bs := parseBackends(v); len(bs) > 0 {
 			cfg.Backends = bs
@@ -216,6 +231,9 @@ func (c Config) Validate() error {
 	if c.Timeout <= 0 {
 		return fmt.Errorf("timeout must be positive")
 	}
+	if c.SourceFactEnabled && c.SourceFact == "" {
+		return fmt.Errorf("source_fact must be non-empty, or set source_fact_enabled to false")
+	}
 	return nil
 }
 
@@ -231,7 +249,8 @@ 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\n")
+		"#   PDBMUX_FRESHNESS_TTL, PDBMUX_BACKENDS (name=url,name=url),\n" +
+		"#   PDBMUX_SOURCE_FACT, PDBMUX_SOURCE_FACT_ENABLED.\n\n")
 	if err := os.WriteFile(path, append(header, data...), 0o644); err != nil {
 		return fmt.Errorf("writing config: %w", err)
 	}

config_test.go

@@ -118,6 +118,69 @@ func TestApplyEnv_Backends(t *testing.T) {
 	}
 }
 
+func TestDefaultConfig_SourceFact(t *testing.T) {
+	cfg := DefaultConfig()
+	if cfg.SourceFact != defaultSourceFact || !cfg.SourceFactEnabled {
+		t.Errorf("source fact defaults to %q enabled=%v, want %q enabled=true",
+			cfg.SourceFact, cfg.SourceFactEnabled, defaultSourceFact)
+	}
+}
+
+func TestApplyEnv_SourceFact(t *testing.T) {
+	cfg := testConfigValid()
+	env := map[string]string{envPrefix + "SOURCE_FACT": "origin_pdb"}
+	applyEnv(&cfg, func(k string) string { return env[k] })
+	if cfg.SourceFact != "origin_pdb" || !cfg.SourceFactEnabled {
+		t.Errorf("name override failed: %q enabled=%v", cfg.SourceFact, cfg.SourceFactEnabled)
+	}
+
+	cfg = testConfigValid()
+	env = map[string]string{envPrefix + "SOURCE_FACT_ENABLED": "false"}
+	applyEnv(&cfg, func(k string) string { return env[k] })
+	if cfg.SourceFactEnabled {
+		t.Error("PDBMUX_SOURCE_FACT_ENABLED=false must disable injection")
+	}
+
+	// A junk boolean leaves the default alone rather than disabling silently.
+	cfg = testConfigValid()
+	env = map[string]string{envPrefix + "SOURCE_FACT_ENABLED": "maybe"}
+	applyEnv(&cfg, func(k string) string { return env[k] })
+	if !cfg.SourceFactEnabled {
+		t.Error("unparseable bool must not change the setting")
+	}
+}
+
+// A config file omitting the key keeps the default; setting it false wins.
+func TestLoad_SourceFactFileOverride(t *testing.T) {
+	dir := t.TempDir()
+	t.Setenv("XDG_CONFIG_HOME", dir)
+	path := filepath.Join(dir, appName, configFileName)
+	if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+		t.Fatal(err)
+	}
+	write := func(body string) Config {
+		t.Helper()
+		if err := os.WriteFile(path, []byte(body), 0o644); err != nil {
+			t.Fatal(err)
+		}
+		cfg, err := Load("")
+		if err != nil {
+			t.Fatal(err)
+		}
+		return cfg
+	}
+
+	cfg := write("backends:\n  - name: a\n    url: http://a:8080\n")
+	if cfg.SourceFact != defaultSourceFact || !cfg.SourceFactEnabled {
+		t.Errorf("omitted keys must keep defaults: %q enabled=%v", cfg.SourceFact, cfg.SourceFactEnabled)
+	}
+
+	cfg = write("backends:\n  - name: a\n    url: http://a:8080\nsource_fact: origin_pdb\nsource_fact_enabled: false\n")
+	if cfg.SourceFact != "origin_pdb" || cfg.SourceFactEnabled {
+		t.Errorf("file override failed: %q enabled=%v", cfg.SourceFact, cfg.SourceFactEnabled)
+	}
+}
+
 func TestValidate(t *testing.T) {
 	cases := []struct {
 		name    string
@@ -130,6 +193,8 @@ func TestValidate(t *testing.T) {
 		{"missing url", func(c *Config) { c.Backends[0].URL = "" }, true},
 		{"bad merge", func(c *Config) { c.Merge = "wrong" }, true},
 		{"zero timeout", func(c *Config) { c.Timeout = 0 }, true},
+		{"empty source fact while enabled", func(c *Config) { c.SourceFact = "" }, true},
+		{"empty source fact while disabled", func(c *Config) { c.SourceFact = ""; c.SourceFactEnabled = false }, false},
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {

main.go

@@ -161,6 +161,11 @@ func printConfig(cfg Config) {
 	fmt.Printf("merge        : %s\n", cfg.Merge)
 	fmt.Printf("timeout      : %s\n", durationString(cfg.Timeout))
 	fmt.Printf("freshness_ttl: %s\n", durationString(cfg.FreshnessTTL))
+	if cfg.SourceFactEnabled {
+		fmt.Printf("source_fact  : %s\n", cfg.SourceFact)
+	} else {
+		fmt.Printf("source_fact  : disabled\n")
+	}
 	fmt.Println("backends:")
 	for _, b := range cfg.Backends {
 		fmt.Printf("  - %-8s %s\n", b.Name, b.URL)

merge.go

@@ -11,12 +11,16 @@ type record struct {
 	Certname        string
 	ReportTimestamp string // only populated for /nodes records
 	Hash            string // only populated for /reports records
+	Name            string // only populated for /facts records
+	Environment     string
 }
 
 type recordMeta struct {
 	Certname        string `json:"certname"`
 	ReportTimestamp string `json:"report_timestamp"`
 	Hash            string `json:"hash"`
+	Name            string `json:"name"`
+	Environment     string `json:"environment"`
 }
 
 func decodeRecords(body []byte) ([]record, error) {
@@ -33,6 +37,8 @@ func decodeRecords(body []byte) ([]record, error) {
 			Certname:        m.Certname,
 			ReportTimestamp: m.ReportTimestamp,
 			Hash:            m.Hash,
+			Name:            m.Name,
+			Environment:     m.Environment,
 		})
 	}
 	return out, nil
@@ -50,10 +56,12 @@ func parseTimestamp(s string) time.Time {
 }
 
 // Ties keep the earlier backend's record — a deterministic tie-break, not a preference.
-func mergeNodes(results []backendResult) []json.RawMessage {
+// A non-nil inject stamps each surviving record with the backend that supplied it.
+func mergeNodes(results []backendResult, inject *sourceInjector) []json.RawMessage {
 	type pick struct {
-		raw json.RawMessage
-		ts  time.Time
+		raw     json.RawMessage
+		ts      time.Time
+		backend string
 	}
 	best := map[string]pick{}
 	var order []string
@@ -62,18 +70,19 @@ func mergeNodes(results []backendResult) []json.RawMessage {
 			ts := parseTimestamp(rec.ReportTimestamp)
 			cur, ok := best[rec.Certname]
 			if !ok {
-				best[rec.Certname] = pick{raw: rec.Raw, ts: ts}
+				best[rec.Certname] = pick{raw: rec.Raw, ts: ts, backend: res.name}
 				order = append(order, rec.Certname)
 				continue
 			}
 			if ts.After(cur.ts) {
-				best[rec.Certname] = pick{raw: rec.Raw, ts: ts}
+				best[rec.Certname] = pick{raw: rec.Raw, ts: ts, backend: res.name}
 			}
 		}
 	}
 	out := make([]json.RawMessage, 0, len(order))
 	for _, cn := range order {
-		out = append(out, best[cn].raw)
+		p := best[cn]
+		out = append(out, inject.stamp(p.raw, p.backend))
 	}
 	return out
 }
@@ -105,16 +114,17 @@ func buildFreshness(results []backendResult) freshness {
 }
 
 // owner names the winning backend per certname; a nil owner (static merge), or one holding no facts for that certname, falls back to configured order.
-func mergeFacts(results []backendResult, owner func(certname string) string) []json.RawMessage {
+// A non-nil inject appends the synthetic source fact after each certname's block, naming the backend that won.
+func mergeFacts(results []backendResult, owner func(certname string) string, inject *sourceInjector) []json.RawMessage {
 	present := map[string][]string{} // certname -> backend names, in configured order
-	byKey := map[string][]json.RawMessage{}
+	byKey := map[string][]record{}
 	for _, res := range results {
 		for _, rec := range res.records {
 			key := rec.Certname + "\x00" + res.name
 			if _, ok := byKey[key]; !ok {
 				present[rec.Certname] = append(present[rec.Certname], res.name)
 			}
-			byKey[key] = append(byKey[key], rec.Raw)
+			byKey[key] = append(byKey[key], rec)
 		}
 	}
 
@@ -140,11 +150,30 @@ func mergeFacts(results []backendResult, owner func(certname string) string) []j
 		if !contains(backends, chosen) {
 			chosen = backends[0]
 		}
-		out = append(out, byKey[cn+"\x00"+chosen]...)
+		recs := byKey[cn+"\x00"+chosen]
+		for _, rec := range recs {
+			// An upstream fact of the same name is dropped: pdbmux's own value is authoritative.
+			if inject.claims(rec.Name) {
+				continue
+			}
+			out = append(out, rec.Raw)
+		}
+		if synth := inject.factRecord(cn, chosen, environmentOf(recs)); synth != nil {
+			out = append(out, synth)
+		}
 	}
 	return out
 }
 
+func environmentOf(recs []record) string {
+	for _, rec := range recs {
+		if rec.Environment != "" {
+			return rec.Environment
+		}
+	}
+	return ""
+}
+
 func contains(s []string, v string) bool {
 	for _, x := range s {
 		if x == v {

merge_test.go

@@ -92,7 +92,7 @@ func TestMergeNodes_NewerWins(t *testing.T) {
 	a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"), node("h2", "2026-07-10T00:00:00Z"))
 	b := recs(t, "b", node("h1", "2026-07-20T00:00:00Z"), node("h3", "2026-07-05T00:00:00Z"))
 
-	merged := mergeNodes([]backendResult{a, b})
+	merged := mergeNodes([]backendResult{a, b}, nil)
 	got := map[string]string{}
 	for _, r := range merged {
 		var m recordMeta
@@ -117,7 +117,7 @@ func TestMergeNodes_OneBackendOnly(t *testing.T) {
 	a := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"))
 	// b returned nothing (e.g. empty result).
 	b := backendResult{name: "b"}
-	merged := mergeNodes([]backendResult{a, b})
+	merged := mergeNodes([]backendResult{a, b}, nil)
 	if len(merged) != 1 || certnames(t, merged)[0] != "h1" {
 		t.Fatalf("expected only h1, got %v", certnames(t, merged))
 	}
@@ -127,7 +127,7 @@ func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) {
 	// Equal timestamps: the backend listed first wins, as a tie-break.
 	first := recs(t, "b", node("h1", "2026-07-01T00:00:00Z"))
 	second := recs(t, "a", node("h1", "2026-07-01T00:00:00Z"))
-	merged := mergeNodes([]backendResult{first, second})
+	merged := mergeNodes([]backendResult{first, second}, nil)
 	if len(merged) != 1 {
 		t.Fatalf("expected 1 record, got %d", len(merged))
 	}
@@ -139,7 +139,7 @@ func TestMergeNodes_TieKeepsEarlierBackend(t *testing.T) {
 
 func TestMergeNodes_PreservesUnknownFields(t *testing.T) {
 	a := recs(t, "a", `{"certname":"h1","report_timestamp":"2026-07-01T00:00:00Z","extra":{"deep":42}}`)
-	merged := mergeNodes([]backendResult{a})
+	merged := mergeNodes([]backendResult{a}, nil)
 	if len(merged) != 1 {
 		t.Fatalf("expected 1 record")
 	}
@@ -156,7 +156,7 @@ func TestMergeFacts_NilOwnerUsesConfiguredOrder(t *testing.T) {
 	a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h2", "role", "db-a", ""))
 	b := recs(t, "b", fact("h1", "role", "web-b", ""))
 
-	merged := mergeFacts([]backendResult{b, a}, nil)
+	merged := mergeFacts([]backendResult{b, a}, nil, nil)
 	got := factValues(t, merged)
 	assertContains(t, got, "h1:role=web-b")
 	assertNotContains(t, got, "h1:role=web-a")
@@ -180,7 +180,7 @@ func TestMergeFacts_Freshness_NewerBackendWins(t *testing.T) {
 		}
 		return "b"
 	}
-	merged := mergeFacts([]backendResult{b, a}, owner)
+	merged := mergeFacts([]backendResult{b, a}, owner, nil)
 	got := factValues(t, merged)
 	// h1 -> all a facts, no b facts.
 	assertContains(t, got, "h1:role=web-a")
@@ -198,7 +198,7 @@ func TestMergeFacts_OwnerMissingFallsBackToConfiguredOrder(t *testing.T) {
 	// backend in the slice that has some.
 	first := recs(t, "b", fact("h1", "role", "web-b", ""))
 	second := recs(t, "a", fact("h1", "role", "web-a", ""))
-	merged := mergeFacts([]backendResult{first, second}, func(string) string { return "ghost" })
+	merged := mergeFacts([]backendResult{first, second}, func(string) string { return "ghost" }, nil)
 	got := factValues(t, merged)
 	assertContains(t, got, "h1:role=web-b") // b is first in slice
 	assertNotContains(t, got, "h1:role=web-a")

server.go

@@ -67,9 +67,9 @@ func (s *Server) handleQuery(w http.ResponseWriter, r *http.Request) {
 	}
 	switch r.URL.Path {
 	case nodesPath:
-		s.serveMerged(w, r, nodesPath, s.mergeNodesResponse)
+		s.serveMerged(w, r, nodesPath, s.mergeNodesResponse(r))
 	case factsPath:
-		s.serveMerged(w, r, factsPath, s.mergeFactsResponse)
+		s.serveMerged(w, r, factsPath, s.mergeFactsResponse(r))
 	case reportsPath:
 		s.serveReports(w, r)
 	case eventsPath:
@@ -224,16 +224,22 @@ func rawRecords(recs []record) []json.RawMessage {
 	return out
 }
 
-func (s *Server) mergeNodesResponse(results []backendResult) []json.RawMessage {
-	return mergeNodes(results)
+func (s *Server) mergeNodesResponse(r *http.Request) func([]backendResult) []json.RawMessage {
+	inject := s.newSourceInjector(r.URL.Query().Get("query"), false)
+	return func(results []backendResult) []json.RawMessage {
+		return mergeNodes(results, inject)
+	}
 }
 
-func (s *Server) mergeFactsResponse(results []backendResult) []json.RawMessage {
-	if s.cfg.Merge == mergeStatic {
-		return mergeFacts(results, nil)
+func (s *Server) mergeFactsResponse(r *http.Request) func([]backendResult) []json.RawMessage {
+	inject := s.newSourceInjector(r.URL.Query().Get("query"), true)
+	return func(results []backendResult) []json.RawMessage {
+		if s.cfg.Merge == mergeStatic {
+			return mergeFacts(results, nil, inject)
+		}
+		fresh := s.freshnessMap(context.Background(), results)
+		return mergeFacts(results, func(cn string) string { return fresh[cn] }, inject)
 	}
-	fresh := s.freshnessMap(context.Background(), results)
-	return mergeFacts(results, func(cn string) string { return fresh[cn] })
 }
 
 // Queries /nodes unfiltered rather than reusing the request's results, because a /facts query's certname set can differ.

server_test.go

@@ -121,11 +121,13 @@ func truncate(t *testing.T, body, limit string) string {
 
 func testConfig(aURL, bURL, merge string) Config {
 	return Config{
-		Listen:       ":0",
-		Backends:     []Backend{{Name: "a", URL: aURL}, {Name: "b", URL: bURL}},
-		Merge:        merge,
-		Timeout:      2 * time.Second,
-		FreshnessTTL: 30 * time.Second,
+		Listen:            ":0",
+		Backends:          []Backend{{Name: "a", URL: aURL}, {Name: "b", URL: bURL}},
+		Merge:             merge,
+		Timeout:           2 * time.Second,
+		FreshnessTTL:      30 * time.Second,
+		SourceFact:        defaultSourceFact,
+		SourceFactEnabled: true,
 	}
 }
 

source.go

@@ -0,0 +1,131 @@
+package main
+
+import "encoding/json"
+
+// sourceInjector synthesises the provenance fact naming the backend whose data
+// won the merge for a given certname. A nil *sourceInjector is the disabled
+// case, so every method is nil-safe and callers need no branch.
+type sourceInjector struct {
+	name string
+}
+
+// newSourceInjector returns nil when injection is off for this request.
+func (s *Server) newSourceInjector(query string, factEntity bool) *sourceInjector {
+	if !s.cfg.SourceFactEnabled || s.cfg.SourceFact == "" {
+		return nil
+	}
+	if !injectable(query, factEntity) {
+		return nil
+	}
+	return &sourceInjector{name: s.cfg.SourceFact}
+}
+
+// claims reports whether an upstream record is the one the injector replaces.
+func (si *sourceInjector) claims(factName string) bool {
+	return si != nil && factName != "" && factName == si.name
+}
+
+// factRecord builds the synthetic /facts record, or nil when disabled.
+// environment is copied from the node's real facts. All four keys of a fact
+// record are always emitted, empty environment included: pypuppetdb indexes them
+// directly (types.py Fact.create_from_dict), so an omitted key is a KeyError.
+func (si *sourceInjector) factRecord(certname, backend, environment string) json.RawMessage {
+	if si == nil {
+		return nil
+	}
+	raw, err := json.Marshal(struct {
+		Certname    string `json:"certname"`
+		Environment string `json:"environment"`
+		Name        string `json:"name"`
+		Value       string `json:"value"`
+	}{Certname: certname, Environment: environment, Name: si.name, Value: backend})
+	if err != nil {
+		return nil
+	}
+	return raw
+}
+
+// stamp adds the provenance key to a /nodes record, overwriting any existing
+// key of that name. A record that is not a JSON object passes through untouched.
+func (si *sourceInjector) stamp(raw json.RawMessage, backend string) json.RawMessage {
+	if si == nil {
+		return raw
+	}
+	var obj map[string]json.RawMessage
+	if json.Unmarshal(raw, &obj) != nil || obj == nil {
+		return raw
+	}
+	value, err := json.Marshal(backend)
+	if err != nil {
+		return raw
+	}
+	obj[si.name] = value
+	out, err := json.Marshal(obj)
+	if err != nil {
+		return raw
+	}
+	return out
+}
+
+// injectable reports whether a response to this query may carry the synthetic
+// record. Two shapes are excluded, both because the client asked for something
+// the synthetic record is not part of:
+//
+//   - a top-level `extract`, which projects a column subset and, with a
+//     `["function", ...]` column, aggregates — injecting there would corrupt the
+//     row shape or silently inflate a count();
+//   - on the facts entity, any outer constraint on `name`, which selects
+//     specific facts. Subquery operands are not descended into: they choose which
+//     nodes match, not which facts come back.
+func injectable(query string, factEntity bool) bool {
+	if query == "" {
+		return true
+	}
+	var ast []json.RawMessage
+	if json.Unmarshal([]byte(query), &ast) != nil || len(ast) == 0 {
+		// Not an AST array pdbmux can reason about; leave the response alone.
+		return false
+	}
+	var op string
+	if json.Unmarshal(ast[0], &op) != nil {
+		return false
+	}
+	if op == "extract" {
+		return false
+	}
+	if !factEntity {
+		return true
+	}
+	return !constrainsField(ast, "name")
+}
+
+// constrainsField walks the boolean skeleton of an AST node looking for a
+// comparison whose field operand is field. Only and/or/not are descended into;
+// anything else, including the subquery operand of `in`, is left alone.
+func constrainsField(parts []json.RawMessage, field string) bool {
+	if len(parts) == 0 {
+		return false
+	}
+	var op string
+	if json.Unmarshal(parts[0], &op) != nil {
+		return false
+	}
+	switch op {
+	case "and", "or", "not":
+		for _, p := range parts[1:] {
+			var sub []json.RawMessage
+			if json.Unmarshal(p, &sub) != nil {
+				continue
+			}
+			if constrainsField(sub, field) {
+				return true
+			}
+		}
+		return false
+	}
+	if len(parts) < 2 {
+		return false
+	}
+	var name string
+	return json.Unmarshal(parts[1], &name) == nil && name == field
+}

source_test.go

@@ -0,0 +1,373 @@
+package main
+
+import (
+	"encoding/json"
+	"net/http"
+	"slices"
+	"testing"
+)
+
+// sourceValues returns certname -> value of the synthetic fact record, and the
+// number of records carrying that fact name.
+func sourceValues(t *testing.T, body []byte, factName string) (map[string]string, int) {
+	t.Helper()
+	var raws []json.RawMessage
+	if err := json.Unmarshal(body, &raws); err != nil {
+		t.Fatalf("unmarshal %s: %v", body, err)
+	}
+	out := map[string]string{}
+	n := 0
+	for _, raw := range raws {
+		var m struct {
+			Certname string `json:"certname"`
+			Name     string `json:"name"`
+			Value    string `json:"value"`
+		}
+		if json.Unmarshal(raw, &m) != nil || m.Name != factName {
+			continue
+		}
+		out[m.Certname] = m.Value
+		n++
+	}
+	return out, n
+}
+
+// nodeSources returns certname -> the stamped provenance field on /nodes records.
+func nodeSources(t *testing.T, body []byte, field string) map[string]string {
+	t.Helper()
+	var raws []json.RawMessage
+	if err := json.Unmarshal(body, &raws); err != nil {
+		t.Fatalf("unmarshal %s: %v", body, err)
+	}
+	out := map[string]string{}
+	for _, raw := range raws {
+		var obj map[string]json.RawMessage
+		if err := json.Unmarshal(raw, &obj); err != nil {
+			t.Fatalf("unmarshal record %s: %v", raw, err)
+		}
+		var m recordMeta
+		_ = json.Unmarshal(raw, &m)
+		v, ok := obj[field]
+		if !ok {
+			continue
+		}
+		var s string
+		if err := json.Unmarshal(v, &s); err != nil {
+			t.Fatalf("provenance field of %s is not a string: %v", raw, err)
+		}
+		out[m.Certname] = s
+	}
+	return out
+}
+
+func factEnv(cn, name, val, env string) string {
+	return `{"certname":"` + cn + `","name":"` + name + `","value":"` + val + `","environment":"` + env + `"}`
+}
+
+// Both backends hold h1; a holds its newer report, so h1's facts and its
+// provenance fact must both come from a.
+func TestHandler_FactsSourceFollowsMergeOwner(t *testing.T) {
+	a := newFakeBackend(t,
+		`[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-01T00:00:00Z")+`]`,
+		`[`+factEnv("h1", "role", "web-a", "production")+`,`+factEnv("h2", "role", "db-a", "production")+`]`)
+	b := newFakeBackend(t,
+		`[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`,
+		`[`+factEnv("h1", "role", "web-b", "staging")+`,`+factEnv("h2", "role", "db-b", "staging")+`]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
+
+	rec := doGet(t, srv.Handler(), factsPath, "")
+	if rec.Code != http.StatusOK {
+		t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+	}
+	got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
+	if n != 2 {
+		t.Fatalf("expected one %s record per certname, got %d: %s", defaultSourceFact, n, rec.Body.String())
+	}
+	if got["h1"] != "a" || got["h2"] != "b" {
+		t.Errorf("provenance must name the backend that won the merge, got %v", got)
+	}
+}
+
+// The synthetic record carries the node's own environment so it groups with the
+// real facts rather than landing in an unrelated environment.
+func TestHandler_FactsSourceCopiesEnvironment(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+factEnv("h1", "role", "web", "staging")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath, "")
+	var raws []json.RawMessage
+	if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil {
+		t.Fatal(err)
+	}
+	var found bool
+	for _, raw := range raws {
+		var m recordMeta
+		if json.Unmarshal(raw, &m) != nil || m.Name != defaultSourceFact {
+			continue
+		}
+		found = true
+		if m.Environment != "staging" {
+			t.Errorf("environment = %q, want staging: %s", m.Environment, raw)
+		}
+	}
+	if !found {
+		t.Fatalf("no %s record: %s", defaultSourceFact, rec.Body.String())
+	}
+}
+
+// A node record's provenance names the backend whose node record won /nodes'
+// own report_timestamp merge.
+func TestHandler_NodesSourceStamped(t *testing.T) {
+	a := newFakeBackend(t,
+		`[`+node("h1", "2026-07-01T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`, `[]`)
+	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: %s", rec.Code, rec.Body.String())
+	}
+	got := nodeSources(t, rec.Body.Bytes(), defaultSourceFact)
+	if got["h1"] != "b" || got["h2"] != "a" {
+		t.Errorf("node provenance = %v, want h1=b h2=a", got)
+	}
+}
+
+// Stamping must not drop unknown upstream fields.
+func TestHandler_NodesSourceKeepsUpstreamFields(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), nodesPath, "")
+	var raws []json.RawMessage
+	if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil {
+		t.Fatal(err)
+	}
+	if len(raws) != 1 {
+		t.Fatalf("expected 1 node, got %d", len(raws))
+	}
+	var obj map[string]json.RawMessage
+	if err := json.Unmarshal(raws[0], &obj); err != nil {
+		t.Fatal(err)
+	}
+	for _, k := range []string{"certname", "report_timestamp", "latest_report_status", defaultSourceFact} {
+		if _, ok := obj[k]; !ok {
+			t.Errorf("field %q missing from stamped record: %s", k, raws[0])
+		}
+	}
+}
+
+func TestHandler_SourceDisabled(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic)
+	cfg.SourceFactEnabled = false
+	srv := newTestServer(cfg)
+
+	facts := doGet(t, srv.Handler(), factsPath, "")
+	if _, n := sourceValues(t, facts.Body.Bytes(), defaultSourceFact); n != 0 {
+		t.Errorf("disabled injection still produced %d records: %s", n, facts.Body.String())
+	}
+	nodes := doGet(t, srv.Handler(), nodesPath, "")
+	if got := nodeSources(t, nodes.Body.Bytes(), defaultSourceFact); len(got) != 0 {
+		t.Errorf("disabled injection still stamped nodes: %v", got)
+	}
+}
+
+func TestHandler_SourceFactNameOverride(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic)
+	cfg.SourceFact = "origin_pdb"
+	srv := newTestServer(cfg)
+
+	facts := doGet(t, srv.Handler(), factsPath, "")
+	got, n := sourceValues(t, facts.Body.Bytes(), "origin_pdb")
+	if n != 1 || got["h1"] != "a" {
+		t.Errorf("override name not honoured: %s", facts.Body.String())
+	}
+	if _, n := sourceValues(t, facts.Body.Bytes(), defaultSourceFact); n != 0 {
+		t.Errorf("default name still emitted alongside the override: %s", facts.Body.String())
+	}
+
+	nodes := doGet(t, srv.Handler(), nodesPath, "")
+	if got := nodeSources(t, nodes.Body.Bytes(), "origin_pdb"); got["h1"] != "a" {
+		t.Errorf("override name not honoured on /nodes: %v", got)
+	}
+}
+
+// An upstream fact of the configured name is replaced, not duplicated: pdbmux's
+// own attribution is authoritative.
+func TestHandler_UpstreamSourceFactOverridden(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`,`+fact("h1", defaultSourceFact, "stale-value", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath, "")
+	got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
+	if n != 1 {
+		t.Fatalf("expected exactly 1 %s record, got %d: %s", defaultSourceFact, n, rec.Body.String())
+	}
+	if got["h1"] != "a" {
+		t.Errorf("upstream value survived: %v", got)
+	}
+}
+
+// A count() must report the backends' real fact count, not one inflated by a
+// record pdbmux invented.
+func TestHandler_SourceNotInjectedOnAggregate(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	a.bodies[factsPath] = `[{"count":3}]`
+	b.bodies[factsPath] = `[{"count":2}]`
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath, `["extract",[["function","count"]]]`)
+	if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 {
+		t.Errorf("aggregate response gained %d synthetic records: %s", n, rec.Body.String())
+	}
+	// Every row must be one an upstream actually returned: an aggregate row pdbmux
+	// invented or rewrote would change the count the client sees.
+	var raws []json.RawMessage
+	if err := json.Unmarshal(rec.Body.Bytes(), &raws); err != nil {
+		t.Fatal(err)
+	}
+	upstream := []string{`{"count":3}`, `{"count":2}`}
+	for _, raw := range raws {
+		if !slices.Contains(upstream, string(raw)) {
+			t.Errorf("aggregate row %s is not an upstream row", raw)
+		}
+	}
+}
+
+// A query naming a specific fact asked for that fact only.
+func TestHandler_SourceNotInjectedWhenNameFiltered(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	for _, q := range []string{
+		`["=","name","role"]`,
+		`["and",["=","certname","h1"],["=","name","role"]]`,
+		`["=","name","` + defaultSourceFact + `"]`,
+	} {
+		rec := doGet(t, srv.Handler(), factsPath, q)
+		if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 0 {
+			t.Errorf("query %s gained %d synthetic records: %s", q, n, rec.Body.String())
+		}
+	}
+}
+
+// A certname filter selects nodes, not facts, so the full fact set — synthetic
+// record included — is still the right answer.
+func TestHandler_SourceInjectedWhenOnlyCertnameFiltered(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", "role", "web", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+	rec := doGet(t, srv.Handler(), factsPath, `["=","certname","h1"]`)
+	if _, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 1 {
+		t.Errorf("expected 1 synthetic record, got %d: %s", n, rec.Body.String())
+	}
+}
+
+func TestInjectable(t *testing.T) {
+	tests := []struct {
+		name       string
+		query      string
+		factEntity bool
+		want       bool
+	}{
+		{"empty query", "", true, true},
+		{"certname filter", `["=","certname","h1"]`, true, true},
+		{"regex certname filter", `["~","certname","^web"]`, true, true},
+		{"name filter", `["=","name","os"]`, true, false},
+		{"name regex filter", `["~","name","^net"]`, true, false},
+		{"name under and", `["and",["=","certname","h1"],["=","name","os"]]`, true, false},
+		{"name under or", `["or",["=","name","os"],["=","name","kernel"]]`, true, false},
+		{"name under not", `["not",["=","name","os"]]`, true, false},
+		{"name in list", `["in","name",["array",["os"]]]`, true, false},
+		// A select_facts subquery narrows which nodes match; the outer response is
+		// still whole fact sets, so the synthetic record belongs in it.
+		{"name only inside subquery", `["in","certname",["extract",["certname"],["select_facts",["=","name","os"]]]]`, true, true},
+		{"top-level extract", `["extract",["certname","value"],["=","certname","h1"]]`, true, false},
+		{"aggregate extract", `["extract",[["function","count"]]]`, true, false},
+		{"nodes name filter is not a fact filter", `["=","name","os"]`, false, true},
+		{"nodes extract", `["extract",["certname"]]`, false, false},
+		{"unparseable query", `not json`, true, false},
+		{"non-array query", `{"a":1}`, true, false},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := injectable(tc.query, tc.factEntity); got != tc.want {
+				t.Errorf("injectable(%s, %v) = %v, want %v", tc.query, tc.factEntity, got, tc.want)
+			}
+		})
+	}
+}
+
+// pypuppetdb reads certname/name/value/environment by direct index, so a
+// missing key is a KeyError there — all four are always present.
+func TestSourceInjector_FactRecordHasEveryFactKey(t *testing.T) {
+	si := &sourceInjector{name: defaultSourceFact}
+	for _, env := range []string{"production", ""} {
+		var obj map[string]json.RawMessage
+		if err := json.Unmarshal(si.factRecord("h1", "a", env), &obj); err != nil {
+			t.Fatal(err)
+		}
+		for _, k := range []string{"certname", "name", "value", "environment"} {
+			if _, ok := obj[k]; !ok {
+				t.Errorf("environment=%q: key %q missing from synthetic fact", env, k)
+			}
+		}
+		if len(obj) != 4 {
+			t.Errorf("synthetic fact has %d keys, want the 4 of a real fact record: %v", len(obj), obj)
+		}
+	}
+}
+
+// A nil injector is the disabled path and must leave every input untouched.
+func TestSourceInjector_NilIsInert(t *testing.T) {
+	var si *sourceInjector
+	if si.claims(defaultSourceFact) {
+		t.Error("nil injector claims a fact name")
+	}
+	if si.factRecord("h1", "a", "production") != nil {
+		t.Error("nil injector produced a record")
+	}
+	raw := json.RawMessage(`{"certname":"h1"}`)
+	if got := si.stamp(raw, "a"); string(got) != string(raw) {
+		t.Errorf("nil injector rewrote %s to %s", raw, got)
+	}
+}
+
+// A response element that is not a JSON object cannot be stamped, and must be
+// passed through rather than dropped or mangled.
+func TestSourceInjector_StampNonObject(t *testing.T) {
+	si := &sourceInjector{name: defaultSourceFact}
+	for _, raw := range []string{`"scalar"`, `[1,2]`, `null`} {
+		if got := si.stamp(json.RawMessage(raw), "a"); string(got) != raw {
+			t.Errorf("stamp(%s) = %s, want unchanged", raw, got)
+		}
+	}
+}
+
+func TestMergeFacts_SourceOrderedAfterOwnersFacts(t *testing.T) {
+	a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h1", "kernel", "Linux", ""))
+	b := recs(t, "b", fact("h1", "role", "web-b", ""))
+	merged := mergeFacts([]backendResult{a, b}, nil, &sourceInjector{name: defaultSourceFact})
+
+	got := factValues(t, merged)
+	want := []string{"h1:role=web-a", "h1:kernel=Linux", "h1:" + defaultSourceFact + "=a"}
+	if !slices.Equal(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}