1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
6 "bytes"
7 "encoding/binary"
8 "errors"
9 "hash/crc32"
10 "slices"
11 "testing"
12)
13
14func 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
23func 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
29func TestRecordRoundTrip(t *testing.T) {
30 ops := sampleOps()
31 second := []op{{kind: opDelete, key: []byte("z")}}
32 enc := appendRecord([]byte("prefix"), 42, ops, second, ops[:1])[6:]
33 size := 0
34 for _, b := range [][]op{ops, second, ops[:1]} {
35 size += batchHeader
36 for _, o := range b {
37 size += opSize(o)
38 }
39 }
40 if want := recordFrame + recordHead + size; len(enc) != want {
41 t.Fatalf("encoded size = %d, want %d", len(enc), want)
42 }
43 seq, got, n, err := decodeRecord(append(enc, 0xff, 0xee))
44 if err != nil {
45 t.Fatal(err)
46 }
47 if seq != 42 || n != len(enc) || len(got) != 3 ||
48 !opsEqual(got[0], ops) || !opsEqual(got[1], second) || !opsEqual(got[2], ops[:1]) {
49 t.Fatalf("got seq=%d n=%d batches=%v", seq, n, got)
50 }
51 owned := ownOps(got[0])
52 clear(enc)
53 if !opsEqual(owned, ops) {
54 t.Fatal("ownOps aliases the source buffer")
55 }
56}
57
58// TestRecordBuilder checks that beginRecord/appendBatch/finishRecord produce
59// the same bytes as appendRecord.
60func TestRecordBuilder(t *testing.T) {
61 ops := sampleOps()
62 rec := beginRecord(nil, 7)
63 rec = appendBatch(rec, ops)
64 rec = appendBatch(rec, ops[:2])
65 finishRecord(rec, 2)
66 if want := appendRecord(nil, 7, ops, ops[:2]); !bytes.Equal(rec, want) {
67 t.Fatal("builder output differs from appendRecord")
68 }
69}
70
71func TestLogHeader(t *testing.T) {
72 h := appendLogHeader(nil, 0x0102030405)
73 if len(h) != logHeaderSize {
74 t.Fatalf("header size %d", len(h))
75 }
76 if base, err := decodeLogHeader(h); err != nil || base != 0x0102030405 {
77 t.Fatalf("base=%d err=%v", base, err)
78 }
79 for i := range h {
80 bad := slices.Clone(h)
81 bad[i] ^= 1
82 if _, err := decodeLogHeader(bad); !errors.Is(err, ErrCorrupt) {
83 t.Fatalf("flip byte %d: err=%v", i, err)
84 }
85 }
86 if _, err := decodeLogHeader(h[:logHeaderSize-1]); !errors.Is(err, ErrCorrupt) {
87 t.Fatalf("short header: err=%v", err)
88 }
89}
90
91// rawRecord frames payload with a correct length and checksum.
92func rawRecord(payload []byte) []byte {
93 b := binary.LittleEndian.AppendUint32(nil, uint32(len(payload)))
94 b = binary.LittleEndian.AppendUint32(b, crc32.Checksum(payload, castagnoli))
95 return append(b, payload...)
96}
97
98// rawPayload builds a payload from the seq, the batch count and raw batch
99// bytes.
100func rawPayload(seq uint64, batches uint32, body ...byte) []byte {
101 p := binary.LittleEndian.AppendUint64(nil, seq)
102 p = binary.LittleEndian.AppendUint32(p, batches)
103 return append(p, body...)
104}
105
106// rawBatch encodes a batch header and raw op bytes.
107func rawBatch(count uint32, ops ...byte) []byte {
108 return append(binary.LittleEndian.AppendUint32(nil, count), ops...)
109}
110
111func cat(parts ...[]byte) []byte { return bytes.Join(parts, nil) }
112
113func TestDecodeRecordUnreadable(t *testing.T) {
114 one := []op{{kind: opPut, key: []byte("k"), value: []byte("v")}}
115 good := appendRecord(nil, 1, one)
116 badCRC := slices.Clone(good)
117 badCRC[len(badCRC)-1] ^= 1
118 tooBig := binary.LittleEndian.AppendUint32(nil, maxRecord+1)
119 tooBig = append(tooBig, make([]byte, 16)...)
120 put := rawBatch(1, opPut, 1, 'k', 1, 'v')
121 del := rawBatch(1, opDelete, 1, 'k')
122 cases := map[string][]byte{
123 "empty": nil,
124 "short frame": good[:7],
125 "short payload": good[:len(good)-1],
126 "bad crc": badCRC,
127 "len too small": rawRecord(make([]byte, 15)),
128 "len too big": tooBig,
129 "zero batches": rawRecord(rawPayload(1, 0, put...)),
130 "batches too big": rawRecord(rawPayload(1, 2, put...)),
131 "batches too few": rawRecord(rawPayload(1, 1, cat(put, del)...)),
132 "zero ops": rawRecord(rawPayload(1, 1, rawBatch(0, opDelete, 1, 'k')...)),
133 "zero ops 2nd": rawRecord(rawPayload(1, 2, cat(put, rawBatch(0))...)),
134 "ops too many": rawRecord(rawPayload(1, 1, rawBatch(2, opPut, 1, 'k', 1, 'v')...)),
135 "ops too few": rawRecord(rawPayload(1, 1, rawBatch(1, opPut, 1, 'k', 1, 'v', opDelete, 1, 'k')...)),
136 "bad kind": rawRecord(rawPayload(1, 1, rawBatch(1, 3, 1, 'k')...)),
137 "empty key": rawRecord(rawPayload(1, 1, rawBatch(1, opDelete, 0, 0)...)),
138 "key past end": rawRecord(rawPayload(1, 1, rawBatch(1, opDelete, 9, 'k')...)),
139 "value past end": rawRecord(rawPayload(1, 1, rawBatch(1, opPut, 1, 'k', 5, 'v')...)),
140 "trailing byte": rawRecord(rawPayload(1, 1, cat(del, []byte{0})...)),
141 "huge batches": rawRecord(rawPayload(1, 0xffffffff, del...)),
142 "huge op count": rawRecord(rawPayload(1, 1, rawBatch(0xffffffff, opDelete, 1, 'k')...)),
143 "truncated batch": rawRecord(rawPayload(1, 2, cat(del, []byte{1, 0})...)),
144 "second batch bad": rawRecord(rawPayload(1, 2, cat(del, rawBatch(1, 9, 1, 'k'))...)),
145 }
146 for name, b := range cases {
147 t.Run(name, func(t *testing.T) {
148 if _, _, _, err := decodeRecord(b); !errors.Is(err, errUnreadable) {
149 t.Fatalf("err = %v, want errUnreadable", err)
150 }
151 })
152 }
153 if _, _, _, err := decodeRecord(rawRecord(rawPayload(1, 1, del...))); err != nil {
154 t.Fatalf("control record: %v", err)
155 }
156}
157
158type snapEntry struct {
159 key, value []byte
160 version uint64
161}
162
163func iterOf(entries []snapEntry) entryIter {
164 return func(yield func(key, value []byte, version uint64) bool) {
165 for _, e := range entries {
166 if !yield(e.key, e.value, e.version) {
167 return
168 }
169 }
170 }
171}
172
173func encodeSnapshot(t testing.TB, seq uint64, entries []snapEntry) []byte {
174 t.Helper()
175 var buf bytes.Buffer
176 if err := writeSnapshot(&buf, seq, uint64(len(entries)), iterOf(entries)); err != nil {
177 t.Fatal(err)
178 }
179 return buf.Bytes()
180}
181
182func TestSnapshotRoundTrip(t *testing.T) {
183 entries := []snapEntry{
184 {[]byte("a"), []byte("1"), 3},
185 {[]byte("b"), nil, 7},
186 {[]byte("c"), bytes.Repeat([]byte("x"), 1000), 1},
187 }
188 for _, in := range [][]snapEntry{nil, entries} {
189 data := encodeSnapshot(t, 9, in)
190 view, err := openSnapshot(data)
191 if err != nil {
192 t.Fatal(err)
193 }
194 if view.seq != 9 || view.count != uint64(len(in)) {
195 t.Fatalf("view = %+v", view)
196 }
197 var got []snapEntry
198 view.each(func(k, v []byte, ver uint64) { got = append(got, snapEntry{k, v, ver}) })
199 if len(got) != len(in) {
200 t.Fatalf("got %d entries, want %d", len(got), len(in))
201 }
202 for i := range in {
203 if !bytes.Equal(got[i].key, in[i].key) || !bytes.Equal(got[i].value, in[i].value) || got[i].version != in[i].version {
204 t.Fatalf("entry %d = %+v, want %+v", i, got[i], in[i])
205 }
206 }
207 }
208}
209
210func TestSnapshotCorrupt(t *testing.T) {
211 entries := []snapEntry{{[]byte("a"), []byte("1"), 1}, {[]byte("b"), []byte("2"), 2}}
212 data := encodeSnapshot(t, 2, entries)
213 for i := range data {
214 bad := slices.Clone(data)
215 bad[i] ^= 0x40
216 if _, err := openSnapshot(bad); !errors.Is(err, ErrCorrupt) {
217 t.Fatalf("flip byte %d: err=%v", i, err)
218 }
219 }
220 for _, n := range []int{0, 1, snapHeaderSize, len(data) - 1} {
221 if _, err := openSnapshot(data[:n]); !errors.Is(err, ErrCorrupt) {
222 t.Fatalf("truncate to %d: err=%v", n, err)
223 }
224 }
225 if _, err := openSnapshot(append(slices.Clone(data), 0)); !errors.Is(err, ErrCorrupt) {
226 t.Fatalf("trailing byte: err=%v", err)
227 }
228}
229
230// forgeSnapshot builds a snapshot with valid checksums around arbitrary body
231// bytes, to reach the structural checks.
232func forgeSnapshot(seq, count uint64, body []byte) []byte {
233 b := append([]byte(snapshotMagic), 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0)
234 binary.LittleEndian.PutUint64(b[8:], seq)
235 binary.LittleEndian.PutUint64(b[16:], count)
236 b = binary.LittleEndian.AppendUint32(b, crc32.Checksum(b, castagnoli))
237 b = append(b, body...)
238 return binary.LittleEndian.AppendUint32(b, crc32.Checksum(body, castagnoli))
239}
240
241func TestSnapshotStructure(t *testing.T) {
242 entry := func(k string, ver uint64, v string) []byte {
243 b := binary.AppendUvarint(nil, uint64(len(k)))
244 b = append(b, k...)
245 b = binary.AppendUvarint(b, ver)
246 b = binary.AppendUvarint(b, uint64(len(v)))
247 return append(b, v...)
248 }
249 cases := map[string][]byte{
250 "not ascending": forgeSnapshot(5, 2, cat(entry("b", 1, ""), entry("a", 1, ""))),
251 "duplicate key": forgeSnapshot(5, 2, cat(entry("a", 1, ""), entry("a", 2, ""))),
252 "version > seq": forgeSnapshot(5, 1, entry("a", 6, "")),
253 "too few": forgeSnapshot(5, 2, entry("a", 1, "")),
254 "too many": forgeSnapshot(5, 1, cat(entry("a", 1, ""), entry("b", 1, ""))),
255 "empty key": forgeSnapshot(5, 1, entry("", 1, "")),
256 "huge count": forgeSnapshot(5, 1<<62, entry("a", 1, "")),
257 "value past end": forgeSnapshot(5, 1, []byte{1, 'a', 1, 9, 'v'}),
258 }
259 for name, b := range cases {
260 t.Run(name, func(t *testing.T) {
261 if _, err := openSnapshot(b); !errors.Is(err, ErrCorrupt) {
262 t.Fatalf("err = %v, want ErrCorrupt", err)
263 }
264 })
265 }
266 if _, err := openSnapshot(forgeSnapshot(5, 2, cat(entry("a", 1, ""), entry("b", 5, "x")))); err != nil {
267 t.Fatalf("control snapshot: %v", err)
268 }
269}
270
271func TestWriteSnapshotRejectsBadInput(t *testing.T) {
272 long := bytes.Repeat([]byte("k"), maxKeyLen+1)
273 cases := map[string]struct {
274 count uint64
275 entries []snapEntry
276 }{
277 "unsorted": {2, []snapEntry{{[]byte("b"), nil, 1}, {[]byte("a"), nil, 1}}},
278 "duplicate": {2, []snapEntry{{[]byte("a"), nil, 1}, {[]byte("a"), nil, 1}}},
279 "empty key": {1, []snapEntry{{nil, nil, 1}}},
280 "long key": {1, []snapEntry{{long, nil, 1}}},
281 "version": {1, []snapEntry{{[]byte("a"), nil, 4}}},
282 "too few": {2, []snapEntry{{[]byte("a"), nil, 1}}},
283 "too many": {1, []snapEntry{{[]byte("a"), nil, 1}, {[]byte("b"), nil, 1}}},
284 "unsorted mid": {3, []snapEntry{{[]byte("a"), nil, 1}, {[]byte("c"), nil, 1}, {[]byte("b"), nil, 1}}},
285 }
286 for name, c := range cases {
287 t.Run(name, func(t *testing.T) {
288 var buf bytes.Buffer
289 if err := writeSnapshot(&buf, 3, c.count, iterOf(c.entries)); err == nil {
290 t.Fatal("want error")
291 }
292 })
293 }
294}
295
296func FuzzDecodeRecord(f *testing.F) {
297 f.Add(appendRecord(nil, 1, sampleOps()))
298 f.Add(appendRecord(nil, 1, sampleOps(), sampleOps()[:2]))
299 f.Add(appendRecord(nil, 1<<40, []op{{kind: opDelete, key: []byte("k")}}))
300 f.Add(rawRecord(rawPayload(1, 1, rawBatch(1, opDelete, 1, 'k')...)))
301 f.Add([]byte{})
302 f.Add(make([]byte, 64))
303 f.Add(appendLogHeader(nil, 7))
304 f.Add(encodeSnapshot(f, 3, []snapEntry{{[]byte("a"), []byte("1"), 2}}))
305 f.Fuzz(func(t *testing.T, b []byte) {
306 // Decoders must not panic, whatever the input.
307 _, _ = decodeLogHeader(b)
308 if view, err := openSnapshot(b); err == nil {
309 view.each(func(k, v []byte, ver uint64) {})
310 }
311 peekRecord(b)
312 seq, batches, n, err := decodeRecord(b)
313 if err != nil {
314 return
315 }
316 // A readable record survives re-encoding (varints may be
317 // non-canonical in the input, so the bytes can differ).
318 if n > len(b) || len(batches) == 0 {
319 t.Fatalf("n=%d len=%d batches=%d", n, len(b), len(batches))
320 }
321 for _, ops := range batches {
322 if len(ops) == 0 {
323 t.Fatal("readable record with an empty batch")
324 }
325 }
326 enc := appendRecord(nil, seq, batches...)
327 seq2, batches2, n2, err := decodeRecord(enc)
328 if err != nil || seq2 != seq || n2 != len(enc) || len(batches2) != len(batches) {
329 t.Fatalf("round trip failed: err=%v", err)
330 }
331 for i := range batches {
332 if !opsEqual(batches[i], batches2[i]) {
333 t.Fatalf("batch %d differs after round trip", i)
334 }
335 }
336 })
337}