Narrow the source-fact /facts fetch to the query's certnames

README.md

@@ -251,7 +251,9 @@ matters more.
   `in` an `array`) selects `pdbmux_source` itself, built only from
   `and`/`or`/`not` over `=`, `~` and `in`-`array` on `certname`, `environment`,
   `name` and `value`: it is still fanned out for the real facts it names, and
-  the synthetic records are added after being filtered by the query locally;
+  the synthetic records are added after being filtered by the query locally.
+  Their fetch carries only the query's top-level `and` conjuncts on
+  `certname`, so a single-host lookup stays single-host;
 - the path is `/facts/<name>` for any other fact. The path segment is the same
   outer `name` constraint, so only the fact's own path carries the record;
 - injection is turned off (see `source_fact_enabled`).

factroutes_test.go

@@ -6,6 +6,8 @@ import (
 	"net/url"
 	"reflect"
 	"slices"
+	"sync"
+	"sync/atomic"
 	"testing"
 )
 
@@ -767,3 +769,65 @@ func TestHandler_FactsQuerySourceFactDisabled(t *testing.T) {
 		t.Errorf("disabled feature synthesised %d records: %s", n, rec.Body.String())
 	}
 }
+
+// The synthetic set is fetched with the query's top-level certname conjuncts
+// only, so a single-host lookup never pulls the whole estate.
+func TestHandler_FactsQuerySourceFactNarrowsUpstream(t *testing.T) {
+	a, b := sourceFactBackends(t)
+	var mu sync.Mutex
+	var got []string
+	h3Only := map[*fakeBackend]string{
+		a: `[` + factEnv("h3", "osfamily", "RedHat", "prod") + `]`,
+		b: `[` + factEnv("h3", "osfamily", "Debian", "dev") + `]`,
+	}
+	for fb, body := range h3Only {
+		fb.handle = func(w http.ResponseWriter, r *http.Request) bool {
+			if r.URL.Path != factsPath {
+				return false
+			}
+			q := r.URL.Query().Get("query")
+			mu.Lock()
+			got = append(got, q)
+			mu.Unlock()
+			if q == `["=","certname","h3"]` {
+				_, _ = w.Write([]byte(body))
+				return true
+			}
+			return false
+		}
+	}
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
+
+	query := `["and",["=","certname","h3"],["=","environment","dev"],["=","name","` + defaultSourceFact + `"]]`
+	rec := doGet(t, srv.Handler(), factsPath, query)
+	if vals, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact); n != 1 || vals["h3"] != "b" {
+		t.Fatalf("got %v (%d records), want h3=b: %s", vals, n, rec.Body.String())
+	}
+	for _, q := range got {
+		if q == "" {
+			t.Fatalf("an upstream /facts fetch was unfiltered: %q", got)
+		}
+	}
+	if !slices.Contains(got, `["=","certname","h3"]`) {
+		t.Fatalf("synthetic fetch was not narrowed to the certname: %q", got)
+	}
+}
+
+// A backend lost between the two fetches is reported as missing.
+func TestHandler_FactsQuerySourceFactCountsSecondFetch(t *testing.T) {
+	a, b := sourceFactBackends(t)
+	var calls atomic.Int32
+	b.handle = func(w http.ResponseWriter, r *http.Request) bool {
+		if r.URL.Path == factsPath && calls.Add(1) > 1 {
+			http.Error(w, "boom", http.StatusInternalServerError)
+			return true
+		}
+		return false
+	}
+	srv := newTestServer(testConfig(a.srv.URL, b.srv.URL, mergeFreshness))
+
+	rec := doGet(t, srv.Handler(), factsPath, `["=","name","`+defaultSourceFact+`"]`)
+	if h := rec.Header().Get(backendsHeader); h != "1/2" {
+		t.Errorf("%s = %q, want 1/2", backendsHeader, h)
+	}
+}

server.go

@@ -9,6 +9,7 @@ import (
 	"log"
 	"net/http"
 	"net/url"
+	"slices"
 	"strconv"
 	"strings"
 	"sync"
@@ -398,8 +399,9 @@ func (s *Server) serveFacts(w http.ResponseWriter, r *http.Request) {
 
 // serveSourceFactQuery answers a /facts query that selects the owned fact by
 // name. The query still goes upstream for the real facts it selects; the
-// synthetic records come from the same merge serveSourceFact uses and are
-// filtered by the query locally, since no backend holds them.
+// synthetic records come from the same merge serveSourceFact uses, fetched for
+// the query's certname scope, and are filtered by the query locally, since no
+// backend holds them.
 func (s *Server) serveSourceFactQuery(w http.ResponseWriter, r *http.Request, inject *sourceInjector, ast []json.RawMessage) {
 	params := queryParams(r.URL.Query().Get("query"))
 	s.serveCached(w, r, factsPath, params, func(ctx context.Context) (cachedResponse, error) {
@@ -408,7 +410,7 @@ func (s *Server) serveSourceFactQuery(w http.ResponseWriter, r *http.Request, in
 			return cachedResponse{}, err
 		}
 		recs := s.mergeFactsWith(inject)(alive)
-		synth, _, err := s.sourceFacts(ctx, s.newSourceInjector("", true), nil)
+		synth, scoped, err := s.sourceFacts(ctx, s.newSourceInjector("", true), queryParams(certnameScope(ast)))
 		if err != nil {
 			return cachedResponse{}, err
 		}
@@ -422,11 +424,21 @@ func (s *Server) serveSourceFactQuery(w http.ResponseWriter, r *http.Request, in
 			}
 		}
 		resp := cachedResponse{Body: encodeRecords(recs), Records: -1}
-		s.countBackends(&resp, alive)
+		s.countBackends(&resp, aliveInBoth(alive, scoped))
 		return resp, nil
 	})
 }
 
+func aliveInBoth(a, b []backendResult) []backendResult {
+	var out []backendResult
+	for _, x := range a {
+		if slices.ContainsFunc(b, func(y backendResult) bool { return y.name == x.name }) {
+			out = append(out, x)
+		}
+	}
+	return out
+}
+
 // sourceFacts is the owned fact's synthetic record set for the nodes params
 // selects, taken from the /facts merge so owner and environment match it.
 func (s *Server) sourceFacts(ctx context.Context, inject *sourceInjector, params url.Values) ([]json.RawMessage, []backendResult, error) {

server_test.go

@@ -44,6 +44,8 @@ type fakeBackend struct {
 	// rawPaths records the still-escaped request paths, so tests can assert an
 	// MBean name's percent-encoding survived the proxy.
 	rawPaths []string
+	// handle, when set, runs first and answers the request itself by returning true.
+	handle func(w http.ResponseWriter, r *http.Request) bool
 }
 
 func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend {
@@ -63,6 +65,9 @@ func newFakeBackend(t *testing.T, nodesBody, factsBody string) *fakeBackend {
 		fb.gotParams[r.URL.Path] = r.URL.Query()
 		fb.rawPaths = append(fb.rawPaths, r.URL.EscapedPath())
 		fb.mu.Unlock()
+		if fb.handle != nil && fb.handle(w, r) {
+			return
+		}
 		if fb.dead {
 			panic(http.ErrAbortHandler)
 		}

source.go

@@ -210,8 +210,9 @@ func constrainsField(parts []json.RawMessage, field string) bool {
 // sourceQuery returns the parsed /facts query when one of its name comparisons
 // selects the owned fact and the whole query can be evaluated locally against a
 // fact record; nil otherwise, which leaves the query on the gated path.
-// ponytail: a negated name match (["not",["=","name","x"]]) does not select the
-// owned fact; add three-valued evaluation if a client needs it.
+// ponytail: only a name match that is true for the owned fact selects it, so
+// ["not",["=","name","osfamily"]] stays on the gated path and synthesises
+// nothing; widen namesFact if a client needs negated selections.
 func (si *sourceInjector) sourceQuery(query string) []json.RawMessage {
 	if si == nil {
 		return nil
@@ -346,3 +347,33 @@ func matchFact(parts []json.RawMessage, f factFields) (match, ok bool) {
 	}
 	return false, false
 }
+
+// certnameScope returns the top-level and's certname comparisons as a query,
+// or "" when there are none. Pushing them upstream keeps every record of a
+// selected node on every backend, so the merge's ownership is unchanged; an
+// environment comparison is not pushed because it can hide the owner's records
+// when backends disagree on a node's environment.
+func certnameScope(ast []json.RawMessage) string {
+	var op string
+	if len(ast) < 2 || json.Unmarshal(ast[0], &op) != nil || op != "and" {
+		return ""
+	}
+	var keep []string
+	for _, p := range ast[1:] {
+		var sub []json.RawMessage
+		var field string
+		if json.Unmarshal(p, &sub) != nil || len(sub) != 3 || json.Unmarshal(sub[1], &field) != nil || field != "certname" {
+			continue
+		}
+		if _, ok := matchFact(sub, factFields{}); ok {
+			keep = append(keep, string(p))
+		}
+	}
+	switch len(keep) {
+	case 0:
+		return ""
+	case 1:
+		return keep[0]
+	}
+	return `["and",` + strings.Join(keep, ",") + `]`
+}

source_test.go

@@ -637,3 +637,25 @@ func TestSourceInjector_SourceQuery(t *testing.T) {
 		t.Error("disabled injector selected a query")
 	}
 }
+
+func TestCertnameScope(t *testing.T) {
+	tests := map[string]string{
+		`["=","name","x"]`: ``,
+		`["and",["=","name","x"],["=","environment","prod"]]`:                                    ``,
+		`["or",["=","certname","h1"],["=","name","x"]]`:                                          ``,
+		`["and",["not",["=","certname","h1"]],["=","name","x"]]`:                                 ``,
+		`["and",["=","certname","h1"],["=","name","x"]]`:                                         `["=","certname","h1"]`,
+		`["and",["~","certname","["],["=","name","x"]]`:                                          ``,
+		`["and",["in","certname",["extract","certname",["=","a","b"]]],["=","name","x"]]`:        ``,
+		`["and",["in","certname",["array",["h1","h2"]]],["~","certname","^h"],["=","name","x"]]`: `["and",["in","certname",["array",["h1","h2"]]],["~","certname","^h"]]`,
+	}
+	for in, want := range tests {
+		var ast []json.RawMessage
+		if err := json.Unmarshal([]byte(in), &ast); err != nil {
+			t.Fatal(err)
+		}
+		if got := certnameScope(ast); got != want {
+			t.Errorf("certnameScope(%s) = %q, want %q", in, got, want)
+		}
+	}
+}