1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
6 "bytes"
7 "context"
8 "errors"
9 "fmt"
10 "io"
11 "path/filepath"
12 "slices"
13 "strings"
14 "testing"
15
16 "github.com/bnema/zerowrap"
17
18 "bnema.dev/ermite-kv/internal/fsys"
19 "bnema.dev/ermite-kv/internal/fsys/faultfs"
20)
21
22const testDir = "/db"
23
24// state is the replayed content of a database.
25type state struct {
26 m map[string]string
27 versions map[string]uint64
28 seqs []uint64 // seqs passed to apply, in order
29}
30
31func (s *state) load(k, v []byte, ver uint64) {
32 s.m[string(k)] = string(v)
33 s.versions[string(k)] = ver
34}
35
36func (s *state) apply(seq uint64, ops []op) {
37 s.seqs = append(s.seqs, seq)
38 for _, o := range ops {
39 switch o.kind {
40 case opPut:
41 s.m[string(o.key)] = string(o.value)
42 s.versions[string(o.key)] = seq
43 case opDelete:
44 delete(s.m, string(o.key))
45 delete(s.versions, string(o.key))
46 }
47 }
48}
49
50// noLog is a disabled logger.
51var noLog = zerowrap.FromCtx(context.Background())
52
53// bufLogger returns a JSON logger at debug level writing to buf.
54func bufLogger(buf *bytes.Buffer) zerowrap.Logger {
55 return zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: buf})
56}
57
58func recoverState(t *testing.T, fx fsys.FS, logger zerowrap.Logger) (*state, recovered, error) {
59 t.Helper()
60 s := &state{m: map[string]string{}, versions: map[string]uint64{}}
61 rec, err := recoverDir(fx, testDir, logger, s.load, s.apply)
62 return s, rec, err
63}
64
65// mustRecover recovers, checks the invariants of a successful recovery and
66// releases the lock and the log file.
67func mustRecover(t *testing.T, fx fsys.FS) (*state, recovered) {
68 t.Helper()
69 s, rec, err := recoverState(t, fx, noLog)
70 if err != nil {
71 t.Fatalf("recoverDir: %v", err)
72 }
73 size, err := rec.log.Size()
74 if err != nil || size != rec.logSize {
75 t.Fatalf("log size = %d (err %v), recovered.logSize = %d", size, err, rec.logSize)
76 }
77 t.Cleanup(func() { _ = rec.log.Close(); _ = rec.lock.Close() })
78 return s, rec
79}
80
81// release closes the log file and the lock of a recovery result.
82func release(rec recovered) {
83 _ = rec.log.Close()
84 _ = rec.lock.Close()
85}
86
87func putOp(k, v string) op { return op{kind: opPut, key: []byte(k), value: []byte(v)} }
88func delOp(k string) op { return op{kind: opDelete, key: []byte(k)} }
89
90// logBytes builds a log file: header for base, then one record per batch with
91// consecutive seqs starting at base+1.
92func logBytes(base uint64, batches ...[]op) []byte {
93 b := appendLogHeader(nil, base)
94 for i, ops := range batches {
95 b = appendRecord(b, base+1+uint64(i), ops)
96 }
97 return b
98}
99
100func put(fx *faultfs.FS, name string, data []byte) { fx.WriteFile(filepath.Join(testDir, name), data) }
101
102func snapshotBytes(t *testing.T, seq uint64, entries ...snapEntry) []byte {
103 t.Helper()
104 return encodeSnapshot(t, seq, entries)
105}
106
107func snapEnt(k, v string, ver uint64) snapEntry { return snapEntry{[]byte(k), []byte(v), ver} }
108
109func wantState(t *testing.T, s *state, m map[string]string) {
110 t.Helper()
111 if len(s.m) != len(m) {
112 t.Fatalf("state = %v, want %v", s.m, m)
113 }
114 for k, v := range m {
115 if got, ok := s.m[k]; !ok || got != v {
116 t.Fatalf("state = %v, want %v", s.m, m)
117 }
118 }
119}
120
121func TestRecoverEmptyDirCreatesLog(t *testing.T) {
122 fx := faultfs.New()
123 s, rec := mustRecover(t, fx)
124 wantState(t, s, nil)
125 if rec.snapSeq != 0 || rec.lastSeq != 0 || rec.logBase != 0 || rec.logSize != logHeaderSize {
126 t.Fatalf("rec = %+v", rec)
127 }
128 // The new log must survive a power loss: file creation rules.
129 _ = rec.log.Close()
130 _ = rec.lock.Close()
131 fx.PowerLoss(faultfs.PowerLossOptions{})
132 names, err := fx.ReadDir(testDir)
133 if err != nil {
134 t.Fatal(err)
135 }
136 if got := strings.Join(names, ","); got != "LOCK,"+logName(0) && got != logName(0) {
137 t.Fatalf("after power loss: %v", names)
138 }
139 data, _ := fx.ReadFile(filepath.Join(testDir, logName(0)))
140 if !bytes.Equal(data, appendLogHeader(nil, 0)) {
141 t.Fatalf("log content %x", data)
142 }
143}
144
145func TestRecoverLockedAndRelease(t *testing.T) {
146 fx := faultfs.New()
147 _, rec := mustRecover(t, fx)
148 if _, _, err := recoverState(t, fx, noLog); !errors.Is(err, ErrLocked) {
149 t.Fatalf("second recover: err = %v, want ErrLocked", err)
150 }
151 _ = rec.log.Close()
152 _ = rec.lock.Close()
153 _, rec2, err := recoverState(t, fx, noLog)
154 if err != nil {
155 t.Fatalf("recover after release: %v", err)
156 }
157 _ = rec2.log.Close()
158 _ = rec2.lock.Close()
159}
160
161func TestRecoverReleasesLockOnError(t *testing.T) {
162 fx := faultfs.New()
163 put(fx, logName(0), []byte("garbage"))
164 if _, _, err := recoverState(t, fx, noLog); !errors.Is(err, ErrCorrupt) {
165 t.Fatalf("err = %v, want ErrCorrupt", err)
166 }
167 l, err := fx.Lock(filepath.Join(testDir, lockName))
168 if err != nil {
169 t.Fatalf("lock still held after failure: %v", err)
170 }
171 _ = l.Close()
172}
173
174func TestRecoverCleanLog(t *testing.T) {
175 fx := faultfs.New()
176 data := logBytes(0,
177 []op{putOp("a", "1"), putOp("b", "2")},
178 []op{delOp("a"), putOp("c", "3")},
179 []op{putOp("b", "4")},
180 )
181 put(fx, logName(0), data)
182 s, rec := mustRecover(t, fx)
183 wantState(t, s, map[string]string{"b": "4", "c": "3"})
184 if got := fmt.Sprint(s.seqs); got != "[1 2 3]" {
185 t.Fatalf("applied seqs %s", got)
186 }
187 if s.versions["b"] != 3 || s.versions["c"] != 2 {
188 t.Fatalf("versions = %v", s.versions)
189 }
190 if rec.snapSeq != 0 || rec.lastSeq != 3 || rec.logBase != 0 || rec.logSize != int64(len(data)) {
191 t.Fatalf("rec = %+v", rec)
192 }
193 // Appends go to the end of the recovered log.
194 if _, err := rec.log.Write([]byte("x")); err != nil {
195 t.Fatal(err)
196 }
197 if size, _ := rec.log.Size(); size != int64(len(data))+1 {
198 t.Fatalf("size after append = %d", size)
199 }
200}
201
202func TestRecoverDeletesTmpFiles(t *testing.T) {
203 fx := faultfs.New()
204 put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
205 put(fx, snapshotName(5)+tmpSuffix, []byte("partial"))
206 put(fx, logName(5)+tmpSuffix, []byte("partial"))
207 s, _ := mustRecover(t, fx)
208 wantState(t, s, map[string]string{"a": "1"})
209 names, _ := fx.ReadDir(testDir)
210 for _, n := range names {
211 if strings.HasSuffix(n, tmpSuffix) {
212 t.Fatalf("tmp file %s left", n)
213 }
214 }
215 // Deletion is durable.
216 fx.PowerLoss(faultfs.PowerLossOptions{})
217 names, _ = fx.ReadDir(testDir)
218 for _, n := range names {
219 if strings.HasSuffix(n, tmpSuffix) {
220 t.Fatalf("tmp file %s back after power loss", n)
221 }
222 }
223}
224
225func TestRecoverTornTail(t *testing.T) {
226 good := logBytes(0, []op{putOp("a", "1")}, []op{putOp("b", "2")})
227 next := appendRecord(nil, 3, []op{putOp("c", "3")})
228 cases := map[string][]byte{
229 "short frame": next[:3],
230 "short record": next[:len(next)-2],
231 "zero-filled tail": make([]byte, 100),
232 "garbage": []byte("this is not a record at all, just junk bytes"),
233 "bad crc": append(append([]byte(nil), next[:len(next)-1]...), next[len(next)-1]^1),
234 "huge length": {0xff, 0xff, 0xff, 0x7f, 1, 2, 3, 4, 5, 6, 7, 8},
235 "one byte": {0x01},
236 }
237 for name, tail := range cases {
238 t.Run(name, func(t *testing.T) {
239 fx := faultfs.New()
240 put(fx, logName(0), append(append([]byte(nil), good...), tail...))
241 var logs bytes.Buffer
242 s, rec, err := recoverState(t, fx, bufLogger(&logs))
243 if err != nil {
244 t.Fatal(err)
245 }
246 defer release(rec)
247 wantState(t, s, map[string]string{"a": "1", "b": "2"})
248 if rec.lastSeq != 2 || rec.logSize != int64(len(good)) {
249 t.Fatalf("rec = %+v, want size %d", rec, len(good))
250 }
251 if got, _ := fx.ReadFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, good) {
252 t.Fatalf("log not truncated: %d bytes, want %d", len(got), len(good))
253 }
254 if got, _ := fx.DurableFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, good) {
255 t.Fatal("truncation not synced")
256 }
257 if !strings.Contains(logs.String(), "dropped torn log tail") {
258 t.Fatalf("no warning logged: %q", logs.String())
259 }
260 })
261 }
262}
263
264func TestRecoverTornTailOnlyHeaderSurvives(t *testing.T) {
265 fx := faultfs.New()
266 put(fx, logName(0), append(logBytes(0), make([]byte, 7)...))
267 s, rec := mustRecover(t, fx)
268 wantState(t, s, nil)
269 if rec.logSize != logHeaderSize {
270 t.Fatalf("logSize = %d", rec.logSize)
271 }
272}
273
274func TestRecoverDamageFollowedByValidRecord(t *testing.T) {
275 r1 := appendRecord(nil, 1, []op{putOp("a", "1")})
276 r2 := appendRecord(nil, 2, []op{putOp("b", "2")})
277 r3 := appendRecord(nil, 3, []op{putOp("c", "3")})
278 flip := func(b []byte, i int) []byte {
279 c := append([]byte(nil), b...)
280 c[i] ^= 0xff
281 return c
282 }
283 hdr := appendLogHeader(nil, 0)
284 cases := map[string][]byte{
285 "bad crc in middle record": cat(hdr, r1, flip(r2, len(r2)-1), r3),
286 "bad payload in first record": cat(hdr, flip(r1, 9), r2),
287 "bad checksum field": cat(hdr, r1, flip(r2, 4), r3),
288 "damaged record before last": cat(hdr, flip(r1, len(r1)-1), r2, r3),
289 "record with wrong count": cat(hdr, rawRecord(rawPayload(1, 5, rawBatch(1, opDelete, 1, 'k')...)), r2),
290 "corrupt but len looks intact": cat(hdr, r1, flip(r2, 12), r3),
291 }
292 for name, data := range cases {
293 t.Run(name, func(t *testing.T) {
294 fx := faultfs.New()
295 put(fx, logName(0), data)
296 _, _, err := recoverState(t, fx, noLog)
297 if !errors.Is(err, ErrCorrupt) {
298 t.Fatalf("err = %v, want ErrCorrupt", err)
299 }
300 // The file must be left untouched.
301 if got, _ := fx.ReadFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, data) {
302 t.Fatal("corrupt log was modified")
303 }
304 })
305 }
306}
307
308func TestRecoverDamageInOlderLog(t *testing.T) {
309 old := logBytes(0, []op{putOp("a", "1")}, []op{putOp("b", "2")})
310 cases := map[string][]byte{
311 "torn tail": old[:len(old)-3],
312 "bad crc": append(append([]byte(nil), old[:len(old)-1]...), old[len(old)-1]^1),
313 "zero tail": append(append([]byte(nil), old...), make([]byte, 30)...),
314 "extra bytes": append(append([]byte(nil), old...), 1, 2, 3),
315 }
316 for name, data := range cases {
317 t.Run(name, func(t *testing.T) {
318 fx := faultfs.New()
319 put(fx, logName(0), data)
320 put(fx, logName(2), logBytes(2, []op{putOp("c", "3")}))
321 if _, _, err := recoverState(t, fx, noLog); !errors.Is(err, ErrCorrupt) {
322 t.Fatalf("err = %v, want ErrCorrupt", err)
323 }
324 })
325 }
326}
327
328func TestRecoverBadLogHeader(t *testing.T) {
329 good := logBytes(0, []op{putOp("a", "1")})
330 flip := func(i int) []byte {
331 c := append([]byte(nil), good...)
332 c[i] ^= 1
333 return c
334 }
335 cases := map[string]struct {
336 name string
337 data []byte
338 }{
339 "empty file": {logName(0), nil},
340 "short header": {logName(0), good[:10]},
341 "bad magic": {logName(0), flip(0)},
342 "bad crc": {logName(0), flip(18)},
343 "flipped base": {logName(0), flip(9)},
344 "base != name": {logName(7), good},
345 "all zero header": {logName(0), make([]byte, logHeaderSize)},
346 }
347 for name, c := range cases {
348 t.Run(name, func(t *testing.T) {
349 fx := faultfs.New()
350 put(fx, c.name, c.data)
351 if _, _, err := recoverState(t, fx, noLog); !errors.Is(err, ErrCorrupt) {
352 t.Fatalf("err = %v, want ErrCorrupt", err)
353 }
354 })
355 }
356 t.Run("bad header in older log", func(t *testing.T) {
357 fx := faultfs.New()
358 put(fx, logName(0), good[:10])
359 put(fx, logName(1), logBytes(1, []op{putOp("b", "2")}))
360 if _, _, err := recoverState(t, fx, noLog); !errors.Is(err, ErrCorrupt) {
361 t.Fatalf("err = %v, want ErrCorrupt", err)
362 }
363 })
364}
365
366func TestRecoverSeqGap(t *testing.T) {
367 cases := map[string]func(fx *faultfs.FS){
368 "gap inside a log": func(fx *faultfs.FS) {
369 b := appendLogHeader(nil, 0)
370 b = appendRecord(b, 1, []op{putOp("a", "1")})
371 b = appendRecord(b, 3, []op{putOp("b", "2")})
372 put(fx, logName(0), b)
373 },
374 "first record not base+1": func(fx *faultfs.FS) {
375 b := appendRecord(appendLogHeader(nil, 0), 2, []op{putOp("a", "1")})
376 put(fx, logName(0), b)
377 },
378 "gap between logs": func(fx *faultfs.FS) {
379 put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
380 put(fx, logName(2), logBytes(2, []op{putOp("c", "3")}))
381 },
382 "first log starts after seq 0": func(fx *faultfs.FS) {
383 put(fx, logName(4), logBytes(4, []op{putOp("a", "1")}))
384 },
385 "log after snapshot has a gap": func(fx *faultfs.FS) {
386 put(fx, snapshotName(3), snapshotBytes(t, 3, snapEnt("a", "1", 3)))
387 put(fx, logName(5), logBytes(5, []op{putOp("b", "2")}))
388 },
389 "logs end before snapshot": func(fx *faultfs.FS) {
390 put(fx, snapshotName(5), snapshotBytes(t, 5, snapEnt("a", "1", 3)))
391 put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
392 },
393 "duplicate seq": func(fx *faultfs.FS) {
394 b := appendLogHeader(nil, 0)
395 b = appendRecord(b, 1, []op{putOp("a", "1")})
396 b = appendRecord(b, 1, []op{putOp("b", "2")})
397 put(fx, logName(0), b)
398 },
399 }
400 for name, setup := range cases {
401 t.Run(name, func(t *testing.T) {
402 fx := faultfs.New()
403 setup(fx)
404 if _, _, err := recoverState(t, fx, noLog); !errors.Is(err, ErrCorrupt) {
405 t.Fatalf("err = %v, want ErrCorrupt", err)
406 }
407 })
408 }
409}
410
411func TestRecoverSnapshotThenLogs(t *testing.T) {
412 fx := faultfs.New()
413 put(fx, snapshotName(3), snapshotBytes(t, 3, snapEnt("a", "1", 1), snapEnt("b", "2", 3)))
414 // log-0 still holds records already in the snapshot; log-3 continues it.
415 put(fx, logName(0), logBytes(0, []op{putOp("a", "old")}, []op{putOp("x", "gone")}, []op{putOp("b", "2")}))
416 put(fx, logName(3), logBytes(3, []op{putOp("c", "3")}, []op{delOp("a")}))
417 s, rec := mustRecover(t, fx)
418 wantState(t, s, map[string]string{"b": "2", "c": "3"})
419 if got := fmt.Sprint(s.seqs); got != "[4 5]" {
420 t.Fatalf("applied seqs %s", got)
421 }
422 if s.versions["b"] != 3 || s.versions["c"] != 4 {
423 t.Fatalf("versions = %v", s.versions)
424 }
425 if rec.snapSeq != 3 || rec.lastSeq != 5 || rec.logBase != 3 {
426 t.Fatalf("rec = %+v", rec)
427 }
428}
429
430func TestRecoverSnapshotWithoutLogCreatesLog(t *testing.T) {
431 fx := faultfs.New()
432 put(fx, snapshotName(7), snapshotBytes(t, 7, snapEnt("a", "1", 7)))
433 s, rec := mustRecover(t, fx)
434 wantState(t, s, map[string]string{"a": "1"})
435 if rec.snapSeq != 7 || rec.lastSeq != 7 || rec.logBase != 7 || rec.logSize != logHeaderSize {
436 t.Fatalf("rec = %+v", rec)
437 }
438 if _, ok := fx.ReadFile(filepath.Join(testDir, logName(7))); !ok {
439 t.Fatal("log-7 not created")
440 }
441}
442
443func TestRecoverInvalidNewestSnapshotFallsBack(t *testing.T) {
444 good := snapshotBytes(t, 2, snapEnt("a", "1", 1), snapEnt("b", "2", 2))
445 newer := snapshotBytes(t, 4, snapEnt("a", "1", 1), snapEnt("z", "9", 4))
446 truncated := newer[:len(newer)-2]
447 flipped := append([]byte(nil), newer...)
448 flipped[30] ^= 1
449 wrongName := snapshotBytes(t, 3, snapEnt("a", "1", 1))
450 for name, bad := range map[string][]byte{"truncated": truncated, "flipped": flipped, "empty": nil, "seq/name mismatch": wrongName} {
451 t.Run(name, func(t *testing.T) {
452 fx := faultfs.New()
453 put(fx, snapshotName(2), good)
454 put(fx, snapshotName(4), bad)
455 put(fx, logName(2), logBytes(2, []op{putOp("c", "3")}, []op{putOp("d", "4")}, []op{delOp("a")}))
456 var logs bytes.Buffer
457 s, rec, err := recoverState(t, fx, bufLogger(&logs))
458 if err != nil {
459 t.Fatal(err)
460 }
461 defer release(rec)
462 wantState(t, s, map[string]string{"b": "2", "c": "3", "d": "4"})
463 if rec.snapSeq != 2 || rec.lastSeq != 5 {
464 t.Fatalf("rec = %+v", rec)
465 }
466 if !strings.Contains(logs.String(), "invalid snapshot") {
467 t.Fatalf("fallback not logged: %q", logs.String())
468 }
469 })
470 }
471}
472
473func TestRecoverInvalidSnapshotNoOtherFallsBackToEmpty(t *testing.T) {
474 fx := faultfs.New()
475 put(fx, snapshotName(3), []byte("junk"))
476 put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
477 s, rec := mustRecover(t, fx)
478 wantState(t, s, map[string]string{"a": "1"})
479 if rec.snapSeq != 0 || rec.lastSeq != 1 {
480 t.Fatalf("rec = %+v", rec)
481 }
482}
483
484func TestRecoverNoSnapshotStartsAtZero(t *testing.T) {
485 fx := faultfs.New()
486 put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
487 put(fx, logName(1), logBytes(1, []op{putOp("b", "2")}))
488 put(fx, "unrelated.txt", []byte("ignored"))
489 put(fx, "log-xyz", []byte("ignored"))
490 s, rec := mustRecover(t, fx)
491 wantState(t, s, map[string]string{"a": "1", "b": "2"})
492 if rec.snapSeq != 0 || rec.lastSeq != 2 || rec.logBase != 1 {
493 t.Fatalf("rec = %+v", rec)
494 }
495}
496
497func TestRecoverSnapshotLoadOrderAndOwnership(t *testing.T) {
498 fx := faultfs.New()
499 put(fx, snapshotName(1), snapshotBytes(t, 1, snapEnt("a", "1", 1), snapEnt("b", "2", 1), snapEnt("c", "3", 1)))
500 var keys []string
501 var held [][]byte
502 _, err := recoverDir(fx, testDir, noLog, func(k, v []byte, ver uint64) {
503 keys = append(keys, string(k))
504 held = append(held, k, v)
505 }, nil)
506 if err != nil {
507 t.Fatal(err)
508 }
509 if strings.Join(keys, "") != "abc" {
510 t.Fatalf("keys = %v", keys)
511 }
512 var all []byte
513 for _, h := range held {
514 all = append(all, h...)
515 }
516 if string(all) != "a1b2c3" {
517 t.Fatalf("retained slices changed: %q", all)
518 }
519}
520
521func TestRecoverPowerLossAfterRecovery(t *testing.T) {
522 // A fresh directory survives a power loss right after recovery and
523 // reopens cleanly with the same state.
524 fx := faultfs.New()
525 _, rec := mustRecover(t, fx)
526 _ = rec.log.Close()
527 _ = rec.lock.Close()
528 fx.PowerLoss(faultfs.PowerLossOptions{})
529 s, rec2 := mustRecover(t, fx)
530 wantState(t, s, nil)
531 if rec2.logBase != 0 {
532 t.Fatalf("rec = %+v", rec2)
533 }
534}
535
536func TestRecoverOSFilesystem(t *testing.T) {
537 dir := filepath.Join(t.TempDir(), "nested", "db")
538 data := logBytes(0, []op{putOp("a", "1")}, []op{putOp("b", "2")})
539 torn := append(append([]byte(nil), data...), 9, 9, 9)
540
541 rec, err := recoverDir(fsys.OS(), dir, noLog, nil, nil)
542 if err != nil {
543 t.Fatal(err)
544 }
545 if _, err := recoverDir(fsys.OS(), dir, noLog, nil, nil); !errors.Is(err, ErrLocked) {
546 t.Fatalf("second recover: err = %v, want ErrLocked", err)
547 }
548 if _, err := rec.log.Write(data[logHeaderSize:]); err != nil {
549 t.Fatal(err)
550 }
551 if _, err := rec.log.Write(torn[len(data):]); err != nil {
552 t.Fatal(err)
553 }
554 if err := rec.log.Sync(); err != nil {
555 t.Fatal(err)
556 }
557 _ = rec.log.Close()
558 _ = rec.lock.Close()
559
560 s := &state{m: map[string]string{}, versions: map[string]uint64{}}
561 rec, err = recoverDir(fsys.OS(), dir, noLog, s.load, s.apply)
562 if err != nil {
563 t.Fatal(err)
564 }
565 defer release(rec)
566 wantState(t, s, map[string]string{"a": "1", "b": "2"})
567 if rec.lastSeq != 2 || rec.logSize != int64(len(data)) {
568 t.Fatalf("rec = %+v", rec)
569 }
570 buf := make([]byte, len(data)+10)
571 n, err := rec.log.ReadAt(buf, 0)
572 if n != len(data) || (err != nil && !errors.Is(err, io.EOF)) {
573 t.Fatalf("ReadAt = %d, %v: tail not truncated", n, err)
574 }
575}
576
577func TestRecoverGroupRecords(t *testing.T) {
578 fx := faultfs.New()
579 b := appendLogHeader(nil, 0)
580 b = appendRecord(b, 1, []op{putOp("a", "1")}, []op{putOp("b", "2"), delOp("a")}, []op{putOp("c", "3")})
581 b = appendRecord(b, 4, []op{putOp("a", "4")})
582 put(fx, logName(0), b)
583 s, rec := mustRecover(t, fx)
584 wantState(t, s, map[string]string{"a": "4", "b": "2", "c": "3"})
585 if got := fmt.Sprint(s.seqs); got != "[1 2 3 4]" {
586 t.Fatalf("applied seqs %s", got)
587 }
588 if s.versions["b"] != 2 || s.versions["c"] != 3 || s.versions["a"] != 4 || rec.lastSeq != 4 {
589 t.Fatalf("versions = %v, rec = %+v", s.versions, rec)
590 }
591}
592
593// TestRecoverRecordStraddlesSnapshot checks that only the batches newer than
594// the snapshot of a record that spans the snapshot seq are replayed.
595func TestRecoverRecordStraddlesSnapshot(t *testing.T) {
596 fx := faultfs.New()
597 put(fx, snapshotName(2), snapshotBytes(t, 2, snapEnt("a", "1", 1), snapEnt("b", "2", 2)))
598 b := appendLogHeader(nil, 0)
599 b = appendRecord(b, 1, []op{putOp("a", "1")}, []op{putOp("b", "2")}, []op{putOp("c", "3")}, []op{delOp("a")})
600 b = appendRecord(b, 5, []op{putOp("d", "5")})
601 put(fx, logName(0), b)
602 s, rec := mustRecover(t, fx)
603 wantState(t, s, map[string]string{"b": "2", "c": "3", "d": "5"})
604 if got := fmt.Sprint(s.seqs); got != "[3 4 5]" {
605 t.Fatalf("applied seqs %s, want [3 4 5]", got)
606 }
607 if rec.snapSeq != 2 || rec.lastSeq != 5 {
608 t.Fatalf("rec = %+v", rec)
609 }
610
611 // A record that ends exactly at the snapshot seq adds nothing.
612 fx = faultfs.New()
613 put(fx, snapshotName(2), snapshotBytes(t, 2, snapEnt("a", "1", 1), snapEnt("b", "2", 2)))
614 put(fx, logName(0), cat(appendLogHeader(nil, 0), appendRecord(nil, 1, []op{putOp("a", "1")}, []op{putOp("b", "2")})))
615 s, rec = mustRecover(t, fx)
616 if len(s.seqs) != 0 || rec.lastSeq != 2 {
617 t.Fatalf("applied %v, rec = %+v", s.seqs, rec)
618 }
619}
620
621func TestRecoverTornGroupRecordFollowedByValidRecord(t *testing.T) {
622 r1 := appendRecord(nil, 1, []op{putOp("a", "1")}, []op{putOp("b", "2")})
623 r2 := appendRecord(nil, 3, []op{putOp("c", "3")}, []op{putOp("d", "4")})
624 r3 := appendRecord(nil, 5, []op{putOp("e", "5")})
625 damaged := slices.Clone(r2)
626 damaged[len(damaged)/2] ^= 0xff
627 fx := faultfs.New()
628 data := cat(appendLogHeader(nil, 0), r1, damaged, r3)
629 put(fx, logName(0), data)
630 if _, _, err := recoverState(t, fx, noLog); !errors.Is(err, ErrCorrupt) {
631 t.Fatalf("err = %v, want ErrCorrupt", err)
632 }
633 if got, _ := fx.ReadFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, data) {
634 t.Fatal("corrupt log was modified")
635 }
636
637 // Without a readable record after it, the same damage is a torn tail
638 // that drops the whole group record, and nothing of it is applied.
639 fx = faultfs.New()
640 put(fx, logName(0), cat(appendLogHeader(nil, 0), r1, damaged))
641 s, rec := mustRecover(t, fx)
642 wantState(t, s, map[string]string{"a": "1", "b": "2"})
643 if rec.lastSeq != 2 {
644 t.Fatalf("rec = %+v", rec)
645 }
646}
647
648// TestRecoverDamagedLenIsTornTail checks that a damaged len field of the last
649// record is handled as a torn tail: the record is dropped.
650func TestRecoverDamagedLenIsTornTail(t *testing.T) {
651 good := logBytes(0, []op{putOp("a", "1")})
652 r2 := appendRecord(nil, 2, []op{putOp("b", "2")})
653 for name, patch := range map[string]func(b []byte){
654 "len beyond EOF": func(b []byte) { b[1] = 0x7f },
655 "len above max": func(b []byte) { b[3] = 0x7f },
656 "len below min": func(b []byte) { clear(b[:4]) },
657 "len off by some": func(b []byte) { b[0] += 3 },
658 } {
659 t.Run(name, func(t *testing.T) {
660 bad := slices.Clone(r2)
661 patch(bad)
662 fx := faultfs.New()
663 put(fx, logName(0), cat(good, bad))
664 s, rec := mustRecover(t, fx)
665 wantState(t, s, map[string]string{"a": "1"})
666 if rec.lastSeq != 1 || rec.logSize != int64(len(good)) {
667 t.Fatalf("rec = %+v", rec)
668 }
669 })
670 }
671}