Fix the bbolt round trip, the blocklist gaps and the deploy set

The cache snapshot never loaded, so every restart was a cold cache. Snapshot packed the whole key while Load decoded the record header out of the key's remainder instead of the value, so every entry was skipped. The two sides are now an explicit encodeRecord/decodeRecord pair with a version byte and a CRC-32C trailer, so a future format change is detected instead of silently misparsed. Load also drops entries that are already past ExpiresAt and rewrites the TTL to the time actually left, so a served TTL never jumps back up across a restart. Blocklist, all six gaps the failing tests were pointing at: a leading UTF-8 BOM is stripped, "!allow example.com" works, SourceMeta exists for a source that only ever contributed a rule, Pattern() reports the bare name again (the new WireName() produces the wildcard form for the API), an RPZ record with no explicit TTL inherits the zone default instead of whatever the reader left in the RR, a hosts line with no address in the first field says so, and the adblock parser handles CRLF and adblock-flavoured lines pasted into a plain list. TestHolderSwapIsAtomic was asserting the wrong invariant — the swap was always atomic, a name blocked by the old set is legitimately unblocked by the new one. Tests added for acl and resolver, both of which turned up real bugs: - a PTR suffix matched across label boundaries, so a group scoped to example.com. also swallowed ample.com. and notexample.com. — a client could land in a group it does not belong to; - the "*" catch-all was dropped from the rendered domain patterns, so the engine's per-query recompile lost the "everything" case; - Upstream.Success never closed an open breaker, so a recovered upstream stayed excluded for the full cooldown; - Exchange fanned out in parallel even under policy: failover. Deploy set: go.mod requires Go 1.27.1 but CI pinned 1.23 and the Dockerfile pinned golang:1.23-alpine, so neither could have built. The Dockerfile also copied the source tree after the console bundle, which clobbered it. The k8s set asked for hostNetwork with two replicas, but the server never sets SO_REUSEPORT, so the second pod could not bind :53 and would crash-loop. It now runs without hostNetwork behind a LoadBalancer Service, with a separate ClusterIP Service for the Ingress (a headless Service cannot back one), a ServiceAccount, an API-key Secret, per-pod PVCs, and a kustomization. The HPA is gone: it cannot drive a StatefulSet. To let the API key come from that Secret, config.Load now expands ${VAR} / ${VAR:-default} / ${NAME-default} and fails to start on an unset variable with no default, so a missing secret is a crash loop rather than an unauthenticated console. Expansion skips YAML comments, so documenting ${VAR} in the ConfigMap does not break the file. Verified: gofmt, go build, go vet clean; go test -race ./internal/... green (acl, app, blocklist, cache, config, resolver, zonefile).
2026-09-27 13:30:43 +00:00
1 parent 3bcc10a029
commit 26134ba437
@@ -1,6 +1,7 @@
 package cache

 import (
+	"encoding/binary"
 	"errors"
 	"math/rand/v2"
 	"net"
@@ -12,6 +13,7 @@ import (
 	"time"

 	"github.com/miekg/dns"
+	bolt "go.etcd.io/bbolt"
 )

 // fakeClock is a manually advanced clock so TTL, ageing and serve-stale
@@ -868,3 +870,231 @@ func BenchmarkManagerGetUnpack(b *testing.B) {
 		}
 	})
 }
+
+// TestStoreSurvivesProcessRestart is the test the warm-restart path exists for:
+// write a snapshot, close the file completely, open a brand new Store on the
+// same path and load it into a fresh LRU. Anything that only works because the
+// same *Store is still holding the file open fails here.
+func TestStoreSurvivesProcessRestart(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "cache.db")
+	clk := newClock()
+	now := clk.Now()
+
+	st, err := OpenStore(StoreOptions{Path: path})
+	if err != nil {
+		t.Fatalf("OpenStore: %v", err)
+	}
+	warm := NewLRU(100, 2)
+	for i := range 5 {
+		n := "r" + strconv.Itoa(i) + ".example.com"
+		wire, _ := answer(t, n, 600, "10.1.0."+strconv.Itoa(i)).Pack()
+		warm.Put(key(n), &Entry{
+			Wire: wire, TTL: 600, Origin: "upstream", Upstream: "9.9.9.9:53",
+			CreatedAt: now, ExpiresAt: now.Add(600 * time.Second), StaleUntil: now.Add(630 * time.Second),
+			Bytes: len(wire),
+		})
+	}
+	if written, err := st.Snapshot(warm, now); err != nil || written != 5 {
+		t.Fatalf("Snapshot = %d, %v; want 5, nil", written, err)
+	}
+	if err := st.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+
+	// Second "process".
+	restarted, err := OpenStore(StoreOptions{Path: path})
+	if err != nil {
+		t.Fatalf("reopen: %v", err)
+	}
+	defer restarted.Close()
+	cold := NewLRU(100, 2)
+	loaded, skipped, err := restarted.Load(cold, clk.Now())
+	if err != nil {
+		t.Fatalf("Load: %v", err)
+	}
+	if loaded != 5 || skipped != 0 {
+		t.Fatalf("Load = %d loaded, %d skipped, want 5/0", loaded, skipped)
+	}
+	e, kind := cold.Get(key("r3.example.com"), clk.Now())
+	if e == nil || kind != Fresh {
+		t.Fatalf("restored entry: entry=%v kind=%v, want non-nil/Fresh", e, kind)
+	}
+	if e.Upstream != "9.9.9.9:53" {
+		t.Errorf("Upstream = %q, want 9.9.9.9:53", e.Upstream)
+	}
+	// The TTL a client sees must not jump back up across the restart: the
+	// loader rewrites it to the time actually left.
+	clk.Advance(120 * time.Second)
+	if got, want := e.AgeTTL(clk.Now()), uint32(480); got != want {
+		t.Errorf("AgeTTL after 120s = %d, want %d", got, want)
+	}
+}
+
+// TestStoreLoadSkipsDamagedRecords feeds Load a file where individual records
+// have been scribbled on, truncated or written by a future version. A snapshot
+// is a cache, not a source of truth: every one of these must cost exactly one
+// entry and none may fail the whole load.
+func TestStoreLoadSkipsDamagedRecords(t *testing.T) {
+	path := filepath.Join(t.TempDir(), "cache.db")
+	clk := newClock()
+	now := clk.Now()
+
+	st, err := OpenStore(StoreOptions{Path: path})
+	if err != nil {
+		t.Fatalf("OpenStore: %v", err)
+	}
+	l := NewLRU(100, 1)
+	for i := range 4 {
+		n := "d" + strconv.Itoa(i) + ".example.com"
+		wire, _ := answer(t, n, 600, "10.2.0."+strconv.Itoa(i)).Pack()
+		l.Put(key(n), &Entry{
+			Wire: wire, TTL: 600, CreatedAt: now, ExpiresAt: now.Add(600 * time.Second),
+			StaleUntil: now.Add(630 * time.Second), Bytes: len(wire),
+		})
+	}
+	if written, err := st.Snapshot(l, now); err != nil || written != 4 {
+		t.Fatalf("Snapshot = %d, %v; want 4, nil", written, err)
+	}
+	// Three different kinds of damage, applied to the first three records
+	// (the keys are packed, so the cursor is the only way to find them).
+	damaged := map[int][]byte{
+		0: futureVersionRecord,                  // written by a hypothetical v2
+		1: {recordVersion, 0, 0x01, 0x02, 0x03}, // truncated mid-varint
+		2: corruptRecord(t, encodeRecord(nil, &Entry{ // flipped bit
+			Wire: []byte{0xde, 0xad}, TTL: 1, CreatedAt: now, ExpiresAt: now, StaleUntil: now,
+		})),
+	}
+	if err := st.db.Update(func(tx *bolt.Tx) error {
+		b := tx.Bucket(bucketCache)
+		cur := b.Cursor()
+		k, _ := cur.First()
+		for i := 0; k != nil; i++ {
+			if bad, ok := damaged[i]; ok {
+				if err := b.Put(k, bad); err != nil {
+					return err
+				}
+			}
+			k, _ = cur.Next()
+		}
+		return nil
+	}); err != nil {
+		t.Fatalf("seed damaged records: %v", err)
+	}
+	if err := st.Close(); err != nil {
+		t.Fatalf("Close: %v", err)
+	}
+
+	again, err := OpenStore(StoreOptions{Path: path})
+	if err != nil {
+		t.Fatalf("reopen: %v", err)
+	}
+	defer again.Close()
+	restored := NewLRU(100, 1)
+	loaded, skipped, err := again.Load(restored, clk.Now())
+	if err != nil {
+		t.Fatalf("Load on a damaged file returned %v, want the good records to survive", err)
+	}
+	if loaded != 1 {
+		t.Errorf("loaded = %d, want 1 (only the untouched record)", loaded)
+	}
+	if skipped != 3 {
+		t.Errorf("skipped = %d, want 3", skipped)
+	}
+	if _, kind := restored.Get(key("d3.example.com"), clk.Now()); kind != Fresh {
+		t.Errorf("intact record did not survive: kind = %v", kind)
+	}
+	if _, kind := restored.Get(key("d0.example.com"), clk.Now()); kind != Miss {
+		t.Errorf("damaged record was loaded anyway: kind = %v", kind)
+	}
+}
+
+// futureVersionRecord is a well-formed record from a hypothetical v2 writer.
+var futureVersionRecord = func() []byte {
+	buf := []byte{recordVersion + 1, 0}
+	buf = binary.AppendUvarint(buf, 0)
+	return buf
+}()
+
+// corruptRecord flips one bit inside the body so the checksum no longer matches.
+func corruptRecord(t *testing.T, rec []byte) []byte {
+	t.Helper()
+	out := append([]byte(nil), rec...)
+	out[len(out)/2] ^= 0x40
+	return out
+}
+
+// TestRecordEncodeDecode pins the on-disk record format itself. The round trip
+// through bbolt used to break because the encoder and decoder disagreed; this
+// test fails at compile-review time rather than at warm-restart time.
+func TestRecordEncodeDecode(t *testing.T) {
+	now := newClock().Now()
+	in := &Entry{
+		Wire:       []byte{0x00, 0x01, 0x02, 0x03},
+		TTL:        600,
+		Origin:     "upstream",
+		Upstream:   "1.1.1.1:53",
+		CreatedAt:  now,
+		ExpiresAt:  now.Add(600 * time.Second),
+		StaleUntil: now.Add(630 * time.Second),
+		Neg:        true,
+		Truncated:  true,
+	}
+	out, err := decodeRecord(encodeRecord(nil, in))
+	if err != nil {
+		t.Fatalf("decodeRecord: %v", err)
+	}
+	if string(out.Wire) != string(in.Wire) {
+		t.Errorf("Wire = % x, want % x", out.Wire, in.Wire)
+	}
+	for _, f := range []struct {
+		name      string
+		got, want any
+	}{
+		{"TTL", out.TTL, in.TTL},
+		{"Origin", out.Origin, in.Origin},
+		{"Upstream", out.Upstream, in.Upstream},
+		{"Neg", out.Neg, in.Neg},
+		{"Truncated", out.Truncated, in.Truncated},
+	} {
+		if f.got != f.want {
+			t.Errorf("%s = %v, want %v", f.name, f.got, f.want)
+		}
+	}
+	// Times come back as wall-clock instants rebuilt from Unix nanoseconds, so
+	// they compare with Equal rather than == (the location bits differ).
+	for _, f := range []struct {
+		name      string
+		got, want time.Time
+	}{
+		{"CreatedAt", out.CreatedAt, in.CreatedAt},
+		{"ExpiresAt", out.ExpiresAt, in.ExpiresAt},
+		{"StaleUntil", out.StaleUntil, in.StaleUntil},
+	} {
+		if !f.got.Equal(f.want) {
+			t.Errorf("%s = %v, want %v", f.name, f.got, f.want)
+		}
+	}
+
+	// A record whose times are all zero must survive too: the varints encode
+	// them as 0 and come back as the zero time, not 1970.
+	zero, err := decodeRecord(encodeRecord(nil, &Entry{Wire: []byte{0xff}}))
+	if err != nil {
+		t.Fatalf("decodeRecord(zero times): %v", err)
+	}
+	if !zero.ExpiresAt.IsZero() || !zero.CreatedAt.IsZero() {
+		t.Errorf("zero times came back as %v/%v", zero.ExpiresAt, zero.CreatedAt)
+	}
+
+	for name, bad := range map[string][]byte{
+		"empty":          nil,
+		"header only":    {recordVersion, 0},
+		"no wire":        encodeRecord(nil, &Entry{}),
+		"bad checksum":   corruptRecord(t, encodeRecord(nil, in)),
+		"bad version":    futureVersionRecord,
+		"truncated tail": {recordVersion, 0, 0x80},
+	} {
+		if _, err := decodeRecord(bad); !errors.Is(err, errBadRecord) {
+			t.Errorf("decodeRecord(%s) error = %v, want errBadRecord", name, err)
+		}
+	}
+}
@@ -3,6 +3,7 @@ package cache
 import (
 	"encoding/binary"
 	"fmt"
+	"hash/crc32"
 	"os"
 	"path/filepath"
 	"sync"
@@ -13,6 +14,154 @@ import (

 var bucketCache = []byte("cache")

+// recordVersion is the on-disk record format version. It is written as the
+// first byte of every value so a future format change can be recognised (and
+// the record dropped) instead of being silently misparsed.
+const recordVersion = 1
+
+// Record flags, stored in byte 1 of a record.
+const (
+	flagNeg       = 1 << 0
+	flagTruncated = 1 << 1
+)
+
+// errBadRecord is returned by decodeRecord for anything it cannot trust: a
+// short buffer, an unknown version, a failed checksum or trailing garbage.
+var errBadRecord = errString("cache: corrupt snapshot record")
+
+var crc32cTable = crc32.MakeTable(crc32.Castagnoli)
+
+// encodeRecord serialises one entry into a snapshot value.
+//
+// The layout is written out here, and not merely implied by the encoder and
+// decoder sitting side by side, because the two drifting apart is exactly what
+// broke the warm-restart path once already:
+//
+//	byte  0         format version (recordVersion)
+//	byte  1         flags (flagNeg, flagTruncated)
+//	uvarint         ExpiresAt as Unix nanoseconds (0 = the zero time)
+//	uvarint         StaleUntil as Unix nanoseconds (0 = the zero time)
+//	uvarint         CreatedAt as Unix nanoseconds (0 = the zero time)
+//	uvarint         TTL in seconds
+//	uvarint         len(Origin)   + Origin
+//	uvarint         len(Upstream) + Upstream
+//	rest            packed wire-format answer
+//	uint32 (LE)     CRC-32C of every byte above
+func encodeRecord(buf []byte, e *Entry) []byte {
+	buf = append(buf, recordVersion)
+	var flags uint8
+	if e.Neg {
+		flags |= flagNeg
+	}
+	if e.Truncated {
+		flags |= flagTruncated
+	}
+	buf = append(buf, flags)
+	buf = binary.AppendUvarint(buf, uint64(nanosOf(e.ExpiresAt)))
+	buf = binary.AppendUvarint(buf, uint64(nanosOf(e.StaleUntil)))
+	buf = binary.AppendUvarint(buf, uint64(nanosOf(e.CreatedAt)))
+	buf = binary.AppendUvarint(buf, uint64(e.TTL))
+	buf = binary.AppendUvarint(buf, uint64(len(e.Origin)))
+	buf = append(buf, e.Origin...)
+	buf = binary.AppendUvarint(buf, uint64(len(e.Upstream)))
+	buf = append(buf, e.Upstream...)
+	buf = append(buf, e.Wire...)
+	return binary.LittleEndian.AppendUint32(buf, crc32.Checksum(buf, crc32cTable))
+}
+
+// decodeRecord is the inverse of encodeRecord. Every failure mode returns
+// errBadRecord: a snapshot is a cache, not a source of truth, so one damaged
+// record must cost one entry and not the whole file.
+func decodeRecord(buf []byte) (*Entry, error) {
+	const header = 2
+	if len(buf) < header+crc32.Size {
+		return nil, errBadRecord
+	}
+	body := buf[:len(buf)-crc32.Size]
+	if got, want := binary.LittleEndian.Uint32(buf[len(buf)-crc32.Size:]), crc32.Checksum(body, crc32cTable); got != want {
+		return nil, errBadRecord
+	}
+	if body[0] != recordVersion {
+		return nil, errBadRecord
+	}
+	e := &Entry{Neg: body[1]&flagNeg != 0, Truncated: body[1]&flagTruncated != 0}
+	rest := body[header:]
+
+	var n int
+	var ok bool
+	if e.ExpiresAt, n, ok = timeFromVarint(rest); !ok {
+		return nil, errBadRecord
+	}
+	rest = rest[n:]
+	if e.StaleUntil, n, ok = timeFromVarint(rest); !ok {
+		return nil, errBadRecord
+	}
+	rest = rest[n:]
+	if e.CreatedAt, n, ok = timeFromVarint(rest); !ok {
+		return nil, errBadRecord
+	}
+	rest = rest[n:]
+
+	ttl, n := binary.Uvarint(rest)
+	if n <= 0 {
+		return nil, errBadRecord
+	}
+	e.TTL = uint32(ttl)
+	rest = rest[n:]
+
+	for _, dst := range []*string{&e.Origin, &e.Upstream} {
+		s, n, ok := readVarintString(rest)
+		if !ok {
+			return nil, errBadRecord
+		}
+		*dst = s
+		rest = rest[n:]
+	}
+	if len(rest) == 0 {
+		return nil, errBadRecord
+	}
+	// The buffer belongs to bbolt's mmap, so copy the wire bytes out instead
+	// of aliasing memory a later transaction may move.
+	e.Wire = append([]byte(nil), rest...)
+	e.Bytes = len(e.Wire) + len(e.Origin) + len(e.Upstream) + 96
+	return e, nil
+}
+
+// originOrSnapshot keeps whatever origin the record carried (the UI shows it)
+// but never stores an empty one.
+func originOrSnapshot(origin string) string {
+	if origin == "" {
+		return "snapshot"
+	}
+	return origin
+}
+
+func nanosOf(t time.Time) int64 {
+	if t.IsZero() {
+		return 0
+	}
+	return t.UnixNano()
+}
+
+func timeFromVarint(buf []byte) (time.Time, int, bool) {
+	v, n := binary.Uvarint(buf)
+	if n <= 0 {
+		return time.Time{}, 0, false
+	}
+	if v == 0 {
+		return time.Time{}, n, true
+	}
+	return time.Unix(0, int64(v)), n, true
+}
+
+func readVarintString(buf []byte) (string, int, bool) {
+	l, n := binary.Uvarint(buf)
+	if n <= 0 || l > uint64(len(buf)-n) {
+		return "", 0, false
+	}
+	return string(buf[n : n+int(l)]), n + int(l), true
+}
+
 // Store persists a snapshot of the LRU to a bbolt file so a restart (or a
 // rolling deploy of a cluster) resumes with a warm cache.
 //
@@ -84,44 +233,34 @@ func (s *Store) Load(l *LRU, now time.Time) (loaded int, skipped int, err error)
 			return nil
 		}
 		cur := b.Cursor()
-		for k, _ := cur.First(); k != nil; k, _ = cur.Next() {
-			key, rest, kerr := unpackKey(k)
+		for k, v := cur.First(); k != nil; k, v = cur.Next() {
+			key, _, kerr := unpackKey(k)
 			if kerr != nil {
 				skipped++
 				continue
 			}
-			expUnix, n := binary.Uvarint(rest)
-			if n <= 0 {
+			e, derr := decodeRecord(v)
+			if derr != nil {
 				skipped++
 				continue
 			}
-			staleUnix, n2 := binary.Uvarint(rest[n:])
-			if n2 <= 0 {
+			if !e.ExpiresAt.After(now) {
+				// The serve-stale window deliberately does not survive a
+				// restart: stale answers bridge a transient upstream failure
+				// inside one process lifetime, they are not a licence to
+				// resurrect an answer from minutes ago. An entry that is
+				// already past its TTL at load time is dropped instead.
 				skipped++
 				continue
 			}
-			wire := rest[n+n2:]
-			if len(wire) == 0 {
-				skipped++
-				continue
-			}
-			expires := time.Unix(0, int64(expUnix))
-			staleUntil := time.Unix(0, int64(staleUnix))
-			if staleUntil.Before(now) {
-				// Already outside the serve-stale window: not worth loading.
-				skipped++
-				continue
-			}
-			ttl := uint32(expires.Sub(now).Seconds())
-			e := &Entry{
-				Wire:       append([]byte(nil), wire...),
-				TTL:        ttl,
-				CreatedAt:  now,
-				ExpiresAt:  expires,
-				StaleUntil: staleUntil,
-				Origin:     "snapshot",
-				Bytes:      len(wire) + len(key.Name) + 96,
-			}
+			// Re-anchor on the loader's clock: the stored CreatedAt belongs to
+			// the previous process, and AgeTTL would otherwise age the answer
+			// by the length of the outage. The TTL is rewritten to the time
+			// actually left, so the TTL a client sees never jumps up across a
+			// restart.
+			e.CreatedAt = now
+			e.TTL = uint32(e.ExpiresAt.Sub(now) / time.Second)
+			e.Origin = originOrSnapshot(e.Origin)
 			l.Put(key, e)
 			loaded++
 		}
@@ -157,11 +296,7 @@ func (s *Store) Snapshot(l *LRU, now time.Time) (written int, err error) {
 			dropped++
 			return true
 		}
-		val := make([]byte, 0, len(e.Wire)+32)
-		val = binary.AppendUvarint(val, uint64(e.ExpiresAt.UnixNano()))
-		val = binary.AppendUvarint(val, uint64(e.StaleUntil.UnixNano()))
-		val = append(val, e.Wire...)
-		rows = append(rows, row{key: kb, val: val})
+		rows = append(rows, row{key: kb, val: encodeRecord(nil, e)})
 		return true
 	})
 	if s.maxEntries > 0 && len(rows) > s.maxEntries {
@@ -170,15 +305,11 @@ func (s *Store) Snapshot(l *LRU, now time.Time) (written int, err error) {
 	}

 	err = s.db.Update(func(tx *bolt.Tx) error {
-		b, err := tx.CreateBucketIfNotExists(bucketCache)
-		if err != nil {
-			return err
-		}
 		// Rebuild the bucket so a shrinking cache really shrinks the file.
 		if err := tx.DeleteBucket(bucketCache); err != nil && err != bolt.ErrBucketNotFound {
 			return err
 		}
-		b, err = tx.CreateBucketIfNotExists(bucketCache)
+		b, err := tx.CreateBucketIfNotExists(bucketCache)
 		if err != nil {
 			return err
 		}