Make the API response contract a test; fix what it found

The console's TypeScript types are hand-written and apiFetch ends in
`JSON.parse(text) as T`, so nothing in the toolchain compared them with the
Go structs. A field that lost its json tag, or a response a struct never had,
showed up as an empty table in a browser and nowhere else: the Go suite was
green and the console was wrong. internal/app/contract_test.go now asserts the
exact JSON key set of every admin endpoint, driven through the real router
with the real hook implementations, plus a test that drives the two write
endpoints with the exact payload web/src builds.

The drift it found, all four fixed here:

blocklist.Source and SourceMeta had no json tags at all, so GET /api/blocklists
sent Go field names ("URL"/"Format") where the console expects snake_case: the
Sources table rendered blank rows and its per-row toggle posted
`name: undefined`. Tagged to match ParseStats.

acl.GroupSummary was flat where the console reads a nested `match` object — the
same shape config.Group has — so every group card read "matches nothing". The
four match fields are nested now, under a new acl.Match.

PUT /api/clients answered 400 `json: unknown field "ptr_suffixes"` to the
console's payload, because config.Group and config.Match carried yaml tags only
and the decoder runs with DisallowUnknownFields. Both structs have json tags
now; the same applies to config.UpstreamServer and UpstreamBootstrap.

PUT /api/upstreams answered 400 `json: unknown field "health"`, because the
console round-trips the whole server object it read, runtime health fields
included. They are accepted and ignored, because the server owns them, and they
are `yaml:"-"` so settings.yaml does not gain runtime state.

Also fixed, each with a test:

A duration on the wire was a nanosecond count. config.Duration had a
MarshalYAML but no MarshalJSON, so bootstraps[].refresh arrived as 30000000000
while the console typed the field as a string, and healthcheck_interval was a
float of seconds in one place and a string in another. Durations on the wire
are duration strings everywhere now — the same spelling config.yaml uses — and a
bare JSON number is still accepted and means seconds, not nanoseconds.
TestDurationJSONForm used to pin the old behaviour and now pins the new one,
saying in its comment that the old behaviour was the bug.

A log row's key set depended on the row. api.LogEntry was omitempty on
client_group, response, upstream and list, so a query with no ACL group
arrived without the key and the console rendered "undefined" where the column
should be empty. The contract test can only pin one shape if the server emits
one, so nothing on that struct is omitempty any more.

A dead peer stayed in the cluster peer list forever. Cluster.seen was only ever
added to and deleted from on an explicit `bye`, so a node that crashed — or
whose Valkey connection died — was reported as a live peer for the life of the
process, and its stale blocklist digest was flagged as divergence forever.
Peers() now reaps peers unheard for peer_ttl*4 and stamps each state with the
receiver's own clock, so clock skew between two machines can neither reap a
healthy peer nor keep a dead one. Discover() ages out the same way on the
publisher's own last_seen stamp, because the keyed Valkey record outlives a
crash: a live node keeps refreshing the key's TTL, so a dead node's field stayed
readable for the life of the fleet.

peer_ttl: 0 would have crashed the process. time.NewTicker(PeerTTL/2) panics on
a non-positive period and `EXPIRE key 0` deletes the heartbeat key, so a config
that left peer_ttl unset would have taken the process down at Start, or made
every instance believe it is alone in the fleet. Both go through peerTTL() and
peerDeadline(), which fall back to the default.

Two cluster tests raced on the wire, not on the product, and one was
environment-dependent rather than wrong. A node that joined after another's
heartbeat was in flight hears about it on the next tick, so "b does not see a"
read immediately after "a sees b" is a coin flip; the divergence test waited for
a digest on the wrong node's view of the fleet; and 127.0.0.1:1 is only
unbindable outside a container that sets
net.ipv4.ip_unprivileged_port_start=0, where it binds and the partial-bind test
fails with both transports up. That test now occupies a port with a live
listener in the test process, which is a deterministic EADDRINUSE on every
machine and every uid.

internal/server/counters_test.go covers the counter-key layer that no other
test touched: the "<transport>.<field>" keyspace, that the aggregate folds only
the known fields, and that the snapshot is safe while counters move.

Two pinned behaviours changed on purpose, each with a test that says so:

An RPZ record did not cover the name it names. rpzRule gave a bare owner name
the subdomain form (*.name), so analytics.example.org. in a policy zone blocked
cdn.analytics.example.org. and left analytics.example.org. resolving normally —
and an rpz-passthru. record did not allowlist its own domain either. RPZ
semantics say the record applies to the domain and everything under it, so a
bare owner now takes the apex form **.name; an owner written with an explicit *.
is still strict, so a zone that wants subdomain-only behaviour can still ask for
it. This changes which queries get blocked for an existing policy zone, which is
why it wanted a decision. TestExampleRPZZoneCoversTheApexItNames pins both
halves and the example zone now says which half it relies on.

metrics.OutcomeLocal had no producer. The engine reported a local-zone answer as
qlog.ActionAllow, which outcomeFor mapped to OutcomeUpstream, so every locally
answered query was counted as an upstream one and the outcome="local" series was
never published — the dashboard's local column was permanently empty.
Response.Local is now set on the local-zone path and outcomeFor consults it
first. That moves published counters, so the README's metrics section says so
too.

internal/cluster/cluster.go

@@ -43,6 +43,12 @@ const (
 	EventBye EventKind = "bye"
 )
 
+// defaultPeerTTL mirrors config.Default()'s cluster.peer_ttl. It is the
+// fallback for a config that leaves it unset: the ticker and the key expiry
+// both divide by it, and a zero period panics NewTicker and turns the key
+// expiry into a delete.
+const defaultPeerTTL = 15 * time.Second
+
 // Event is a cluster message.
 type Event struct {
 	Kind    EventKind      `json:"kind"`
@@ -70,6 +76,25 @@ type NodeState struct {
 	Listeners       []string  `json:"listeners"`
 	// Leader is true when this node holds the list-refresh lease.
 	Leader bool `json:"leader"`
+	// LastSeen is when this node was last alive, as recorded by whoever holds
+	// this state: the publishing node stamps it into the keyed record it
+	// leaves in Valkey, and a receiver stamps it again with its own clock when
+	// the heartbeat arrives. It exists so a state can be aged out (see
+	// Cluster.Peers) and so the console can show how stale a peer is. A zero
+	// value means "unknown" — a record written by a build without this field —
+	// and is treated as live rather than stale, so a rolling upgrade never
+	// empties the peer list.
+	LastSeen time.Time `json:"last_seen,omitempty"`
+}
+
+// peer is a node's last published state plus the local clock reading when it
+// arrived. seenAt is deliberately kept out of the wire format and out of any
+// comparison against a remote clock: ageing is decided by how long *this*
+// process has heard nothing, so clock skew between two machines can neither
+// reap a healthy peer nor keep a dead one.
+type peer struct {
+	state  NodeState
+	seenAt time.Time
 }
 
 // Handler receives cluster events from peers.
@@ -113,7 +138,7 @@ type Cluster struct {
 	lastEvent atomic.Int64
 	// seen is the last state per node id.
 	seenMu sync.RWMutex
-	seen   map[string]NodeState
+	seen   map[string]peer
 
 	stop chan struct{}
 	wg   sync.WaitGroup
@@ -130,7 +155,7 @@ func New(opt Options) *Cluster {
 		log:  log.With("component", "cluster"),
 		opt:  opt,
 		node: opt.Version,
-		seen: make(map[string]NodeState),
+		seen: make(map[string]peer),
 		stop: make(chan struct{}),
 	}
 	if cfg := opt.Config.Get(); cfg != nil && cfg.Cluster.NodeID != "" {
@@ -276,7 +301,10 @@ func (c *Cluster) runSession(ctx context.Context) error {
 	c.connected.Store(true)
 	c.log.Info("cluster subscribed", "channel", c.channel(EventHeartbeat))
 
-	hb := time.NewTicker(c.cfg.Get().Cluster.PeerTTL.D() / 2)
+	// c.peerTTL() rather than the raw config value: time.NewTicker panics on a
+	// non-positive period, so a peer_ttl of 0 would take the process down at
+	// Start rather than falling back to the default.
+	hb := time.NewTicker(c.peerTTL() / 2)
 	defer hb.Stop()
 
 	ch := sub.Channel()
@@ -298,6 +326,10 @@ func (c *Cluster) runSession(ctx context.Context) error {
 				continue
 			}
 			c.connected.Store(true)
+			// Age out peers nobody has heard from in peerDeadline. The reap
+			// also runs from Peers, so this is about not growing the map
+			// while nobody is reading the API — not about correctness.
+			c.reapStalePeers()
 			c.renewLease(ctx)
 		case msg, ok := <-ch:
 			if !ok {
@@ -312,9 +344,7 @@ func (c *Cluster) runSession(ctx context.Context) error {
 			}
 			c.lastEvent.Store(time.Now().Unix())
 			if ev.State != nil {
-				c.seenMu.Lock()
-				c.seen[ev.NodeID] = *ev.State
-				c.seenMu.Unlock()
+				c.remember(ev.NodeID, *ev.State)
 			}
 			if ev.Kind == EventBye {
 				c.seenMu.Lock()
@@ -369,9 +399,54 @@ func (c *Cluster) State() *NodeState {
 		st.BlocklistBuilt = built
 		st.ConfigDigest = cfgDigest
 	}
+	// Stamped last, so a StateFunc cannot overwrite it with its own idea of
+	// when this node was last alive.
+	st.LastSeen = time.Now()
 	return st
 }
 
+// peerTTL is the configured heartbeat interval, or the default when it is
+// unset — a zero peer_ttl must not make every peer instantly stale.
+func (c *Cluster) peerTTL() time.Duration {
+	if c.cfg != nil {
+		if d := c.cfg.Get().Cluster.PeerTTL.D(); d > 0 {
+			return d
+		}
+	}
+	return defaultPeerTTL
+}
+
+// peerDeadline is how long a peer may go unheard before it is reaped: four
+// missed heartbeats. Long enough that a GC pause, a blocked scheduler or a
+// Valkey restart does not empty the peer list; short enough that a crashed pod
+// is gone from the UI before an operator goes looking for it.
+func (c *Cluster) peerDeadline() time.Duration { return c.peerTTL() * 4 }
+
+// remember records a peer's state together with the time it arrived.
+func (c *Cluster) remember(id string, st NodeState) {
+	c.seenMu.Lock()
+	c.seen[id] = peer{state: st, seenAt: time.Now()}
+	c.seenMu.Unlock()
+}
+
+// reapStalePeers forgets peers unheard for longer than peerDeadline and
+// reports how many went. It runs on the heartbeat tick so the map cannot grow
+// without bound when nobody reads the API, and from Peers so a reader can never
+// observe a dead node.
+func (c *Cluster) reapStalePeers() int {
+	cutoff := time.Now().Add(-c.peerDeadline())
+	c.seenMu.Lock()
+	defer c.seenMu.Unlock()
+	n := 0
+	for id, p := range c.seen {
+		if p.seenAt.Before(cutoff) {
+			delete(c.seen, id)
+			n++
+		}
+	}
+	return n
+}
+
 // publishState broadcasts this instance's heartbeat. It reports whether the
 // write reached Valkey: the heartbeat is the only liveness signal this package
 // has, because go-redis's Subscribe reconnects transparently and runSession
@@ -392,7 +467,10 @@ func (c *Cluster) publishState(ctx context.Context) error {
 	// Also keep a keyed record so a late-joining instance can discover the
 	// existing set without waiting a full heartbeat interval.
 	pipe.HSet(ctx, c.heartbeatKey(), c.node, mustJSON(st))
-	pipe.Expire(ctx, c.heartbeatKey(), c.cfg.Get().Cluster.PeerTTL.D()*4)
+	// c.peerDeadline() so an unset peer_ttl cannot turn this into
+	// EXPIRE key 0, which *deletes* the key and would make every instance
+	// believe it is the only one in the fleet.
+	pipe.Expire(ctx, c.heartbeatKey(), c.peerDeadline())
 	if _, err := pipe.Exec(ctx); err != nil {
 		return err
 	}
@@ -407,12 +485,22 @@ func mustJSON(v any) string {
 	return string(b)
 }
 
-// Peers returns the last known state of other instances.
+// Peers returns the last known state of every other instance heard from within
+// peerDeadline. A node that stops heartbeating is dropped: a peer list that
+// only ever grows reports a fleet that is not there any more, and a dead peer
+// carrying an old blocklist digest is flagged as divergent forever.
 func (c *Cluster) Peers() []NodeState {
+	// Reap here as well as on the tick, so what a caller reads is reaped
+	// whether or not the maintain loop is still ticking — it is not, during a
+	// Valkey outage, and that is exactly when a peer that died at the same
+	// moment is most likely to be missed.
+	c.reapStalePeers()
 	c.seenMu.RLock()
 	defer c.seenMu.RUnlock()
 	out := make([]NodeState, 0, len(c.seen))
-	for _, s := range c.seen {
+	for _, p := range c.seen {
+		s := p.state
+		s.LastSeen = p.seenAt
 		out = append(out, s)
 	}
 	sort.Slice(out, func(i, j int) bool { return out[i].NodeID < out[j].NodeID })
@@ -605,12 +693,19 @@ func (c *Cluster) Discover(ctx context.Context) []NodeState {
 		ids = append(ids, id)
 	}
 	sort.Strings(ids)
+	cutoff := time.Now().Add(-c.peerDeadline())
 	for _, id := range ids {
 		if id == c.node {
 			continue
 		}
 		var st NodeState
 		if err := json.Unmarshal([]byte(fields[id]), &st); err == nil {
+			// The hash outlives a crash: a live node keeps refreshing the
+			// key's TTL, so a dead node's field would stay readable for the
+			// life of the fleet. Age it out on the publisher's own stamp.
+			if !st.LastSeen.IsZero() && st.LastSeen.Before(cutoff) {
+				continue
+			}
 			out = append(out, st)
 		}
 	}

internal/cluster/cluster_test.go

@@ -488,8 +488,8 @@ func TestStatusReportsDivergentPeers(t *testing.T) {
 	// And a peer with no blocklist at all (empty digest) is not divergence:
 	// the rule requires both sides to be non-empty.
 	db.set("", 0)
-	waitFor(t, "b's state to be republished with an empty digest", 3*time.Second, func() bool {
-		return peerDigest(b, a.NodeID()) == ""
+	waitFor(t, "a's view of b to be republished with an empty digest", 3*time.Second, func() bool {
+		return peerDigest(a, b.NodeID()) == ""
 	})
 	if got := a.Status(ctx).Divergent; len(got) != 0 {
 		t.Errorf("a peer with no blocklist is reported divergent: %v", got)
@@ -599,13 +599,19 @@ func TestHeartbeatAppearsInPeersWithItsIdentity(t *testing.T) {
 	if hasPeer(a.Peers(), a.NodeID()) {
 		t.Error("a node appears in its own peer list")
 	}
-	if !hasPeer(b.Peers(), a.NodeID()) {
-		t.Error("b does not see a in Peers()")
-	}
+	// The other direction has to be waited for rather than read: a joined
+	// *after* a's heartbeat was already in flight, so a hearing b is
+	// immediate while b hears a only on a's next tick. Reading b here is a
+	// race that loses almost every run, and it has nothing to do with the
+	// behaviour under test.
+	waitFor(t, "b to see a in Peers()", 3*time.Second, func() bool {
+		return hasPeer(b.Peers(), a.NodeID())
+	})
 	// The blocklist digest travels with the heartbeat: that is what the
-	// divergence rule compares.
-	waitFor(t, "a to learn b's blocklist digest", 3*time.Second, func() bool {
-		return peerDigest(a, b.NodeID()) == "rules-a" || peerDigest(a, b.NodeID()) != ""
+	// divergence rule compares. b has no digests configured, so a's view of
+	// b is legitimately empty and b's view of a is the one that must arrive.
+	waitFor(t, "b to learn a's blocklist digest", 3*time.Second, func() bool {
+		return peerDigest(b, a.NodeID()) == "rules-a"
 	})
 	if got := peerDigest(a, b.NodeID()); got != "" {
 		t.Errorf("unexpected digest %q from a node with no digests configured", got)