fix(kv): protect record length with a header checksum

the 4-byte length was outside the CRC, so a bit flip in it was misread as a torn tail and silently truncated every later record. Files now carry a magic+version header and each record header carries its own CRC; legacy files replay with the original semantics and are upgraded in place through the existing compaction protocol.

commit: fd389f8054552358d660084d696f49bd90058811
author: brice · 2mo ago · 2026-07-26 04:54 UTC
parent: bd37e91
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

3 files · +554 -71

M pkg/kv/codec.go +44 -6


@@ -12,7 +12,32 @@ const (
12	12		opDel
13	13		opBatch
14	14
15		-	headerLen = 8
	15	+	// Legacy record framing (files with no file header): a 4-byte big-endian
	16	+	// payload length followed by CRC32(payload). The length itself is
	17	+	// unprotected, so a bit flip in it is indistinguishable from a torn tail.
	18	+	// These files are replayed with the original semantics and upgraded in
	19	+	// place by Open; nothing ever writes this framing again.
	20	+	legacyHeaderLen = 8
	21	+
	22	+	// Current framing. Every file starts with fileMagic plus a uint16 format
	23	+	// version, and every record carries a self-verifying header:
	24	+	//
	25	+	// [0:4] payload length, big endian
	26	+	// [4:8] CRC32(payload)
	27	+	// [8:12] CRC32(header[0:8])
	28	+	//
	29	+	// The header checksum is what makes the length trustworthy: a corrupt
	30	+	// length now fails closed instead of masquerading as a torn tail and
	31	+	// silently truncating every record that follows it.
	32	+	recordHeaderLen = 12
	33	+	fileMagicLen = 4
	34	+	fileHeaderLen = fileMagicLen + 2
	35	+	formatVersion = 1
	36	+
	37	+	// maxPayloadLen bounds any allocation driven by a length field. Catalogue
	38	+	// records and batches are orders of magnitude smaller than this.
	39	+	maxPayloadLen = 64 << 20
	40	+
16	41		payloadPrefixLen = 3
17	42		batchEntryPrefixLen = 7
18	43		compactThreshold = 64

@@ -20,6 +45,10 @@ const (
20	45		maxKeyLen = 1<<16 - 1
21	46		)
22	47
	48	+	// fileMagic cannot be mistaken for a legacy record header: read as a legacy
	49	+	// big-endian length it is ~1.4 GiB, far beyond maxPayloadLen.
	50	+	var fileMagic = [fileMagicLen]byte{'V', 'E', 'V', 'K'}
	51	+
23	52		var (
24	53		errBadRecord = errors.New("bad record")
25	54		ErrCorruptWAL = errors.New("kv: corrupt WAL")

@@ -49,7 +78,7 @@ func encodeRecord(op byte, key, value []byte) ([]byte, error) {
49	78		if op == opSet {
50	79		payloadLen += len(value)
51	80		}
52		-	if uint64(payloadLen) > math.MaxUint32 {
	81	+	if payloadLen > maxPayloadLen {
53	82		return nil, errors.New("payload too large")
54	83		}
55	84		payload := make([]byte, payloadLen)

@@ -86,7 +115,7 @@ func encodeBatch(changes []BatchChange) ([]byte, error) {
86	115		}
87	116		}
88	117		payloadLen += 1 + 2 + len(change.Key) + 4 + valueLen
89		-	if uint64(payloadLen) > math.MaxUint32 {
	118	+	if payloadLen > maxPayloadLen {
90	119		return nil, errors.New("payload too large")
91	120		}
92	121		}

@@ -112,11 +141,20 @@ func encodeBatch(changes []BatchChange) ([]byte, error) {
112	141		return framePayload(payload), nil
113	142		}
114	143
	144	+	// fileHeader is the fixed prefix of every file written in the current format.
	145	+	func fileHeader() []byte {
	146	+	buf := make([]byte, fileHeaderLen)
	147	+	copy(buf, fileMagic[:])
	148	+	binary.BigEndian.PutUint16(buf[fileMagicLen:], formatVersion)
	149	+	return buf
	150	+	}
	151	+
115	152		func framePayload(payload []byte) []byte {
116		-	buf := make([]byte, headerLen+len(payload))
117		-	binary.BigEndian.PutUint32(buf[:4], uint32(len(payload)))
	153	+	buf := make([]byte, recordHeaderLen+len(payload))
	154	+	binary.BigEndian.PutUint32(buf[0:4], uint32(len(payload)))
118	155		binary.BigEndian.PutUint32(buf[4:8], crc32.ChecksumIEEE(payload))
119		-	copy(buf[headerLen:], payload)
	156	+	binary.BigEndian.PutUint32(buf[8:recordHeaderLen], crc32.ChecksumIEEE(buf[0:8]))
	157	+	copy(buf[recordHeaderLen:], payload)
120	158		return buf
121	159		}
122	160

 	opDel
 	opBatch
-	headerLen           = 8
+	// Legacy record framing (files with no file header): a 4-byte big-endian
+	// payload length followed by CRC32(payload). The length itself is
+	// unprotected, so a bit flip in it is indistinguishable from a torn tail.
+	// These files are replayed with the original semantics and upgraded in
+	// place by Open; nothing ever writes this framing again.
+	legacyHeaderLen = 8
+	// Current framing. Every file starts with fileMagic plus a uint16 format
+	// version, and every record carries a self-verifying header:
+	//
+	//	[0:4]   payload length, big endian
+	//	[4:8]   CRC32(payload)
+	//	[8:12]  CRC32(header[0:8])
+	//
+	// The header checksum is what makes the length trustworthy: a corrupt
+	// length now fails closed instead of masquerading as a torn tail and
+	// silently truncating every record that follows it.
+	recordHeaderLen = 12
+	fileMagicLen    = 4
+	fileHeaderLen   = fileMagicLen + 2
+	formatVersion   = 1
+	// maxPayloadLen bounds any allocation driven by a length field. Catalogue
+	// records and batches are orders of magnitude smaller than this.
+	maxPayloadLen = 64 << 20
 	payloadPrefixLen    = 3
 	batchEntryPrefixLen = 7
 	compactThreshold    = 64
 	maxKeyLen           = 1<<16 - 1
 )
+// fileMagic cannot be mistaken for a legacy record header: read as a legacy
+// big-endian length it is ~1.4 GiB, far beyond maxPayloadLen.
+var fileMagic = [fileMagicLen]byte{'V', 'E', 'V', 'K'}
 var (
 	errBadRecord  = errors.New("bad record")
 	ErrCorruptWAL = errors.New("kv: corrupt WAL")
 	if op == opSet {
 		payloadLen += len(value)
 	}
-	if uint64(payloadLen) > math.MaxUint32 {
+	if payloadLen > maxPayloadLen {
 		return nil, errors.New("payload too large")
 	}
 	payload := make([]byte, payloadLen)
 			}
 		}
 		payloadLen += 1 + 2 + len(change.Key) + 4 + valueLen
-		if uint64(payloadLen) > math.MaxUint32 {
+		if payloadLen > maxPayloadLen {
 			return nil, errors.New("payload too large")
 		}
 	}
 	return framePayload(payload), nil
 }
+// fileHeader is the fixed prefix of every file written in the current format.
+func fileHeader() []byte {
+	buf := make([]byte, fileHeaderLen)
+	copy(buf, fileMagic[:])
+	binary.BigEndian.PutUint16(buf[fileMagicLen:], formatVersion)
+	return buf
+}
 func framePayload(payload []byte) []byte {
-	buf := make([]byte, headerLen+len(payload))
+	buf := make([]byte, recordHeaderLen+len(payload))
-	binary.BigEndian.PutUint32(buf[:4], uint32(len(payload)))
+	binary.BigEndian.PutUint32(buf[0:4], uint32(len(payload)))
 	binary.BigEndian.PutUint32(buf[4:8], crc32.ChecksumIEEE(payload))
-	copy(buf[headerLen:], payload)
+	binary.BigEndian.PutUint32(buf[8:recordHeaderLen], crc32.ChecksumIEEE(buf[0:8]))
+	copy(buf[recordHeaderLen:], payload)
 	return buf
 }

M pkg/kv/kv.go +134 -21

@@ -1,6 +1,7 @@
1	1		package kv
2	2
3	3		import (
	4	+	"bytes"
4	5		"encoding/binary"
5	6		"errors"
6	7		"fmt"

@@ -15,7 +16,9 @@ import (
15	16		"github.com/bnema/vev/pkg/safedir"
16	17		)
17	18
18		-	// ReplayResult describes a strictly validated WAL prefix.
	19	+	// ReplayResult describes a strictly validated WAL prefix. LastGood is an
	20	+	// absolute file offset, so truncating a torn tail to it preserves the file
	21	+	// header of a current-format file.
19	22		type ReplayResult struct {
20	23		Data map[string][]byte
21	24		EntrySize map[string]int64

@@ -23,8 +26,34 @@ type ReplayResult struct {
23	26		Live int64
24	27		LastGood int64
25	28		TornTail bool
	29	+	Format walFormat
26	30		}
27	31
	32	+	// walFormat is the on-disk framing a file was written with.
	33	+	type walFormat uint8
	34	+
	35	+	const (
	36	+	// formatLegacy is a file written before the record header carried its own
	37	+	// checksum. It is read with the original semantics and upgraded at Open.
	38	+	formatLegacy walFormat = iota
	39	+	formatCurrent
	40	+	)
	41	+
	42	+	func (f walFormat) headerLen() int64 {
	43	+	if f == formatCurrent {
	44	+	return recordHeaderLen
	45	+	}
	46	+	return legacyHeaderLen
	47	+	}
	48	+
	49	+	type headerStatus uint8
	50	+
	51	+	const (
	52	+	headerOK headerStatus = iota
	53	+	headerEOF
	54	+	headerTorn
	55	+	)
	56	+
28	57		type fsHooks struct {
29	58		Open func(string, int, os.FileMode) (*os.File, error)
30	59		Write func(*os.File, []byte) (int, error)

@@ -97,7 +126,15 @@ func openWithHooks(path string, hooks fsHooks) (*Store, error) {
97	126		return cleanupLock(err)
98	127		}
99	128		s := &Store{path: path, file: f, lockFile: lockFile, data: result.Data, entrySize: result.EntrySize, total: result.Total, live: result.Live, hooks: hooks}
100		-	if s.shouldCompact() {
	129	+	// A legacy file has already replayed with its original semantics, so the
	130	+	// data is exactly what the previous release would have loaded. Rewriting it
	131	+	// through the compaction protocol upgrades the framing without a second
	132	+	// rewrite path: every crash state it can leave behind is one recoverCompaction
	133	+	// already resolves, so a restart sees either the whole legacy file or the
	134	+	// whole upgraded one. A failed compaction poisons the store, so there is no
	135	+	// safe way to continue in legacy mode; Open fails closed and the next start
	136	+	// retries from whichever complete file recovery selects.
	137	+	if result.Format != formatCurrent \|\| s.shouldCompact() {
101	138		if err := s.compactLocked(); err != nil {
102	139		_ = f.Close()
103	140		return cleanupLock(err)

@@ -266,6 +303,16 @@ func (s *Store) compactLocked() (retErr error) {
266	303		if err != nil {
267	304		return err
268	305		}
	306	+	// Every replacement is written in the current format, which is also how a
	307	+	// legacy file is upgraded.
	308	+	header := fileHeader()
	309	+	if n, err := s.hooks.Write(f, header); err != nil \|\| n != len(header) {
	310	+	_ = s.hooks.Close(f)
	311	+	if err == nil {
	312	+	err = io.ErrShortWrite
	313	+	}
	314	+	return err
	315	+	}
269	316		keys := make([]string, 0, len(s.data))
270	317		for k := range s.data {
271	318		keys = append(keys, k)

@@ -356,7 +403,7 @@ func (s *Store) compactLocked() (retErr error) {
356	403		return err
357	404		}
358	405		result, err := replayFile(current)
359		-	if err != nil \|\| result.TornTail {
	406	+	if err != nil \|\| result.TornTail \|\| result.Format != formatCurrent {
360	407		_ = s.hooks.Close(current)
361	408		if err == nil {
362	409		err = ErrCorruptWAL

@@ -402,6 +449,10 @@ func (s *Store) poisonLocked() error {
402	449		return errors.Join(fileErr, releaseLock(lockFile))
403	450		}
404	451
	452	+	// replayFile validates a WAL prefix in whichever framing the file uses. A torn
	453	+	// tail is the only truncatable outcome; every other inconsistency is corruption
	454	+	// and fails closed, so a damaged record can never be silently dropped along
	455	+	// with everything written after it.
405	456		func replayFile(f *os.File) (ReplayResult, error) {
406	457		info, err := f.Stat()
407	458		if err != nil {

@@ -411,34 +462,34 @@ func replayFile(f *os.File) (ReplayResult, error) {
411	462		return ReplayResult{}, err
412	463		}
413	464		fileSize := info.Size()
414		-	result := ReplayResult{Data: make(map[string][]byte), EntrySize: make(map[string]int64)}
415		-	header := make([]byte, headerLen)
	465	+	format, err := readFileFormat(f, fileSize)
	466	+	if err != nil {
	467	+	return ReplayResult{}, err
	468	+	}
	469	+	result := ReplayResult{Data: make(map[string][]byte), EntrySize: make(map[string]int64), Format: format}
	470	+	if format == formatCurrent {
	471	+	result.LastGood = fileHeaderLen
	472	+	}
416	473		for {
417		-	n, err := io.ReadFull(f, header)
418		-	if errors.Is(err, io.EOF) {
419		-	return result, nil
420		-	}
421		-	if errors.Is(err, io.ErrUnexpectedEOF) {
422		-	_ = n
423		-	result.TornTail = true
424		-	return result, nil
425		-	}
	474	+	header, status, err := readRecordHeader(f, format, fileSize, result.LastGood)
426	475		if err != nil {
427	476		return ReplayResult{}, err
428	477		}
429		-	payloadLen := binary.BigEndian.Uint32(header[:4])
430		-	if int64(payloadLen) > fileSize-result.LastGood-headerLen {
	478	+	if status == headerEOF {
	479	+	return result, nil
	480	+	}
	481	+	if status == headerTorn {
431	482		result.TornTail = true
432	483		return result, nil
433	484		}
434		-	payload := make([]byte, int(payloadLen))
	485	+	payload := make([]byte, int(header.payloadLen))
435	486		if _, err := io.ReadFull(f, payload); errors.Is(err, io.EOF) \|\| errors.Is(err, io.ErrUnexpectedEOF) {
436	487		result.TornTail = true
437	488		return result, nil
438	489		} else if err != nil {
439	490		return ReplayResult{}, err
440	491		}
441		-	if crc32.ChecksumIEEE(payload) != binary.BigEndian.Uint32(header[4:8]) {
	492	+	if crc32.ChecksumIEEE(payload) != header.payloadCRC {
442	493		return ReplayResult{}, fmt.Errorf("%w: CRC mismatch at offset %d", ErrCorruptWAL, result.LastGood)
443	494		}
444	495		var records []record

@@ -452,13 +503,73 @@ func replayFile(f *os.File) (ReplayResult, error) {
452	503		if err != nil {
453	504		return ReplayResult{}, fmt.Errorf("%w: invalid record at offset %d: %v", ErrCorruptWAL, result.LastGood, err)
454	505		}
455		-	size := int64(headerLen + len(payload))
	506	+	size := format.headerLen() + int64(len(payload))
456	507		applyRecords(result.Data, result.EntrySize, &result.Live, records, size)
457	508		result.Total += size
458	509		result.LastGood += size
459	510		}
460	511		}
461	512
	513	+	// readFileFormat consumes the file header when one is present and leaves the
	514	+	// reader positioned at the first record either way.
	515	+	func readFileFormat(f io.ReadSeeker, fileSize int64) (walFormat, error) {
	516	+	if fileSize < fileHeaderLen {
	517	+	return formatLegacy, nil
	518	+	}
	519	+	header := make([]byte, fileHeaderLen)
	520	+	if _, err := io.ReadFull(f, header); err != nil {
	521	+	return formatLegacy, err
	522	+	}
	523	+	if !bytes.Equal(header[:fileMagicLen], fileMagic[:]) {
	524	+	if _, err := f.Seek(0, io.SeekStart); err != nil {
	525	+	return formatLegacy, err
	526	+	}
	527	+	return formatLegacy, nil
	528	+	}
	529	+	if version := binary.BigEndian.Uint16(header[fileMagicLen:]); version != formatVersion {
	530	+	return formatLegacy, fmt.Errorf("%w: unsupported format version %d", ErrCorruptWAL, version)
	531	+	}
	532	+	return formatCurrent, nil
	533	+	}
	534	+
	535	+	type recordHeader struct {
	536	+	payloadLen uint32
	537	+	payloadCRC uint32
	538	+	}
	539	+
	540	+	// readRecordHeader implements the replay decision table. In the current format
	541	+	// the header checksum is verified before the length is trusted for anything:
	542	+	// an unverifiable header is corruption, and only a header that checks out may
	543	+	// declare the record short (a torn tail).
	544	+	func readRecordHeader(f io.Reader, format walFormat, fileSize, offset int64) (recordHeader, headerStatus, error) {
	545	+	buf := make([]byte, format.headerLen())
	546	+	if _, err := io.ReadFull(f, buf); errors.Is(err, io.EOF) {
	547	+	return recordHeader{}, headerEOF, nil
	548	+	} else if errors.Is(err, io.ErrUnexpectedEOF) {
	549	+	return recordHeader{}, headerTorn, nil
	550	+	} else if err != nil {
	551	+	return recordHeader{}, headerOK, err
	552	+	}
	553	+	header := recordHeader{
	554	+	payloadLen: binary.BigEndian.Uint32(buf[0:4]),
	555	+	payloadCRC: binary.BigEndian.Uint32(buf[4:8]),
	556	+	}
	557	+	if format == formatCurrent {
	558	+	if crc32.ChecksumIEEE(buf[0:8]) != binary.BigEndian.Uint32(buf[8:recordHeaderLen]) {
	559	+	return recordHeader{}, headerOK, fmt.Errorf("%w: record header CRC mismatch at offset %d", ErrCorruptWAL, offset)
	560	+	}
	561	+	// The length is now trustworthy, so an implausible one is corruption
	562	+	// rather than a short tail. Checked before any allocation.
	563	+	if header.payloadLen > maxPayloadLen {
	564	+	return recordHeader{}, headerOK, fmt.Errorf("%w: record length %d exceeds maximum at offset %d", ErrCorruptWAL, header.payloadLen, offset)
	565	+	}
	566	+	}
	567	+	if int64(header.payloadLen) > fileSize-offset-format.headerLen() {
	568	+	return recordHeader{}, headerTorn, nil
	569	+	}
	570	+	return header, headerOK, nil
	571	+	}
	572	+
462	573		func applyRecords(data map[string][]byte, sizes map[string]int64, live *int64, records []record, size int64) {
463	574		setCount := int64(0)
464	575		for _, rec := range records {

@@ -495,7 +606,7 @@ func recoverCompaction(path string, hooks fsHooks) error {
495	606		}
496	607		dir := filepath.Dir(path)
497	608		if currentExists && currentErr == nil {
498		-	if nextExists && !prevExists {
	609	+	if nextExists {
499	610		if err := hooks.Remove(path + ".next"); err != nil {
500	611		return err
501	612		}

@@ -580,7 +691,9 @@ func validatePath(path string) error {
580	691		if err != nil {
581	692		return err
582	693		}
583		-	if !exists \|\| result.TornTail {
	694	+	// A replacement this package just wrote must be complete and in the current
	695	+	// format; anything else means the write did not land as intended.
	696	+	if !exists \|\| result.TornTail \|\| result.Format != formatCurrent {
584	697		return ErrCorruptWAL
585	698		}
586	699		return nil

@@ -1,6 +1,7 @@
1		package kv	1		package kv
2			2
3		import (	3		import (
			4	+	"bytes"
4		"encoding/binary"	5		"encoding/binary"
5		"errors"	6		"errors"
6		"fmt"	7		"fmt"

@@ -15,7 +16,9 @@ import (
15		"github.com/bnema/vev/pkg/safedir"	16		"github.com/bnema/vev/pkg/safedir"
16		)	17		)
17			18
18	-	// ReplayResult describes a strictly validated WAL prefix.	19	+	// ReplayResult describes a strictly validated WAL prefix. LastGood is an
			20	+	// absolute file offset, so truncating a torn tail to it preserves the file
			21	+	// header of a current-format file.
19		type ReplayResult struct {	22		type ReplayResult struct {
20		Data map[string][]byte	23		Data map[string][]byte
21		EntrySize map[string]int64	24		EntrySize map[string]int64

@@ -23,8 +26,34 @@ type ReplayResult struct {
23		Live int64	26		Live int64
24		LastGood int64	27		LastGood int64
25		TornTail bool	28		TornTail bool
			29	+	Format walFormat
26		}	30		}
27			31
			32	+	// walFormat is the on-disk framing a file was written with.
			33	+	type walFormat uint8
			34	+
			35	+	const (
			36	+	// formatLegacy is a file written before the record header carried its own
			37	+	// checksum. It is read with the original semantics and upgraded at Open.
			38	+	formatLegacy walFormat = iota
			39	+	formatCurrent
			40	+	)
			41	+
			42	+	func (f walFormat) headerLen() int64 {
			43	+	if f == formatCurrent {
			44	+	return recordHeaderLen
			45	+	}
			46	+	return legacyHeaderLen
			47	+	}
			48	+
			49	+	type headerStatus uint8
			50	+
			51	+	const (
			52	+	headerOK headerStatus = iota
			53	+	headerEOF
			54	+	headerTorn
			55	+	)
			56	+
28		type fsHooks struct {	57		type fsHooks struct {
29		Open func(string, int, os.FileMode) (*os.File, error)	58		Open func(string, int, os.FileMode) (*os.File, error)
30		Write func(*os.File, []byte) (int, error)	59		Write func(*os.File, []byte) (int, error)

@@ -97,7 +126,15 @@ func openWithHooks(path string, hooks fsHooks) (*Store, error) {
97		return cleanupLock(err)	126		return cleanupLock(err)
98		}	127		}
99		s := &Store{path: path, file: f, lockFile: lockFile, data: result.Data, entrySize: result.EntrySize, total: result.Total, live: result.Live, hooks: hooks}	128		s := &Store{path: path, file: f, lockFile: lockFile, data: result.Data, entrySize: result.EntrySize, total: result.Total, live: result.Live, hooks: hooks}
100	-	if s.shouldCompact() {	129	+	// A legacy file has already replayed with its original semantics, so the
			130	+	// data is exactly what the previous release would have loaded. Rewriting it
			131	+	// through the compaction protocol upgrades the framing without a second
			132	+	// rewrite path: every crash state it can leave behind is one recoverCompaction
			133	+	// already resolves, so a restart sees either the whole legacy file or the
			134	+	// whole upgraded one. A failed compaction poisons the store, so there is no
			135	+	// safe way to continue in legacy mode; Open fails closed and the next start
			136	+	// retries from whichever complete file recovery selects.
			137	+	if result.Format != formatCurrent \|\| s.shouldCompact() {
101		if err := s.compactLocked(); err != nil {	138		if err := s.compactLocked(); err != nil {
102		_ = f.Close()	139		_ = f.Close()
103		return cleanupLock(err)	140		return cleanupLock(err)

@@ -266,6 +303,16 @@ func (s *Store) compactLocked() (retErr error) {
266		if err != nil {	303		if err != nil {
267		return err	304		return err
268		}	305		}
			306	+	// Every replacement is written in the current format, which is also how a
			307	+	// legacy file is upgraded.
			308	+	header := fileHeader()
			309	+	if n, err := s.hooks.Write(f, header); err != nil \|\| n != len(header) {
			310	+	_ = s.hooks.Close(f)
			311	+	if err == nil {
			312	+	err = io.ErrShortWrite
			313	+	}
			314	+	return err
			315	+	}
269		keys := make([]string, 0, len(s.data))	316		keys := make([]string, 0, len(s.data))
270		for k := range s.data {	317		for k := range s.data {
271		keys = append(keys, k)	318		keys = append(keys, k)

@@ -356,7 +403,7 @@ func (s *Store) compactLocked() (retErr error) {
356		return err	403		return err
357		}	404		}
358		result, err := replayFile(current)	405		result, err := replayFile(current)
359	-	if err != nil \|\| result.TornTail {	406	+	if err != nil \|\| result.TornTail \|\| result.Format != formatCurrent {
360		_ = s.hooks.Close(current)	407		_ = s.hooks.Close(current)
361		if err == nil {	408		if err == nil {
362		err = ErrCorruptWAL	409		err = ErrCorruptWAL

@@ -402,6 +449,10 @@ func (s *Store) poisonLocked() error {
402		return errors.Join(fileErr, releaseLock(lockFile))	449		return errors.Join(fileErr, releaseLock(lockFile))
403		}	450		}
404			451
			452	+	// replayFile validates a WAL prefix in whichever framing the file uses. A torn
			453	+	// tail is the only truncatable outcome; every other inconsistency is corruption
			454	+	// and fails closed, so a damaged record can never be silently dropped along
			455	+	// with everything written after it.
405		func replayFile(f *os.File) (ReplayResult, error) {	456		func replayFile(f *os.File) (ReplayResult, error) {
406		info, err := f.Stat()	457		info, err := f.Stat()
407		if err != nil {	458		if err != nil {

@@ -411,34 +462,34 @@ func replayFile(f *os.File) (ReplayResult, error) {
411		return ReplayResult{}, err	462		return ReplayResult{}, err
412		}	463		}
413		fileSize := info.Size()	464		fileSize := info.Size()
414	-	result := ReplayResult{Data: make(map[string][]byte), EntrySize: make(map[string]int64)}	465	+	format, err := readFileFormat(f, fileSize)
415	-	header := make([]byte, headerLen)	466	+	if err != nil {
			467	+	return ReplayResult{}, err
			468	+	}
			469	+	result := ReplayResult{Data: make(map[string][]byte), EntrySize: make(map[string]int64), Format: format}
			470	+	if format == formatCurrent {
			471	+	result.LastGood = fileHeaderLen
			472	+	}
416		for {	473		for {
417	-	n, err := io.ReadFull(f, header)	474	+	header, status, err := readRecordHeader(f, format, fileSize, result.LastGood)
418	-	if errors.Is(err, io.EOF) {
419	-	return result, nil
420	-	}
421	-	if errors.Is(err, io.ErrUnexpectedEOF) {
422	-	_ = n
423	-	result.TornTail = true
424	-	return result, nil
425	-	}
426		if err != nil {	475		if err != nil {
427		return ReplayResult{}, err	476		return ReplayResult{}, err
428		}	477		}
429	-	payloadLen := binary.BigEndian.Uint32(header[:4])	478	+	if status == headerEOF {
430	-	if int64(payloadLen) > fileSize-result.LastGood-headerLen {	479	+	return result, nil
			480	+	}
			481	+	if status == headerTorn {
431		result.TornTail = true	482		result.TornTail = true
432		return result, nil	483		return result, nil
433		}	484		}
434	-	payload := make([]byte, int(payloadLen))	485	+	payload := make([]byte, int(header.payloadLen))
435		if _, err := io.ReadFull(f, payload); errors.Is(err, io.EOF) \|\| errors.Is(err, io.ErrUnexpectedEOF) {	486		if _, err := io.ReadFull(f, payload); errors.Is(err, io.EOF) \|\| errors.Is(err, io.ErrUnexpectedEOF) {
436		result.TornTail = true	487		result.TornTail = true
437		return result, nil	488		return result, nil
438		} else if err != nil {	489		} else if err != nil {
439		return ReplayResult{}, err	490		return ReplayResult{}, err
440		}	491		}
441	-	if crc32.ChecksumIEEE(payload) != binary.BigEndian.Uint32(header[4:8]) {	492	+	if crc32.ChecksumIEEE(payload) != header.payloadCRC {
442		return ReplayResult{}, fmt.Errorf("%w: CRC mismatch at offset %d", ErrCorruptWAL, result.LastGood)	493		return ReplayResult{}, fmt.Errorf("%w: CRC mismatch at offset %d", ErrCorruptWAL, result.LastGood)
443		}	494		}
444		var records []record	495		var records []record

@@ -452,13 +503,73 @@ func replayFile(f *os.File) (ReplayResult, error) {
452		if err != nil {	503		if err != nil {
453		return ReplayResult{}, fmt.Errorf("%w: invalid record at offset %d: %v", ErrCorruptWAL, result.LastGood, err)	504		return ReplayResult{}, fmt.Errorf("%w: invalid record at offset %d: %v", ErrCorruptWAL, result.LastGood, err)
454		}	505		}
455	-	size := int64(headerLen + len(payload))	506	+	size := format.headerLen() + int64(len(payload))
456		applyRecords(result.Data, result.EntrySize, &result.Live, records, size)	507		applyRecords(result.Data, result.EntrySize, &result.Live, records, size)
457		result.Total += size	508		result.Total += size
458		result.LastGood += size	509		result.LastGood += size
459		}	510		}
460		}	511		}
461			512
			513	+	// readFileFormat consumes the file header when one is present and leaves the
			514	+	// reader positioned at the first record either way.
			515	+	func readFileFormat(f io.ReadSeeker, fileSize int64) (walFormat, error) {
			516	+	if fileSize < fileHeaderLen {
			517	+	return formatLegacy, nil
			518	+	}
			519	+	header := make([]byte, fileHeaderLen)
			520	+	if _, err := io.ReadFull(f, header); err != nil {
			521	+	return formatLegacy, err
			522	+	}
			523	+	if !bytes.Equal(header[:fileMagicLen], fileMagic[:]) {
			524	+	if _, err := f.Seek(0, io.SeekStart); err != nil {
			525	+	return formatLegacy, err
			526	+	}
			527	+	return formatLegacy, nil
			528	+	}
			529	+	if version := binary.BigEndian.Uint16(header[fileMagicLen:]); version != formatVersion {
			530	+	return formatLegacy, fmt.Errorf("%w: unsupported format version %d", ErrCorruptWAL, version)
			531	+	}
			532	+	return formatCurrent, nil
			533	+	}
			534	+
			535	+	type recordHeader struct {
			536	+	payloadLen uint32
			537	+	payloadCRC uint32
			538	+	}
			539	+
			540	+	// readRecordHeader implements the replay decision table. In the current format
			541	+	// the header checksum is verified before the length is trusted for anything:
			542	+	// an unverifiable header is corruption, and only a header that checks out may
			543	+	// declare the record short (a torn tail).
			544	+	func readRecordHeader(f io.Reader, format walFormat, fileSize, offset int64) (recordHeader, headerStatus, error) {
			545	+	buf := make([]byte, format.headerLen())
			546	+	if _, err := io.ReadFull(f, buf); errors.Is(err, io.EOF) {
			547	+	return recordHeader{}, headerEOF, nil
			548	+	} else if errors.Is(err, io.ErrUnexpectedEOF) {
			549	+	return recordHeader{}, headerTorn, nil
			550	+	} else if err != nil {
			551	+	return recordHeader{}, headerOK, err
			552	+	}
			553	+	header := recordHeader{
			554	+	payloadLen: binary.BigEndian.Uint32(buf[0:4]),
			555	+	payloadCRC: binary.BigEndian.Uint32(buf[4:8]),
			556	+	}
			557	+	if format == formatCurrent {
			558	+	if crc32.ChecksumIEEE(buf[0:8]) != binary.BigEndian.Uint32(buf[8:recordHeaderLen]) {
			559	+	return recordHeader{}, headerOK, fmt.Errorf("%w: record header CRC mismatch at offset %d", ErrCorruptWAL, offset)
			560	+	}
			561	+	// The length is now trustworthy, so an implausible one is corruption
			562	+	// rather than a short tail. Checked before any allocation.
			563	+	if header.payloadLen > maxPayloadLen {
			564	+	return recordHeader{}, headerOK, fmt.Errorf("%w: record length %d exceeds maximum at offset %d", ErrCorruptWAL, header.payloadLen, offset)
			565	+	}
			566	+	}
			567	+	if int64(header.payloadLen) > fileSize-offset-format.headerLen() {
			568	+	return recordHeader{}, headerTorn, nil
			569	+	}
			570	+	return header, headerOK, nil
			571	+	}
			572	+
462		func applyRecords(data map[string][]byte, sizes map[string]int64, live *int64, records []record, size int64) {	573		func applyRecords(data map[string][]byte, sizes map[string]int64, live *int64, records []record, size int64) {
463		setCount := int64(0)	574		setCount := int64(0)
464		for _, rec := range records {	575		for _, rec := range records {

@@ -495,7 +606,7 @@ func recoverCompaction(path string, hooks fsHooks) error {
495		}	606		}
496		dir := filepath.Dir(path)	607		dir := filepath.Dir(path)
497		if currentExists && currentErr == nil {	608		if currentExists && currentErr == nil {
498	-	if nextExists && !prevExists {	609	+	if nextExists {
499		if err := hooks.Remove(path + ".next"); err != nil {	610		if err := hooks.Remove(path + ".next"); err != nil {
500		return err	611		return err
501		}	612		}

@@ -580,7 +691,9 @@ func validatePath(path string) error {
580		if err != nil {	691		if err != nil {
581		return err	692		return err
582		}	693		}
583	-	if !exists \|\| result.TornTail {	694	+	// A replacement this package just wrote must be complete and in the current
			695	+	// format; anything else means the write did not land as intended.
			696	+	if !exists \|\| result.TornTail \|\| result.Format != formatCurrent {
584		return ErrCorruptWAL	697		return ErrCorruptWAL
585		}	698		}
586		return nil	699		return nil

M pkg/kv/kv_test.go +376 -44


@@ -16,6 +16,41 @@ func privatePath(t *testing.T) string {
16	16		return filepath.Join(t.TempDir(), "vev", "store.kv")
17	17		}
18	18
	19	+	// currentFile frames records the way the store writes them today: a file
	20	+	// header followed by self-checksummed record headers.
	21	+	func currentFile(records ...[]byte) []byte {
	22	+	buf := fileHeader()
	23	+	for _, record := range records {
	24	+	buf = append(buf, record...)
	25	+	}
	26	+	return buf
	27	+	}
	28	+
	29	+	// legacyFile reproduces the framing written by the released version: no file
	30	+	// header, and a record header of length plus payload CRC only. Compatibility
	31	+	// with these files is a hard requirement, so the fixture stays in the tests.
	32	+	func legacyFile(records ...[]byte) []byte {
	33	+	var buf []byte
	34	+	for _, record := range records {
	35	+	payload := record[recordHeaderLen:]
	36	+	header := make([]byte, legacyHeaderLen)
	37	+	binary.BigEndian.PutUint32(header[0:4], uint32(len(payload)))
	38	+	binary.BigEndian.PutUint32(header[4:8], crc32.ChecksumIEEE(payload))
	39	+	buf = append(append(buf, header...), payload...)
	40	+	}
	41	+	return buf
	42	+	}
	43	+
	44	+	func writeFile(t *testing.T, path string, data []byte) {
	45	+	t.Helper()
	46	+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
	47	+	t.Fatal(err)
	48	+	}
	49	+	if err := os.WriteFile(path, data, 0o600); err != nil {
	50	+	t.Fatal(err)
	51	+	}
	52	+	}
	53	+
19	54		func closeStore(t testing.T, s Store) {
20	55		t.Helper()
21	56		if err := s.Close(); err != nil {

@@ -128,7 +163,7 @@ func TestOpenTruncatesMidRecordTail(t *testing.T) {
128	163		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
129	164		t.Fatal(err)
130	165		}
131		-	if err := os.WriteFile(path, append(first, second[:len(second)/2]...), 0o600); err != nil {
	166	+	if err := os.WriteFile(path, currentFile(first, second[:len(second)/2]), 0o600); err != nil {
132	167		t.Fatal(err)
133	168		}
134	169

@@ -147,8 +182,8 @@ func TestOpenTruncatesMidRecordTail(t *testing.T) {
147	182		if err != nil {
148	183		t.Fatal(err)
149	184		}
150		-	if info.Size() != int64(len(first)) {
151		-	t.Fatalf("tail not truncated: size=%d want=%d", info.Size(), len(first))
	185	+	if want := int64(fileHeaderLen + len(first)); info.Size() != want {
	186	+	t.Fatalf("tail not truncated: size=%d want=%d", info.Size(), want)
152	187		}
153	188		}
154	189

@@ -156,8 +191,8 @@ func TestCRCcorruptionOfLastRecordFailsClosed(t *testing.T) {
156	191		path := privatePath(t)
157	192		first, _ := encodeRecord(opSet, []byte("k"), []byte("old"))
158	193		second, _ := encodeRecord(opSet, []byte("k"), []byte("new"))
159		-	buf := append(append([]byte{}, first...), second...)
160		-	buf[len(first)+headerLen+payloadPrefixLen+1] ^= 0xff
	194	+	buf := currentFile(first, second)
	195	+	buf[fileHeaderLen+len(first)+recordHeaderLen+payloadPrefixLen+1] ^= 0xff
161	196		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
162	197		t.Fatal(err)
163	198		}

@@ -251,14 +286,14 @@ func TestReplayStrict(t *testing.T) {
251	286		torn bool
252	287		corrupt bool
253	288		}{
254		-	{name: "torn-final-header", wal: append(append([]byte{}, valid...), batch[:headerLen-1]...), torn: true},
255		-	{name: "torn-final-payload", wal: append(append([]byte{}, valid...), batch[:len(batch)-1]...), torn: true},
256		-	{name: "final-crc", wal: corruptRecord(append(append([]byte{}, valid...), batch...), len(valid)+4), corrupt: true},
257		-	{name: "middle-crc", wal: corruptRecord(append(append([]byte{}, valid...), batch...), 4), corrupt: true},
258		-	{name: "invalid-op", wal: rawRecord([]byte{0xff, 0, 0}), corrupt: true},
259		-	{name: "impossible-batch-count", wal: rawRecord([]byte{opBatch, 0xff, 0xff, 0xff, 0xff}), corrupt: true},
260		-	{name: "duplicate-key-in-batch", wal: rawRecord(rawBatchPayload([]BatchChange{{Key: []byte("x")}, {Key: []byte("x"), Delete: true}})), corrupt: true},
261		-	{name: "trailing-batch-garbage", wal: rawRecord(append(rawBatchPayload([]BatchChange{{Key: []byte("x")}}), 0)), corrupt: true},
	289	+	{name: "torn-final-header", wal: currentFile(valid, batch[:recordHeaderLen-1]), torn: true},
	290	+	{name: "torn-final-payload", wal: currentFile(valid, batch[:len(batch)-1]), torn: true},
	291	+	{name: "final-crc", wal: corruptRecord(currentFile(valid, batch), fileHeaderLen+len(valid)+recordHeaderLen), corrupt: true},
	292	+	{name: "middle-crc", wal: corruptRecord(currentFile(valid, batch), fileHeaderLen+recordHeaderLen), corrupt: true},
	293	+	{name: "invalid-op", wal: currentFile(rawRecord([]byte{0xff, 0, 0})), corrupt: true},
	294	+	{name: "impossible-batch-count", wal: currentFile(rawRecord([]byte{opBatch, 0xff, 0xff, 0xff, 0xff})), corrupt: true},
	295	+	{name: "duplicate-key-in-batch", wal: currentFile(rawRecord(rawBatchPayload([]BatchChange{{Key: []byte("x")}, {Key: []byte("x"), Delete: true}}))), corrupt: true},
	296	+	{name: "trailing-batch-garbage", wal: currentFile(rawRecord(append(rawBatchPayload([]BatchChange{{Key: []byte("x")}}), 0))), corrupt: true},
262	297		}
263	298		for _, tt := range tests {
264	299		t.Run(tt.name, func(t *testing.T) {

@@ -287,13 +322,62 @@ func TestReplayStrict(t *testing.T) {
287	322		if string(result.Data["old"]) != "value" \|\| result.Data["new"] != nil {
288	323		t.Fatalf("torn batch was partially applied: %#v", result.Data)
289	324		}
290		-	if !result.TornTail \|\| result.LastGood != int64(len(valid)) {
	325	+	if !result.TornTail \|\| result.LastGood != int64(fileHeaderLen+len(valid)) {
291	326		t.Fatalf("result = %+v", result)
292	327		}
293	328		})
294	329		}
295	330		}
296	331
	332	+	// TestRecoverCompactionCleansStaleNextRegardlessOfPrev covers the branch
	333	+	// where current is valid: any stale .next must be removed even when a .prev
	334	+	// also exists (previously the "nextExists && !prevExists" guard left a .next
	335	+	// behind whenever .prev was present too, so spec §3's complete-cleanup
	336	+	// requirement was not met after certain crash-then-retry sequences).
	337	+	func TestRecoverCompactionCleansStaleNextRegardlessOfPrev(t *testing.T) {
	338	+	oldRecord, _ := encodeRecord(opSet, []byte("version"), []byte("old"))
	339	+	newRecord, _ := encodeRecord(opSet, []byte("version"), []byte("new"))
	340	+	oldFile, newFile := currentFile(oldRecord), currentFile(newRecord)
	341	+	tests := []struct {
	342	+	name string
	343	+	current, next, prev []byte
	344	+	}{
	345	+	{name: "current-next-only", current: oldFile, next: newFile},
	346	+	{name: "current-next-and-prev", current: oldFile, next: newFile, prev: oldFile},
	347	+	}
	348	+	for _, tt := range tests {
	349	+	t.Run(tt.name, func(t *testing.T) {
	350	+	path := privatePath(t)
	351	+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
	352	+	t.Fatal(err)
	353	+	}
	354	+	for suffix, data := range map[string][]byte{"": tt.current, ".next": tt.next, ".prev": tt.prev} {
	355	+	if data != nil {
	356	+	if err := os.WriteFile(path+suffix, data, 0o600); err != nil {
	357	+	t.Fatal(err)
	358	+	}
	359	+	}
	360	+	}
	361	+	if err := recoverCompaction(path, defaultFSHooks()); err != nil {
	362	+	t.Fatal(err)
	363	+	}
	364	+	if _, err := os.Stat(path + ".next"); !errors.Is(err, os.ErrNotExist) {
	365	+	t.Fatalf(".next survived recovery: err = %v", err)
	366	+	}
	367	+	if _, err := os.Stat(path + ".prev"); !errors.Is(err, os.ErrNotExist) {
	368	+	t.Fatalf(".prev survived recovery: err = %v", err)
	369	+	}
	370	+	got, err := Replay(path)
	371	+	if err != nil {
	372	+	t.Fatal(err)
	373	+	}
	374	+	if string(got["version"]) != "old" {
	375	+	t.Fatalf("version = %q, want %q (current must remain authoritative)", got["version"], "old")
	376	+	}
	377	+	})
	378	+	}
	379	+	}
	380	+
297	381		func TestDecodeBatchRejectsImpossibleCountBeforeAllocating(t *testing.T) {
298	382		payload := []byte{opBatch, 0xff, 0xff, 0xff, 0xff}
299	383		allocs := testing.AllocsPerRun(10, func() {

@@ -340,19 +424,20 @@ func TestBatchAtomicity(t *testing.T) {
340	424		func TestCompactionRecoveryMatrix(t *testing.T) {
341	425		oldRecord, _ := encodeRecord(opSet, []byte("version"), []byte("old"))
342	426		newRecord, _ := encodeRecord(opSet, []byte("version"), []byte("new"))
	427	+	oldFile, newFile := currentFile(oldRecord), currentFile(newRecord)
343	428		tests := []struct {
344	429		name string
345	430		current, next, prev []byte
346	431		want string
347	432		}{
348		-	{name: "current-alone", current: oldRecord, want: "old"},
349		-	{name: "current-next", current: oldRecord, next: newRecord, want: "old"},
350		-	{name: "prev-next", prev: oldRecord, next: newRecord, want: "new"},
351		-	{name: "prev-invalid-next", prev: oldRecord, next: rawRecord([]byte{0xff, 0, 0}), want: "old"},
352		-	{name: "prev-torn-next", prev: oldRecord, next: newRecord[:len(newRecord)-1], want: "old"},
353		-	{name: "current-prev", current: newRecord, prev: oldRecord, want: "new"},
354		-	{name: "invalid-current-prev", current: rawRecord([]byte{0xff, 0, 0}), prev: oldRecord, want: "old"},
355		-	{name: "prev-alone", prev: oldRecord, want: "old"},
	433	+	{name: "current-alone", current: oldFile, want: "old"},
	434	+	{name: "current-next", current: oldFile, next: newFile, want: "old"},
	435	+	{name: "prev-next", prev: oldFile, next: newFile, want: "new"},
	436	+	{name: "prev-invalid-next", prev: oldFile, next: currentFile(rawRecord([]byte{0xff, 0, 0})), want: "old"},
	437	+	{name: "prev-torn-next", prev: oldFile, next: newFile[:len(newFile)-1], want: "old"},
	438	+	{name: "current-prev", current: newFile, prev: oldFile, want: "new"},
	439	+	{name: "invalid-current-prev", current: currentFile(rawRecord([]byte{0xff, 0, 0})), prev: oldFile, want: "old"},
	440	+	{name: "prev-alone", prev: oldFile, want: "old"},
356	441		}
357	442		for _, tt := range tests {
358	443		t.Run(tt.name, func(t *testing.T) {

@@ -392,7 +477,7 @@ func TestCompactionRecoveryMatrix(t *testing.T) {
392	477		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
393	478		t.Fatal(err)
394	479		}
395		-	if err := os.WriteFile(path, oldRecord, 0o600); err != nil {
	480	+	if err := os.WriteFile(path, oldFile, 0o600); err != nil {
396	481		t.Fatal(err)
397	482		}
398	483		s, err := Open(path)

@@ -488,13 +573,14 @@ func TestCompactionFilesystemOperationFailuresPoisonAndRecover(t *testing.T) {
488	573		if err != nil {
489	574		t.Fatal(err)
490	575		}
	576	+	oldFile := currentFile(oldRecord)
491	577		for _, tt := range tests {
492	578		t.Run(tt.name, func(t *testing.T) {
493	579		path := privatePath(t)
494	580		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
495	581		t.Fatal(err)
496	582		}
497		-	if err := os.WriteFile(path, oldRecord, 0o600); err != nil {
	583	+	if err := os.WriteFile(path, oldFile, 0o600); err != nil {
498	584		t.Fatal(err)
499	585		}
500	586		s, err := Open(path)

@@ -599,13 +685,7 @@ func requireCandidateBytes(t *testing.T, path string, want map[string][]byte) {
599	685		}
600	686
601	687		func corruptRecord(buf []byte, offset int) []byte { buf[offset] ^= 0xff; return buf }
602		-	func rawRecord(payload []byte) []byte {
603		-	buf := make([]byte, headerLen+len(payload))
604		-	binary.BigEndian.PutUint32(buf[:4], uint32(len(payload)))
605		-	binary.BigEndian.PutUint32(buf[4:8], crc32.ChecksumIEEE(payload))
606		-	copy(buf[8:], payload)
607		-	return buf
608		-	}
	688	+	func rawRecord(payload []byte) []byte { return framePayload(payload) }
609	689		func rawBatchPayload(changes []BatchChange) []byte {
610	690		payload := []byte{opBatch, 0, 0, 0, byte(len(changes))}
611	691		for _, change := range changes {

@@ -621,23 +701,275 @@ func rawBatchPayload(changes []BatchChange) []byte {
621	701		return payload
622	702		}
623	703
624		-	func TestReplayStopsAtBadPayloadLengthWithoutAllocatingHugeBuffer(t *testing.T) {
625		-	path := privatePath(t)
	704	+	// An oversized length must never drive an allocation. In a legacy file the
	705	+	// length is unprotected, so it stays a torn tail; in the current format the
	706	+	// header checksum proves the length was written that way, which is corruption.
	707	+	func TestReplayRejectsOversizedLengthWithoutAllocating(t *testing.T) {
626	708		first, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
627		-	header := make([]byte, headerLen)
628		-	binary.BigEndian.PutUint32(header[0:4], 1<<31)
629		-	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
	709	+	legacyHeader := make([]byte, legacyHeaderLen)
	710	+	binary.BigEndian.PutUint32(legacyHeader[0:4], 1<<31)
	711	+
	712	+	t.Run("legacy length is a torn tail", func(t *testing.T) {
	713	+	path := privatePath(t)
	714	+	writeFile(t, path, append(legacyFile(first), legacyHeader...))
	715	+	s, err := Open(path)
	716	+	if err != nil {
	717	+	t.Fatal(err)
	718	+	}
	719	+	defer closeStore(t, s)
	720	+	if got, ok := s.Get([]byte("a")); !ok \|\| string(got) != "one" {
	721	+	t.Fatalf("good record missing before bad length: %q %v", got, ok)
	722	+	}
	723	+	})
	724	+
	725	+	t.Run("current oversized length fails closed", func(t *testing.T) {
	726	+	path := privatePath(t)
	727	+	wal := currentFile(first, oversizedRecordHeader(maxPayloadLen+1))
	728	+	writeFile(t, path, wal)
	729	+	if _, err := Open(path); !errors.Is(err, ErrCorruptWAL) {
	730	+	t.Fatalf("Open error = %v, want ErrCorruptWAL", err)
	731	+	}
	732	+	requireFileBytes(t, path, wal)
	733	+	})
	734	+	}
	735	+
	736	+	// oversizedRecordHeader is a record header whose own checksum is valid, so the
	737	+	// length is provably what the writer put there rather than a bit flip.
	738	+	func oversizedRecordHeader(payloadLen uint32) []byte {
	739	+	header := make([]byte, recordHeaderLen)
	740	+	binary.BigEndian.PutUint32(header[0:4], payloadLen)
	741	+	binary.BigEndian.PutUint32(header[8:recordHeaderLen], crc32.ChecksumIEEE(header[0:8]))
	742	+	return header
	743	+	}
	744	+
	745	+	func requireFileBytes(t *testing.T, path string, want []byte) {
	746	+	t.Helper()
	747	+	got, err := os.ReadFile(path)
	748	+	if err != nil {
630	749		t.Fatal(err)
631	750		}
632		-	if err := os.WriteFile(path, append(first, header...), 0o600); err != nil {
633		-	t.Fatal(err)
	751	+	if !bytes.Equal(got, want) {
	752	+	t.Fatalf("file was mutated: got %d bytes, want %d", len(got), len(want))
634	753		}
635		-	s, err := Open(path)
636		-	if err != nil {
637		-	t.Fatal(err)
	754	+	}
	755	+
	756	+	// The regression this format exists for: a bit flip in the length field of an
	757	+	// early record used to look exactly like a torn tail, so Open truncated every
	758	+	// record written after it and reopened a smaller, structurally valid file.
	759	+	func TestCorruptLengthOfEarlyRecordFailsClosedWithoutTruncating(t *testing.T) {
	760	+	first, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
	761	+	second, _ := encodeRecord(opSet, []byte("b"), []byte("two"))
	762	+	third, _ := encodeRecord(opSet, []byte("c"), []byte("three"))
	763	+
	764	+	tests := []struct {
	765	+	name string
	766	+	offset int
	767	+	}{
	768	+	{name: "first-record-length", offset: fileHeaderLen},
	769	+	{name: "second-record-length", offset: fileHeaderLen + len(first) + 3},
638	770		}
639		-	defer closeStore(t, s)
640		-	if got, ok := s.Get([]byte("a")); !ok \|\| string(got) != "one" {
641		-	t.Fatalf("good record missing before bad length: %q %v", got, ok)
	771	+	for _, tt := range tests {
	772	+	t.Run(tt.name, func(t *testing.T) {
	773	+	path := privatePath(t)
	774	+	wal := currentFile(first, second, third)
	775	+	// Flip a high bit so the length claims far more than the file holds:
	776	+	// the exact shape that used to be misread as an incomplete tail.
	777	+	wal[tt.offset] ^= 0x40
	778	+	writeFile(t, path, wal)
	779	+
	780	+	if _, err := Open(path); !errors.Is(err, ErrCorruptWAL) {
	781	+	t.Fatalf("Open error = %v, want ErrCorruptWAL", err)
	782	+	}
	783	+	requireFileBytes(t, path, wal)
	784	+	if _, err := Replay(path); !errors.Is(err, ErrCorruptWAL) {
	785	+	t.Fatalf("Replay error = %v, want ErrCorruptWAL", err)
	786	+	}
	787	+	})
	788	+	}
	789	+	}
	790	+
	791	+	// Every row of the replay decision table for the current format. A torn tail is
	792	+	// the only truncatable outcome.
	793	+	func TestCurrentFormatReplayDecisionTable(t *testing.T) {
	794	+	first, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
	795	+	second, _ := encodeRecord(opSet, []byte("b"), []byte("two"))
	796	+	invalidOp := rawRecord([]byte{0xff, 0, 0})
	797	+
	798	+	tests := []struct {
	799	+	name string
	800	+	wal []byte
	801	+	torn bool
	802	+	corrupt bool
	803	+	}{
	804	+	{name: "partial-header", wal: currentFile(first, second[:recordHeaderLen-1]), torn: true},
	805	+	{name: "header-checksum-mismatch", wal: corruptRecord(currentFile(first, second), fileHeaderLen+len(first)+1), corrupt: true},
	806	+	{name: "short-payload", wal: currentFile(first, second[:len(second)-1]), torn: true},
	807	+	{name: "payload-checksum-mismatch", wal: corruptRecord(currentFile(first, second), fileHeaderLen+len(first)+recordHeaderLen), corrupt: true},
	808	+	{name: "undecodable-payload", wal: currentFile(first, invalidOp), corrupt: true},
	809	+	{name: "oversized-length", wal: currentFile(first, oversizedRecordHeader(maxPayloadLen+1)), corrupt: true},
	810	+	{name: "complete-records", wal: currentFile(first, second)},
	811	+	}
	812	+	for _, tt := range tests {
	813	+	t.Run(tt.name, func(t *testing.T) {
	814	+	path := privatePath(t)
	815	+	writeFile(t, path, tt.wal)
	816	+	f, err := os.Open(path)
	817	+	if err != nil {
	818	+	t.Fatal(err)
	819	+	}
	820	+	result, err := replayFile(f)
	821	+	_ = f.Close()
	822	+
	823	+	if tt.corrupt {
	824	+	if !errors.Is(err, ErrCorruptWAL) {
	825	+	t.Fatalf("replay error = %v, want ErrCorruptWAL", err)
	826	+	}
	827	+	return
	828	+	}
	829	+	if err != nil {
	830	+	t.Fatal(err)
	831	+	}
	832	+	if result.Format != formatCurrent {
	833	+	t.Fatalf("format = %d, want current", result.Format)
	834	+	}
	835	+	if result.TornTail != tt.torn {
	836	+	t.Fatalf("TornTail = %v, want %v", result.TornTail, tt.torn)
	837	+	}
	838	+	if string(result.Data["a"]) != "one" {
	839	+	t.Fatalf("prefix lost: %#v", result.Data)
	840	+	}
	841	+	wantLastGood := int64(fileHeaderLen + len(first))
	842	+	if !tt.torn {
	843	+	wantLastGood += int64(len(second))
	844	+	}
	845	+	if result.LastGood != wantLastGood {
	846	+	t.Fatalf("LastGood = %d, want %d", result.LastGood, wantLastGood)
	847	+	}
	848	+	})
	849	+	}
	850	+	}
	851	+
	852	+	// A file written by the released version must keep working: it opens with the
	853	+	// original semantics, is upgraded in place, and reopens identically.
	854	+	func TestLegacyFileOpensAndUpgradesInPlace(t *testing.T) {
	855	+	setA, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
	856	+	setB, _ := encodeRecord(opSet, []byte("b"), []byte("two"))
	857	+	replaceA, _ := encodeRecord(opSet, []byte("a"), []byte("uno"))
	858	+	delB, _ := encodeRecord(opDel, []byte("b"), nil)
	859	+	torn, _ := encodeRecord(opSet, []byte("c"), []byte("three"))
	860	+
	861	+	tests := []struct {
	862	+	name string
	863	+	wal []byte
	864	+	want map[string]string
	865	+	}{
	866	+	{name: "complete", wal: legacyFile(setA, setB, replaceA), want: map[string]string{"a": "uno", "b": "two"}},
	867	+	{name: "tombstone", wal: legacyFile(setA, setB, delB), want: map[string]string{"a": "one"}},
	868	+	{name: "torn-tail", wal: append(legacyFile(setA, setB), legacyFile(torn)[:legacyHeaderLen+2]...), want: map[string]string{"a": "one", "b": "two"}},
	869	+	{name: "empty", wal: nil, want: map[string]string{}},
	870	+	}
	871	+	for _, tt := range tests {
	872	+	t.Run(tt.name, func(t *testing.T) {
	873	+	path := privatePath(t)
	874	+	writeFile(t, path, tt.wal)
	875	+
	876	+	s, err := Open(path)
	877	+	if err != nil {
	878	+	t.Fatalf("legacy file failed to open: %v", err)
	879	+	}
	880	+	requireContents(t, s, tt.want)
	881	+	if err := s.Set([]byte("added"), []byte("after-upgrade")); err != nil {
	882	+	t.Fatal(err)
	883	+	}
	884	+	closeStore(t, s)
	885	+
	886	+	upgraded, err := os.ReadFile(path)
	887	+	if err != nil {
	888	+	t.Fatal(err)
	889	+	}
	890	+	if !bytes.HasPrefix(upgraded, fileMagic[:]) {
	891	+	t.Fatal("legacy file was not upgraded in place")
	892	+	}
	893	+	reopened, err := Open(path)
	894	+	if err != nil {
	895	+	t.Fatal(err)
	896	+	}
	897	+	defer closeStore(t, reopened)
	898	+	want := map[string]string{"added": "after-upgrade"}
	899	+	for k, v := range tt.want {
	900	+	want[k] = v
	901	+	}
	902	+	requireContents(t, reopened, want)
	903	+	replayed, err := Replay(path)
	904	+	if err != nil {
	905	+	t.Fatal(err)
	906	+	}
	907	+	if len(replayed) != len(want) {
	908	+	t.Fatalf("Replay = %#v, want %#v", replayed, want)
	909	+	}
	910	+	})
	911	+	}
	912	+	}
	913	+
	914	+	func requireContents(t testing.T, s Store, want map[string]string) {
	915	+	t.Helper()
	916	+	for key, value := range want {
	917	+	got, ok := s.Get([]byte(key))
	918	+	if !ok \|\| string(got) != value {
	919	+	t.Fatalf("Get(%q) = %q, %v; want %q", key, got, ok, value)
	920	+	}
	921	+	}
	922	+	count := 0
	923	+	s.Range(func([]byte, []byte) bool { count++; return true })
	924	+	if count != len(want) {
	925	+	t.Fatalf("store holds %d keys, want %d", count, len(want))
	926	+	}
	927	+	}
	928	+
	929	+	// The upgrade runs through the compaction protocol, so a crash at any of its
	930	+	// boundaries must leave either the whole legacy file or the whole upgraded one.
	931	+	func TestLegacyUpgradeCrashSafety(t *testing.T) {
	932	+	setA, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
	933	+	setB, _ := encodeRecord(opSet, []byte("b"), []byte("two"))
	934	+	legacy := legacyFile(setA, setB)
	935	+	want := map[string]string{"a": "one", "b": "two"}
	936	+
	937	+	faultPoints := []string{
	938	+	"before-write-next", "after-write-next", "before-sync-next", "after-sync-next",
	939	+	"after-sync-dir-next", "after-remove-stale-prev", "after-rename-current-prev",
	940	+	"after-sync-dir-prev", "after-rename-next-current", "after-sync-dir-current",
	941	+	"before-reopen-current", "after-reopen-current", "after-remove-prev", "after-final-sync-dir",
	942	+	}
	943	+	for _, point := range faultPoints {
	944	+	t.Run("fault-"+point, func(t *testing.T) {
	945	+	path := privatePath(t)
	946	+	writeFile(t, path, legacy)
	947	+	hooks := defaultFSHooks()
	948	+	hooks.Fault = func(got string) error {
	949	+	if got == point {
	950	+	return errors.New("injected " + point)
	951	+	}
	952	+	return nil
	953	+	}
	954	+	if _, err := openWithHooks(path, hooks); err == nil {
	955	+	t.Fatalf("fault %q did not interrupt the upgrade", point)
	956	+	}
	957	+
	958	+	// Restart: recovery selects a complete file, and the data is intact
	959	+	// whichever format that file happens to be in.
	960	+	s, err := Open(path)
	961	+	if err != nil {
	962	+	t.Fatalf("restart after fault %q: %v", point, err)
	963	+	}
	964	+	requireContents(t, s, want)
	965	+	closeStore(t, s)
	966	+	upgraded, err := os.ReadFile(path)
	967	+	if err != nil {
	968	+	t.Fatal(err)
	969	+	}
	970	+	if !bytes.HasPrefix(upgraded, fileMagic[:]) {
	971	+	t.Fatal("restart did not complete the upgrade")
	972	+	}
	973	+	})
642	974		}
643	975		}


@@ -16,6 +16,41 @@ func privatePath(t *testing.T) string {
16		return filepath.Join(t.TempDir(), "vev", "store.kv")	16		return filepath.Join(t.TempDir(), "vev", "store.kv")
17		}	17		}
18			18
			19	+	// currentFile frames records the way the store writes them today: a file
			20	+	// header followed by self-checksummed record headers.
			21	+	func currentFile(records ...[]byte) []byte {
			22	+	buf := fileHeader()
			23	+	for _, record := range records {
			24	+	buf = append(buf, record...)
			25	+	}
			26	+	return buf
			27	+	}
			28	+
			29	+	// legacyFile reproduces the framing written by the released version: no file
			30	+	// header, and a record header of length plus payload CRC only. Compatibility
			31	+	// with these files is a hard requirement, so the fixture stays in the tests.
			32	+	func legacyFile(records ...[]byte) []byte {
			33	+	var buf []byte
			34	+	for _, record := range records {
			35	+	payload := record[recordHeaderLen:]
			36	+	header := make([]byte, legacyHeaderLen)
			37	+	binary.BigEndian.PutUint32(header[0:4], uint32(len(payload)))
			38	+	binary.BigEndian.PutUint32(header[4:8], crc32.ChecksumIEEE(payload))
			39	+	buf = append(append(buf, header...), payload...)
			40	+	}
			41	+	return buf
			42	+	}
			43	+
			44	+	func writeFile(t *testing.T, path string, data []byte) {
			45	+	t.Helper()
			46	+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
			47	+	t.Fatal(err)
			48	+	}
			49	+	if err := os.WriteFile(path, data, 0o600); err != nil {
			50	+	t.Fatal(err)
			51	+	}
			52	+	}
			53	+
19		func closeStore(t testing.T, s Store) {	54		func closeStore(t testing.T, s Store) {
20		t.Helper()	55		t.Helper()
21		if err := s.Close(); err != nil {	56		if err := s.Close(); err != nil {

@@ -128,7 +163,7 @@ func TestOpenTruncatesMidRecordTail(t *testing.T) {
128		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {	163		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
129		t.Fatal(err)	164		t.Fatal(err)
130		}	165		}
131	-	if err := os.WriteFile(path, append(first, second[:len(second)/2]...), 0o600); err != nil {	166	+	if err := os.WriteFile(path, currentFile(first, second[:len(second)/2]), 0o600); err != nil {
132		t.Fatal(err)	167		t.Fatal(err)
133		}	168		}
134			169

@@ -147,8 +182,8 @@ func TestOpenTruncatesMidRecordTail(t *testing.T) {
147		if err != nil {	182		if err != nil {
148		t.Fatal(err)	183		t.Fatal(err)
149		}	184		}
150	-	if info.Size() != int64(len(first)) {	185	+	if want := int64(fileHeaderLen + len(first)); info.Size() != want {
151	-	t.Fatalf("tail not truncated: size=%d want=%d", info.Size(), len(first))	186	+	t.Fatalf("tail not truncated: size=%d want=%d", info.Size(), want)
152		}	187		}
153		}	188		}
154			189

@@ -156,8 +191,8 @@ func TestCRCcorruptionOfLastRecordFailsClosed(t *testing.T) {
156		path := privatePath(t)	191		path := privatePath(t)
157		first, _ := encodeRecord(opSet, []byte("k"), []byte("old"))	192		first, _ := encodeRecord(opSet, []byte("k"), []byte("old"))
158		second, _ := encodeRecord(opSet, []byte("k"), []byte("new"))	193		second, _ := encodeRecord(opSet, []byte("k"), []byte("new"))
159	-	buf := append(append([]byte{}, first...), second...)	194	+	buf := currentFile(first, second)
160	-	buf[len(first)+headerLen+payloadPrefixLen+1] ^= 0xff	195	+	buf[fileHeaderLen+len(first)+recordHeaderLen+payloadPrefixLen+1] ^= 0xff
161		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {	196		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
162		t.Fatal(err)	197		t.Fatal(err)
163		}	198		}

@@ -251,14 +286,14 @@ func TestReplayStrict(t *testing.T) {
251		torn bool	286		torn bool
252		corrupt bool	287		corrupt bool
253		}{	288		}{
254	-	{name: "torn-final-header", wal: append(append([]byte{}, valid...), batch[:headerLen-1]...), torn: true},	289	+	{name: "torn-final-header", wal: currentFile(valid, batch[:recordHeaderLen-1]), torn: true},
255	-	{name: "torn-final-payload", wal: append(append([]byte{}, valid...), batch[:len(batch)-1]...), torn: true},	290	+	{name: "torn-final-payload", wal: currentFile(valid, batch[:len(batch)-1]), torn: true},
256	-	{name: "final-crc", wal: corruptRecord(append(append([]byte{}, valid...), batch...), len(valid)+4), corrupt: true},	291	+	{name: "final-crc", wal: corruptRecord(currentFile(valid, batch), fileHeaderLen+len(valid)+recordHeaderLen), corrupt: true},
257	-	{name: "middle-crc", wal: corruptRecord(append(append([]byte{}, valid...), batch...), 4), corrupt: true},	292	+	{name: "middle-crc", wal: corruptRecord(currentFile(valid, batch), fileHeaderLen+recordHeaderLen), corrupt: true},
258	-	{name: "invalid-op", wal: rawRecord([]byte{0xff, 0, 0}), corrupt: true},	293	+	{name: "invalid-op", wal: currentFile(rawRecord([]byte{0xff, 0, 0})), corrupt: true},
259	-	{name: "impossible-batch-count", wal: rawRecord([]byte{opBatch, 0xff, 0xff, 0xff, 0xff}), corrupt: true},	294	+	{name: "impossible-batch-count", wal: currentFile(rawRecord([]byte{opBatch, 0xff, 0xff, 0xff, 0xff})), corrupt: true},
260	-	{name: "duplicate-key-in-batch", wal: rawRecord(rawBatchPayload([]BatchChange{{Key: []byte("x")}, {Key: []byte("x"), Delete: true}})), corrupt: true},	295	+	{name: "duplicate-key-in-batch", wal: currentFile(rawRecord(rawBatchPayload([]BatchChange{{Key: []byte("x")}, {Key: []byte("x"), Delete: true}}))), corrupt: true},
261	-	{name: "trailing-batch-garbage", wal: rawRecord(append(rawBatchPayload([]BatchChange{{Key: []byte("x")}}), 0)), corrupt: true},	296	+	{name: "trailing-batch-garbage", wal: currentFile(rawRecord(append(rawBatchPayload([]BatchChange{{Key: []byte("x")}}), 0))), corrupt: true},
262		}	297		}
263		for _, tt := range tests {	298		for _, tt := range tests {
264		t.Run(tt.name, func(t *testing.T) {	299		t.Run(tt.name, func(t *testing.T) {

@@ -287,13 +322,62 @@ func TestReplayStrict(t *testing.T) {
287		if string(result.Data["old"]) != "value" \|\| result.Data["new"] != nil {	322		if string(result.Data["old"]) != "value" \|\| result.Data["new"] != nil {
288		t.Fatalf("torn batch was partially applied: %#v", result.Data)	323		t.Fatalf("torn batch was partially applied: %#v", result.Data)
289		}	324		}
290	-	if !result.TornTail \|\| result.LastGood != int64(len(valid)) {	325	+	if !result.TornTail \|\| result.LastGood != int64(fileHeaderLen+len(valid)) {
291		t.Fatalf("result = %+v", result)	326		t.Fatalf("result = %+v", result)
292		}	327		}
293		})	328		})
294		}	329		}
295		}	330		}
296			331
			332	+	// TestRecoverCompactionCleansStaleNextRegardlessOfPrev covers the branch
			333	+	// where current is valid: any stale .next must be removed even when a .prev
			334	+	// also exists (previously the "nextExists && !prevExists" guard left a .next
			335	+	// behind whenever .prev was present too, so spec §3's complete-cleanup
			336	+	// requirement was not met after certain crash-then-retry sequences).
			337	+	func TestRecoverCompactionCleansStaleNextRegardlessOfPrev(t *testing.T) {
			338	+	oldRecord, _ := encodeRecord(opSet, []byte("version"), []byte("old"))
			339	+	newRecord, _ := encodeRecord(opSet, []byte("version"), []byte("new"))
			340	+	oldFile, newFile := currentFile(oldRecord), currentFile(newRecord)
			341	+	tests := []struct {
			342	+	name string
			343	+	current, next, prev []byte
			344	+	}{
			345	+	{name: "current-next-only", current: oldFile, next: newFile},
			346	+	{name: "current-next-and-prev", current: oldFile, next: newFile, prev: oldFile},
			347	+	}
			348	+	for _, tt := range tests {
			349	+	t.Run(tt.name, func(t *testing.T) {
			350	+	path := privatePath(t)
			351	+	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
			352	+	t.Fatal(err)
			353	+	}
			354	+	for suffix, data := range map[string][]byte{"": tt.current, ".next": tt.next, ".prev": tt.prev} {
			355	+	if data != nil {
			356	+	if err := os.WriteFile(path+suffix, data, 0o600); err != nil {
			357	+	t.Fatal(err)
			358	+	}
			359	+	}
			360	+	}
			361	+	if err := recoverCompaction(path, defaultFSHooks()); err != nil {
			362	+	t.Fatal(err)
			363	+	}
			364	+	if _, err := os.Stat(path + ".next"); !errors.Is(err, os.ErrNotExist) {
			365	+	t.Fatalf(".next survived recovery: err = %v", err)
			366	+	}
			367	+	if _, err := os.Stat(path + ".prev"); !errors.Is(err, os.ErrNotExist) {
			368	+	t.Fatalf(".prev survived recovery: err = %v", err)
			369	+	}
			370	+	got, err := Replay(path)
			371	+	if err != nil {
			372	+	t.Fatal(err)
			373	+	}
			374	+	if string(got["version"]) != "old" {
			375	+	t.Fatalf("version = %q, want %q (current must remain authoritative)", got["version"], "old")
			376	+	}
			377	+	})
			378	+	}
			379	+	}
			380	+
297		func TestDecodeBatchRejectsImpossibleCountBeforeAllocating(t *testing.T) {	381		func TestDecodeBatchRejectsImpossibleCountBeforeAllocating(t *testing.T) {
298		payload := []byte{opBatch, 0xff, 0xff, 0xff, 0xff}	382		payload := []byte{opBatch, 0xff, 0xff, 0xff, 0xff}
299		allocs := testing.AllocsPerRun(10, func() {	383		allocs := testing.AllocsPerRun(10, func() {

@@ -340,19 +424,20 @@ func TestBatchAtomicity(t *testing.T) {
340		func TestCompactionRecoveryMatrix(t *testing.T) {	424		func TestCompactionRecoveryMatrix(t *testing.T) {
341		oldRecord, _ := encodeRecord(opSet, []byte("version"), []byte("old"))	425		oldRecord, _ := encodeRecord(opSet, []byte("version"), []byte("old"))
342		newRecord, _ := encodeRecord(opSet, []byte("version"), []byte("new"))	426		newRecord, _ := encodeRecord(opSet, []byte("version"), []byte("new"))
			427	+	oldFile, newFile := currentFile(oldRecord), currentFile(newRecord)
343		tests := []struct {	428		tests := []struct {
344		name string	429		name string
345		current, next, prev []byte	430		current, next, prev []byte
346		want string	431		want string
347		}{	432		}{
348	-	{name: "current-alone", current: oldRecord, want: "old"},	433	+	{name: "current-alone", current: oldFile, want: "old"},
349	-	{name: "current-next", current: oldRecord, next: newRecord, want: "old"},	434	+	{name: "current-next", current: oldFile, next: newFile, want: "old"},
350	-	{name: "prev-next", prev: oldRecord, next: newRecord, want: "new"},	435	+	{name: "prev-next", prev: oldFile, next: newFile, want: "new"},
351	-	{name: "prev-invalid-next", prev: oldRecord, next: rawRecord([]byte{0xff, 0, 0}), want: "old"},	436	+	{name: "prev-invalid-next", prev: oldFile, next: currentFile(rawRecord([]byte{0xff, 0, 0})), want: "old"},
352	-	{name: "prev-torn-next", prev: oldRecord, next: newRecord[:len(newRecord)-1], want: "old"},	437	+	{name: "prev-torn-next", prev: oldFile, next: newFile[:len(newFile)-1], want: "old"},
353	-	{name: "current-prev", current: newRecord, prev: oldRecord, want: "new"},	438	+	{name: "current-prev", current: newFile, prev: oldFile, want: "new"},
354	-	{name: "invalid-current-prev", current: rawRecord([]byte{0xff, 0, 0}), prev: oldRecord, want: "old"},	439	+	{name: "invalid-current-prev", current: currentFile(rawRecord([]byte{0xff, 0, 0})), prev: oldFile, want: "old"},
355	-	{name: "prev-alone", prev: oldRecord, want: "old"},	440	+	{name: "prev-alone", prev: oldFile, want: "old"},
356		}	441		}
357		for _, tt := range tests {	442		for _, tt := range tests {
358		t.Run(tt.name, func(t *testing.T) {	443		t.Run(tt.name, func(t *testing.T) {

@@ -392,7 +477,7 @@ func TestCompactionRecoveryMatrix(t *testing.T) {
392		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {	477		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
393		t.Fatal(err)	478		t.Fatal(err)
394		}	479		}
395	-	if err := os.WriteFile(path, oldRecord, 0o600); err != nil {	480	+	if err := os.WriteFile(path, oldFile, 0o600); err != nil {
396		t.Fatal(err)	481		t.Fatal(err)
397		}	482		}
398		s, err := Open(path)	483		s, err := Open(path)

@@ -488,13 +573,14 @@ func TestCompactionFilesystemOperationFailuresPoisonAndRecover(t *testing.T) {
488		if err != nil {	573		if err != nil {
489		t.Fatal(err)	574		t.Fatal(err)
490		}	575		}
			576	+	oldFile := currentFile(oldRecord)
491		for _, tt := range tests {	577		for _, tt := range tests {
492		t.Run(tt.name, func(t *testing.T) {	578		t.Run(tt.name, func(t *testing.T) {
493		path := privatePath(t)	579		path := privatePath(t)
494		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {	580		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
495		t.Fatal(err)	581		t.Fatal(err)
496		}	582		}
497	-	if err := os.WriteFile(path, oldRecord, 0o600); err != nil {	583	+	if err := os.WriteFile(path, oldFile, 0o600); err != nil {
498		t.Fatal(err)	584		t.Fatal(err)
499		}	585		}
500		s, err := Open(path)	586		s, err := Open(path)

@@ -599,13 +685,7 @@ func requireCandidateBytes(t *testing.T, path string, want map[string][]byte) {
599		}	685		}
600			686
601		func corruptRecord(buf []byte, offset int) []byte { buf[offset] ^= 0xff; return buf }	687		func corruptRecord(buf []byte, offset int) []byte { buf[offset] ^= 0xff; return buf }
602	-	func rawRecord(payload []byte) []byte {	688	+	func rawRecord(payload []byte) []byte { return framePayload(payload) }
603	-	buf := make([]byte, headerLen+len(payload))
604	-	binary.BigEndian.PutUint32(buf[:4], uint32(len(payload)))
605	-	binary.BigEndian.PutUint32(buf[4:8], crc32.ChecksumIEEE(payload))
606	-	copy(buf[8:], payload)
607	-	return buf
608	-	}
609		func rawBatchPayload(changes []BatchChange) []byte {	689		func rawBatchPayload(changes []BatchChange) []byte {
610		payload := []byte{opBatch, 0, 0, 0, byte(len(changes))}	690		payload := []byte{opBatch, 0, 0, 0, byte(len(changes))}
611		for _, change := range changes {	691		for _, change := range changes {

@@ -621,23 +701,275 @@ func rawBatchPayload(changes []BatchChange) []byte {
621		return payload	701		return payload
622		}	702		}
623			703
624	-	func TestReplayStopsAtBadPayloadLengthWithoutAllocatingHugeBuffer(t *testing.T) {	704	+	// An oversized length must never drive an allocation. In a legacy file the
625	-	path := privatePath(t)	705	+	// length is unprotected, so it stays a torn tail; in the current format the
			706	+	// header checksum proves the length was written that way, which is corruption.
			707	+	func TestReplayRejectsOversizedLengthWithoutAllocating(t *testing.T) {
626		first, _ := encodeRecord(opSet, []byte("a"), []byte("one"))	708		first, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
627	-	header := make([]byte, headerLen)	709	+	legacyHeader := make([]byte, legacyHeaderLen)
628	-	binary.BigEndian.PutUint32(header[0:4], 1<<31)	710	+	binary.BigEndian.PutUint32(legacyHeader[0:4], 1<<31)
629	-	if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {	711	+
			712	+	t.Run("legacy length is a torn tail", func(t *testing.T) {
			713	+	path := privatePath(t)
			714	+	writeFile(t, path, append(legacyFile(first), legacyHeader...))
			715	+	s, err := Open(path)
			716	+	if err != nil {
			717	+	t.Fatal(err)
			718	+	}
			719	+	defer closeStore(t, s)
			720	+	if got, ok := s.Get([]byte("a")); !ok \|\| string(got) != "one" {
			721	+	t.Fatalf("good record missing before bad length: %q %v", got, ok)
			722	+	}
			723	+	})
			724	+
			725	+	t.Run("current oversized length fails closed", func(t *testing.T) {
			726	+	path := privatePath(t)
			727	+	wal := currentFile(first, oversizedRecordHeader(maxPayloadLen+1))
			728	+	writeFile(t, path, wal)
			729	+	if _, err := Open(path); !errors.Is(err, ErrCorruptWAL) {
			730	+	t.Fatalf("Open error = %v, want ErrCorruptWAL", err)
			731	+	}
			732	+	requireFileBytes(t, path, wal)
			733	+	})
			734	+	}
			735	+
			736	+	// oversizedRecordHeader is a record header whose own checksum is valid, so the
			737	+	// length is provably what the writer put there rather than a bit flip.
			738	+	func oversizedRecordHeader(payloadLen uint32) []byte {
			739	+	header := make([]byte, recordHeaderLen)
			740	+	binary.BigEndian.PutUint32(header[0:4], payloadLen)
			741	+	binary.BigEndian.PutUint32(header[8:recordHeaderLen], crc32.ChecksumIEEE(header[0:8]))
			742	+	return header
			743	+	}
			744	+
			745	+	func requireFileBytes(t *testing.T, path string, want []byte) {
			746	+	t.Helper()
			747	+	got, err := os.ReadFile(path)
			748	+	if err != nil {
630		t.Fatal(err)	749		t.Fatal(err)
631		}	750		}
632	-	if err := os.WriteFile(path, append(first, header...), 0o600); err != nil {	751	+	if !bytes.Equal(got, want) {
633	-	t.Fatal(err)	752	+	t.Fatalf("file was mutated: got %d bytes, want %d", len(got), len(want))
634		}	753		}
635	-	s, err := Open(path)	754	+	}
636	-	if err != nil {	755	+
637	-	t.Fatal(err)	756	+	// The regression this format exists for: a bit flip in the length field of an
			757	+	// early record used to look exactly like a torn tail, so Open truncated every
			758	+	// record written after it and reopened a smaller, structurally valid file.
			759	+	func TestCorruptLengthOfEarlyRecordFailsClosedWithoutTruncating(t *testing.T) {
			760	+	first, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
			761	+	second, _ := encodeRecord(opSet, []byte("b"), []byte("two"))
			762	+	third, _ := encodeRecord(opSet, []byte("c"), []byte("three"))
			763	+
			764	+	tests := []struct {
			765	+	name string
			766	+	offset int
			767	+	}{
			768	+	{name: "first-record-length", offset: fileHeaderLen},
			769	+	{name: "second-record-length", offset: fileHeaderLen + len(first) + 3},
638		}	770		}
639	-	defer closeStore(t, s)	771	+	for _, tt := range tests {
640	-	if got, ok := s.Get([]byte("a")); !ok \|\| string(got) != "one" {	772	+	t.Run(tt.name, func(t *testing.T) {
641	-	t.Fatalf("good record missing before bad length: %q %v", got, ok)	773	+	path := privatePath(t)
			774	+	wal := currentFile(first, second, third)
			775	+	// Flip a high bit so the length claims far more than the file holds:
			776	+	// the exact shape that used to be misread as an incomplete tail.
			777	+	wal[tt.offset] ^= 0x40
			778	+	writeFile(t, path, wal)
			779	+
			780	+	if _, err := Open(path); !errors.Is(err, ErrCorruptWAL) {
			781	+	t.Fatalf("Open error = %v, want ErrCorruptWAL", err)
			782	+	}
			783	+	requireFileBytes(t, path, wal)
			784	+	if _, err := Replay(path); !errors.Is(err, ErrCorruptWAL) {
			785	+	t.Fatalf("Replay error = %v, want ErrCorruptWAL", err)
			786	+	}
			787	+	})
			788	+	}
			789	+	}
			790	+
			791	+	// Every row of the replay decision table for the current format. A torn tail is
			792	+	// the only truncatable outcome.
			793	+	func TestCurrentFormatReplayDecisionTable(t *testing.T) {
			794	+	first, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
			795	+	second, _ := encodeRecord(opSet, []byte("b"), []byte("two"))
			796	+	invalidOp := rawRecord([]byte{0xff, 0, 0})
			797	+
			798	+	tests := []struct {
			799	+	name string
			800	+	wal []byte
			801	+	torn bool
			802	+	corrupt bool
			803	+	}{
			804	+	{name: "partial-header", wal: currentFile(first, second[:recordHeaderLen-1]), torn: true},
			805	+	{name: "header-checksum-mismatch", wal: corruptRecord(currentFile(first, second), fileHeaderLen+len(first)+1), corrupt: true},
			806	+	{name: "short-payload", wal: currentFile(first, second[:len(second)-1]), torn: true},
			807	+	{name: "payload-checksum-mismatch", wal: corruptRecord(currentFile(first, second), fileHeaderLen+len(first)+recordHeaderLen), corrupt: true},
			808	+	{name: "undecodable-payload", wal: currentFile(first, invalidOp), corrupt: true},
			809	+	{name: "oversized-length", wal: currentFile(first, oversizedRecordHeader(maxPayloadLen+1)), corrupt: true},
			810	+	{name: "complete-records", wal: currentFile(first, second)},
			811	+	}
			812	+	for _, tt := range tests {
			813	+	t.Run(tt.name, func(t *testing.T) {
			814	+	path := privatePath(t)
			815	+	writeFile(t, path, tt.wal)
			816	+	f, err := os.Open(path)
			817	+	if err != nil {
			818	+	t.Fatal(err)
			819	+	}
			820	+	result, err := replayFile(f)
			821	+	_ = f.Close()
			822	+
			823	+	if tt.corrupt {
			824	+	if !errors.Is(err, ErrCorruptWAL) {
			825	+	t.Fatalf("replay error = %v, want ErrCorruptWAL", err)
			826	+	}
			827	+	return
			828	+	}
			829	+	if err != nil {
			830	+	t.Fatal(err)
			831	+	}
			832	+	if result.Format != formatCurrent {
			833	+	t.Fatalf("format = %d, want current", result.Format)
			834	+	}
			835	+	if result.TornTail != tt.torn {
			836	+	t.Fatalf("TornTail = %v, want %v", result.TornTail, tt.torn)
			837	+	}
			838	+	if string(result.Data["a"]) != "one" {
			839	+	t.Fatalf("prefix lost: %#v", result.Data)
			840	+	}
			841	+	wantLastGood := int64(fileHeaderLen + len(first))
			842	+	if !tt.torn {
			843	+	wantLastGood += int64(len(second))
			844	+	}
			845	+	if result.LastGood != wantLastGood {
			846	+	t.Fatalf("LastGood = %d, want %d", result.LastGood, wantLastGood)
			847	+	}
			848	+	})
			849	+	}
			850	+	}
			851	+
			852	+	// A file written by the released version must keep working: it opens with the
			853	+	// original semantics, is upgraded in place, and reopens identically.
			854	+	func TestLegacyFileOpensAndUpgradesInPlace(t *testing.T) {
			855	+	setA, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
			856	+	setB, _ := encodeRecord(opSet, []byte("b"), []byte("two"))
			857	+	replaceA, _ := encodeRecord(opSet, []byte("a"), []byte("uno"))
			858	+	delB, _ := encodeRecord(opDel, []byte("b"), nil)
			859	+	torn, _ := encodeRecord(opSet, []byte("c"), []byte("three"))
			860	+
			861	+	tests := []struct {
			862	+	name string
			863	+	wal []byte
			864	+	want map[string]string
			865	+	}{
			866	+	{name: "complete", wal: legacyFile(setA, setB, replaceA), want: map[string]string{"a": "uno", "b": "two"}},
			867	+	{name: "tombstone", wal: legacyFile(setA, setB, delB), want: map[string]string{"a": "one"}},
			868	+	{name: "torn-tail", wal: append(legacyFile(setA, setB), legacyFile(torn)[:legacyHeaderLen+2]...), want: map[string]string{"a": "one", "b": "two"}},
			869	+	{name: "empty", wal: nil, want: map[string]string{}},
			870	+	}
			871	+	for _, tt := range tests {
			872	+	t.Run(tt.name, func(t *testing.T) {
			873	+	path := privatePath(t)
			874	+	writeFile(t, path, tt.wal)
			875	+
			876	+	s, err := Open(path)
			877	+	if err != nil {
			878	+	t.Fatalf("legacy file failed to open: %v", err)
			879	+	}
			880	+	requireContents(t, s, tt.want)
			881	+	if err := s.Set([]byte("added"), []byte("after-upgrade")); err != nil {
			882	+	t.Fatal(err)
			883	+	}
			884	+	closeStore(t, s)
			885	+
			886	+	upgraded, err := os.ReadFile(path)
			887	+	if err != nil {
			888	+	t.Fatal(err)
			889	+	}
			890	+	if !bytes.HasPrefix(upgraded, fileMagic[:]) {
			891	+	t.Fatal("legacy file was not upgraded in place")
			892	+	}
			893	+	reopened, err := Open(path)
			894	+	if err != nil {
			895	+	t.Fatal(err)
			896	+	}
			897	+	defer closeStore(t, reopened)
			898	+	want := map[string]string{"added": "after-upgrade"}
			899	+	for k, v := range tt.want {
			900	+	want[k] = v
			901	+	}
			902	+	requireContents(t, reopened, want)
			903	+	replayed, err := Replay(path)
			904	+	if err != nil {
			905	+	t.Fatal(err)
			906	+	}
			907	+	if len(replayed) != len(want) {
			908	+	t.Fatalf("Replay = %#v, want %#v", replayed, want)
			909	+	}
			910	+	})
			911	+	}
			912	+	}
			913	+
			914	+	func requireContents(t testing.T, s Store, want map[string]string) {
			915	+	t.Helper()
			916	+	for key, value := range want {
			917	+	got, ok := s.Get([]byte(key))
			918	+	if !ok \|\| string(got) != value {
			919	+	t.Fatalf("Get(%q) = %q, %v; want %q", key, got, ok, value)
			920	+	}
			921	+	}
			922	+	count := 0
			923	+	s.Range(func([]byte, []byte) bool { count++; return true })
			924	+	if count != len(want) {
			925	+	t.Fatalf("store holds %d keys, want %d", count, len(want))
			926	+	}
			927	+	}
			928	+
			929	+	// The upgrade runs through the compaction protocol, so a crash at any of its
			930	+	// boundaries must leave either the whole legacy file or the whole upgraded one.
			931	+	func TestLegacyUpgradeCrashSafety(t *testing.T) {
			932	+	setA, _ := encodeRecord(opSet, []byte("a"), []byte("one"))
			933	+	setB, _ := encodeRecord(opSet, []byte("b"), []byte("two"))
			934	+	legacy := legacyFile(setA, setB)
			935	+	want := map[string]string{"a": "one", "b": "two"}
			936	+
			937	+	faultPoints := []string{
			938	+	"before-write-next", "after-write-next", "before-sync-next", "after-sync-next",
			939	+	"after-sync-dir-next", "after-remove-stale-prev", "after-rename-current-prev",
			940	+	"after-sync-dir-prev", "after-rename-next-current", "after-sync-dir-current",
			941	+	"before-reopen-current", "after-reopen-current", "after-remove-prev", "after-final-sync-dir",
			942	+	}
			943	+	for _, point := range faultPoints {
			944	+	t.Run("fault-"+point, func(t *testing.T) {
			945	+	path := privatePath(t)
			946	+	writeFile(t, path, legacy)
			947	+	hooks := defaultFSHooks()
			948	+	hooks.Fault = func(got string) error {
			949	+	if got == point {
			950	+	return errors.New("injected " + point)
			951	+	}
			952	+	return nil
			953	+	}
			954	+	if _, err := openWithHooks(path, hooks); err == nil {
			955	+	t.Fatalf("fault %q did not interrupt the upgrade", point)
			956	+	}
			957	+
			958	+	// Restart: recovery selects a complete file, and the data is intact
			959	+	// whichever format that file happens to be in.
			960	+	s, err := Open(path)
			961	+	if err != nil {
			962	+	t.Fatalf("restart after fault %q: %v", point, err)
			963	+	}
			964	+	requireContents(t, s, want)
			965	+	closeStore(t, s)
			966	+	upgraded, err := os.ReadFile(path)
			967	+	if err != nil {
			968	+	t.Fatal(err)
			969	+	}
			970	+	if !bytes.HasPrefix(upgraded, fileMagic[:]) {
			971	+	t.Fatal("restart did not complete the upgrade")
			972	+	}
			973	+	})
642		}	974		}
643		}	975		}