1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
6 "context"
7 "errors"
8 "fmt"
9 "math/rand/v2"
10 "os"
11 "path/filepath"
12 "slices"
13 "strings"
14 "sync"
15 "sync/atomic"
16 "testing"
17 "testing/synctest"
18 "time"
19
20 "github.com/bnema/zerowrap"
21
22 "bnema.dev/ermite-kv/internal/fsys"
23 "bnema.dev/ermite-kv/internal/fsys/faultfs"
24)
25
26var errCrash = errors.New("test: power was cut")
27
28// hookFS wraps an FS and intercepts the operations that change files:
29// hook (if set) may block or fail them, and a power loss can be cut at a
30// given step. With serial set, operations run one at a time under mu, so a
31// cut is atomic with respect to every other operation and nothing touches the
32// file system afterwards: once cut, every operation fails with errCrash, like
33// a process that no longer exists.
34type hookFS struct {
35 fsys.FS
36 ff *faultfs.FS // underlying FS, needed by cut
37 serial bool // run operations under mu
38 hook func(op, path string) error // optional; set before use
39
40 mu sync.Mutex // guards the fields below when serial
41 armed bool // record the operations in trace
42 trace []string
43 crashAt int // cut the power before the crashAt-th traced operation (0: never)
44 plo faultfs.PowerLossOptions
45 dead bool
46}
47
48func (h *hookFS) run(op, path string, fn func() error) error {
49 if h.serial {
50 h.mu.Lock()
51 defer h.mu.Unlock()
52 if h.dead {
53 return errCrash
54 }
55 if h.armed {
56 h.trace = append(h.trace, op+" "+filepath.Base(path))
57 if len(h.trace) == h.crashAt {
58 h.cutLocked(h.plo)
59 return errCrash
60 }
61 }
62 }
63 if h.hook != nil {
64 if err := h.hook(op, path); err != nil {
65 return err
66 }
67 }
68 return fn()
69}
70
71func (h *hookFS) arm() {
72 h.mu.Lock()
73 h.armed = true
74 h.mu.Unlock()
75}
76
77func (h *hookFS) isDead() bool {
78 h.mu.Lock()
79 defer h.mu.Unlock()
80 return h.dead
81}
82
83func (h *hookFS) steps() []string {
84 h.mu.Lock()
85 defer h.mu.Unlock()
86 return slices.Clone(h.trace)
87}
88
89// cut cuts the power: the unsynced state is dropped as plo says and every
90// later operation fails.
91func (h *hookFS) cut(plo faultfs.PowerLossOptions) {
92 h.mu.Lock()
93 defer h.mu.Unlock()
94 if !h.dead {
95 h.cutLocked(plo)
96 }
97}
98
99func (h *hookFS) cutLocked(plo faultfs.PowerLossOptions) {
100 h.dead = true
101 h.ff.PowerLoss(plo)
102}
103
104func (h *hookFS) OpenFile(name string, flag int, perm os.FileMode) (fsys.File, error) {
105 var f fsys.File
106 err := h.run("open", name, func() (err error) {
107 f, err = h.FS.OpenFile(name, flag, perm)
108 return err
109 })
110 if err != nil {
111 return nil, err
112 }
113 return &hookFile{File: f, fs: h, path: name}, nil
114}
115
116func (h *hookFS) Rename(oldpath, newpath string) error {
117 return h.run("rename", newpath, func() error { return h.FS.Rename(oldpath, newpath) })
118}
119
120func (h *hookFS) Remove(name string) error {
121 return h.run("remove", name, func() error { return h.FS.Remove(name) })
122}
123
124func (h *hookFS) SyncDir(dir string) error {
125 return h.run("syncdir", dir, func() error { return h.FS.SyncDir(dir) })
126}
127
128type hookFile struct {
129 fsys.File
130 fs *hookFS
131 path string
132}
133
134func (f *hookFile) Write(p []byte) (n int, err error) {
135 err = f.fs.run("write", f.path, func() (err error) {
136 n, err = f.File.Write(p)
137 return err
138 })
139 return n, err
140}
141
142func (f *hookFile) Sync() error { return f.fs.run("sync", f.path, f.File.Sync) }
143func (f *hookFile) SyncAll() error { return f.fs.run("syncall", f.path, f.File.SyncAll) }
144
145func (f *hookFile) Truncate(size int64) error {
146 return f.fs.run("truncate", f.path, func() error { return f.File.Truncate(size) })
147}
148
149// The sequential workload: batch i (which gets seq i) puts one key.
150const seqPad = "................................"
151
152func seqKey(i int) []byte { return fmt.Appendf(nil, "k%06d", i) }
153func seqValue(i int) []byte { return fmt.Appendf(nil, "v%06d%s", i, seqPad) }
154
155// recSize is the encoded size of the log record of one sequential batch; it
156// does not depend on i. Options.CompactAfter = n*recSize makes the (n+1)-th
157// record of a log trigger a compaction.
158var recSize = int64(len(appendRecord(nil, 1, []op{{kind: opPut, key: seqKey(1), value: seqValue(1)}})))
159
160// compactOpts compacts when the 6th record of a log is written.
161func compactOpts() Options { return Options{CompactAfter: 5 * recSize} }
162
163// applySeq applies batch i and checks that it gets seq i.
164func applySeq(db *DB, i int) error {
165 var b Batch
166 b.Put(seqKey(i), seqValue(i))
167 seq, err := db.Apply(bg, &b)
168 if err == nil && seq != uint64(i) {
169 err = fmt.Errorf("batch %d got seq %d", i, seq)
170 }
171 return err
172}
173
174func mustApply(t testing.TB, db *DB, i int) {
175 t.Helper()
176 if err := applySeq(db, i); err != nil {
177 t.Fatalf("apply %d: %v", i, err)
178 }
179}
180
181// barrier returns once the writer has finished everything it did for the
182// batches applied so far, including a log switch: an empty batch is only
183// taken after the previous loop iteration, which ends with the compaction
184// trigger.
185func barrier(t testing.TB, db *DB) {
186 t.Helper()
187 if _, err := db.Apply(bg, new(Batch)); err != nil {
188 t.Fatalf("barrier: %v", err)
189 }
190}
191
192// checkSequential checks that s holds exactly the keys of batches 1..N for
193// some N ≥ lastAcked, with the right values and versions, and returns N.
194func checkSequential(s *Snapshot, lastAcked int) (int, error) {
195 n := s.Len()
196 if s.Seq() != uint64(n) {
197 return n, fmt.Errorf("seq %d but %d keys", s.Seq(), n)
198 }
199 if n < lastAcked {
200 return n, fmt.Errorf("acknowledged batch %d lost: recovered %d batches", lastAcked, n)
201 }
202 for i := 1; i <= n; i++ {
203 it, ok := s.Get(bg, seqKey(i))
204 if !ok || string(it.Value) != string(seqValue(i)) || it.Version != uint64(i) {
205 return n, fmt.Errorf("key %d: %q at version %d, present %v", i, it.Value, it.Version, ok)
206 }
207 }
208 return n, nil
209}
210
211func mustCheck(t testing.TB, db *DB, lastAcked int) int {
212 t.Helper()
213 n, err := checkSequential(db.Snapshot(bg), lastAcked)
214 if err != nil {
215 t.Fatal(err)
216 }
217 return n
218}
219
220// listFiles returns the seqs of the snapshot and log files in testDir.
221func listFiles(t testing.TB, fx fsys.FS) (snaps, logs []uint64) {
222 t.Helper()
223 names, err := fx.ReadDir(testDir)
224 if err != nil {
225 t.Fatal(err)
226 }
227 for _, name := range names {
228 if s, ok := parseSeqName(name, snapshotPrefix); ok {
229 snaps = append(snaps, s)
230 } else if b, ok := parseSeqName(name, logPrefix); ok {
231 logs = append(logs, b)
232 }
233 }
234 return snaps, logs
235}
236
237// settled reports whether the directory holds exactly snapshot-S and log-S.
238func settled(t testing.TB, fx fsys.FS, s uint64) bool {
239 snaps, logs := listFiles(t, fx)
240 return slices.Equal(snaps, []uint64{s}) && slices.Equal(logs, []uint64{s})
241}
242
243// settle returns once every goroutine of the bubble is durably blocked: the
244// writer and any compactor have finished all the work they can do. It must be
245// called inside a synctest bubble.
246func settle() { synctest.Wait() }
247
248// applyUntil applies batches i, i+1, … until cond holds, settling after each
249// one so that a compaction the batch triggered has finished. It returns the
250// next unused i.
251func applyUntil(t testing.TB, db *DB, i int, what string, cond func() bool) int {
252 t.Helper()
253 settle()
254 for ; !cond(); i++ {
255 if i > 1000 {
256 t.Fatalf("%s did not happen", what)
257 }
258 mustApply(t, db, i)
259 settle()
260 }
261 return i
262}
263
264// failedDB reports whether the database is in the failed state.
265func failedDB(db *DB) bool {
266 select {
267 case <-db.failed:
268 return true
269 default:
270 return false
271 }
272}
273
274// debugCtx returns a context whose logger writes JSON at debug level to the
275// returned buffer.
276func debugCtx() (context.Context, *syncBuffer) {
277 buf := new(syncBuffer)
278 return zerowrap.WithCtx(bg, zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: buf})), buf
279}
280
281func logged(buf *syncBuffer, msg string) int {
282 return strings.Count(buf.String(), `"message":"`+msg+`"`)
283}
284
285func TestCompactTrigger(t *testing.T) {
286 synctest.Test(t, func(t *testing.T) {
287 fx := faultfs.New()
288 ctx, logs := debugCtx()
289 db := mustOpenCtx(t, ctx, fx, compactOpts())
290
291 // Exactly CompactAfter bytes after the header: not above the threshold.
292 for i := 1; i <= 5; i++ {
293 mustApply(t, db, i)
294 }
295 barrier(t, db)
296 if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
297 t.Fatalf("below the threshold: snapshots %v, logs %v", snaps, files)
298 }
299
300 mustApply(t, db, 6)
301 settle() // compaction: the background work has finished
302 if logged(logs, "compaction done") != 1 {
303 t.Fatalf("compaction did not happen: %s", logs.String())
304 }
305 if !settled(t, fx, 6) {
306 snaps, files := listFiles(t, fx)
307 t.Fatalf("after compaction: snapshots %v, logs %v; want snapshot-6 and log-6 only", snaps, files)
308 }
309 if n := logged(logs, "log switch"); n != 1 {
310 t.Fatalf("%d log switch messages, want 1", n)
311 }
312 for _, key := range []string{`"seq":6`, `"keys":6`, `"bytes":`, `"duration":`} {
313 if !strings.Contains(logs.String(), key) {
314 t.Errorf("compaction message lacks %s: %s", key, logs.String())
315 }
316 }
317
318 // The new log takes the following batches; the old files stay gone.
319 mustApply(t, db, 7)
320 if err := db.Close(bg); err != nil {
321 t.Fatal(err)
322 }
323 db = mustOpen(t, fx, Options{})
324 if n := mustCheck(t, db, 7); n != 7 {
325 t.Fatalf("recovered %d batches, want 7", n)
326 }
327 })
328}
329
330func TestCompactDisabled(t *testing.T) {
331 fx := faultfs.New()
332 db := mustOpen(t, fx, Options{CompactAfter: -1})
333 for i := 1; i <= 50; i++ {
334 mustApply(t, db, i)
335 }
336 barrier(t, db)
337 if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
338 t.Fatalf("snapshots %v, logs %v", snaps, files)
339 }
340}
341
342func TestCompactSingleFlight(t *testing.T) {
343 synctest.Test(t, func(t *testing.T) {
344 fx := faultfs.New()
345 started, release := make(chan struct{}), make(chan struct{})
346 var once sync.Once
347 h := &hookFS{FS: fx, hook: func(op, path string) error {
348 if op == "open" && strings.Contains(path, snapshotPrefix) {
349 once.Do(func() { close(started); <-release })
350 }
351 return nil
352 }}
353 ctx, logs := debugCtx()
354 db := mustOpenCtx(t, ctx, h, compactOpts())
355 for i := 1; i <= 6; i++ {
356 mustApply(t, db, i)
357 }
358 settle() // the compactor is parked in the hook
359 select {
360 case <-started:
361 default:
362 t.Fatal("the compactor did not start")
363 }
364
365 // The compactor is stuck. The writer is not: it goes on far past the
366 // threshold without starting a second compaction.
367 for i := 7; i <= 60; i++ {
368 mustApply(t, db, i)
369 }
370 barrier(t, db)
371 if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0, 6}) {
372 t.Fatalf("while compacting: snapshots %v, logs %v", snaps, files)
373 }
374 if n := logged(logs, "log switch"); n != 1 {
375 t.Fatalf("%d log switches while a compaction was running, want 1", n)
376 }
377
378 close(release)
379 settle() // first compaction: the background work has finished
380 if logged(logs, "compaction done") != 1 {
381 t.Fatalf("first compaction did not happen: %s", logs.String())
382 }
383 // The log is still over the threshold: the next commit compacts again.
384 i := 61
385 i = applyUntil(t, db, i, "second compaction", func() bool { return logged(logs, "compaction done") >= 2 })
386 if n := logged(logs, "log switch"); n != 2 {
387 t.Fatalf("%d log switches, want 2", n)
388 }
389 if err := db.Close(bg); err != nil {
390 t.Fatal(err)
391 }
392 snaps, files := listFiles(t, fx)
393 if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
394 t.Fatalf("after the second compaction: snapshots %v, logs %v", snaps, files)
395 }
396 db = mustOpen(t, fx, Options{})
397 mustCheck(t, db, i-1)
398 })
399}
400
401func TestCompactRemovesOldFiles(t *testing.T) {
402 fx := faultfs.New()
403 db := mustOpen(t, fx, compactOpts())
404 const total = 200
405 for i := 1; i <= total; i++ {
406 mustApply(t, db, i)
407 }
408 if err := db.Close(bg); err != nil { // waits for the last compaction
409 t.Fatal(err)
410 }
411 snaps, files := listFiles(t, fx)
412 if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] == 0 {
413 t.Fatalf("snapshots %v, logs %v: want only the newest pair", snaps, files)
414 }
415 names, _ := fx.ReadDir(testDir)
416 for _, n := range names {
417 if strings.HasSuffix(n, tmpSuffix) {
418 t.Errorf("leftover %s", n)
419 }
420 }
421 db = mustOpen(t, fx, compactOpts())
422 if n := mustCheck(t, db, total); n != total {
423 t.Fatalf("recovered %d batches, want %d", n, total)
424 }
425 // A reopened database keeps compacting.
426 for i := total + 1; i <= 2*total; i++ {
427 mustApply(t, db, i)
428 }
429 if err := db.Close(bg); err != nil {
430 t.Fatal(err)
431 }
432 db = mustOpen(t, fx, Options{})
433 mustCheck(t, db, 2*total)
434}
435
436// TestCompactOS runs compaction on the real file system.
437func TestCompactOS(t *testing.T) {
438 dir := filepath.Join(t.TempDir(), "db")
439 ctx, logs := debugCtx()
440 db, err := Open(ctx, dir, compactOpts())
441 if err != nil {
442 t.Fatal(err)
443 }
444 for i := 1; i <= 100; i++ {
445 mustApply(t, db, i)
446 }
447 if err := db.Close(bg); err != nil {
448 t.Fatal(err)
449 }
450 if logged(logs, "compaction done") == 0 {
451 t.Fatal("no compaction on the OS file system")
452 }
453 db, err = Open(bg, dir, Options{})
454 if err != nil {
455 t.Fatal(err)
456 }
457 defer func() { _ = db.Close(bg) }()
458 mustCheck(t, db, 100)
459}
460
461// compactorFaults fail one step of the compactor, which runs after the log
462// switch. The injection happens before batch 6 is applied, so the numbering
463// counts from the group write: Write 1 is the group, 2 the header of log-6, 3
464// the snapshot; the syncs count Sync and SyncAll (1 the group, 2 the log
465// before the switch, 3 log-6, 4 the snapshot); SyncDir 1 is log-6, 2 the
466// snapshot, 3 the cleanup; Rename 1 is log-6, 2 the snapshot; OpenFile 1 is
467// log-6, 2 the snapshot; Remove 1 is log-0.
468var compactorFaults = []struct {
469 name string
470 inject func(fx *faultfs.FS)
471}{
472 {"open snapshot", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpOpenFile, 2, nil) }},
473 {"write snapshot", func(fx *faultfs.FS) { fx.FailWrite(3, nil) }},
474 {"short write snapshot", func(fx *faultfs.FS) { fx.ShortWrite(3, 10) }},
475 {"sync snapshot", func(fx *faultfs.FS) { fx.FailSync(4, nil) }},
476 {"rename snapshot", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpRename, 2, nil) }},
477 {"syncdir snapshot", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpSyncDir, 2, nil) }},
478 {"remove old log", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpRemove, 1, nil) }},
479 {"syncdir cleanup", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpSyncDir, 3, nil) }},
480}
481
482// TestCompactorFailure checks that a failing compactor is logged, leaves the
483// database usable and a recoverable directory, and that the next trigger
484// compacts.
485func TestCompactorFailure(t *testing.T) {
486 for _, f := range compactorFaults {
487 t.Run(f.name, func(t *testing.T) {
488 synctest.Test(t, func(t *testing.T) {
489 fx := faultfs.New()
490 ctx, logs := debugCtx()
491 db := mustOpenCtx(t, ctx, fx, compactOpts())
492 for i := 1; i <= 5; i++ {
493 mustApply(t, db, i)
494 }
495 f.inject(fx)
496 mustApply(t, db, 6)
497 settle() // compaction failure: the background work has finished
498 if logged(logs, "compaction failed") != 1 {
499 t.Fatalf("compaction failure did not happen: %s", logs.String())
500 }
501 if !strings.Contains(logs.String(), `"level":"warn"`) {
502 t.Errorf("failure not logged as a warning: %s", logs.String())
503 }
504 mustApply(t, db, 7) // the database is not failed
505 if err := db.Close(bg); err != nil {
506 t.Fatalf("Close: %v", err)
507 }
508
509 // Reopen the partial state: it recovers and compacts again.
510 ctx, logs = debugCtx()
511 db = mustOpenCtx(t, ctx, fx, compactOpts())
512 mustCheck(t, db, 7)
513 i := 8
514 i = applyUntil(t, db, i, "retried compaction", func() bool { return logged(logs, "compaction done") >= 1 })
515 if err := db.Close(bg); err != nil {
516 t.Fatal(err)
517 }
518 snaps, files := listFiles(t, fx)
519 if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
520 t.Fatalf("after the retry: snapshots %v, logs %v", snaps, files)
521 }
522 db = mustOpen(t, fx, Options{})
523 mustCheck(t, db, i-1)
524 })
525 })
526 }
527}
528
529// TestCompactorRetryWithoutReopen checks the retry in the same process.
530func TestCompactorRetryWithoutReopen(t *testing.T) {
531 synctest.Test(t, func(t *testing.T) {
532 fx := faultfs.New()
533 ctx, logs := debugCtx()
534 db := mustOpenCtx(t, ctx, fx, compactOpts())
535 for i := 1; i <= 5; i++ {
536 mustApply(t, db, i)
537 }
538 fx.FailOp(faultfs.OpRename, 2, nil) // the snapshot
539 mustApply(t, db, 6)
540 settle() // compaction failure: the background work has finished
541 if logged(logs, "compaction failed") != 1 {
542 t.Fatalf("compaction failure did not happen: %s", logs.String())
543 }
544 applyUntil(t, db, 7, "retried compaction", func() bool { return logged(logs, "compaction done") >= 1 })
545 if err := db.Close(bg); err != nil {
546 t.Fatal(err)
547 }
548 snaps, files := listFiles(t, fx)
549 if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
550 t.Fatalf("snapshots %v, logs %v", snaps, files)
551 }
552 })
553}
554
555// switchFaults fail the creation of log-6 in the writer. The numbering is
556// the one of compactorFaults.
557var switchFaults = []struct {
558 name string
559 inject func(fx *faultfs.FS)
560}{
561 {"open", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpOpenFile, 1, nil) }},
562 {"write", func(fx *faultfs.FS) { fx.FailWrite(2, nil) }},
563 {"sync", func(fx *faultfs.FS) { fx.FailSync(3, nil) }},
564 {"rename", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpRename, 1, nil) }},
565 {"syncdir", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpSyncDir, 1, nil) }},
566}
567
568// TestLogSwitchFailure checks that failing to create the new log only
569// postpones the compaction: the database goes on appending to the current
570// log and no log-6 is left behind.
571func TestLogSwitchFailure(t *testing.T) {
572 for _, f := range switchFaults {
573 t.Run(f.name, func(t *testing.T) {
574 synctest.Test(t, func(t *testing.T) {
575 fx := faultfs.New()
576 ctx, logs := debugCtx()
577 db := mustOpenCtx(t, ctx, fx, compactOpts())
578 for i := 1; i <= 5; i++ {
579 mustApply(t, db, i)
580 }
581 f.inject(fx)
582 mustApply(t, db, 6)
583 barrier(t, db)
584 if logged(logs, "log switch failed") != 1 {
585 t.Fatalf("switch failure not logged: %s", logs.String())
586 }
587 snaps, files := listFiles(t, fx)
588 if len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
589 t.Fatalf("after the failure: snapshots %v, logs %v", snaps, files)
590 }
591 names, _ := fx.ReadDir(testDir)
592 for _, n := range names {
593 if strings.HasSuffix(n, tmpSuffix) {
594 t.Errorf("leftover %s", n)
595 }
596 }
597
598 // No retry on every commit: a failure costs another CompactAfter bytes.
599 mustApply(t, db, 7)
600 barrier(t, db)
601 if n := logged(logs, "log switch failed"); n != 1 {
602 t.Fatalf("%d switch attempts for 2 commits, want 1", n)
603 }
604 i := 8
605 i = applyUntil(t, db, i, "compaction after the failed switch", func() bool { return logged(logs, "compaction done") >= 1 })
606 if err := db.Close(bg); err != nil {
607 t.Fatal(err)
608 }
609 db = mustOpen(t, fx, Options{})
610 mustCheck(t, db, i-1)
611 })
612 })
613 }
614}
615
616// TestLogSwitchFailsDatabase covers the failures that must fail the database:
617// the sync of the current log, and a stray log-S that cannot be removed.
618func TestLogSwitchFailsDatabase(t *testing.T) {
619 cases := []struct {
620 name string
621 inject func(fx *faultfs.FS)
622 }{
623 {"syncall of the current log", func(fx *faultfs.FS) { fx.FailSync(2, nil) }},
624 {"syncdir then remove", func(fx *faultfs.FS) {
625 fx.FailOp(faultfs.OpSyncDir, 1, nil)
626 fx.FailOp(faultfs.OpRemove, 1, nil)
627 }},
628 }
629 for _, c := range cases {
630 t.Run(c.name, func(t *testing.T) {
631 synctest.Test(t, func(t *testing.T) {
632 fx := faultfs.New()
633 ctx, logs := debugCtx()
634 db := mustOpenCtx(t, ctx, fx, compactOpts())
635 for i := 1; i <= 5; i++ {
636 mustApply(t, db, i)
637 }
638 c.inject(fx)
639 mustApply(t, db, 6) // acknowledged: the failure comes after the reply
640 settle() // the writer has seen the failure
641 if !failedDB(db) {
642 t.Fatal("the database did not fail")
643 }
644 var b Batch
645 b.Put([]byte("later"), nil)
646 if _, err := db.Apply(bg, &b); !errors.Is(err, ErrFailed) {
647 t.Fatalf("Apply after the failure: %v", err)
648 }
649 if !strings.Contains(logs.String(), "database failed") {
650 t.Errorf("failure not logged: %s", logs.String())
651 }
652 if err := db.Close(bg); !errors.Is(err, ErrFailed) {
653 t.Fatalf("Close = %v, want the failure", err)
654 }
655 db = mustOpen(t, fx, Options{})
656 mustCheck(t, db, 6)
657 })
658 })
659 }
660}
661
662// TestLogSwitchSyncFailureDropsUnsyncedData is the fsyncgate case with
663// NoSync: the SyncAll before the switch is the first sync of the log. When it
664// fails and the kernel drops the pages, the batches acknowledged without a
665// sync are gone; the database must fail rather than continue, and what is
666// left must open as a prefix.
667func TestLogSwitchSyncFailureDropsUnsyncedData(t *testing.T) {
668 synctest.Test(t, func(t *testing.T) {
669 fx := faultfs.New()
670 fx.DropOnSyncFailure(true)
671 opts := compactOpts()
672 opts.NoSync = true
673 db := mustOpen(t, fx, opts)
674 for i := 1; i <= 5; i++ {
675 mustApply(t, db, i)
676 }
677 fx.FailSync(1, nil)
678 mustApply(t, db, 6)
679 settle() // the writer has seen the failure
680 if !failedDB(db) {
681 t.Fatal("the database did not fail")
682 }
683 if err := db.Close(bg); !errors.Is(err, ErrFailed) {
684 t.Fatalf("Close = %v, want the failure", err)
685 }
686 db = mustOpen(t, fx, Options{})
687 if n := mustCheck(t, db, 0); n != 0 {
688 t.Fatalf("recovered %d batches, want 0 (all were dropped)", n)
689 }
690 })
691}
692
693// TestCloseWaitsForCompaction checks that Close does not cancel a running
694// compaction, waits for it, and leaves a directory that opens.
695func TestCloseWaitsForCompaction(t *testing.T) {
696 synctest.Test(t, func(t *testing.T) {
697 fx := faultfs.New()
698 started, release := make(chan struct{}), make(chan struct{})
699 var once sync.Once
700 h := &hookFS{FS: fx, hook: func(op, path string) error {
701 if op == "open" && strings.Contains(path, snapshotPrefix) {
702 once.Do(func() { close(started); <-release })
703 }
704 return nil
705 }}
706 ctx, logs := debugCtx()
707 db := mustOpenCtx(t, ctx, h, compactOpts())
708 for i := 1; i <= 6; i++ {
709 mustApply(t, db, i)
710 }
711 settle() // the compactor is parked in the hook
712 select {
713 case <-started:
714 default:
715 t.Fatal("the compactor did not start")
716 }
717 closed := make(chan error, 1)
718 go func() { closed <- db.Close(bg) }()
719 settle() // Close is blocked waiting for the compactor, which is parked in the hook
720 select {
721 case err := <-closed:
722 t.Fatalf("Close returned during the compaction: %v", err)
723 default:
724 }
725 if _, err := db.Apply(bg, new(Batch)); !errors.Is(err, ErrClosed) {
726 t.Fatalf("Apply while closing: %v", err)
727 }
728 close(release)
729 settle() // the compaction finishes and Close returns
730 select {
731 case err := <-closed:
732 if err != nil {
733 t.Fatalf("Close: %v", err)
734 }
735 default:
736 t.Fatal("Close did not return after the compaction was released")
737 }
738 if logged(logs, "compaction done") != 1 || !settled(t, fx, 6) {
739 t.Fatalf("the compaction did not complete: %s", logs.String())
740 }
741 db = mustOpen(t, fx, Options{})
742 mustCheck(t, db, 6)
743 })
744}
745
746// TestCloseDuringCompaction (AC9) closes the database while writers run and
747// compactions are frequent: Close returns, and the next Open finds every
748// acknowledged batch.
749func TestCloseDuringCompaction(t *testing.T) {
750 var switches atomic.Int64
751 for i := range short(12, 4) {
752 t.Run(fmt.Sprint(i), func(t *testing.T) {
753 fx := faultfs.New()
754 fx.SetSyncDelay(50 * time.Microsecond)
755 ctx, logs := debugCtx()
756 db := mustOpenCtx(t, ctx, fx, Options{CompactAfter: 2 * recSize})
757 const writers = 16
758 acked := make([]int, writers) // highest acknowledged n+1
759 var wg sync.WaitGroup
760 for g := range writers {
761 wg.Go(func() {
762 for n := 0; ; n++ {
763 var b Batch
764 b.Put(fmt.Appendf(nil, "g%02d", g), seqValue(n))
765 b.Put(fmt.Appendf(nil, "p%02d/%06d", g, n), nil)
766 _, err := db.Apply(bg, &b)
767 if errors.Is(err, ErrClosed) {
768 return
769 }
770 if err != nil {
771 t.Errorf("writer %d: %v", g, err)
772 return
773 }
774 acked[g] = n + 1
775 }
776 })
777 }
778 time.Sleep(time.Duration(5+3*i) * time.Millisecond)
779 closed := make(chan error, 1)
780 go func() { closed <- db.Close(bg) }()
781 select {
782 case err := <-closed:
783 if err != nil {
784 t.Errorf("Close: %v", err)
785 }
786 case <-time.After(20 * time.Second):
787 t.Fatal("Close did not return")
788 }
789 wg.Wait()
790 switches.Add(int64(logged(logs, "log switch")))
791
792 fx.SetSyncDelay(0)
793 s := mustOpen(t, fx, Options{}).Snapshot(bg)
794 for g, n := range acked {
795 it, ok := s.Get(bg, fmt.Appendf(nil, "g%02d", g))
796 if n == 0 {
797 continue
798 }
799 // The batch after the last acknowledged one may have been
800 // committed by the final group.
801 if !ok || string(it.Value) != string(seqValue(n-1)) && string(it.Value) != string(seqValue(n)) {
802 t.Errorf("writer %d: acked %d batches, recovered %q (%v)", g, n, it.Value, ok)
803 }
804 if _, ok := s.Get(bg, fmt.Appendf(nil, "p%02d/%06d", g, n-1)); !ok {
805 t.Errorf("writer %d: acked batch %d lost", g, n-1)
806 }
807 }
808 })
809 }
810 if switches.Load() == 0 {
811 t.Error("no log switch happened: the test did not exercise compaction")
812 }
813}
814
815// compactSteps runs a workload of three rounds of six batches (each round
816// ends with a compaction) on a fresh file system and cuts the power before
817// the crashAt-th file operation after Open (0 and out of range: after the
818// workload). It then reopens the database and reports a violation of AC2.
819// The workload is sequential and waits for each compaction, so the sequence
820// of operations is the same in every run.
821func compactSteps(t *testing.T, plo faultfs.PowerLossOptions, crashAt int) (steps []string, err error) {
822 ff := faultfs.New()
823 h := &hookFS{FS: ff, ff: ff, serial: true, crashAt: crashAt, plo: plo}
824 db, oerr := open(testCtx(t), h, testDir, compactOpts())
825 if oerr != nil {
826 return nil, fmt.Errorf("open: %w", oerr)
827 }
828 h.arm()
829 lastAcked := 0
830workload:
831 for round := 1; round <= 3; round++ {
832 for k := 1; k <= 6; k++ {
833 i := (round-1)*6 + k
834 if applySeq(db, i) != nil {
835 break workload
836 }
837 lastAcked = i
838 }
839 // Wait for the compaction of this round.
840 deadline := time.Now().Add(10 * time.Second)
841 for !h.isDead() && (!settled(t, ff, uint64(6*round)) || db.compacting.Load()) {
842 if time.Now().After(deadline) {
843 return h.steps(), errors.New("timeout waiting for the compaction")
844 }
845 time.Sleep(100 * time.Microsecond)
846 }
847 }
848 steps = h.steps()
849 _ = db.Close(bg)
850 h.cut(plo) // no-op if the power was already cut
851
852 for range 2 {
853 db2, err := open(testCtx(t), ff, testDir, Options{})
854 if err != nil {
855 return steps, fmt.Errorf("reopen: %w", err)
856 }
857 n, err := checkSequential(db2.Snapshot(bg), lastAcked)
858 if err == nil {
859 // The recovered log accepts new batches.
860 err = applySeq(db2, n+1)
861 lastAcked = n + 1
862 }
863 if cerr := db2.Close(bg); err == nil {
864 err = cerr
865 }
866 if err != nil {
867 return steps, err
868 }
869 }
870 return steps, nil
871}
872
873// TestCompactPowerLossAtEachStep cuts the power before each file operation
874// of three compactions in turn, in every power-loss mode, and checks AC2:
875// the reopened database holds every acknowledged batch and is a prefix.
876func TestCompactPowerLossAtEachStep(t *testing.T) {
877 modes := []struct {
878 name string
879 plo faultfs.PowerLossOptions
880 }{
881 {"drop", faultfs.PowerLossOptions{}},
882 {"zerofill", faultfs.PowerLossOptions{ZeroFill: true}},
883 {"keeppages", faultfs.PowerLossOptions{KeepPages: true, PageSize: 64}},
884 {"keeppages+zerofill", faultfs.PowerLossOptions{KeepPages: true, ZeroFill: true, PageSize: 64}},
885 }
886 base := powerLossBase(t)
887 for _, m := range modes {
888 t.Run(m.name, func(t *testing.T) {
889 plo := m.plo
890 plo.Rand = rand.New(rand.NewPCG(base, 7))
891 steps, err := compactSteps(t, plo, 0)
892 if err != nil {
893 t.Fatalf("without a cut: %v", err)
894 }
895 t.Logf("%d file operations in the workload", len(steps))
896 for step := 1; step <= len(steps); step++ {
897 plo.Rand = rand.New(rand.NewPCG(base+uint64(step), 7))
898 if _, err := compactSteps(t, plo, step); err != nil {
899 t.Fatalf("power cut before step %d (%s), seed %d (ERMITE_KV_SEED=%d reproduces it): %v",
900 step, steps[step-1], base, base, err)
901 }
902 }
903 })
904 }
905}
906
907// TestCompactPowerLoss cuts the power at a random time during a workload
908// that compacts all the time, then checks AC2. With NoSync the acknowledged
909// batches may be lost, but the directory must open and be a prefix; only the
910// modes that cannot make Open fail with ErrCorrupt are used (see
911// Options.NoSync).
912func TestCompactPowerLoss(t *testing.T) {
913 modes := []struct {
914 name string
915 nosync bool
916 plo faultfs.PowerLossOptions
917 }{
918 {"drop", false, faultfs.PowerLossOptions{}},
919 {"zerofill", false, faultfs.PowerLossOptions{ZeroFill: true}},
920 {"keeppages", false, faultfs.PowerLossOptions{KeepPages: true}},
921 {"keeppages+zerofill", false, faultfs.PowerLossOptions{KeepPages: true, ZeroFill: true}},
922 {"nosync drop", true, faultfs.PowerLossOptions{}},
923 {"nosync zerofill", true, faultfs.PowerLossOptions{ZeroFill: true}},
924 }
925 base := powerLossBase(t)
926 iterations := short(100, 10)
927 for _, m := range modes {
928 t.Run(m.name, func(t *testing.T) {
929 for i := range iterations {
930 seed := base + uint64(i)
931 if err := runCompactPowerLoss(t, seed, m.plo, m.nosync); err != nil {
932 t.Fatalf("seed %d (ERMITE_KV_SEED=%d reproduces it): %v", seed, seed, err)
933 }
934 }
935 })
936 }
937}
938
939func runCompactPowerLoss(t *testing.T, seed uint64, plo faultfs.PowerLossOptions, nosync bool) error {
940 rng := rand.New(rand.NewPCG(seed, 0x9e3779b97f4a7c15))
941 ff := faultfs.New()
942 h := &hookFS{FS: ff, ff: ff, serial: true}
943 opts := Options{NoSync: nosync, CompactAfter: int64(2+rng.IntN(10)) * recSize}
944 db, err := open(testCtx(t), h, testDir, opts)
945 if err != nil {
946 return fmt.Errorf("open: %w", err)
947 }
948 ff.SetSyncDelay(time.Duration(rng.IntN(100)) * time.Microsecond)
949 var lastAcked atomic.Int64
950 done := make(chan struct{})
951 go func() {
952 defer close(done)
953 for i := 1; applySeq(db, i) == nil; i++ {
954 lastAcked.Store(int64(i))
955 }
956 }()
957 time.Sleep(time.Duration(rng.IntN(6000)) * time.Microsecond)
958 plo.Rand = rand.New(rand.NewPCG(seed, 7))
959 if plo.KeepPages {
960 plo.PageSize = 64
961 }
962 h.cut(plo)
963 <-done
964 _ = db.Close(bg)
965 ff.SetSyncDelay(0)
966
967 floor := int(lastAcked.Load())
968 if nosync {
969 floor = 0
970 }
971 for range 2 {
972 db, err = open(testCtx(t), ff, testDir, Options{})
973 if err != nil {
974 return fmt.Errorf("reopen: %w", err)
975 }
976 n, err := checkSequential(db.Snapshot(bg), floor)
977 if err == nil {
978 err = applySeq(db, n+1) // appends to the right log
979 floor = n + 1
980 }
981 if cerr := db.Close(bg); err == nil {
982 err = cerr
983 }
984 if err != nil {
985 return err
986 }
987 }
988 return nil
989}