Stop the source-fact drilldown fanning out per pinned value

The <value> path segment is client-supplied and reached a full unfiltered
/facts fan-out, keyed per value, so every distinct value was a fresh
whole-estate query and a fresh cache entry.

- validate <value> against the configured backend names, answering [] with
  no fan-out when it names none
- key the drilldown's fetch on the fact name alone and apply <value> to the
  shared record set, so all values share one entry and one fan-out
- report every configured backend on the no-fan-out empty response, which is
  complete rather than partial

README.md

@@ -182,12 +182,18 @@ records from the `/facts` merge that produces them, which makes the `certname`
 set, the owner and the `environment` identical to the ones an unfiltered `/facts`
 response reports, and lets the request's own `query` narrow the result upstream.
 That costs one `/facts` fan-out per cache miss — the widest fan-out `pdbmux`
-makes — on a rare, user-initiated path. `/facts/pdbmux_source/<value>` pins the
-backend name, so it answers with the nodes that backend owns, and with `[]` for a
-value naming no backend. An `extract`/`count` query still takes the summing
-branch, and the gated query shapes above still answer `[]`, as does every form
-while injection is off — with the name kept out of `/fact-names`, since nothing
-then produces it.
+makes — on a rare, user-initiated path.
+
+`/facts/pdbmux_source/<value>` pins the backend name, so it answers with the
+nodes that backend owns. The `<value>` segment never reaches that fan-out: the
+synthetic record's value is always a backend name, so a value naming none is
+answered `[]` from the configured names alone, with no fan-out at all, and a
+value naming one filters a record set fetched under a key the value is not part
+of. The record set is a property of the estate rather than of the filter, so
+every value of it — and the unfiltered path — share one entry and one fetch.
+An `extract`/`count` query still takes the summing branch, and the gated query
+shapes above still answer `[]`, as does every form while injection is off — with
+the name kept out of `/fact-names`, since nothing then produces it.
 
 **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
@@ -407,7 +413,9 @@ backend later without further handler changes.
   requests can only share an entry when they share a key, and the key is path
   plus query, which is exactly what decides whether injection applies; a
   name-filtered `/facts` query and a plain one therefore cache separately and
-  neither is ever served the other's shape. `source_fact` and
+  neither is ever served the other's shape. The one path that keys on less than
+  it is asked is the `pdbmux_source` drilldown, whose `<value>` is dropped from
+  the key and applied to the shared entry instead. `source_fact` and
   `source_fact_enabled` are read once at startup, and the cache lives for the
   same process, so changing either cannot leave differently-shaped entries
   behind.

cache_test.go

@@ -96,6 +96,18 @@ func (cb *countingBackend) hitCount(path string) int {
 	return cb.hits[path]
 }
 
+// totalHits counts requests across every path, so a test can assert a route
+// reached no backend at all.
+func (cb *countingBackend) totalHits() int {
+	cb.mu.Lock()
+	defer cb.mu.Unlock()
+	n := 0
+	for _, c := range cb.hits {
+		n += c
+	}
+	return n
+}
+
 func (cb *countingBackend) setFail(v bool) {
 	cb.mu.Lock()
 	defer cb.mu.Unlock()

factroutes_test.go

@@ -337,6 +337,66 @@ func TestHandler_FactsBySourceNameAndValueFiltersByOwner(t *testing.T) {
 	}
 }
 
+// The pinned value is client-supplied and the drilldown's fetch is the widest
+// query pdbmux makes, so a value naming no backend is answered from the
+// configured names alone: no fan-out, and so no per-value cache key either.
+func TestHandler_FactsBySourceNameUnknownValueSkipsFanOut(t *testing.T) {
+	facts := map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`}
+	a := newCountingBackend(t, facts)
+	b := newCountingBackend(t, facts)
+	srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
+
+	for _, value := range []string{"nosuchbackend", "nonce-1", "nonce-2"} {
+		rec := doGet(t, srv.Handler(), sourceFactURL+"/"+value, "")
+		if rec.Code != http.StatusOK {
+			t.Fatalf("status %d: %s", rec.Code, rec.Body.String())
+		}
+		if got := rec.Body.String(); got != "[]\n" {
+			t.Errorf("%s/%s = %q, want %q", sourceFactURL, value, got, "[]\n")
+		}
+		// The answer is complete, not built from a subset of backends.
+		if h := rec.Header().Get(backendsHeader); h != "2/2" {
+			t.Errorf("%s/%s %s = %q, want 2/2", sourceFactURL, value, backendsHeader, h)
+		}
+	}
+	for _, cb := range []*countingBackend{a, b} {
+		if got := cb.totalHits(); got != 0 {
+			t.Errorf("unknown value reached a backend %d times, want 0", got)
+		}
+	}
+}
+
+// The record set is the estate's, not the pinned value's, so every value shares
+// one entry: two valid values must not cost two whole-estate fan-outs.
+func TestHandler_FactsBySourceNameValuesShareOneFetch(t *testing.T) {
+	a := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h1", "osfamily", "RedHat", "") + `]`})
+	b := newCountingBackend(t, map[string]string{factsPath: `[` + fact("h2", "osfamily", "Debian", "") + `]`})
+	srv, _ := newCachedServer(t, cacheTestConfig(a.srv.URL, b.srv.URL))
+
+	for _, tc := range []struct {
+		path string
+		want map[string]string
+	}{
+		{sourceFactURL + "/a", map[string]string{"h1": "a"}},
+		{sourceFactURL + "/b", map[string]string{"h2": "b"}},
+		{sourceFactURL, map[string]string{"h1": "a", "h2": "b"}},
+	} {
+		rec := doGet(t, srv.Handler(), tc.path, "")
+		got, n := sourceValues(t, rec.Body.Bytes(), defaultSourceFact)
+		if n != len(tc.want) || !reflect.DeepEqual(got, tc.want) {
+			t.Fatalf("%s = %v (%d records), want %v", tc.path, got, n, tc.want)
+		}
+	}
+	for _, cb := range []*countingBackend{a, b} {
+		if got := cb.hitCount(factsPath); got != 1 {
+			t.Errorf("%s fetched %d times, want 1 shared fetch", factsPath, got)
+		}
+		if got := cb.hitCount(sourceFactURL); got != 0 {
+			t.Errorf("%s was fanned out %d times, want 0", sourceFactURL, got)
+		}
+	}
+}
+
 // Nothing produces the fact while injection is off, so the drilldown is empty
 // and the name is not advertised for a client to click through to.
 func TestHandler_FactsBySourceNamePathEmptyWhenDisabled(t *testing.T) {

server.go

@@ -335,22 +335,55 @@ func (s *Server) serveFactsByName(w http.ResponseWriter, r *http.Request, name,
 // records are taken from the /facts merge that produces them instead, which
 // makes the certname set, the owner and the environment identical to the ones an
 // unfiltered /facts response reports, and lets the request's query narrow the
-// result upstream. That costs one extra fan-out on a rare, user-initiated path.
+// result upstream.
+//
+// That merge is the widest query pdbmux makes and the pinned value is
+// client-supplied, so the value never reaches it: a value naming no configured
+// backend is answered empty without any fan-out, and a value naming one filters
+// a record set fetched under a value-independent key. Otherwise each distinct
+// value would be a fresh cache key, a fresh flight and a fresh whole-estate
+// fan-out.
 func (s *Server) serveSourceFact(w http.ResponseWriter, r *http.Request, inject *sourceInjector, value string, valued bool) {
+	// The synthetic record's value is always a backend name, so any other value
+	// matches zero records. The response is complete rather than degraded, so it
+	// reports every configured backend.
+	if valued && !s.hasBackend(value) {
+		n := len(s.cfg.Backends)
+		writeCached(w, cachedResponse{Records: -1, Backends: n, Configured: n})
+		return
+	}
+
 	params := queryParams(r.URL.Query().Get("query"))
 	merge := s.mergeFactsWith(inject)
-	s.serveCached(w, r, r.URL.Path, params, func(ctx context.Context) (cachedResponse, error) {
+	var filter recordFilter
+	if valued {
+		filter = func(recs []json.RawMessage) []json.RawMessage {
+			return sourceFactRecords(recs, inject.name, value, true)
+		}
+	}
+	// The stored set is the whole owned-fact record set, so every value of it
+	// keys, and waits on, the same fetch.
+	s.serveFiltered(w, r, factsPath+"/"+inject.name, params, filter, func(ctx context.Context) (cachedResponse, error) {
 		alive, err := s.aliveResults(ctx, factsPath, params)
 		if err != nil {
 			return cachedResponse{}, err
 		}
-		recs := sourceFactRecords(merge(alive), inject.name, value, valued)
+		recs := sourceFactRecords(merge(alive), inject.name, "", false)
 		resp := cachedResponse{Body: encodeRecords(recs), Records: -1}
 		s.countBackends(&resp, alive)
 		return resp, nil
 	})
 }
 
+func (s *Server) hasBackend(name string) bool {
+	for _, b := range s.cfg.Backends {
+		if b.Name == name {
+			return true
+		}
+	}
+	return false
+}
+
 // A count row carries no certname, so the certname-keyed merge would collapse every backend's count into one backend's; aggregates take the summing path instead.
 func (s *Server) serveNodes(w http.ResponseWriter, r *http.Request) {
 	if spec := parseAggregate(r.URL.Query().Get("query")); spec != nil {
@@ -457,12 +490,36 @@ type cachedResponse struct {
 	Configured int             `json:"configured"` // backends configured at build time
 }
 
+// recordFilter narrows a response's records after it has been built or read back
+// from the cache, so requests differing only in the filter share one stored entry
+// and one fan-out. It leaves Records alone, so it only suits responses that set
+// no X-Records.
+type recordFilter func([]json.RawMessage) []json.RawMessage
+
+func (f recordFilter) apply(resp cachedResponse) cachedResponse {
+	if f == nil {
+		return resp
+	}
+	var recs []json.RawMessage
+	if json.Unmarshal(resp.Body, &recs) != nil {
+		return resp
+	}
+	resp.Body = encodeRecords(f(recs))
+	return resp
+}
+
 // serveCached answers from the cache when the entry is fresh, otherwise runs
 // build — single-flighted, so N concurrent identical requests cause one upstream
 // fan-out — and stores the result. A build failure falls back to a stale entry
 // when one exists; that is the only path on which stale data is served. Paths
 // with no cache configured run build directly, unchanged.
 func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string, params url.Values, build func(context.Context) (cachedResponse, error)) {
+	s.serveFiltered(w, r, path, params, nil, build)
+}
+
+// serveFiltered is serveCached with a per-request narrowing applied to whatever
+// the shared entry holds.
+func (s *Server) serveFiltered(w http.ResponseWriter, r *http.Request, path string, params url.Values, filter recordFilter, build func(context.Context) (cachedResponse, error)) {
 	cache, enabled := s.cacheFor(path, params)
 	if !enabled {
 		resp, err := build(r.Context())
@@ -470,7 +527,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string
 			http.Error(w, err.Error(), http.StatusBadGateway)
 			return
 		}
-		writeCached(w, resp)
+		writeCached(w, filter.apply(resp))
 		return
 	}
 
@@ -482,7 +539,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string
 		s.log.Printf("warning: cache lookup for %s failed: %v", key, err)
 	case status == CacheFresh:
 		s.stale.markFresh()
-		s.writeStored(w, ent, CacheFresh)
+		s.writeStored(w, ent, CacheFresh, filter)
 		return
 	case status == CacheStale:
 		stale = &ent
@@ -523,7 +580,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string
 			s.stale.markStale(s.now())
 			s.log.Printf("warning: serving stale %s from cache (stored %s): %v",
 				path, stale.StoredAt.UTC().Format(time.RFC3339), err)
-			s.writeStored(w, *stale, CacheStale)
+			s.writeStored(w, *stale, CacheStale, filter)
 			return
 		}
 		http.Error(w, err.Error(), http.StatusBadGateway)
@@ -531,7 +588,7 @@ func (s *Server) serveCached(w http.ResponseWriter, r *http.Request, path string
 	}
 	s.stale.markFresh()
 	s.setCacheHeaders(w, CacheMiss, time.Time{})
-	writeCached(w, resp)
+	writeCached(w, filter.apply(resp))
 }
 
 // http.Client reads a zero Timeout as "no deadline", but it would expire a
@@ -543,7 +600,7 @@ func (s *Server) flightTimeout() time.Duration {
 	return defaultTimeout
 }
 
-func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status CacheStatus) {
+func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status CacheStatus, filter recordFilter) {
 	var resp cachedResponse
 	if err := json.Unmarshal(ent.Body, &resp); err != nil {
 		s.log.Printf("warning: unreadable cache entry: %v", err)
@@ -551,7 +608,7 @@ func (s *Server) writeStored(w http.ResponseWriter, ent CacheEntry, status Cache
 		return
 	}
 	s.setCacheHeaders(w, status, ent.StoredAt)
-	writeCached(w, resp)
+	writeCached(w, filter.apply(resp))
 }
 
 // setCacheHeaders labels a response from a cache-backed path: X-Cache is