Add a raw cache file format

cache-blob.go

@@ -38,11 +38,11 @@ import (
 // blob be matched, or hashed, without decoding it back into an lruKey.
 type cacheBlob []byte
 
-// blobVersion is the version of the layout above. It is reserved rather than
-// used: every blob is freshly allocated and so carries 0, and nothing reads it
-// back. It is here so that a later change to the layout has somewhere to say
-// so, and can tell its own records from these. A writer that bumps it has to
-// start setting it explicitly.
+// blobVersion is the version of the layout above. A blob is freshly allocated
+// and so carries 0 without being written, but it is read back: a blob read
+// from the cache file is rejected unless it carries this version, so a later
+// change to the layout keeps old records out of new accessors. A writer that
+// bumps it has to start setting it explicitly.
 //
 // Note that 1 is already taken: it is binaryFormatVersion, the Redis record
 // format, which is a different layout in the same first byte. A stored form

cache-file-raw.go

@@ -0,0 +1,141 @@
+package rdns
+
+import (
+	"bufio"
+	"encoding/binary"
+	"errors"
+	"fmt"
+	"io"
+
+	"github.com/miekg/dns"
+)
+
+// The raw cache file stores entries in the form the cache already holds them,
+// so writing is a copy and reading needs no parsing:
+//
+//	header:  magic, then a version byte
+//	record:  uint32 length, then the blob, repeated to end of file
+//
+// The magic starts with a NUL so it can't be mistaken for the JSON format,
+// whose records start with '{', and so anything that opens the file can see
+// at once that it is binary.
+//
+// The version belongs to the file rather than to each record: it is checked
+// once, and a file cannot hold a mix. The version byte inside a blob stays
+// reserved for a record that travels on its own, without a file around it.
+const (
+	rawCacheMagic     = "\x00RDC"
+	rawCacheVersion   = 1
+	rawCacheHeaderLen = len(rawCacheMagic) + 1
+
+	// An upper bound on a stored record, so a corrupt length can't ask for an
+	// unbounded allocation. This is a sanity limit rather than a property of
+	// the data: a stored entry is not bounded by the 64KB wire limit, because
+	// the cache packs without name compression and a response that arrived
+	// near the limit can be twice that once stored. The theoretical worst case
+	// is a few megabytes, so this sits well above anything real while still
+	// keeping a bad length to a bounded allocation.
+	maxRawCacheRecord = 16 << 20
+)
+
+// isRawCacheFile reports whether the reader is positioned at a raw cache file,
+// leaving it unread either way so the right decoder can take it from the top.
+func isRawCacheFile(r *bufio.Reader) bool {
+	magic, err := r.Peek(len(rawCacheMagic))
+	return err == nil && string(magic) == rawCacheMagic
+}
+
+func (c *lruCache) serializeRaw(w io.Writer) error {
+	header := make([]byte, 0, rawCacheHeaderLen)
+	header = append(header, rawCacheMagic...)
+	header = append(header, rawCacheVersion)
+	if _, err := w.Write(header); err != nil {
+		return err
+	}
+
+	var length [4]byte
+	for item := c.tail.prev; item != c.head; item = item.prev {
+		if len(item.blob) > maxRawCacheRecord {
+			// Nothing on the store path bounds an entry this far, so this
+			// should not happen; say so rather than dropping it in silence.
+			Log.Warn("cache entry too large for the cache file, skipping",
+				"size", len(item.blob), "name", item.blob.key().Question.Name)
+			continue
+		}
+		binary.BigEndian.PutUint32(length[:], uint32(len(item.blob)))
+		if _, err := w.Write(length[:]); err != nil {
+			return err
+		}
+		if _, err := w.Write(item.blob); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// deserializeRaw reads a raw cache file. A record that can't be used is
+// skipped, as in the JSON format. Damage to the framing is different: there is
+// no way to find the next record once a length is untrustworthy, so reading
+// stops there and keeps the entries it already has.
+func (c *lruCache) deserializeRaw(r io.Reader) error {
+	header := make([]byte, rawCacheHeaderLen)
+	if _, err := io.ReadFull(r, header); err != nil {
+		return fmt.Errorf("failed to read cache file header: %w", err)
+	}
+	if string(header[:len(rawCacheMagic)]) != rawCacheMagic {
+		return errors.New("not a raw cache file")
+	}
+	if version := header[len(rawCacheMagic)]; version != rawCacheVersion {
+		return fmt.Errorf("unsupported raw cache file version %d", version)
+	}
+
+	var length [4]byte
+	for {
+		if _, err := io.ReadFull(r, length[:]); err != nil {
+			if errors.Is(err, io.EOF) {
+				return nil // clean end of file
+			}
+			Log.Warn("cache file ends mid-record, keeping the entries read so far", "error", err)
+			return nil
+		}
+
+		n := binary.BigEndian.Uint32(length[:])
+		if n < blobHdrLen || n > maxRawCacheRecord {
+			Log.Warn("cache file record has an implausible length, stopping", "length", n)
+			return nil
+		}
+
+		blob := make(cacheBlob, n)
+		if _, err := io.ReadFull(r, blob); err != nil {
+			Log.Warn("cache file ends mid-record, keeping the entries read so far", "error", err)
+			return nil
+		}
+
+		key, ok := blobFromFile(blob)
+		if !ok {
+			continue // skip the record, the framing is still good
+		}
+		c.addKey(key, blob)
+	}
+}
+
+// blobFromFile checks a blob read from disk far enough to be sure the
+// accessors on it are safe, and returns the key it is stored under.
+func blobFromFile(blob cacheBlob) (lruKey, bool) {
+	if len(blob) < blobHdrLen || blob.version() != blobVersion {
+		return lruKey{}, false
+	}
+	if blobHdrLen+blob.netLen()+blob.nameLen() > len(blob) {
+		return lruKey{}, false
+	}
+	// Unpack once here so a corrupt message is kept out of the cache rather
+	// than taking up an entry until the lookup that finds it evicts it.
+	if err := new(dns.Msg).Unpack(blob.message()); err != nil {
+		return lruKey{}, false
+	}
+	key := blob.key()
+	if key.Question.Name == "" {
+		return lruKey{}, false
+	}
+	return key, true
+}

cache-file-raw_test.go

@@ -0,0 +1,212 @@
+package rdns
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/miekg/dns"
+	"github.com/stretchr/testify/require"
+)
+
+func rawTestEntry(t *testing.T, i int) (*dns.Msg, *cacheAnswer) {
+	t.Helper()
+	name := fmt.Sprintf("host%d.example.com.", i)
+	q := new(dns.Msg)
+	q.SetQuestion(name, dns.TypeA)
+	a := new(dns.Msg)
+	a.SetReply(q)
+	rr, err := dns.NewRR(fmt.Sprintf("%s 3600 IN A 192.0.%d.%d", name, i/256, i%256))
+	require.NoError(t, err)
+	a.Answer = []dns.RR{rr}
+
+	now := time.Now()
+	return q, &cacheAnswer{Msg: a, Timestamp: now, Expiry: now.Add(time.Hour)}
+}
+
+func fillCache(t *testing.T, c *lruCache, n int) []*dns.Msg {
+	t.Helper()
+	queries := make([]*dns.Msg, n)
+	for i := range n {
+		q, answer := rawTestEntry(t, i)
+		queries[i] = q
+		key := lruKeyFromQuery(q)
+		blob, err := newCacheBlob(key, answer)
+		require.NoError(t, err)
+		c.addKey(key, blob)
+	}
+	return queries
+}
+
+func TestRawCacheFileRoundTrip(t *testing.T) {
+	src := newLRUCache(0)
+	queries := fillCache(t, src, 20)
+
+	var buf bytes.Buffer
+	require.NoError(t, src.serializeRaw(&buf))
+
+	dst := newLRUCache(0)
+	require.NoError(t, dst.deserializeRaw(bytes.NewReader(buf.Bytes())))
+	require.Equal(t, len(queries), dst.size())
+	for _, q := range queries {
+		key := lruKeyFromQuery(q)
+		require.NotNil(t, dst.find(dst.hash(key), key), "%s must be findable again", key.Question.Name)
+	}
+
+	// Writing it back out reproduces the file, so the queue order survives too
+	// and the least recently used entry is still the first to go.
+	var again bytes.Buffer
+	require.NoError(t, dst.serializeRaw(&again))
+	require.Equal(t, buf.Bytes(), again.Bytes())
+}
+
+// The option picks the format to write; the format to read comes from the file
+// itself, so an existing cache survives the option being changed either way.
+func TestCacheFileFormatIsDetectedNotConfigured(t *testing.T) {
+	for _, tc := range []struct {
+		wrote, prefix, thenRunsAs string
+	}{
+		{"", "{", CacheFileFormatRaw},                            // unset writes JSON
+		{CacheFileFormatRaw, rawCacheMagic, CacheFileFormatJSON}, // raw read back by a JSON-configured run
+	} {
+		t.Run(tc.wrote+"-then-"+tc.thenRunsAs, func(t *testing.T) {
+			filename := filepath.Join(t.TempDir(), "cache")
+
+			first := NewMemoryBackend(MemoryBackendOptions{
+				GCPeriod: time.Hour, Filename: filename, FileFormat: tc.wrote,
+			})
+			queries := make([]*dns.Msg, 5)
+			for i := range queries {
+				q, answer := rawTestEntry(t, i)
+				queries[i] = q
+				first.Store(q, answer)
+			}
+			require.NoError(t, first.Close())
+
+			content, err := os.ReadFile(filename)
+			require.NoError(t, err)
+			require.True(t, bytes.HasPrefix(content, []byte(tc.prefix)), "%q wrote the wrong format", tc.wrote)
+
+			second := NewMemoryBackend(MemoryBackendOptions{
+				GCPeriod: time.Hour, Filename: filename, FileFormat: tc.thenRunsAs,
+			})
+			defer second.Close()
+			require.Equal(t, len(queries), second.Size())
+			for _, q := range queries {
+				_, _, ok := second.Lookup(q)
+				require.True(t, ok, "%s must survive the format switch", q.Question[0].Name)
+			}
+		})
+	}
+}
+
+// Downgrading the binary has to cost a cold cache and nothing worse, which is
+// what the magic buys: a build that predates the format can't parse it.
+func TestRawCacheFileRejectedByJSONReader(t *testing.T) {
+	src := newLRUCache(0)
+	fillCache(t, src, 3)
+	var buf bytes.Buffer
+	require.NoError(t, src.serializeRaw(&buf))
+
+	dst := newLRUCache(0)
+	require.Error(t, dst.deserialize(bytes.NewReader(buf.Bytes())))
+	require.Zero(t, dst.size())
+}
+
+// Damage to a record costs that record; damage to the framing costs the rest
+// of the file, because there is no way to find the next record from it. Either
+// way what was read before it is kept.
+func TestRawCacheFileCorruption(t *testing.T) {
+	src := newLRUCache(0)
+	fillCache(t, src, 10)
+	var buf bytes.Buffer
+	require.NoError(t, src.serializeRaw(&buf))
+	good := buf.Bytes()
+	firstLen := int(binary.BigEndian.Uint32(good[rawCacheHeaderLen:]))
+
+	t.Run("corrupt message skips one record", func(t *testing.T) {
+		// Scribble on the packed message, leaving the framing intact.
+		damaged := bytes.Clone(good)
+		msg := damaged[rawCacheHeaderLen+4 : rawCacheHeaderLen+4+firstLen]
+		for i := len(msg) - 8; i < len(msg); i++ {
+			msg[i] = 0xff
+		}
+		dst := newLRUCache(0)
+		require.NoError(t, dst.deserializeRaw(bytes.NewReader(damaged)))
+		require.Equal(t, 9, dst.size(), "the other records still load")
+	})
+
+	t.Run("truncated file keeps what came before", func(t *testing.T) {
+		dst := newLRUCache(0)
+		require.NoError(t, dst.deserializeRaw(bytes.NewReader(good[:len(good)-20])))
+		require.Equal(t, 9, dst.size())
+	})
+
+	t.Run("implausible length is refused not allocated", func(t *testing.T) {
+		damaged := bytes.Clone(good)
+		binary.BigEndian.PutUint32(damaged[rawCacheHeaderLen+4+firstLen:], 3_000_000_000)
+		dst := newLRUCache(0)
+		require.NoError(t, dst.deserializeRaw(bytes.NewReader(damaged)))
+		require.Equal(t, 1, dst.size(), "only the record before the bad length is kept")
+	})
+}
+
+// A stored entry is not bounded by the 64KB wire limit: the cache packs
+// without name compression, so a response that arrived comfortably inside the
+// limit can be twice that once stored. The record limit has to leave room for
+// it rather than dropping the largest entries on the way to disk.
+func TestRawCacheFileKeepsOversizedEntries(t *testing.T) {
+	q := new(dns.Msg)
+	q.SetQuestion("big.example.com.", dns.TypeTXT)
+	a := new(dns.Msg)
+	a.SetReply(q)
+	name := strings.Repeat("averylonglabelusedtomakethisnamebig.", 6) + "example.com."
+	for range 290 {
+		rr, err := dns.NewRR(fmt.Sprintf(`%s 3600 IN TXT "%s"`, name, strings.Repeat("x", 200)))
+		require.NoError(t, err)
+		a.Answer = append(a.Answer, rr)
+	}
+
+	a.Compress = true
+	onTheWire, err := a.Pack()
+	require.NoError(t, err)
+	require.Less(t, len(onTheWire), dns.MaxMsgSize, "this has to be a response that could arrive")
+	a.Compress = false
+
+	now := time.Now()
+	key := lruKeyFromQuery(q)
+	blob, err := newCacheBlob(key, &cacheAnswer{Msg: a, Timestamp: now, Expiry: now.Add(time.Hour)})
+	require.NoError(t, err)
+	require.Greater(t, len(blob), dns.MaxMsgSize, "stored uncompressed, it outgrows the wire limit")
+
+	src := newLRUCache(0)
+	src.addKey(key, blob)
+	var buf bytes.Buffer
+	require.NoError(t, src.serializeRaw(&buf))
+
+	dst := newLRUCache(0)
+	require.NoError(t, dst.deserializeRaw(bytes.NewReader(buf.Bytes())))
+	require.Equal(t, 1, dst.size(), "the entry must survive the file, not be skipped")
+}
+
+// Persisting blobs makes their version byte part of an on-disk format, so a
+// record laid out by a later version has to be refused rather than read
+// through accessors that no longer match it.
+func TestRawCacheFileRejectsUnknownBlobVersion(t *testing.T) {
+	src := newLRUCache(0)
+	fillCache(t, src, 3)
+	var buf bytes.Buffer
+	require.NoError(t, src.serializeRaw(&buf))
+
+	damaged := bytes.Clone(buf.Bytes())
+	damaged[rawCacheHeaderLen+4+blobOffVersion] = blobVersion + 1
+
+	dst := newLRUCache(0)
+	require.NoError(t, dst.deserializeRaw(bytes.NewReader(damaged)))
+	require.Equal(t, 2, dst.size(), "the record with an unknown layout is skipped, the rest load")
+}