detached 15cd1c5

feat(record): add log and snapshot codec with fuzz test

commit
15cd1c533fa64fa05d013dbceeffcccdd1811922
author
brice · · 2026-10-10 01:52 UTC
committer
brice · · 2026-10-10 01:57 UTC
parent
72a9140
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +666 -0
Unified
UnifiedSplit
A errors.go +24 -0
@@ -0,0 +1,24 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import "errors"
6+
7+var (
8+ // ErrClosed is returned when the database is closed.
9+ ErrClosed = errors.New("kv: closed")
10+ // ErrFailed is returned once an I/O error put the database in the failed
11+ // state. It wraps the first I/O error.
12+ ErrFailed = errors.New("kv: database failed")
13+ // ErrConflict is returned when a batch precondition does not hold.
14+ ErrConflict = errors.New("kv: precondition failed")
15+ // ErrBatchTooLarge is returned when an encoded batch exceeds the limit.
16+ ErrBatchTooLarge = errors.New("kv: batch too large")
17+ // ErrInvalidKey is returned for an empty key or one longer than 64 KiB.
18+ ErrInvalidKey = errors.New("kv: invalid key")
19+ // ErrLocked is returned when another process holds the directory lock.
20+ ErrLocked = errors.New("kv: directory locked by another process")
21+ // ErrCorrupt is returned when on-disk data is damaged. It is wrapped with
22+ // the file and offset.
23+ ErrCorrupt = errors.New("kv: corrupt data")
24+)
@@ -0,0 +1,24 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import "errors"
6+
7+var (
8+ // ErrClosed is returned when the database is closed.
9+ ErrClosed = errors.New("kv: closed")
10+ // ErrFailed is returned once an I/O error put the database in the failed
11+ // state. It wraps the first I/O error.
12+ ErrFailed = errors.New("kv: database failed")
13+ // ErrConflict is returned when a batch precondition does not hold.
14+ ErrConflict = errors.New("kv: precondition failed")
15+ // ErrBatchTooLarge is returned when an encoded batch exceeds the limit.
16+ ErrBatchTooLarge = errors.New("kv: batch too large")
17+ // ErrInvalidKey is returned for an empty key or one longer than 64 KiB.
18+ ErrInvalidKey = errors.New("kv: invalid key")
19+ // ErrLocked is returned when another process holds the directory lock.
20+ ErrLocked = errors.New("kv: directory locked by another process")
21+ // ErrCorrupt is returned when on-disk data is damaged. It is wrapped with
22+ // the file and offset.
23+ ErrCorrupt = errors.New("kv: corrupt data")
24+)
A record.go +352 -0
@@ -0,0 +1,352 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bufio"
7+ "bytes"
8+ "encoding/binary"
9+ "errors"
10+ "fmt"
11+ "hash/crc32"
12+ "io"
13+)
14+
15+// Operation kinds.
16+const (
17+ opPut byte = 1
18+ opDelete byte = 2
19+)
20+
21+// Format limits.
22+const (
23+ maxRecord = 64 << 20 // maximum record payload size
24+ maxKeyLen = 64 << 10 // maximum key length
25+)
26+
27+const (
28+ logMagic = "EKVLOG01"
29+ snapshotMagic = "EKVSNP01"
30+ logHeaderSize = 8 + 8 + 4
31+ recordFrame = 4 + 4 // len + crc
32+ recordMinBody = 8 + 4 // seq + count
33+ snapHeaderSize = 8 + 8 + 8 + 4
34+ minOpSize = 3 // kind + klen + 1 key byte
35+)
36+
37+var castagnoli = crc32.MakeTable(crc32.Castagnoli)
38+
39+// errUnreadable marks a log record that is not readable (see decodeRecord).
40+var errUnreadable = errors.New("unreadable record")
41+
42+// op is one mutation of a batch. value is nil for opDelete.
43+type op struct {
44+ kind byte
45+ key, value []byte
46+}
47+
48+// opSize returns the encoded size of o.
49+func opSize(o op) int {
50+ n := 1 + uvarintLen(uint64(len(o.key))) + len(o.key)
51+ if o.kind == opPut {
52+ n += uvarintLen(uint64(len(o.value))) + len(o.value)
53+ }
54+ return n
55+}
56+
57+func uvarintLen(v uint64) int {
58+ n := 1
59+ for ; v >= 0x80; v >>= 7 {
60+ n++
61+ }
62+ return n
63+}
64+
65+// appendLogHeader appends the 20-byte log header for base to dst.
66+func appendLogHeader(dst []byte, base uint64) []byte {
67+ start := len(dst)
68+ dst = append(dst, logMagic...)
69+ dst = binary.LittleEndian.AppendUint64(dst, base)
70+ return binary.LittleEndian.AppendUint32(dst, crc32.Checksum(dst[start:], castagnoli))
71+}
72+
73+// decodeLogHeader validates the log header at the start of b and returns its
74+// base. Errors wrap ErrCorrupt.
75+func decodeLogHeader(b []byte) (base uint64, err error) {
76+ if len(b) < logHeaderSize {
77+ return 0, fmt.Errorf("%w: short log header", ErrCorrupt)
78+ }
79+ if string(b[:8]) != logMagic {
80+ return 0, fmt.Errorf("%w: bad log magic", ErrCorrupt)
81+ }
82+ if crc32.Checksum(b[:16], castagnoli) != binary.LittleEndian.Uint32(b[16:20]) {
83+ return 0, fmt.Errorf("%w: bad log header checksum", ErrCorrupt)
84+ }
85+ return binary.LittleEndian.Uint64(b[8:16]), nil
86+}
87+
88+// appendRecord appends one encoded record (len, crc, seq, count, ops) to dst.
89+// The payload must not exceed maxRecord and ops must be non-empty with valid
90+// keys; the caller enforces this (batches are validated before encoding).
91+func appendRecord(dst []byte, seq uint64, ops []op) []byte {
92+ start := len(dst)
93+ dst = append(dst, make([]byte, recordFrame)...)
94+ dst = binary.LittleEndian.AppendUint64(dst, seq)
95+ dst = binary.LittleEndian.AppendUint32(dst, uint32(len(ops)))
96+ for _, o := range ops {
97+ dst = append(dst, o.kind)
98+ dst = binary.AppendUvarint(dst, uint64(len(o.key)))
99+ dst = append(dst, o.key...)
100+ if o.kind == opPut {
101+ dst = binary.AppendUvarint(dst, uint64(len(o.value)))
102+ dst = append(dst, o.value...)
103+ }
104+ }
105+ payload := dst[start+recordFrame:]
106+ binary.LittleEndian.PutUint32(dst[start:], uint32(len(payload)))
107+ binary.LittleEndian.PutUint32(dst[start+4:], crc32.Checksum(payload, castagnoli))
108+ return dst
109+}
110+
111+// peekRecord returns the total encoded size (frame + payload) announced by
112+// the record frame at the start of b. ok is false if fewer than 8 bytes are
113+// present or the announced payload length is outside [12, maxRecord]. It does
114+// not check that the payload is present.
115+func peekRecord(b []byte) (total int, ok bool) {
116+ if len(b) < recordFrame {
117+ return 0, false
118+ }
119+ n := binary.LittleEndian.Uint32(b)
120+ if n < recordMinBody || n > maxRecord {
121+ return 0, false
122+ }
123+ return recordFrame + int(n), true
124+}
125+
126+// decodeRecord decodes the record at the start of b. A record is readable
127+// only if the frame is present, 12 ≤ len ≤ maxRecord, len payload bytes are
128+// present, the CRC matches, and the payload holds exactly count ≥ 1 valid
129+// ops consuming exactly len bytes. Otherwise the error wraps errUnreadable.
130+//
131+// The decoded ops are appended to ops[:0]; their keys and values alias b.
132+// n is the number of bytes consumed.
133+func decodeRecord(b []byte, ops []op) (seq uint64, out []op, n int, err error) {
134+ total, ok := peekRecord(b)
135+ if !ok {
136+ return 0, nil, 0, fmt.Errorf("%w: bad or short frame", errUnreadable)
137+ }
138+ if len(b) < total {
139+ return 0, nil, 0, fmt.Errorf("%w: short payload", errUnreadable)
140+ }
141+ payload := b[recordFrame:total]
142+ if crc32.Checksum(payload, castagnoli) != binary.LittleEndian.Uint32(b[4:8]) {
143+ return 0, nil, 0, fmt.Errorf("%w: checksum mismatch", errUnreadable)
144+ }
145+ seq, out, err = decodePayload(payload, ops)
146+ if err != nil {
147+ return 0, nil, 0, err
148+ }
149+ return seq, out, total, nil
150+}
151+
152+// decodePayload decodes seq, count and ops from a whole payload.
153+func decodePayload(p []byte, ops []op) (uint64, []op, error) {
154+ seq := binary.LittleEndian.Uint64(p)
155+ count := binary.LittleEndian.Uint32(p[8:])
156+ p = p[recordMinBody:]
157+ if count == 0 || uint64(count) > uint64(len(p)/minOpSize) {
158+ return 0, nil, fmt.Errorf("%w: bad op count %d", errUnreadable, count)
159+ }
160+ ops = ops[:0]
161+ for range count {
162+ if len(p) == 0 {
163+ return 0, nil, fmt.Errorf("%w: truncated op", errUnreadable)
164+ }
165+ o := op{kind: p[0]}
166+ if o.kind != opPut && o.kind != opDelete {
167+ return 0, nil, fmt.Errorf("%w: bad op kind %d", errUnreadable, o.kind)
168+ }
169+ var ok bool
170+ if o.key, p, ok = cutBytes(p[1:], maxKeyLen); !ok || len(o.key) == 0 {
171+ return 0, nil, fmt.Errorf("%w: bad key", errUnreadable)
172+ }
173+ if o.kind == opPut {
174+ if o.value, p, ok = cutBytes(p, maxRecord); !ok {
175+ return 0, nil, fmt.Errorf("%w: bad value", errUnreadable)
176+ }
177+ }
178+ ops = append(ops, o)
179+ }
180+ if len(p) != 0 {
181+ return 0, nil, fmt.Errorf("%w: %d trailing payload bytes", errUnreadable, len(p))
182+ }
183+ return seq, ops, nil
184+}
185+
186+// cutBytes reads a uvarint length (at most limit) and that many bytes from
187+// the start of b. rest is what follows. The returned slice has no spare
188+// capacity.
189+func cutBytes(b []byte, limit int) (v, rest []byte, ok bool) {
190+ l, n := binary.Uvarint(b)
191+ if n <= 0 || l > uint64(limit) || l > uint64(len(b)-n) {
192+ return nil, nil, false
193+ }
194+ end := n + int(l)
195+ return b[n:end:end], b[end:], true
196+}
197+
198+// ownOps returns a copy of ops whose keys and values live in one new
199+// allocation, so the result does not pin the buffer the ops were decoded from.
200+func ownOps(ops []op) []op {
201+ size := 0
202+ for _, o := range ops {
203+ size += len(o.key) + len(o.value)
204+ }
205+ buf := make([]byte, 0, size)
206+ out := make([]op, len(ops))
207+ for i, o := range ops {
208+ out[i].kind = o.kind
209+ buf = append(buf, o.key...)
210+ out[i].key = buf[len(buf)-len(o.key) : len(buf) : len(buf)]
211+ if o.kind == opPut {
212+ buf = append(buf, o.value...)
213+ out[i].value = buf[len(buf)-len(o.value) : len(buf) : len(buf)]
214+ }
215+ }
216+ return out
217+}
218+
219+// entryIter yields snapshot entries in strictly ascending key order. It stops
220+// when yield returns false. The slices passed to yield need only stay valid
221+// until yield returns.
222+type entryIter = func(yield func(key, value []byte, version uint64) bool)
223+
224+// writeSnapshot writes a snapshot file of count entries at seq to w. entries
225+// must yield exactly count entries sorted strictly ascending by key, with
226+// 1 ≤ len(key) ≤ maxKeyLen and version ≤ seq; otherwise an error is returned
227+// (the output is then incomplete and must be discarded).
228+func writeSnapshot(w io.Writer, seq, count uint64, entries entryIter) error {
229+ hdr := append(make([]byte, 0, snapHeaderSize), snapshotMagic...)
230+ hdr = binary.LittleEndian.AppendUint64(hdr, seq)
231+ hdr = binary.LittleEndian.AppendUint64(hdr, count)
232+ hdr = binary.LittleEndian.AppendUint32(hdr, crc32.Checksum(hdr, castagnoli))
233+ bw := bufio.NewWriterSize(w, 64<<10)
234+ if _, err := bw.Write(hdr); err != nil {
235+ return err
236+ }
237+
238+ var (
239+ sum uint32
240+ buf []byte
241+ prev []byte
242+ written uint64
243+ err error
244+ )
245+ entries(func(key, value []byte, version uint64) bool {
246+ switch {
247+ case written == count:
248+ err = errors.New("snapshot: more entries than count")
249+ case len(key) == 0 || len(key) > maxKeyLen:
250+ err = fmt.Errorf("snapshot: invalid key length %d", len(key))
251+ case written > 0 && bytes.Compare(prev, key) >= 0:
252+ err = errors.New("snapshot: keys not strictly ascending")
253+ case version > seq:
254+ err = fmt.Errorf("snapshot: version %d above seq %d", version, seq)
255+ }
256+ if err != nil {
257+ return false
258+ }
259+ buf = binary.AppendUvarint(buf[:0], uint64(len(key)))
260+ buf = append(buf, key...)
261+ buf = binary.AppendUvarint(buf, version)
262+ buf = binary.AppendUvarint(buf, uint64(len(value)))
263+ buf = append(buf, value...)
264+ sum = crc32.Update(sum, castagnoli, buf)
265+ if _, err = bw.Write(buf); err != nil {
266+ return false
267+ }
268+ prev = append(prev[:0], key...)
269+ written++
270+ return true
271+ })
272+ if err != nil {
273+ return err
274+ }
275+ if written != count {
276+ return fmt.Errorf("snapshot: %d entries yielded, want %d", written, count)
277+ }
278+ if _, err := bw.Write(binary.LittleEndian.AppendUint32(nil, sum)); err != nil {
279+ return err
280+ }
281+ return bw.Flush()
282+}
283+
284+// snapshotView is a fully validated snapshot file held in memory.
285+type snapshotView struct {
286+ seq, count uint64
287+ body []byte // entries, without header and trailer
288+}
289+
290+// openSnapshot validates the snapshot file contents b: header and trailer
291+// checksums, exactly count entries, strictly ascending non-empty keys,
292+// versions ≤ seq and no trailing bytes. Errors wrap ErrCorrupt.
293+func openSnapshot(b []byte) (snapshotView, error) {
294+ if len(b) < snapHeaderSize+4 {
295+ return snapshotView{}, fmt.Errorf("%w: short snapshot", ErrCorrupt)
296+ }
297+ if string(b[:8]) != snapshotMagic {
298+ return snapshotView{}, fmt.Errorf("%w: bad snapshot magic", ErrCorrupt)
299+ }
300+ if crc32.Checksum(b[:snapHeaderSize-4], castagnoli) != binary.LittleEndian.Uint32(b[snapHeaderSize-4:]) {
301+ return snapshotView{}, fmt.Errorf("%w: bad snapshot header checksum", ErrCorrupt)
302+ }
303+ v := snapshotView{
304+ seq: binary.LittleEndian.Uint64(b[8:]),
305+ count: binary.LittleEndian.Uint64(b[16:]),
306+ body: b[snapHeaderSize : len(b)-4],
307+ }
308+ if crc32.Checksum(v.body, castagnoli) != binary.LittleEndian.Uint32(b[len(b)-4:]) {
309+ return snapshotView{}, fmt.Errorf("%w: bad snapshot checksum", ErrCorrupt)
310+ }
311+ if err := v.walk(nil); err != nil {
312+ return snapshotView{}, err
313+ }
314+ return v, nil
315+}
316+
317+// each calls fn for every entry in ascending key order. The slices alias the
318+// file contents: fn must not modify them and must copy them to retain them
319+// independently of the file buffer.
320+func (v snapshotView) each(fn func(key, value []byte, version uint64)) {
321+ _ = v.walk(fn) // cannot fail: the view was validated
322+}
323+
324+func (v snapshotView) walk(fn func(key, value []byte, version uint64)) error {
325+ body := v.body
326+ var prev []byte
327+ for i := uint64(0); i < v.count; i++ {
328+ var key, value []byte
329+ var ok bool
330+ if key, body, ok = cutBytes(body, maxKeyLen); !ok || len(key) == 0 {
331+ return fmt.Errorf("%w: bad snapshot key at entry %d", ErrCorrupt, i)
332+ }
333+ if i > 0 && bytes.Compare(prev, key) >= 0 {
334+ return fmt.Errorf("%w: snapshot keys not ascending at entry %d", ErrCorrupt, i)
335+ }
336+ version, n := binary.Uvarint(body)
337+ if n <= 0 || version > v.seq {
338+ return fmt.Errorf("%w: bad snapshot version at entry %d", ErrCorrupt, i)
339+ }
340+ if value, body, ok = cutBytes(body[n:], maxRecord); !ok {
341+ return fmt.Errorf("%w: bad snapshot value at entry %d", ErrCorrupt, i)
342+ }
343+ if fn != nil {
344+ fn(key, value, version)
345+ }
346+ prev = key
347+ }
348+ if len(body) != 0 {
349+ return fmt.Errorf("%w: %d trailing snapshot bytes", ErrCorrupt, len(body))
350+ }
351+ return nil
352+}
@@ -0,0 +1,352 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bufio"
7+ "bytes"
8+ "encoding/binary"
9+ "errors"
10+ "fmt"
11+ "hash/crc32"
12+ "io"
13+)
14+
15+// Operation kinds.
16+const (
17+ opPut byte = 1
18+ opDelete byte = 2
19+)
20+
21+// Format limits.
22+const (
23+ maxRecord = 64 << 20 // maximum record payload size
24+ maxKeyLen = 64 << 10 // maximum key length
25+)
26+
27+const (
28+ logMagic = "EKVLOG01"
29+ snapshotMagic = "EKVSNP01"
30+ logHeaderSize = 8 + 8 + 4
31+ recordFrame = 4 + 4 // len + crc
32+ recordMinBody = 8 + 4 // seq + count
33+ snapHeaderSize = 8 + 8 + 8 + 4
34+ minOpSize = 3 // kind + klen + 1 key byte
35+)
36+
37+var castagnoli = crc32.MakeTable(crc32.Castagnoli)
38+
39+// errUnreadable marks a log record that is not readable (see decodeRecord).
40+var errUnreadable = errors.New("unreadable record")
41+
42+// op is one mutation of a batch. value is nil for opDelete.
43+type op struct {
44+ kind byte
45+ key, value []byte
46+}
47+
48+// opSize returns the encoded size of o.
49+func opSize(o op) int {
50+ n := 1 + uvarintLen(uint64(len(o.key))) + len(o.key)
51+ if o.kind == opPut {
52+ n += uvarintLen(uint64(len(o.value))) + len(o.value)
53+ }
54+ return n
55+}
56+
57+func uvarintLen(v uint64) int {
58+ n := 1
59+ for ; v >= 0x80; v >>= 7 {
60+ n++
61+ }
62+ return n
63+}
64+
65+// appendLogHeader appends the 20-byte log header for base to dst.
66+func appendLogHeader(dst []byte, base uint64) []byte {
67+ start := len(dst)
68+ dst = append(dst, logMagic...)
69+ dst = binary.LittleEndian.AppendUint64(dst, base)
70+ return binary.LittleEndian.AppendUint32(dst, crc32.Checksum(dst[start:], castagnoli))
71+}
72+
73+// decodeLogHeader validates the log header at the start of b and returns its
74+// base. Errors wrap ErrCorrupt.
75+func decodeLogHeader(b []byte) (base uint64, err error) {
76+ if len(b) < logHeaderSize {
77+ return 0, fmt.Errorf("%w: short log header", ErrCorrupt)
78+ }
79+ if string(b[:8]) != logMagic {
80+ return 0, fmt.Errorf("%w: bad log magic", ErrCorrupt)
81+ }
82+ if crc32.Checksum(b[:16], castagnoli) != binary.LittleEndian.Uint32(b[16:20]) {
83+ return 0, fmt.Errorf("%w: bad log header checksum", ErrCorrupt)
84+ }
85+ return binary.LittleEndian.Uint64(b[8:16]), nil
86+}
87+
88+// appendRecord appends one encoded record (len, crc, seq, count, ops) to dst.
89+// The payload must not exceed maxRecord and ops must be non-empty with valid
90+// keys; the caller enforces this (batches are validated before encoding).
91+func appendRecord(dst []byte, seq uint64, ops []op) []byte {
92+ start := len(dst)
93+ dst = append(dst, make([]byte, recordFrame)...)
94+ dst = binary.LittleEndian.AppendUint64(dst, seq)
95+ dst = binary.LittleEndian.AppendUint32(dst, uint32(len(ops)))
96+ for _, o := range ops {
97+ dst = append(dst, o.kind)
98+ dst = binary.AppendUvarint(dst, uint64(len(o.key)))
99+ dst = append(dst, o.key...)
100+ if o.kind == opPut {
101+ dst = binary.AppendUvarint(dst, uint64(len(o.value)))
102+ dst = append(dst, o.value...)
103+ }
104+ }
105+ payload := dst[start+recordFrame:]
106+ binary.LittleEndian.PutUint32(dst[start:], uint32(len(payload)))
107+ binary.LittleEndian.PutUint32(dst[start+4:], crc32.Checksum(payload, castagnoli))
108+ return dst
109+}
110+
111+// peekRecord returns the total encoded size (frame + payload) announced by
112+// the record frame at the start of b. ok is false if fewer than 8 bytes are
113+// present or the announced payload length is outside [12, maxRecord]. It does
114+// not check that the payload is present.
115+func peekRecord(b []byte) (total int, ok bool) {
116+ if len(b) < recordFrame {
117+ return 0, false
118+ }
119+ n := binary.LittleEndian.Uint32(b)
120+ if n < recordMinBody || n > maxRecord {
121+ return 0, false
122+ }
123+ return recordFrame + int(n), true
124+}
125+
126+// decodeRecord decodes the record at the start of b. A record is readable
127+// only if the frame is present, 12 ≤ len ≤ maxRecord, len payload bytes are
128+// present, the CRC matches, and the payload holds exactly count ≥ 1 valid
129+// ops consuming exactly len bytes. Otherwise the error wraps errUnreadable.
130+//
131+// The decoded ops are appended to ops[:0]; their keys and values alias b.
132+// n is the number of bytes consumed.
133+func decodeRecord(b []byte, ops []op) (seq uint64, out []op, n int, err error) {
134+ total, ok := peekRecord(b)
135+ if !ok {
136+ return 0, nil, 0, fmt.Errorf("%w: bad or short frame", errUnreadable)
137+ }
138+ if len(b) < total {
139+ return 0, nil, 0, fmt.Errorf("%w: short payload", errUnreadable)
140+ }
141+ payload := b[recordFrame:total]
142+ if crc32.Checksum(payload, castagnoli) != binary.LittleEndian.Uint32(b[4:8]) {
143+ return 0, nil, 0, fmt.Errorf("%w: checksum mismatch", errUnreadable)
144+ }
145+ seq, out, err = decodePayload(payload, ops)
146+ if err != nil {
147+ return 0, nil, 0, err
148+ }
149+ return seq, out, total, nil
150+}
151+
152+// decodePayload decodes seq, count and ops from a whole payload.
153+func decodePayload(p []byte, ops []op) (uint64, []op, error) {
154+ seq := binary.LittleEndian.Uint64(p)
155+ count := binary.LittleEndian.Uint32(p[8:])
156+ p = p[recordMinBody:]
157+ if count == 0 || uint64(count) > uint64(len(p)/minOpSize) {
158+ return 0, nil, fmt.Errorf("%w: bad op count %d", errUnreadable, count)
159+ }
160+ ops = ops[:0]
161+ for range count {
162+ if len(p) == 0 {
163+ return 0, nil, fmt.Errorf("%w: truncated op", errUnreadable)
164+ }
165+ o := op{kind: p[0]}
166+ if o.kind != opPut && o.kind != opDelete {
167+ return 0, nil, fmt.Errorf("%w: bad op kind %d", errUnreadable, o.kind)
168+ }
169+ var ok bool
170+ if o.key, p, ok = cutBytes(p[1:], maxKeyLen); !ok || len(o.key) == 0 {
171+ return 0, nil, fmt.Errorf("%w: bad key", errUnreadable)
172+ }
173+ if o.kind == opPut {
174+ if o.value, p, ok = cutBytes(p, maxRecord); !ok {
175+ return 0, nil, fmt.Errorf("%w: bad value", errUnreadable)
176+ }
177+ }
178+ ops = append(ops, o)
179+ }
180+ if len(p) != 0 {
181+ return 0, nil, fmt.Errorf("%w: %d trailing payload bytes", errUnreadable, len(p))
182+ }
183+ return seq, ops, nil
184+}
185+
186+// cutBytes reads a uvarint length (at most limit) and that many bytes from
187+// the start of b. rest is what follows. The returned slice has no spare
188+// capacity.
189+func cutBytes(b []byte, limit int) (v, rest []byte, ok bool) {
190+ l, n := binary.Uvarint(b)
191+ if n <= 0 || l > uint64(limit) || l > uint64(len(b)-n) {
192+ return nil, nil, false
193+ }
194+ end := n + int(l)
195+ return b[n:end:end], b[end:], true
196+}
197+
198+// ownOps returns a copy of ops whose keys and values live in one new
199+// allocation, so the result does not pin the buffer the ops were decoded from.
200+func ownOps(ops []op) []op {
201+ size := 0
202+ for _, o := range ops {
203+ size += len(o.key) + len(o.value)
204+ }
205+ buf := make([]byte, 0, size)
206+ out := make([]op, len(ops))
207+ for i, o := range ops {
208+ out[i].kind = o.kind
209+ buf = append(buf, o.key...)
210+ out[i].key = buf[len(buf)-len(o.key) : len(buf) : len(buf)]
211+ if o.kind == opPut {
212+ buf = append(buf, o.value...)
213+ out[i].value = buf[len(buf)-len(o.value) : len(buf) : len(buf)]
214+ }
215+ }
216+ return out
217+}
218+
219+// entryIter yields snapshot entries in strictly ascending key order. It stops
220+// when yield returns false. The slices passed to yield need only stay valid
221+// until yield returns.
222+type entryIter = func(yield func(key, value []byte, version uint64) bool)
223+
224+// writeSnapshot writes a snapshot file of count entries at seq to w. entries
225+// must yield exactly count entries sorted strictly ascending by key, with
226+// 1 ≤ len(key) ≤ maxKeyLen and version ≤ seq; otherwise an error is returned
227+// (the output is then incomplete and must be discarded).
228+func writeSnapshot(w io.Writer, seq, count uint64, entries entryIter) error {
229+ hdr := append(make([]byte, 0, snapHeaderSize), snapshotMagic...)
230+ hdr = binary.LittleEndian.AppendUint64(hdr, seq)
231+ hdr = binary.LittleEndian.AppendUint64(hdr, count)
232+ hdr = binary.LittleEndian.AppendUint32(hdr, crc32.Checksum(hdr, castagnoli))
233+ bw := bufio.NewWriterSize(w, 64<<10)
234+ if _, err := bw.Write(hdr); err != nil {
235+ return err
236+ }
237+
238+ var (
239+ sum uint32
240+ buf []byte
241+ prev []byte
242+ written uint64
243+ err error
244+ )
245+ entries(func(key, value []byte, version uint64) bool {
246+ switch {
247+ case written == count:
248+ err = errors.New("snapshot: more entries than count")
249+ case len(key) == 0 || len(key) > maxKeyLen:
250+ err = fmt.Errorf("snapshot: invalid key length %d", len(key))
251+ case written > 0 && bytes.Compare(prev, key) >= 0:
252+ err = errors.New("snapshot: keys not strictly ascending")
253+ case version > seq:
254+ err = fmt.Errorf("snapshot: version %d above seq %d", version, seq)
255+ }
256+ if err != nil {
257+ return false
258+ }
259+ buf = binary.AppendUvarint(buf[:0], uint64(len(key)))
260+ buf = append(buf, key...)
261+ buf = binary.AppendUvarint(buf, version)
262+ buf = binary.AppendUvarint(buf, uint64(len(value)))
263+ buf = append(buf, value...)
264+ sum = crc32.Update(sum, castagnoli, buf)
265+ if _, err = bw.Write(buf); err != nil {
266+ return false
267+ }
268+ prev = append(prev[:0], key...)
269+ written++
270+ return true
271+ })
272+ if err != nil {
273+ return err
274+ }
275+ if written != count {
276+ return fmt.Errorf("snapshot: %d entries yielded, want %d", written, count)
277+ }
278+ if _, err := bw.Write(binary.LittleEndian.AppendUint32(nil, sum)); err != nil {
279+ return err
280+ }
281+ return bw.Flush()
282+}
283+
284+// snapshotView is a fully validated snapshot file held in memory.
285+type snapshotView struct {
286+ seq, count uint64
287+ body []byte // entries, without header and trailer
288+}
289+
290+// openSnapshot validates the snapshot file contents b: header and trailer
291+// checksums, exactly count entries, strictly ascending non-empty keys,
292+// versions ≤ seq and no trailing bytes. Errors wrap ErrCorrupt.
293+func openSnapshot(b []byte) (snapshotView, error) {
294+ if len(b) < snapHeaderSize+4 {
295+ return snapshotView{}, fmt.Errorf("%w: short snapshot", ErrCorrupt)
296+ }
297+ if string(b[:8]) != snapshotMagic {
298+ return snapshotView{}, fmt.Errorf("%w: bad snapshot magic", ErrCorrupt)
299+ }
300+ if crc32.Checksum(b[:snapHeaderSize-4], castagnoli) != binary.LittleEndian.Uint32(b[snapHeaderSize-4:]) {
301+ return snapshotView{}, fmt.Errorf("%w: bad snapshot header checksum", ErrCorrupt)
302+ }
303+ v := snapshotView{
304+ seq: binary.LittleEndian.Uint64(b[8:]),
305+ count: binary.LittleEndian.Uint64(b[16:]),
306+ body: b[snapHeaderSize : len(b)-4],
307+ }
308+ if crc32.Checksum(v.body, castagnoli) != binary.LittleEndian.Uint32(b[len(b)-4:]) {
309+ return snapshotView{}, fmt.Errorf("%w: bad snapshot checksum", ErrCorrupt)
310+ }
311+ if err := v.walk(nil); err != nil {
312+ return snapshotView{}, err
313+ }
314+ return v, nil
315+}
316+
317+// each calls fn for every entry in ascending key order. The slices alias the
318+// file contents: fn must not modify them and must copy them to retain them
319+// independently of the file buffer.
320+func (v snapshotView) each(fn func(key, value []byte, version uint64)) {
321+ _ = v.walk(fn) // cannot fail: the view was validated
322+}
323+
324+func (v snapshotView) walk(fn func(key, value []byte, version uint64)) error {
325+ body := v.body
326+ var prev []byte
327+ for i := uint64(0); i < v.count; i++ {
328+ var key, value []byte
329+ var ok bool
330+ if key, body, ok = cutBytes(body, maxKeyLen); !ok || len(key) == 0 {
331+ return fmt.Errorf("%w: bad snapshot key at entry %d", ErrCorrupt, i)
332+ }
333+ if i > 0 && bytes.Compare(prev, key) >= 0 {
334+ return fmt.Errorf("%w: snapshot keys not ascending at entry %d", ErrCorrupt, i)
335+ }
336+ version, n := binary.Uvarint(body)
337+ if n <= 0 || version > v.seq {
338+ return fmt.Errorf("%w: bad snapshot version at entry %d", ErrCorrupt, i)
339+ }
340+ if value, body, ok = cutBytes(body[n:], maxRecord); !ok {
341+ return fmt.Errorf("%w: bad snapshot value at entry %d", ErrCorrupt, i)
342+ }
343+ if fn != nil {
344+ fn(key, value, version)
345+ }
346+ prev = key
347+ }
348+ if len(body) != 0 {
349+ return fmt.Errorf("%w: %d trailing snapshot bytes", ErrCorrupt, len(body))
350+ }
351+ return nil
352+}
A record_test.go +290 -0
@@ -0,0 +1,290 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bytes"
7+ "encoding/binary"
8+ "errors"
9+ "hash/crc32"
10+ "slices"
11+ "testing"
12+)
13+
14+func sampleOps() []op {
15+ return []op{
16+ {kind: opPut, key: []byte("a"), value: []byte("1")},
17+ {kind: opDelete, key: []byte("b")},
18+ {kind: opPut, key: []byte("c"), value: nil},
19+ {kind: opPut, key: bytes.Repeat([]byte("k"), 200), value: bytes.Repeat([]byte("v"), 300)},
20+ }
21+}
22+
23+func opsEqual(a, b []op) bool {
24+ return slices.EqualFunc(a, b, func(x, y op) bool {
25+ return x.kind == y.kind && bytes.Equal(x.key, y.key) && bytes.Equal(x.value, y.value)
26+ })
27+}
28+
29+func TestRecordRoundTrip(t *testing.T) {
30+ ops := sampleOps()
31+ enc := appendRecord([]byte("prefix"), 42, ops)[6:]
32+ size := 0
33+ for _, o := range ops {
34+ size += opSize(o)
35+ }
36+ if want := recordFrame + recordMinBody + size; len(enc) != want {
37+ t.Fatalf("encoded size = %d, want %d", len(enc), want)
38+ }
39+ seq, got, n, err := decodeRecord(append(enc, 0xff, 0xee), nil)
40+ if err != nil {
41+ t.Fatal(err)
42+ }
43+ if seq != 42 || n != len(enc) || !opsEqual(got, ops) {
44+ t.Fatalf("got seq=%d n=%d ops=%v", seq, n, got)
45+ }
46+ owned := ownOps(got)
47+ clear(enc)
48+ if !opsEqual(owned, ops) {
49+ t.Fatal("ownOps aliases the source buffer")
50+ }
51+}
52+
53+func TestLogHeader(t *testing.T) {
54+ h := appendLogHeader(nil, 0x0102030405)
55+ if len(h) != logHeaderSize {
56+ t.Fatalf("header size %d", len(h))
57+ }
58+ if base, err := decodeLogHeader(h); err != nil || base != 0x0102030405 {
59+ t.Fatalf("base=%d err=%v", base, err)
60+ }
61+ for i := range h {
62+ bad := slices.Clone(h)
63+ bad[i] ^= 1
64+ if _, err := decodeLogHeader(bad); !errors.Is(err, ErrCorrupt) {
65+ t.Fatalf("flip byte %d: err=%v", i, err)
66+ }
67+ }
68+ if _, err := decodeLogHeader(h[:logHeaderSize-1]); !errors.Is(err, ErrCorrupt) {
69+ t.Fatalf("short header: err=%v", err)
70+ }
71+}
72+
73+// rawRecord frames payload with a correct length and checksum.
74+func rawRecord(payload []byte) []byte {
75+ b := binary.LittleEndian.AppendUint32(nil, uint32(len(payload)))
76+ b = binary.LittleEndian.AppendUint32(b, crc32.Checksum(payload, castagnoli))
77+ return append(b, payload...)
78+}
79+
80+func rawPayload(seq uint64, count uint32, body ...byte) []byte {
81+ p := binary.LittleEndian.AppendUint64(nil, seq)
82+ p = binary.LittleEndian.AppendUint32(p, count)
83+ return append(p, body...)
84+}
85+
86+func TestDecodeRecordUnreadable(t *testing.T) {
87+ good := appendRecord(nil, 1, []op{{kind: opPut, key: []byte("k"), value: []byte("v")}})
88+ badCRC := slices.Clone(good)
89+ badCRC[len(badCRC)-1] ^= 1
90+ tooBig := binary.LittleEndian.AppendUint32(nil, maxRecord+1)
91+ tooBig = append(tooBig, make([]byte, 16)...)
92+ cases := map[string][]byte{
93+ "empty": nil,
94+ "short frame": good[:7],
95+ "short payload": good[:len(good)-1],
96+ "bad crc": badCRC,
97+ "len too small": rawRecord(make([]byte, 11)),
98+ "len too big": tooBig,
99+ "zero count": rawRecord(rawPayload(1, 0)),
100+ "count too big": rawRecord(rawPayload(1, 2, opPut, 1, 'k', 1, 'v')),
101+ "count too small": rawRecord(rawPayload(1, 1, opPut, 1, 'k', 1, 'v', opDelete, 1, 'k')),
102+ "bad kind": rawRecord(rawPayload(1, 1, 3, 1, 'k')),
103+ "empty key": rawRecord(rawPayload(1, 1, opDelete, 0, 0)),
104+ "key past end": rawRecord(rawPayload(1, 1, opDelete, 9, 'k')),
105+ "value past end": rawRecord(rawPayload(1, 1, opPut, 1, 'k', 5, 'v')),
106+ "trailing byte": rawRecord(rawPayload(1, 1, opDelete, 1, 'k', 0)),
107+ "huge count": rawRecord(rawPayload(1, 0xffffffff, opDelete, 1, 'k')),
108+ }
109+ for name, b := range cases {
110+ t.Run(name, func(t *testing.T) {
111+ if _, _, _, err := decodeRecord(b, nil); !errors.Is(err, errUnreadable) {
112+ t.Fatalf("err = %v, want errUnreadable", err)
113+ }
114+ })
115+ }
116+ if _, _, _, err := decodeRecord(rawRecord(rawPayload(1, 1, opDelete, 1, 'k')), nil); err != nil {
117+ t.Fatalf("control record: %v", err)
118+ }
119+}
120+
121+type snapEntry struct {
122+ key, value []byte
123+ version uint64
124+}
125+
126+func iterOf(entries []snapEntry) entryIter {
127+ return func(yield func(key, value []byte, version uint64) bool) {
128+ for _, e := range entries {
129+ if !yield(e.key, e.value, e.version) {
130+ return
131+ }
132+ }
133+ }
134+}
135+
136+func encodeSnapshot(t testing.TB, seq uint64, entries []snapEntry) []byte {
137+ t.Helper()
138+ var buf bytes.Buffer
139+ if err := writeSnapshot(&buf, seq, uint64(len(entries)), iterOf(entries)); err != nil {
140+ t.Fatal(err)
141+ }
142+ return buf.Bytes()
143+}
144+
145+func TestSnapshotRoundTrip(t *testing.T) {
146+ entries := []snapEntry{
147+ {[]byte("a"), []byte("1"), 3},
148+ {[]byte("b"), nil, 7},
149+ {[]byte("c"), bytes.Repeat([]byte("x"), 1000), 1},
150+ }
151+ for _, in := range [][]snapEntry{nil, entries} {
152+ data := encodeSnapshot(t, 9, in)
153+ view, err := openSnapshot(data)
154+ if err != nil {
155+ t.Fatal(err)
156+ }
157+ if view.seq != 9 || view.count != uint64(len(in)) {
158+ t.Fatalf("view = %+v", view)
159+ }
160+ var got []snapEntry
161+ view.each(func(k, v []byte, ver uint64) { got = append(got, snapEntry{k, v, ver}) })
162+ if len(got) != len(in) {
163+ t.Fatalf("got %d entries, want %d", len(got), len(in))
164+ }
165+ for i := range in {
166+ if !bytes.Equal(got[i].key, in[i].key) || !bytes.Equal(got[i].value, in[i].value) || got[i].version != in[i].version {
167+ t.Fatalf("entry %d = %+v, want %+v", i, got[i], in[i])
168+ }
169+ }
170+ }
171+}
172+
173+func TestSnapshotCorrupt(t *testing.T) {
174+ entries := []snapEntry{{[]byte("a"), []byte("1"), 1}, {[]byte("b"), []byte("2"), 2}}
175+ data := encodeSnapshot(t, 2, entries)
176+ for i := range data {
177+ bad := slices.Clone(data)
178+ bad[i] ^= 0x40
179+ if _, err := openSnapshot(bad); !errors.Is(err, ErrCorrupt) {
180+ t.Fatalf("flip byte %d: err=%v", i, err)
181+ }
182+ }
183+ for _, n := range []int{0, 1, snapHeaderSize, len(data) - 1} {
184+ if _, err := openSnapshot(data[:n]); !errors.Is(err, ErrCorrupt) {
185+ t.Fatalf("truncate to %d: err=%v", n, err)
186+ }
187+ }
188+ if _, err := openSnapshot(append(slices.Clone(data), 0)); !errors.Is(err, ErrCorrupt) {
189+ t.Fatalf("trailing byte: err=%v", err)
190+ }
191+}
192+
193+// forgeSnapshot builds a snapshot with valid checksums around arbitrary body
194+// bytes, to reach the structural checks.
195+func forgeSnapshot(seq, count uint64, body []byte) []byte {
196+ b := append([]byte(snapshotMagic), 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
197+ binary.LittleEndian.PutUint64(b[8:], seq)
198+ binary.LittleEndian.PutUint64(b[16:], count)
199+ b = binary.LittleEndian.AppendUint32(b, crc32.Checksum(b, castagnoli))
200+ b = append(b, body...)
201+ return binary.LittleEndian.AppendUint32(b, crc32.Checksum(body, castagnoli))
202+}
203+
204+func TestSnapshotStructure(t *testing.T) {
205+ entry := func(k string, ver uint64, v string) []byte {
206+ b := binary.AppendUvarint(nil, uint64(len(k)))
207+ b = append(b, k...)
208+ b = binary.AppendUvarint(b, ver)
209+ b = binary.AppendUvarint(b, uint64(len(v)))
210+ return append(b, v...)
211+ }
212+ cat := func(parts ...[]byte) []byte { return bytes.Join(parts, nil) }
213+ cases := map[string][]byte{
214+ "not ascending": forgeSnapshot(5, 2, cat(entry("b", 1, ""), entry("a", 1, ""))),
215+ "duplicate key": forgeSnapshot(5, 2, cat(entry("a", 1, ""), entry("a", 2, ""))),
216+ "version > seq": forgeSnapshot(5, 1, entry("a", 6, "")),
217+ "too few": forgeSnapshot(5, 2, entry("a", 1, "")),
218+ "too many": forgeSnapshot(5, 1, cat(entry("a", 1, ""), entry("b", 1, ""))),
219+ "empty key": forgeSnapshot(5, 1, entry("", 1, "")),
220+ "huge count": forgeSnapshot(5, 1<<62, entry("a", 1, "")),
221+ "value past end": forgeSnapshot(5, 1, []byte{1, 'a', 1, 9, 'v'}),
222+ }
223+ for name, b := range cases {
224+ t.Run(name, func(t *testing.T) {
225+ if _, err := openSnapshot(b); !errors.Is(err, ErrCorrupt) {
226+ t.Fatalf("err = %v, want ErrCorrupt", err)
227+ }
228+ })
229+ }
230+ if _, err := openSnapshot(forgeSnapshot(5, 2, cat(entry("a", 1, ""), entry("b", 5, "x")))); err != nil {
231+ t.Fatalf("control snapshot: %v", err)
232+ }
233+}
234+
235+func TestWriteSnapshotRejectsBadInput(t *testing.T) {
236+ long := bytes.Repeat([]byte("k"), maxKeyLen+1)
237+ cases := map[string]struct {
238+ count uint64
239+ entries []snapEntry
240+ }{
241+ "unsorted": {2, []snapEntry{{[]byte("b"), nil, 1}, {[]byte("a"), nil, 1}}},
242+ "duplicate": {2, []snapEntry{{[]byte("a"), nil, 1}, {[]byte("a"), nil, 1}}},
243+ "empty key": {1, []snapEntry{{nil, nil, 1}}},
244+ "long key": {1, []snapEntry{{long, nil, 1}}},
245+ "version": {1, []snapEntry{{[]byte("a"), nil, 4}}},
246+ "too few": {2, []snapEntry{{[]byte("a"), nil, 1}}},
247+ "too many": {1, []snapEntry{{[]byte("a"), nil, 1}, {[]byte("b"), nil, 1}}},
248+ "unsorted mid": {3, []snapEntry{{[]byte("a"), nil, 1}, {[]byte("c"), nil, 1}, {[]byte("b"), nil, 1}}},
249+ }
250+ for name, c := range cases {
251+ t.Run(name, func(t *testing.T) {
252+ var buf bytes.Buffer
253+ if err := writeSnapshot(&buf, 3, c.count, iterOf(c.entries)); err == nil {
254+ t.Fatal("want error")
255+ }
256+ })
257+ }
258+}
259+
260+func FuzzDecodeRecord(f *testing.F) {
261+ f.Add(appendRecord(nil, 1, sampleOps()))
262+ f.Add(appendRecord(nil, 1<<40, []op{{kind: opDelete, key: []byte("k")}}))
263+ f.Add(rawRecord(rawPayload(1, 1, opDelete, 1, 'k')))
264+ f.Add([]byte{})
265+ f.Add(make([]byte, 64))
266+ f.Add(appendLogHeader(nil, 7))
267+ f.Add(encodeSnapshot(f, 3, []snapEntry{{[]byte("a"), []byte("1"), 2}}))
268+ f.Fuzz(func(t *testing.T, b []byte) {
269+ // Decoders must not panic, whatever the input.
270+ _, _ = decodeLogHeader(b)
271+ if view, err := openSnapshot(b); err == nil {
272+ view.each(func(k, v []byte, ver uint64) {})
273+ }
274+ peekRecord(b)
275+ seq, ops, n, err := decodeRecord(b, nil)
276+ if err != nil {
277+ return
278+ }
279+ // A readable record survives re-encoding (varints may be
280+ // non-canonical in the input, so the bytes can differ).
281+ if n > len(b) || len(ops) == 0 {
282+ t.Fatalf("n=%d len=%d ops=%d", n, len(b), len(ops))
283+ }
284+ enc := appendRecord(nil, seq, ops)
285+ seq2, ops2, n2, err := decodeRecord(enc, nil)
286+ if err != nil || seq2 != seq || n2 != len(enc) || !opsEqual(ops2, ops) {
287+ t.Fatalf("round trip failed: err=%v", err)
288+ }
289+ })
290+}
@@ -0,0 +1,290 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bytes"
7+ "encoding/binary"
8+ "errors"
9+ "hash/crc32"
10+ "slices"
11+ "testing"
12+)
13+
14+func sampleOps() []op {
15+ return []op{
16+ {kind: opPut, key: []byte("a"), value: []byte("1")},
17+ {kind: opDelete, key: []byte("b")},
18+ {kind: opPut, key: []byte("c"), value: nil},
19+ {kind: opPut, key: bytes.Repeat([]byte("k"), 200), value: bytes.Repeat([]byte("v"), 300)},
20+ }
21+}
22+
23+func opsEqual(a, b []op) bool {
24+ return slices.EqualFunc(a, b, func(x, y op) bool {
25+ return x.kind == y.kind && bytes.Equal(x.key, y.key) && bytes.Equal(x.value, y.value)
26+ })
27+}
28+
29+func TestRecordRoundTrip(t *testing.T) {
30+ ops := sampleOps()
31+ enc := appendRecord([]byte("prefix"), 42, ops)[6:]
32+ size := 0
33+ for _, o := range ops {
34+ size += opSize(o)
35+ }
36+ if want := recordFrame + recordMinBody + size; len(enc) != want {
37+ t.Fatalf("encoded size = %d, want %d", len(enc), want)
38+ }
39+ seq, got, n, err := decodeRecord(append(enc, 0xff, 0xee), nil)
40+ if err != nil {
41+ t.Fatal(err)
42+ }
43+ if seq != 42 || n != len(enc) || !opsEqual(got, ops) {
44+ t.Fatalf("got seq=%d n=%d ops=%v", seq, n, got)
45+ }
46+ owned := ownOps(got)
47+ clear(enc)
48+ if !opsEqual(owned, ops) {
49+ t.Fatal("ownOps aliases the source buffer")
50+ }
51+}
52+
53+func TestLogHeader(t *testing.T) {
54+ h := appendLogHeader(nil, 0x0102030405)
55+ if len(h) != logHeaderSize {
56+ t.Fatalf("header size %d", len(h))
57+ }
58+ if base, err := decodeLogHeader(h); err != nil || base != 0x0102030405 {
59+ t.Fatalf("base=%d err=%v", base, err)
60+ }
61+ for i := range h {
62+ bad := slices.Clone(h)
63+ bad[i] ^= 1
64+ if _, err := decodeLogHeader(bad); !errors.Is(err, ErrCorrupt) {
65+ t.Fatalf("flip byte %d: err=%v", i, err)
66+ }
67+ }
68+ if _, err := decodeLogHeader(h[:logHeaderSize-1]); !errors.Is(err, ErrCorrupt) {
69+ t.Fatalf("short header: err=%v", err)
70+ }
71+}
72+
73+// rawRecord frames payload with a correct length and checksum.
74+func rawRecord(payload []byte) []byte {
75+ b := binary.LittleEndian.AppendUint32(nil, uint32(len(payload)))
76+ b = binary.LittleEndian.AppendUint32(b, crc32.Checksum(payload, castagnoli))
77+ return append(b, payload...)
78+}
79+
80+func rawPayload(seq uint64, count uint32, body ...byte) []byte {
81+ p := binary.LittleEndian.AppendUint64(nil, seq)
82+ p = binary.LittleEndian.AppendUint32(p, count)
83+ return append(p, body...)
84+}
85+
86+func TestDecodeRecordUnreadable(t *testing.T) {
87+ good := appendRecord(nil, 1, []op{{kind: opPut, key: []byte("k"), value: []byte("v")}})
88+ badCRC := slices.Clone(good)
89+ badCRC[len(badCRC)-1] ^= 1
90+ tooBig := binary.LittleEndian.AppendUint32(nil, maxRecord+1)
91+ tooBig = append(tooBig, make([]byte, 16)...)
92+ cases := map[string][]byte{
93+ "empty": nil,
94+ "short frame": good[:7],
95+ "short payload": good[:len(good)-1],
96+ "bad crc": badCRC,
97+ "len too small": rawRecord(make([]byte, 11)),
98+ "len too big": tooBig,
99+ "zero count": rawRecord(rawPayload(1, 0)),
100+ "count too big": rawRecord(rawPayload(1, 2, opPut, 1, 'k', 1, 'v')),
101+ "count too small": rawRecord(rawPayload(1, 1, opPut, 1, 'k', 1, 'v', opDelete, 1, 'k')),
102+ "bad kind": rawRecord(rawPayload(1, 1, 3, 1, 'k')),
103+ "empty key": rawRecord(rawPayload(1, 1, opDelete, 0, 0)),
104+ "key past end": rawRecord(rawPayload(1, 1, opDelete, 9, 'k')),
105+ "value past end": rawRecord(rawPayload(1, 1, opPut, 1, 'k', 5, 'v')),
106+ "trailing byte": rawRecord(rawPayload(1, 1, opDelete, 1, 'k', 0)),
107+ "huge count": rawRecord(rawPayload(1, 0xffffffff, opDelete, 1, 'k')),
108+ }
109+ for name, b := range cases {
110+ t.Run(name, func(t *testing.T) {
111+ if _, _, _, err := decodeRecord(b, nil); !errors.Is(err, errUnreadable) {
112+ t.Fatalf("err = %v, want errUnreadable", err)
113+ }
114+ })
115+ }
116+ if _, _, _, err := decodeRecord(rawRecord(rawPayload(1, 1, opDelete, 1, 'k')), nil); err != nil {
117+ t.Fatalf("control record: %v", err)
118+ }
119+}
120+
121+type snapEntry struct {
122+ key, value []byte
123+ version uint64
124+}
125+
126+func iterOf(entries []snapEntry) entryIter {
127+ return func(yield func(key, value []byte, version uint64) bool) {
128+ for _, e := range entries {
129+ if !yield(e.key, e.value, e.version) {
130+ return
131+ }
132+ }
133+ }
134+}
135+
136+func encodeSnapshot(t testing.TB, seq uint64, entries []snapEntry) []byte {
137+ t.Helper()
138+ var buf bytes.Buffer
139+ if err := writeSnapshot(&buf, seq, uint64(len(entries)), iterOf(entries)); err != nil {
140+ t.Fatal(err)
141+ }
142+ return buf.Bytes()
143+}
144+
145+func TestSnapshotRoundTrip(t *testing.T) {
146+ entries := []snapEntry{
147+ {[]byte("a"), []byte("1"), 3},
148+ {[]byte("b"), nil, 7},
149+ {[]byte("c"), bytes.Repeat([]byte("x"), 1000), 1},
150+ }
151+ for _, in := range [][]snapEntry{nil, entries} {
152+ data := encodeSnapshot(t, 9, in)
153+ view, err := openSnapshot(data)
154+ if err != nil {
155+ t.Fatal(err)
156+ }
157+ if view.seq != 9 || view.count != uint64(len(in)) {
158+ t.Fatalf("view = %+v", view)
159+ }
160+ var got []snapEntry
161+ view.each(func(k, v []byte, ver uint64) { got = append(got, snapEntry{k, v, ver}) })
162+ if len(got) != len(in) {
163+ t.Fatalf("got %d entries, want %d", len(got), len(in))
164+ }
165+ for i := range in {
166+ if !bytes.Equal(got[i].key, in[i].key) || !bytes.Equal(got[i].value, in[i].value) || got[i].version != in[i].version {
167+ t.Fatalf("entry %d = %+v, want %+v", i, got[i], in[i])
168+ }
169+ }
170+ }
171+}
172+
173+func TestSnapshotCorrupt(t *testing.T) {
174+ entries := []snapEntry{{[]byte("a"), []byte("1"), 1}, {[]byte("b"), []byte("2"), 2}}
175+ data := encodeSnapshot(t, 2, entries)
176+ for i := range data {
177+ bad := slices.Clone(data)
178+ bad[i] ^= 0x40
179+ if _, err := openSnapshot(bad); !errors.Is(err, ErrCorrupt) {
180+ t.Fatalf("flip byte %d: err=%v", i, err)
181+ }
182+ }
183+ for _, n := range []int{0, 1, snapHeaderSize, len(data) - 1} {
184+ if _, err := openSnapshot(data[:n]); !errors.Is(err, ErrCorrupt) {
185+ t.Fatalf("truncate to %d: err=%v", n, err)
186+ }
187+ }
188+ if _, err := openSnapshot(append(slices.Clone(data), 0)); !errors.Is(err, ErrCorrupt) {
189+ t.Fatalf("trailing byte: err=%v", err)
190+ }
191+}
192+
193+// forgeSnapshot builds a snapshot with valid checksums around arbitrary body
194+// bytes, to reach the structural checks.
195+func forgeSnapshot(seq, count uint64, body []byte) []byte {
196+ b := append([]byte(snapshotMagic), 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
197+ binary.LittleEndian.PutUint64(b[8:], seq)
198+ binary.LittleEndian.PutUint64(b[16:], count)
199+ b = binary.LittleEndian.AppendUint32(b, crc32.Checksum(b, castagnoli))
200+ b = append(b, body...)
201+ return binary.LittleEndian.AppendUint32(b, crc32.Checksum(body, castagnoli))
202+}
203+
204+func TestSnapshotStructure(t *testing.T) {
205+ entry := func(k string, ver uint64, v string) []byte {
206+ b := binary.AppendUvarint(nil, uint64(len(k)))
207+ b = append(b, k...)
208+ b = binary.AppendUvarint(b, ver)
209+ b = binary.AppendUvarint(b, uint64(len(v)))
210+ return append(b, v...)
211+ }
212+ cat := func(parts ...[]byte) []byte { return bytes.Join(parts, nil) }
213+ cases := map[string][]byte{
214+ "not ascending": forgeSnapshot(5, 2, cat(entry("b", 1, ""), entry("a", 1, ""))),
215+ "duplicate key": forgeSnapshot(5, 2, cat(entry("a", 1, ""), entry("a", 2, ""))),
216+ "version > seq": forgeSnapshot(5, 1, entry("a", 6, "")),
217+ "too few": forgeSnapshot(5, 2, entry("a", 1, "")),
218+ "too many": forgeSnapshot(5, 1, cat(entry("a", 1, ""), entry("b", 1, ""))),
219+ "empty key": forgeSnapshot(5, 1, entry("", 1, "")),
220+ "huge count": forgeSnapshot(5, 1<<62, entry("a", 1, "")),
221+ "value past end": forgeSnapshot(5, 1, []byte{1, 'a', 1, 9, 'v'}),
222+ }
223+ for name, b := range cases {
224+ t.Run(name, func(t *testing.T) {
225+ if _, err := openSnapshot(b); !errors.Is(err, ErrCorrupt) {
226+ t.Fatalf("err = %v, want ErrCorrupt", err)
227+ }
228+ })
229+ }
230+ if _, err := openSnapshot(forgeSnapshot(5, 2, cat(entry("a", 1, ""), entry("b", 5, "x")))); err != nil {
231+ t.Fatalf("control snapshot: %v", err)
232+ }
233+}
234+
235+func TestWriteSnapshotRejectsBadInput(t *testing.T) {
236+ long := bytes.Repeat([]byte("k"), maxKeyLen+1)
237+ cases := map[string]struct {
238+ count uint64
239+ entries []snapEntry
240+ }{
241+ "unsorted": {2, []snapEntry{{[]byte("b"), nil, 1}, {[]byte("a"), nil, 1}}},
242+ "duplicate": {2, []snapEntry{{[]byte("a"), nil, 1}, {[]byte("a"), nil, 1}}},
243+ "empty key": {1, []snapEntry{{nil, nil, 1}}},
244+ "long key": {1, []snapEntry{{long, nil, 1}}},
245+ "version": {1, []snapEntry{{[]byte("a"), nil, 4}}},
246+ "too few": {2, []snapEntry{{[]byte("a"), nil, 1}}},
247+ "too many": {1, []snapEntry{{[]byte("a"), nil, 1}, {[]byte("b"), nil, 1}}},
248+ "unsorted mid": {3, []snapEntry{{[]byte("a"), nil, 1}, {[]byte("c"), nil, 1}, {[]byte("b"), nil, 1}}},
249+ }
250+ for name, c := range cases {
251+ t.Run(name, func(t *testing.T) {
252+ var buf bytes.Buffer
253+ if err := writeSnapshot(&buf, 3, c.count, iterOf(c.entries)); err == nil {
254+ t.Fatal("want error")
255+ }
256+ })
257+ }
258+}
259+
260+func FuzzDecodeRecord(f *testing.F) {
261+ f.Add(appendRecord(nil, 1, sampleOps()))
262+ f.Add(appendRecord(nil, 1<<40, []op{{kind: opDelete, key: []byte("k")}}))
263+ f.Add(rawRecord(rawPayload(1, 1, opDelete, 1, 'k')))
264+ f.Add([]byte{})
265+ f.Add(make([]byte, 64))
266+ f.Add(appendLogHeader(nil, 7))
267+ f.Add(encodeSnapshot(f, 3, []snapEntry{{[]byte("a"), []byte("1"), 2}}))
268+ f.Fuzz(func(t *testing.T, b []byte) {
269+ // Decoders must not panic, whatever the input.
270+ _, _ = decodeLogHeader(b)
271+ if view, err := openSnapshot(b); err == nil {
272+ view.each(func(k, v []byte, ver uint64) {})
273+ }
274+ peekRecord(b)
275+ seq, ops, n, err := decodeRecord(b, nil)
276+ if err != nil {
277+ return
278+ }
279+ // A readable record survives re-encoding (varints may be
280+ // non-canonical in the input, so the bytes can differ).
281+ if n > len(b) || len(ops) == 0 {
282+ t.Fatalf("n=%d len=%d ops=%d", n, len(b), len(ops))
283+ }
284+ enc := appendRecord(nil, seq, ops)
285+ seq2, ops2, n2, err := decodeRecord(enc, nil)
286+ if err != nil || seq2 != seq || n2 != len(enc) || !opsEqual(ops2, ops) {
287+ t.Fatalf("round trip failed: err=%v", err)
288+ }
289+ })
290+}

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 1

Tags 2

↑↓ moveenter openesc close