Override the source fact on every query shape

- drop upstream facts of the configured name whenever the feature is enabled,
  independent of the per-query injection gate, and log the drop once per request
- walk the whole AST for a nested extract and skip injection when one is found
  outside an in subquery
- skip the environment scan on /facts when nothing is injected
- document the override rule and that PQL-syntax queries never get the fact

README.md

@@ -114,17 +114,25 @@ 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.
+While the fact is enabled the configured name is `pdbmux`'s alone. A `/facts`
+record of that name coming from a backend is **always dropped**, on every query
+shape — including the shapes below, where nothing is injected in its place — so
+the fact means exactly one thing and a node never carries two of it. Each request
+that drops one logs it once. Only `source_fact_enabled: false` restores upstream
+records of that name; rename the synthetic fact via `source_fact` if the real one
+matters more.
 
-**Injection is skipped**, and the response passes through untouched, when:
+**Injection is skipped**, and no synthetic record is added, 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**;
+- the query contains an `extract` anywhere outside an `in` subquery — 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**. The whole query is walked, so an `extract` nested
+  under `and`/`or`/`not`/`from` skips injection too; an `extract` inside an `in`
+  operand projects the subquery rather than the response, so it does not;
+- the query is not an AST array — every **PQL-syntax** query (`facts { certname
+  = "web1" }`) lands here. `pdbmux` cannot tell what such a query projects, so it
+  never injects into a PQL response. Use the AST form to get the fact;
 - (`/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`

merge.go

@@ -114,7 +114,7 @@ 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.
-// A non-nil inject appends the synthetic source fact after each certname's block, naming the backend that won.
+// inject appends the synthetic source fact after each certname's block, naming the backend that won, and always drops upstream facts of that name.
 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][]record{}
@@ -152,12 +152,17 @@ func mergeFacts(results []backendResult, owner func(certname string) string, inj
 		}
 		recs := byKey[cn+"\x00"+chosen]
 		for _, rec := range recs {
-			// An upstream fact of the same name is dropped: pdbmux's own value is authoritative.
+			// An upstream fact of the configured name is dropped on every query shape,
+			// injected or not: while the feature is on the name is pdbmux's alone.
 			if inject.claims(rec.Name) {
+				inject.suppressed++
 				continue
 			}
 			out = append(out, rec.Raw)
 		}
+		if !inject.injects() {
+			continue
+		}
 		if synth := inject.factRecord(cn, chosen, environmentOf(recs)); synth != nil {
 			out = append(out, synth)
 		}

server.go

@@ -258,11 +258,15 @@ func (s *Server) mergeNodesResponse(r *http.Request) func([]backendResult) []jso
 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 {
+		var merged []json.RawMessage
 		if s.cfg.Merge == mergeStatic {
-			return mergeFacts(results, nil, inject)
+			merged = mergeFacts(results, nil, inject)
+		} else {
+			fresh := s.freshnessMap(context.Background(), results)
+			merged = 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] }, inject)
+		inject.logSuppressed(s.log)
+		return merged
 	}
 }
 

source.go

@@ -1,36 +1,56 @@
 package main
 
-import "encoding/json"
+import (
+	"encoding/json"
+	"log"
+)
 
-// 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.
+// sourceInjector owns the configured fact name for one request. A nil
+// *sourceInjector is the feature-disabled case, so every method is nil-safe and
+// callers need no branch.
 type sourceInjector struct {
 	name string
+	// inject is false when the query shape rules synthesis out. Suppression of an
+	// upstream fact of the same name does not depend on it.
+	inject     bool
+	suppressed int
 }
 
-// newSourceInjector returns nil when injection is off for this request.
+// newSourceInjector returns nil only when the feature is off; a gated query
+// yields an injector that suppresses but does not synthesise.
 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}
+	return &sourceInjector{name: s.cfg.SourceFact, inject: injectable(query, factEntity)}
 }
 
-// claims reports whether an upstream record is the one the injector replaces.
+// claims reports whether an upstream record carries the name pdbmux owns. While
+// the feature is enabled the name means one thing on every query shape, so such
+// a record is dropped even when the query gate has ruled synthesis out.
 func (si *sourceInjector) claims(factName string) bool {
 	return si != nil && factName != "" && factName == si.name
 }
 
+// injects reports whether this response may carry the synthetic record.
+func (si *sourceInjector) injects() bool {
+	return si != nil && si.inject
+}
+
+// logSuppressed reports, once per request, that upstream records were dropped.
+func (si *sourceInjector) logSuppressed(l *log.Logger) {
+	if si == nil || si.suppressed == 0 || l == nil {
+		return
+	}
+	l.Printf("info: dropped %d upstream %q fact record(s); pdbmux owns that fact name", si.suppressed, 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 {
+	if !si.injects() {
 		return nil
 	}
 	raw, err := json.Marshal(struct {
@@ -48,7 +68,7 @@ func (si *sourceInjector) factRecord(certname, backend, environment string) json
 // 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 {
+	if !si.injects() {
 		return raw
 	}
 	var obj map[string]json.RawMessage
@@ -68,12 +88,14 @@ func (si *sourceInjector) stamp(raw json.RawMessage, backend string) json.RawMes
 }
 
 // injectable reports whether a response to this query may carry the synthetic
-// record. Two shapes are excluded, both because the client asked for something
+// record. Three shapes are excluded, each 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();
+//   - a query that is not an AST array, which includes every PQL-syntax query:
+//     pdbmux cannot tell what it projects, so it changes nothing;
+//   - an `extract` anywhere in the query's own projection scope, 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.
@@ -90,7 +112,7 @@ func injectable(query string, factEntity bool) bool {
 	if json.Unmarshal(ast[0], &op) != nil {
 		return false
 	}
-	if op == "extract" {
+	if hasExtract(ast) {
 		return false
 	}
 	if !factEntity {
@@ -99,6 +121,41 @@ func injectable(query string, factEntity bool) bool {
 	return !constrainsField(ast, "name")
 }
 
+// hasExtract reports whether an extract appears anywhere in the query's own
+// projection scope. openvoxdb accepts an extract as an operand of a boolean
+// operator — engine.clj's user-node->plan-node sends every and/or/not operand
+// back through itself (src/puppetlabs/puppetdb/query_eng/engine.clj:2697-2733)
+// and valid-operator? lists "extract" (:2780-2784) — so the row shape can be
+// rewritten below the top level, and the whole tree is walked to fail closed.
+// `in` is the one operator not descended into: its operand becomes
+// InExpression's :subquery (:2705-2712), projecting the subquery rather than
+// the response.
+func hasExtract(parts []json.RawMessage) bool {
+	if len(parts) == 0 {
+		return false
+	}
+	var op string
+	if json.Unmarshal(parts[0], &op) != nil {
+		return false
+	}
+	switch op {
+	case "extract":
+		return true
+	case "in":
+		return false
+	}
+	for _, p := range parts[1:] {
+		var sub []json.RawMessage
+		if json.Unmarshal(p, &sub) != nil {
+			continue
+		}
+		if hasExtract(sub) {
+			return true
+		}
+	}
+	return false
+}
+
 // 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.

source_test.go

@@ -1,7 +1,10 @@
 package main
 
 import (
+	"bytes"
 	"encoding/json"
+	"io"
+	"log"
 	"net/http"
 	"slices"
 	"strings"
@@ -220,9 +223,90 @@ func TestHandler_UpstreamSourceFactOverridden(t *testing.T) {
 	}
 }
 
-// A count() must report the backends' real fact count, not one inflated by a
-// record pdbmux invented.
-func TestHandler_SourceNotInjectedOnAggregate(t *testing.T) {
+// The configured name means one thing on every query shape: an upstream fact of
+// that name is dropped whether or not the query gate allows synthesis.
+func TestHandler_UpstreamSourceFactSuppressedOnEveryGateState(t *testing.T) {
+	const upstream = "REAL-UPSTREAM-VALUE"
+	tests := []struct {
+		name  string
+		query string
+		want  string // synthetic value, or "" when the gate blocks injection
+	}{
+		{"injection on", "", "a"},
+		{"gated by extract", `["extract",["certname","name","value"],["=","certname","h1"]]`, ""},
+		{"gated by name filter", `["=","name","` + defaultSourceFact + `"]`, ""},
+		{"gated by nested extract", `["and",["=","certname","h1"],["extract",["certname"]]]`, ""},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+				`[`+fact("h1", "role", "web", "")+`,`+fact("h1", defaultSourceFact, upstream, "")+`]`)
+			b := newFakeBackend(t, `[]`, `[]`)
+			srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic))
+
+			rec := doGet(t, srv.Handler(), factsPath, tc.query)
+			if rec.Code != http.StatusOK {
+				t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+			}
+			if strings.Contains(rec.Body.String(), upstream) {
+				t.Fatalf("upstream value survived: %s", rec.Body.String())
+			}
+			got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
+			if tc.want == "" {
+				if n != 0 {
+					t.Errorf("gated query produced %d %s records: %s", n, defaultSourceFact, rec.Body.String())
+				}
+			} else if n != 1 || got["h1"] != tc.want {
+				t.Errorf("provenance = %v (%d records), want h1=%s: %s", got, n, tc.want, rec.Body.String())
+			}
+			if !strings.Contains(rec.Body.String(), `"role"`) {
+				t.Errorf("real facts were dropped: %s", rec.Body.String())
+			}
+		})
+	}
+}
+
+// Dropping an upstream record is invisible in the response, so it is logged —
+// once per request, not once per record.
+func TestHandler_SuppressedUpstreamFactLoggedOncePerRequest(t *testing.T) {
+	a := newFakeBackend(t,
+		`[`+node("h1", "2026-07-20T00:00:00Z")+`,`+node("h2", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", defaultSourceFact, "old-h1", "")+`,`+fact("h2", defaultSourceFact, "old-h2", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	var buf bytes.Buffer
+	srv := NewServer(testConfig(a.srv.URL, b.srv.URL, mergeStatic), log.New(&buf, "", 0))
+
+	doGet(t, srv.Handler(), factsPath, "")
+	if n := strings.Count(buf.String(), defaultSourceFact); n != 1 {
+		t.Fatalf("expected 1 log line naming the fact, got %d: %s", n, buf.String())
+	}
+	if !strings.Contains(buf.String(), "dropped 2") {
+		t.Errorf("log does not report the number dropped: %s", buf.String())
+	}
+}
+
+// Disabled means untouched: an upstream fact of the configured name is served as
+// the backend reported it.
+func TestHandler_SourceDisabledKeepsUpstreamFact(t *testing.T) {
+	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`,
+		`[`+fact("h1", defaultSourceFact, "upstream-value", "")+`]`)
+	b := newFakeBackend(t, `[]`, `[]`)
+	cfg := testConfig(a.srv.URL, b.srv.URL, mergeStatic)
+	cfg.SourceFactEnabled = false
+	srv := newTestServer(cfg)
+
+	rec := doGet(t, srv.Handler(), factsPath, "")
+	got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
+	if n != 1 || got["h1"] != "upstream-value" {
+		t.Errorf("disabled injection altered the upstream fact: %s", rec.Body.String())
+	}
+}
+
+// Covers only that an aggregate /facts response gains no synthetic record and no
+// rewritten row. It does not cover whether the aggregate rows are correct:
+// /facts has no parseAggregate branch, so its rows take the certname merge
+// instead of serveSummed and are not summed across backends.
+func TestHandler_SourceNotInjectedOnFactsAggregate(t *testing.T) {
 	a := newFakeBackend(t, `[`+node("h1", "2026-07-20T00:00:00Z")+`]`, `[]`)
 	b := newFakeBackend(t, `[]`, `[]`)
 	a.bodies[factsPath] = `[{"count":3}]`
@@ -333,6 +417,16 @@ func TestInjectable(t *testing.T) {
 		{"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},
+		// openvoxdb hands each boolean operand back to user-node->plan-node, which
+		// builds an extract node from it, so a nested extract can reshape the rows.
+		{"extract under and", `["and",["=","certname","h1"],["extract",["certname"]]]`, true, false},
+		{"extract under or", `["or",["extract",["certname"]],["=","certname","h1"]]`, true, false},
+		{"extract under not", `["not",["extract",["certname"]]]`, true, false},
+		{"extract nested two deep", `["and",["or",["extract",[["function","count"]]]]]`, true, false},
+		{"extract under from", `["from","facts",["extract",["certname"]]]`, true, false},
+		{"nested extract on nodes", `["and",["extract",["certname"]]]`, false, false},
+		// An extract inside an `in` operand projects the subquery, not the response.
+		{"extract under in stays injectable", `["in","certname",["extract",["certname"],["select_facts",["=","name","os"]]]]`, true, true},
 		{"nodes name filter is not a fact filter", `["=","name","os"]`, false, true},
 		{"nodes extract", `["extract",["certname"]]`, false, false},
 		{"unparseable query", `not json`, true, false},
@@ -350,7 +444,7 @@ func TestInjectable(t *testing.T) {
 // 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}
+	si := &sourceInjector{name: defaultSourceFact, inject: true}
 	for _, env := range []string{"production", ""} {
 		var obj map[string]json.RawMessage
 		if err := json.Unmarshal(si.factRecord("h1", "a", env), &obj); err != nil {
@@ -373,6 +467,10 @@ func TestSourceInjector_NilIsInert(t *testing.T) {
 	if si.claims(defaultSourceFact) {
 		t.Error("nil injector claims a fact name")
 	}
+	if si.injects() {
+		t.Error("nil injector injects")
+	}
+	si.logSuppressed(log.New(io.Discard, "", 0))
 	if si.factRecord("h1", "a", "production") != nil {
 		t.Error("nil injector produced a record")
 	}
@@ -385,7 +483,7 @@ func TestSourceInjector_NilIsInert(t *testing.T) {
 // 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}
+	si := &sourceInjector{name: defaultSourceFact, inject: true}
 	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)
@@ -393,10 +491,53 @@ func TestSourceInjector_StampNonObject(t *testing.T) {
 	}
 }
 
+// A gated injector suppresses without synthesising anything, on either endpoint.
+func TestSourceInjector_GatedSuppressesButDoesNotInject(t *testing.T) {
+	si := &sourceInjector{name: defaultSourceFact}
+	if !si.claims(defaultSourceFact) {
+		t.Error("gated injector does not claim its own fact name")
+	}
+	if si.factRecord("h1", "a", "production") != nil {
+		t.Error("gated injector produced a record")
+	}
+	raw := json.RawMessage(`{"certname":"h1"}`)
+	if got := si.stamp(raw, "a"); string(got) != string(raw) {
+		t.Errorf("gated injector rewrote %s to %s", raw, got)
+	}
+}
+
+// The disabled path must return the backends' records verbatim, source fact
+// included.
+func TestMergeFacts_DisabledIsUntouched(t *testing.T) {
+	a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h1", defaultSourceFact, "upstream", ""))
+	merged := mergeFacts([]backendResult{a}, nil, nil)
+
+	got := factValues(t, merged)
+	want := []string{"h1:role=web-a", "h1:" + defaultSourceFact + "=upstream"}
+	if !slices.Equal(got, want) {
+		t.Errorf("merged = %v, want %v", got, want)
+	}
+}
+
+// A gated query still gets the upstream record removed, and nothing added.
+func TestMergeFacts_GatedSuppressesUpstream(t *testing.T) {
+	a := recs(t, "a", fact("h1", "role", "web-a", ""), fact("h1", defaultSourceFact, "upstream", ""))
+	si := &sourceInjector{name: defaultSourceFact}
+	merged := mergeFacts([]backendResult{a}, nil, si)
+
+	got := factValues(t, merged)
+	if !slices.Equal(got, []string{"h1:role=web-a"}) {
+		t.Errorf("merged = %v, want only the real fact", got)
+	}
+	if si.suppressed != 1 {
+		t.Errorf("suppressed = %d, want 1", si.suppressed)
+	}
+}
+
 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})
+	merged := mergeFacts([]backendResult{a, b}, nil, &sourceInjector{name: defaultSourceFact, inject: true})
 
 	got := factValues(t, merged)
 	want := []string{"h1:role=web-a", "h1:kernel=Linux", "h1:" + defaultSourceFact + "=a"}