detached 21d0a80

fix(writer): allocate the Apply reply channel per call so tests can use synctest

The pooled request kept a reply channel created inside one synctest bubble,
which crashed the next bubble that reused it. Apply now allocates its request
and channel each call (about 3-5% on BenchmarkApply, small next to fdatasync).

Timing-based tests (group window, context before dequeue, compaction, close,
faultfs sync delay) now run in synctest bubbles instead of polling and sleeping.
commit
21d0a80b779c2edfee0a91ec22b0ee93d96c38bd
author
brice · · 2026-10-11 18:37 UTC
parent
5e09b48
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
5 files · +487 -451
Unified
UnifiedSplit
M compact_test.go +300 -283
@@ -14,6 +14,7 @@ import (
1414 "sync"
1515 "sync/atomic"
1616 "testing"
17+ "testing/synctest"
1718 "time"
1819
1920 "github.com/bnema/zerowrap"
@@ -239,14 +240,34 @@ func settled(t testing.TB, fx fsys.FS, s uint64) bool {
239240 return slices.Equal(snaps, []uint64{s}) && slices.Equal(logs, []uint64{s})
240241 }
241242
242-func waitFor(t testing.TB, what string, cond func() bool) {
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.
246+func 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.
251+func applyUntil(t testing.TB, db *DB, i int, what string, cond func() bool) int {
243252 t.Helper()
244- deadline := time.Now().Add(10 * time.Second)
245- for !cond() {
246- if time.Now().After(deadline) {
247- t.Fatalf("timeout waiting for %s", what)
253+ settle()
254+ for ; !cond(); i++ {
255+ if i > 1000 {
256+ t.Fatalf("%s did not happen", what)
248257 }
249- time.Sleep(time.Millisecond)
258+ mustApply(t, db, i)
259+ settle()
260+ }
261+ return i
262+}
263+
264+// failedDB reports whether the database is in the failed state.
265+func failedDB(db *DB) bool {
266+ select {
267+ case <-db.failed:
268+ return true
269+ default:
270+ return false
250271 }
251272 }
252273
@@ -262,43 +283,48 @@ func logged(buf *syncBuffer, msg string) int {
262283 }
263284
264285 func TestCompactTrigger(t *testing.T) {
265- fx := faultfs.New()
266- ctx, logs := debugCtx()
267- db := mustOpenCtx(t, ctx, fx, compactOpts())
268-
269- // Exactly CompactAfter bytes after the header: not above the threshold.
270- for i := 1; i <= 5; i++ {
271- mustApply(t, db, i)
272- }
273- barrier(t, db)
274- if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
275- t.Fatalf("below the threshold: snapshots %v, logs %v", snaps, files)
276- }
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+ }
277299
278- mustApply(t, db, 6)
279- waitFor(t, "compaction", func() bool { return logged(logs, "compaction done") == 1 })
280- if !settled(t, fx, 6) {
281- snaps, files := listFiles(t, fx)
282- t.Fatalf("after compaction: snapshots %v, logs %v; want snapshot-6 and log-6 only", snaps, files)
283- }
284- if n := logged(logs, "log switch"); n != 1 {
285- t.Fatalf("%d log switch messages, want 1", n)
286- }
287- for _, key := range []string{`"seq":6`, `"keys":6`, `"bytes":`, `"duration":`} {
288- if !strings.Contains(logs.String(), key) {
289- t.Errorf("compaction message lacks %s: %s", key, logs.String())
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+ }
290316 }
291- }
292317
293- // The new log takes the following batches; the old files stay gone.
294- mustApply(t, db, 7)
295- if err := db.Close(bg); err != nil {
296- t.Fatal(err)
297- }
298- db = mustOpen(t, fx, Options{})
299- if n := mustCheck(t, db, 7); n != 7 {
300- t.Fatalf("recovered %d batches, want 7", n)
301- }
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+ })
302328 }
303329
304330 func TestCompactDisabled(t *testing.T) {
@@ -314,63 +340,62 @@ func TestCompactDisabled(t *testing.T) {
314340 }
315341
316342 func TestCompactSingleFlight(t *testing.T) {
317- fx := faultfs.New()
318- started, release := make(chan struct{}), make(chan struct{})
319- var once sync.Once
320- h := &hookFS{FS: fx, hook: func(op, path string) error {
321- if op == "open" && strings.Contains(path, snapshotPrefix) {
322- once.Do(func() { close(started); <-release })
323- }
324- return nil
325- }}
326- ctx, logs := debugCtx()
327- db := mustOpenCtx(t, ctx, h, compactOpts())
328- for i := 1; i <= 6; i++ {
329- mustApply(t, db, i)
330- }
331- select {
332- case <-started:
333- case <-time.After(10 * time.Second):
334- t.Fatal("the compactor did not start")
335- }
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+ }
336364
337- // The compactor is stuck. The writer is not: it goes on far past the
338- // threshold without starting a second compaction.
339- for i := 7; i <= 60; i++ {
340- mustApply(t, db, i)
341- }
342- barrier(t, db)
343- if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0, 6}) {
344- t.Fatalf("while compacting: snapshots %v, logs %v", snaps, files)
345- }
346- if n := logged(logs, "log switch"); n != 1 {
347- t.Fatalf("%d log switches while a compaction was running, want 1", n)
348- }
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+ }
349377
350- close(release)
351- waitFor(t, "first compaction", func() bool { return logged(logs, "compaction done") == 1 })
352- // The log is still over the threshold: the next commit compacts again.
353- i := 61
354- waitFor(t, "second compaction", func() bool {
355- if logged(logs, "compaction done") >= 2 {
356- return true
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())
357382 }
358- mustApply(t, db, i)
359- i++
360- return false
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)
361398 })
362- if n := logged(logs, "log switch"); n != 2 {
363- t.Fatalf("%d log switches, want 2", n)
364- }
365- if err := db.Close(bg); err != nil {
366- t.Fatal(err)
367- }
368- snaps, files := listFiles(t, fx)
369- if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
370- t.Fatalf("after the second compaction: snapshots %v, logs %v", snaps, files)
371- }
372- db = mustOpen(t, fx, Options{})
373- mustCheck(t, db, i-1)
374399 }
375400
376401 func TestCompactRemovesOldFiles(t *testing.T) {
@@ -460,76 +485,72 @@ var compactorFaults = []struct {
460485 func TestCompactorFailure(t *testing.T) {
461486 for _, f := range compactorFaults {
462487 t.Run(f.name, func(t *testing.T) {
463- fx := faultfs.New()
464- ctx, logs := debugCtx()
465- db := mustOpenCtx(t, ctx, fx, compactOpts())
466- for i := 1; i <= 5; i++ {
467- mustApply(t, db, i)
468- }
469- f.inject(fx)
470- mustApply(t, db, 6)
471- waitFor(t, "compaction failure", func() bool { return logged(logs, "compaction failed") == 1 })
472- if !strings.Contains(logs.String(), `"level":"warn"`) {
473- t.Errorf("failure not logged as a warning: %s", logs.String())
474- }
475- mustApply(t, db, 7) // the database is not failed
476- if err := db.Close(bg); err != nil {
477- t.Fatalf("Close: %v", err)
478- }
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+ }
479508
480- // Reopen the partial state: it recovers and compacts again.
481- ctx, logs = debugCtx()
482- db = mustOpenCtx(t, ctx, fx, compactOpts())
483- mustCheck(t, db, 7)
484- i := 8
485- waitFor(t, "retried compaction", func() bool {
486- if logged(logs, "compaction done") >= 1 {
487- return true
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)
488517 }
489- mustApply(t, db, i)
490- i++
491- return false
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)
492524 })
493- if err := db.Close(bg); err != nil {
494- t.Fatal(err)
495- }
496- snaps, files := listFiles(t, fx)
497- if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
498- t.Fatalf("after the retry: snapshots %v, logs %v", snaps, files)
499- }
500- db = mustOpen(t, fx, Options{})
501- mustCheck(t, db, i-1)
502525 })
503526 }
504527 }
505528
506529 // TestCompactorRetryWithoutReopen checks the retry in the same process.
507530 func TestCompactorRetryWithoutReopen(t *testing.T) {
508- fx := faultfs.New()
509- ctx, logs := debugCtx()
510- db := mustOpenCtx(t, ctx, fx, compactOpts())
511- for i := 1; i <= 5; i++ {
512- mustApply(t, db, i)
513- }
514- fx.FailOp(faultfs.OpRename, 2, nil) // the snapshot
515- mustApply(t, db, 6)
516- waitFor(t, "compaction failure", func() bool { return logged(logs, "compaction failed") == 1 })
517- i := 7
518- waitFor(t, "retried compaction", func() bool {
519- if logged(logs, "compaction done") >= 1 {
520- return true
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+ i := 7
545+ i = applyUntil(t, db, i, "retried compaction", func() bool { return logged(logs, "compaction done") >= 1 })
546+ if err := db.Close(bg); err != nil {
547+ t.Fatal(err)
548+ }
549+ snaps, files := listFiles(t, fx)
550+ if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
551+ t.Fatalf("snapshots %v, logs %v", snaps, files)
521552 }
522- mustApply(t, db, i)
523- i++
524- return false
525553 })
526- if err := db.Close(bg); err != nil {
527- t.Fatal(err)
528- }
529- snaps, files := listFiles(t, fx)
530- if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
531- t.Fatalf("snapshots %v, logs %v", snaps, files)
532- }
533554 }
534555
535556 // switchFaults fail the creation of log-6 in the writer. The numbering is
@@ -551,49 +572,44 @@ var switchFaults = []struct {
551572 func TestLogSwitchFailure(t *testing.T) {
552573 for _, f := range switchFaults {
553574 t.Run(f.name, func(t *testing.T) {
554- fx := faultfs.New()
555- ctx, logs := debugCtx()
556- db := mustOpenCtx(t, ctx, fx, compactOpts())
557- for i := 1; i <= 5; i++ {
558- mustApply(t, db, i)
559- }
560- f.inject(fx)
561- mustApply(t, db, 6)
562- barrier(t, db)
563- if logged(logs, "log switch failed") != 1 {
564- t.Fatalf("switch failure not logged: %s", logs.String())
565- }
566- snaps, files := listFiles(t, fx)
567- if len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
568- t.Fatalf("after the failure: snapshots %v, logs %v", snaps, files)
569- }
570- names, _ := fx.ReadDir(testDir)
571- for _, n := range names {
572- if strings.HasSuffix(n, tmpSuffix) {
573- t.Errorf("leftover %s", n)
575+ synctest.Test(t, func(t *testing.T) {
576+ fx := faultfs.New()
577+ ctx, logs := debugCtx()
578+ db := mustOpenCtx(t, ctx, fx, compactOpts())
579+ for i := 1; i <= 5; i++ {
580+ mustApply(t, db, i)
581+ }
582+ f.inject(fx)
583+ mustApply(t, db, 6)
584+ barrier(t, db)
585+ if logged(logs, "log switch failed") != 1 {
586+ t.Fatalf("switch failure not logged: %s", logs.String())
587+ }
588+ snaps, files := listFiles(t, fx)
589+ if len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
590+ t.Fatalf("after the failure: snapshots %v, logs %v", snaps, files)
591+ }
592+ names, _ := fx.ReadDir(testDir)
593+ for _, n := range names {
594+ if strings.HasSuffix(n, tmpSuffix) {
595+ t.Errorf("leftover %s", n)
596+ }
574597 }
575- }
576598
577- // No retry on every commit: a failure costs another CompactAfter bytes.
578- mustApply(t, db, 7)
579- barrier(t, db)
580- if n := logged(logs, "log switch failed"); n != 1 {
581- t.Fatalf("%d switch attempts for 2 commits, want 1", n)
582- }
583- i := 8
584- waitFor(t, "compaction after the failed switch", func() bool {
585- if logged(logs, "compaction done") >= 1 {
586- return true
599+ // No retry on every commit: a failure costs another CompactAfter bytes.
600+ mustApply(t, db, 7)
601+ barrier(t, db)
602+ if n := logged(logs, "log switch failed"); n != 1 {
603+ t.Fatalf("%d switch attempts for 2 commits, want 1", n)
604+ }
605+ i := 8
606+ i = applyUntil(t, db, i, "compaction after the failed switch", func() bool { return logged(logs, "compaction done") >= 1 })
607+ if err := db.Close(bg); err != nil {
608+ t.Fatal(err)
587609 }
588- mustApply(t, db, i)
589- i++
590- return false
610+ db = mustOpen(t, fx, Options{})
611+ mustCheck(t, db, i-1)
591612 })
592- if err := db.Close(bg); err != nil {
593- t.Fatal(err)
594- }
595- db = mustOpen(t, fx, Options{})
596- mustCheck(t, db, i-1)
597613 })
598614 }
599615 }
@@ -613,35 +629,33 @@ func TestLogSwitchFailsDatabase(t *testing.T) {
613629 }
614630 for _, c := range cases {
615631 t.Run(c.name, func(t *testing.T) {
616- fx := faultfs.New()
617- ctx, logs := debugCtx()
618- db := mustOpenCtx(t, ctx, fx, compactOpts())
619- for i := 1; i <= 5; i++ {
620- mustApply(t, db, i)
621- }
622- c.inject(fx)
623- mustApply(t, db, 6) // acknowledged: the failure comes after the reply
624- waitFor(t, "failed state", func() bool {
625- select {
626- case <-db.failed:
627- return true
628- default:
629- return false
632+ synctest.Test(t, func(t *testing.T) {
633+ fx := faultfs.New()
634+ ctx, logs := debugCtx()
635+ db := mustOpenCtx(t, ctx, fx, compactOpts())
636+ for i := 1; i <= 5; i++ {
637+ mustApply(t, db, i)
638+ }
639+ c.inject(fx)
640+ mustApply(t, db, 6) // acknowledged: the failure comes after the reply
641+ settle() // the writer has seen the failure
642+ if !failedDB(db) {
643+ t.Fatal("the database did not fail")
644+ }
645+ var b Batch
646+ b.Put([]byte("later"), nil)
647+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrFailed) {
648+ t.Fatalf("Apply after the failure: %v", err)
649+ }
650+ if !strings.Contains(logs.String(), "database failed") {
651+ t.Errorf("failure not logged: %s", logs.String())
630652 }
653+ if err := db.Close(bg); !errors.Is(err, ErrFailed) {
654+ t.Fatalf("Close = %v, want the failure", err)
655+ }
656+ db = mustOpen(t, fx, Options{})
657+ mustCheck(t, db, 6)
631658 })
632- var b Batch
633- b.Put([]byte("later"), nil)
634- if _, err := db.Apply(bg, &b); !errors.Is(err, ErrFailed) {
635- t.Fatalf("Apply after the failure: %v", err)
636- }
637- if !strings.Contains(logs.String(), "database failed") {
638- t.Errorf("failure not logged: %s", logs.String())
639- }
640- if err := db.Close(bg); !errors.Is(err, ErrFailed) {
641- t.Fatalf("Close = %v, want the failure", err)
642- }
643- db = mustOpen(t, fx, Options{})
644- mustCheck(t, db, 6)
645659 })
646660 }
647661 }
@@ -652,79 +666,82 @@ func TestLogSwitchFailsDatabase(t *testing.T) {
652666 // sync are gone; the database must fail rather than continue, and what is
653667 // left must open as a prefix.
654668 func TestLogSwitchSyncFailureDropsUnsyncedData(t *testing.T) {
655- fx := faultfs.New()
656- fx.DropOnSyncFailure(true)
657- opts := compactOpts()
658- opts.NoSync = true
659- db := mustOpen(t, fx, opts)
660- for i := 1; i <= 5; i++ {
661- mustApply(t, db, i)
662- }
663- fx.FailSync(1, nil)
664- mustApply(t, db, 6)
665- waitFor(t, "failed state", func() bool {
666- select {
667- case <-db.failed:
668- return true
669- default:
670- return false
669+ synctest.Test(t, func(t *testing.T) {
670+ fx := faultfs.New()
671+ fx.DropOnSyncFailure(true)
672+ opts := compactOpts()
673+ opts.NoSync = true
674+ db := mustOpen(t, fx, opts)
675+ for i := 1; i <= 5; i++ {
676+ mustApply(t, db, i)
677+ }
678+ fx.FailSync(1, nil)
679+ mustApply(t, db, 6)
680+ settle() // the writer has seen the failure
681+ if !failedDB(db) {
682+ t.Fatal("the database did not fail")
683+ }
684+ if err := db.Close(bg); !errors.Is(err, ErrFailed) {
685+ t.Fatalf("Close = %v, want the failure", err)
686+ }
687+ db = mustOpen(t, fx, Options{})
688+ if n := mustCheck(t, db, 0); n != 0 {
689+ t.Fatalf("recovered %d batches, want 0 (all were dropped)", n)
671690 }
672691 })
673- if err := db.Close(bg); !errors.Is(err, ErrFailed) {
674- t.Fatalf("Close = %v, want the failure", err)
675- }
676- db = mustOpen(t, fx, Options{})
677- if n := mustCheck(t, db, 0); n != 0 {
678- t.Fatalf("recovered %d batches, want 0 (all were dropped)", n)
679- }
680692 }
681693
682694 // TestCloseWaitsForCompaction checks that Close does not cancel a running
683695 // compaction, waits for it, and leaves a directory that opens.
684696 func TestCloseWaitsForCompaction(t *testing.T) {
685- fx := faultfs.New()
686- started, release := make(chan struct{}), make(chan struct{})
687- var once sync.Once
688- h := &hookFS{FS: fx, hook: func(op, path string) error {
689- if op == "open" && strings.Contains(path, snapshotPrefix) {
690- once.Do(func() { close(started); <-release })
691- }
692- return nil
693- }}
694- ctx, logs := debugCtx()
695- db := mustOpenCtx(t, ctx, h, compactOpts())
696- for i := 1; i <= 6; i++ {
697- mustApply(t, db, i)
698- }
699- select {
700- case <-started:
701- case <-time.After(10 * time.Second):
702- t.Fatal("the compactor did not start")
703- }
704- closed := make(chan error, 1)
705- go func() { closed <- db.Close(bg) }()
706- select {
707- case err := <-closed:
708- t.Fatalf("Close returned during the compaction: %v", err)
709- case <-time.After(100 * time.Millisecond):
710- }
711- if _, err := db.Apply(bg, new(Batch)); !errors.Is(err, ErrClosed) {
712- t.Fatalf("Apply while closing: %v", err)
713- }
714- close(release)
715- select {
716- case err := <-closed:
717- if err != nil {
718- t.Fatalf("Close: %v", err)
697+ synctest.Test(t, func(t *testing.T) {
698+ fx := faultfs.New()
699+ started, release := make(chan struct{}), make(chan struct{})
700+ var once sync.Once
701+ h := &hookFS{FS: fx, hook: func(op, path string) error {
702+ if op == "open" && strings.Contains(path, snapshotPrefix) {
703+ once.Do(func() { close(started); <-release })
704+ }
705+ return nil
706+ }}
707+ ctx, logs := debugCtx()
708+ db := mustOpenCtx(t, ctx, h, compactOpts())
709+ for i := 1; i <= 6; i++ {
710+ mustApply(t, db, i)
719711 }
720- case <-time.After(10 * time.Second):
721- t.Fatal("Close did not return after the compaction was released")
722- }
723- if logged(logs, "compaction done") != 1 || !settled(t, fx, 6) {
724- t.Fatalf("the compaction did not complete: %s", logs.String())
725- }
726- db = mustOpen(t, fx, Options{})
727- mustCheck(t, db, 6)
712+ settle() // the compactor is parked in the hook
713+ select {
714+ case <-started:
715+ default:
716+ t.Fatal("the compactor did not start")
717+ }
718+ closed := make(chan error, 1)
719+ go func() { closed <- db.Close(bg) }()
720+ settle() // Close is blocked waiting for the compactor, which is parked in the hook
721+ select {
722+ case err := <-closed:
723+ t.Fatalf("Close returned during the compaction: %v", err)
724+ default:
725+ }
726+ if _, err := db.Apply(bg, new(Batch)); !errors.Is(err, ErrClosed) {
727+ t.Fatalf("Apply while closing: %v", err)
728+ }
729+ close(release)
730+ settle() // the compaction finishes and Close returns
731+ select {
732+ case err := <-closed:
733+ if err != nil {
734+ t.Fatalf("Close: %v", err)
735+ }
736+ default:
737+ t.Fatal("Close did not return after the compaction was released")
738+ }
739+ if logged(logs, "compaction done") != 1 || !settled(t, fx, 6) {
740+ t.Fatalf("the compaction did not complete: %s", logs.String())
741+ }
742+ db = mustOpen(t, fx, Options{})
743+ mustCheck(t, db, 6)
744+ })
728745 }
729746
730747 // TestCloseDuringCompaction (AC9) closes the database while writers run and
@@ -14,6 +14,7 @@ import (
14 "sync"14 "sync"
15 "sync/atomic"15 "sync/atomic"
16 "testing"16 "testing"
17+ "testing/synctest"
17 "time"18 "time"
18 19
19 "github.com/bnema/zerowrap"20 "github.com/bnema/zerowrap"
@@ -239,14 +240,34 @@ func settled(t testing.TB, fx fsys.FS, s uint64) bool {
239 return slices.Equal(snaps, []uint64{s}) && slices.Equal(logs, []uint64{s})240 return slices.Equal(snaps, []uint64{s}) && slices.Equal(logs, []uint64{s})
240 }241 }
241 242
242-func waitFor(t testing.TB, what string, cond func() bool) {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.
246+func 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.
251+func applyUntil(t testing.TB, db *DB, i int, what string, cond func() bool) int {
243 t.Helper()252 t.Helper()
244- deadline := time.Now().Add(10 * time.Second)253+ settle()
245- for !cond() {254+ for ; !cond(); i++ {
246- if time.Now().After(deadline) {255+ if i > 1000 {
247- t.Fatalf("timeout waiting for %s", what)256+ t.Fatalf("%s did not happen", what)
248 }257 }
249- time.Sleep(time.Millisecond)258+ mustApply(t, db, i)
259+ settle()
260+ }
261+ return i
262+}
263+
264+// failedDB reports whether the database is in the failed state.
265+func failedDB(db *DB) bool {
266+ select {
267+ case <-db.failed:
268+ return true
269+ default:
270+ return false
250 }271 }
251 }272 }
252 273
@@ -262,43 +283,48 @@ func logged(buf *syncBuffer, msg string) int {
262 }283 }
263 284
264 func TestCompactTrigger(t *testing.T) {285 func TestCompactTrigger(t *testing.T) {
265- fx := faultfs.New()286+ synctest.Test(t, func(t *testing.T) {
266- ctx, logs := debugCtx()287+ fx := faultfs.New()
267- db := mustOpenCtx(t, ctx, fx, compactOpts())288+ ctx, logs := debugCtx()
268-289+ db := mustOpenCtx(t, ctx, fx, compactOpts())
269- // Exactly CompactAfter bytes after the header: not above the threshold.290+
270- for i := 1; i <= 5; i++ {291+ // Exactly CompactAfter bytes after the header: not above the threshold.
271- mustApply(t, db, i)292+ for i := 1; i <= 5; i++ {
272- }293+ mustApply(t, db, i)
273- barrier(t, db)294+ }
274- if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {295+ barrier(t, db)
275- t.Fatalf("below the threshold: snapshots %v, logs %v", snaps, files)296+ if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
276- }297+ t.Fatalf("below the threshold: snapshots %v, logs %v", snaps, files)
298+ }
277 299
278- mustApply(t, db, 6)300+ mustApply(t, db, 6)
279- waitFor(t, "compaction", func() bool { return logged(logs, "compaction done") == 1 })301+ settle() // compaction: the background work has finished
280- if !settled(t, fx, 6) {302+ if logged(logs, "compaction done") != 1 {
281- snaps, files := listFiles(t, fx)303+ t.Fatalf("compaction did not happen: %s", logs.String())
282- t.Fatalf("after compaction: snapshots %v, logs %v; want snapshot-6 and log-6 only", snaps, files)304+ }
283- }305+ if !settled(t, fx, 6) {
284- if n := logged(logs, "log switch"); n != 1 {306+ snaps, files := listFiles(t, fx)
285- t.Fatalf("%d log switch messages, want 1", n)307+ t.Fatalf("after compaction: snapshots %v, logs %v; want snapshot-6 and log-6 only", snaps, files)
286- }308+ }
287- for _, key := range []string{`"seq":6`, `"keys":6`, `"bytes":`, `"duration":`} {309+ if n := logged(logs, "log switch"); n != 1 {
288- if !strings.Contains(logs.String(), key) {310+ t.Fatalf("%d log switch messages, want 1", n)
289- t.Errorf("compaction message lacks %s: %s", key, logs.String())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+ }
290 }316 }
291- }
292 317
293- // The new log takes the following batches; the old files stay gone.318+ // The new log takes the following batches; the old files stay gone.
294- mustApply(t, db, 7)319+ mustApply(t, db, 7)
295- if err := db.Close(bg); err != nil {320+ if err := db.Close(bg); err != nil {
296- t.Fatal(err)321+ t.Fatal(err)
297- }322+ }
298- db = mustOpen(t, fx, Options{})323+ db = mustOpen(t, fx, Options{})
299- if n := mustCheck(t, db, 7); n != 7 {324+ if n := mustCheck(t, db, 7); n != 7 {
300- t.Fatalf("recovered %d batches, want 7", n)325+ t.Fatalf("recovered %d batches, want 7", n)
301- }326+ }
327+ })
302 }328 }
303 329
304 func TestCompactDisabled(t *testing.T) {330 func TestCompactDisabled(t *testing.T) {
@@ -314,63 +340,62 @@ func TestCompactDisabled(t *testing.T) {
314 }340 }
315 341
316 func TestCompactSingleFlight(t *testing.T) {342 func TestCompactSingleFlight(t *testing.T) {
317- fx := faultfs.New()343+ synctest.Test(t, func(t *testing.T) {
318- started, release := make(chan struct{}), make(chan struct{})344+ fx := faultfs.New()
319- var once sync.Once345+ started, release := make(chan struct{}), make(chan struct{})
320- h := &hookFS{FS: fx, hook: func(op, path string) error {346+ var once sync.Once
321- if op == "open" && strings.Contains(path, snapshotPrefix) {347+ h := &hookFS{FS: fx, hook: func(op, path string) error {
322- once.Do(func() { close(started); <-release })348+ if op == "open" && strings.Contains(path, snapshotPrefix) {
323- }349+ once.Do(func() { close(started); <-release })
324- return nil350+ }
325- }}351+ return nil
326- ctx, logs := debugCtx()352+ }}
327- db := mustOpenCtx(t, ctx, h, compactOpts())353+ ctx, logs := debugCtx()
328- for i := 1; i <= 6; i++ {354+ db := mustOpenCtx(t, ctx, h, compactOpts())
329- mustApply(t, db, i)355+ for i := 1; i <= 6; i++ {
330- }356+ mustApply(t, db, i)
331- select {357+ }
332- case <-started:358+ settle() // the compactor is parked in the hook
333- case <-time.After(10 * time.Second):359+ select {
334- t.Fatal("the compactor did not start")360+ case <-started:
335- }361+ default:
362+ t.Fatal("the compactor did not start")
363+ }
336 364
337- // The compactor is stuck. The writer is not: it goes on far past the365+ // The compactor is stuck. The writer is not: it goes on far past the
338- // threshold without starting a second compaction.366+ // threshold without starting a second compaction.
339- for i := 7; i <= 60; i++ {367+ for i := 7; i <= 60; i++ {
340- mustApply(t, db, i)368+ mustApply(t, db, i)
341- }369+ }
342- barrier(t, db)370+ barrier(t, db)
343- if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0, 6}) {371+ if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0, 6}) {
344- t.Fatalf("while compacting: snapshots %v, logs %v", snaps, files)372+ t.Fatalf("while compacting: snapshots %v, logs %v", snaps, files)
345- }373+ }
346- if n := logged(logs, "log switch"); n != 1 {374+ if n := logged(logs, "log switch"); n != 1 {
347- t.Fatalf("%d log switches while a compaction was running, want 1", n)375+ t.Fatalf("%d log switches while a compaction was running, want 1", n)
348- }376+ }
349 377
350- close(release)378+ close(release)
351- waitFor(t, "first compaction", func() bool { return logged(logs, "compaction done") == 1 })379+ settle() // first compaction: the background work has finished
352- // The log is still over the threshold: the next commit compacts again.380+ if logged(logs, "compaction done") != 1 {
353- i := 61381+ t.Fatalf("first compaction did not happen: %s", logs.String())
354- waitFor(t, "second compaction", func() bool {
355- if logged(logs, "compaction done") >= 2 {
356- return true
357 }382 }
358- mustApply(t, db, i)383+ // The log is still over the threshold: the next commit compacts again.
359- i++384+ i := 61
360- return false385+ 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)
361 })398 })
362- if n := logged(logs, "log switch"); n != 2 {
363- t.Fatalf("%d log switches, want 2", n)
364- }
365- if err := db.Close(bg); err != nil {
366- t.Fatal(err)
367- }
368- snaps, files := listFiles(t, fx)
369- if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
370- t.Fatalf("after the second compaction: snapshots %v, logs %v", snaps, files)
371- }
372- db = mustOpen(t, fx, Options{})
373- mustCheck(t, db, i-1)
374 }399 }
375 400
376 func TestCompactRemovesOldFiles(t *testing.T) {401 func TestCompactRemovesOldFiles(t *testing.T) {
@@ -460,76 +485,72 @@ var compactorFaults = []struct {
460 func TestCompactorFailure(t *testing.T) {485 func TestCompactorFailure(t *testing.T) {
461 for _, f := range compactorFaults {486 for _, f := range compactorFaults {
462 t.Run(f.name, func(t *testing.T) {487 t.Run(f.name, func(t *testing.T) {
463- fx := faultfs.New()488+ synctest.Test(t, func(t *testing.T) {
464- ctx, logs := debugCtx()489+ fx := faultfs.New()
465- db := mustOpenCtx(t, ctx, fx, compactOpts())490+ ctx, logs := debugCtx()
466- for i := 1; i <= 5; i++ {491+ db := mustOpenCtx(t, ctx, fx, compactOpts())
467- mustApply(t, db, i)492+ for i := 1; i <= 5; i++ {
468- }493+ mustApply(t, db, i)
469- f.inject(fx)494+ }
470- mustApply(t, db, 6)495+ f.inject(fx)
471- waitFor(t, "compaction failure", func() bool { return logged(logs, "compaction failed") == 1 })496+ mustApply(t, db, 6)
472- if !strings.Contains(logs.String(), `"level":"warn"`) {497+ settle() // compaction failure: the background work has finished
473- t.Errorf("failure not logged as a warning: %s", logs.String())498+ if logged(logs, "compaction failed") != 1 {
474- }499+ t.Fatalf("compaction failure did not happen: %s", logs.String())
475- mustApply(t, db, 7) // the database is not failed500+ }
476- if err := db.Close(bg); err != nil {501+ if !strings.Contains(logs.String(), `"level":"warn"`) {
477- t.Fatalf("Close: %v", err)502+ t.Errorf("failure not logged as a warning: %s", logs.String())
478- }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+ }
479 508
480- // Reopen the partial state: it recovers and compacts again.509+ // Reopen the partial state: it recovers and compacts again.
481- ctx, logs = debugCtx()510+ ctx, logs = debugCtx()
482- db = mustOpenCtx(t, ctx, fx, compactOpts())511+ db = mustOpenCtx(t, ctx, fx, compactOpts())
483- mustCheck(t, db, 7)512+ mustCheck(t, db, 7)
484- i := 8513+ i := 8
485- waitFor(t, "retried compaction", func() bool {514+ i = applyUntil(t, db, i, "retried compaction", func() bool { return logged(logs, "compaction done") >= 1 })
486- if logged(logs, "compaction done") >= 1 {515+ if err := db.Close(bg); err != nil {
487- return true516+ t.Fatal(err)
488 }517 }
489- mustApply(t, db, i)518+ snaps, files := listFiles(t, fx)
490- i++519+ if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
491- return false520+ t.Fatalf("after the retry: snapshots %v, logs %v", snaps, files)
521+ }
522+ db = mustOpen(t, fx, Options{})
523+ mustCheck(t, db, i-1)
492 })524 })
493- if err := db.Close(bg); err != nil {
494- t.Fatal(err)
495- }
496- snaps, files := listFiles(t, fx)
497- if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
498- t.Fatalf("after the retry: snapshots %v, logs %v", snaps, files)
499- }
500- db = mustOpen(t, fx, Options{})
501- mustCheck(t, db, i-1)
502 })525 })
503 }526 }
504 }527 }
505 528
506 // TestCompactorRetryWithoutReopen checks the retry in the same process.529 // TestCompactorRetryWithoutReopen checks the retry in the same process.
507 func TestCompactorRetryWithoutReopen(t *testing.T) {530 func TestCompactorRetryWithoutReopen(t *testing.T) {
508- fx := faultfs.New()531+ synctest.Test(t, func(t *testing.T) {
509- ctx, logs := debugCtx()532+ fx := faultfs.New()
510- db := mustOpenCtx(t, ctx, fx, compactOpts())533+ ctx, logs := debugCtx()
511- for i := 1; i <= 5; i++ {534+ db := mustOpenCtx(t, ctx, fx, compactOpts())
512- mustApply(t, db, i)535+ for i := 1; i <= 5; i++ {
513- }536+ mustApply(t, db, i)
514- fx.FailOp(faultfs.OpRename, 2, nil) // the snapshot537+ }
515- mustApply(t, db, 6)538+ fx.FailOp(faultfs.OpRename, 2, nil) // the snapshot
516- waitFor(t, "compaction failure", func() bool { return logged(logs, "compaction failed") == 1 })539+ mustApply(t, db, 6)
517- i := 7540+ settle() // compaction failure: the background work has finished
518- waitFor(t, "retried compaction", func() bool {541+ if logged(logs, "compaction failed") != 1 {
519- if logged(logs, "compaction done") >= 1 {542+ t.Fatalf("compaction failure did not happen: %s", logs.String())
520- return true543+ }
544+ i := 7
545+ i = applyUntil(t, db, i, "retried compaction", func() bool { return logged(logs, "compaction done") >= 1 })
546+ if err := db.Close(bg); err != nil {
547+ t.Fatal(err)
548+ }
549+ snaps, files := listFiles(t, fx)
550+ if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
551+ t.Fatalf("snapshots %v, logs %v", snaps, files)
521 }552 }
522- mustApply(t, db, i)
523- i++
524- return false
525 })553 })
526- if err := db.Close(bg); err != nil {
527- t.Fatal(err)
528- }
529- snaps, files := listFiles(t, fx)
530- if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] <= 6 {
531- t.Fatalf("snapshots %v, logs %v", snaps, files)
532- }
533 }554 }
534 555
535 // switchFaults fail the creation of log-6 in the writer. The numbering is556 // switchFaults fail the creation of log-6 in the writer. The numbering is
@@ -551,49 +572,44 @@ var switchFaults = []struct {
551 func TestLogSwitchFailure(t *testing.T) {572 func TestLogSwitchFailure(t *testing.T) {
552 for _, f := range switchFaults {573 for _, f := range switchFaults {
553 t.Run(f.name, func(t *testing.T) {574 t.Run(f.name, func(t *testing.T) {
554- fx := faultfs.New()575+ synctest.Test(t, func(t *testing.T) {
555- ctx, logs := debugCtx()576+ fx := faultfs.New()
556- db := mustOpenCtx(t, ctx, fx, compactOpts())577+ ctx, logs := debugCtx()
557- for i := 1; i <= 5; i++ {578+ db := mustOpenCtx(t, ctx, fx, compactOpts())
558- mustApply(t, db, i)579+ for i := 1; i <= 5; i++ {
559- }580+ mustApply(t, db, i)
560- f.inject(fx)581+ }
561- mustApply(t, db, 6)582+ f.inject(fx)
562- barrier(t, db)583+ mustApply(t, db, 6)
563- if logged(logs, "log switch failed") != 1 {584+ barrier(t, db)
564- t.Fatalf("switch failure not logged: %s", logs.String())585+ if logged(logs, "log switch failed") != 1 {
565- }586+ t.Fatalf("switch failure not logged: %s", logs.String())
566- snaps, files := listFiles(t, fx)587+ }
567- if len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {588+ snaps, files := listFiles(t, fx)
568- t.Fatalf("after the failure: snapshots %v, logs %v", snaps, files)589+ if len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
569- }590+ t.Fatalf("after the failure: snapshots %v, logs %v", snaps, files)
570- names, _ := fx.ReadDir(testDir)591+ }
571- for _, n := range names {592+ names, _ := fx.ReadDir(testDir)
572- if strings.HasSuffix(n, tmpSuffix) {593+ for _, n := range names {
573- t.Errorf("leftover %s", n)594+ if strings.HasSuffix(n, tmpSuffix) {
595+ t.Errorf("leftover %s", n)
596+ }
574 }597 }
575- }
576 598
577- // No retry on every commit: a failure costs another CompactAfter bytes.599+ // No retry on every commit: a failure costs another CompactAfter bytes.
578- mustApply(t, db, 7)600+ mustApply(t, db, 7)
579- barrier(t, db)601+ barrier(t, db)
580- if n := logged(logs, "log switch failed"); n != 1 {602+ if n := logged(logs, "log switch failed"); n != 1 {
581- t.Fatalf("%d switch attempts for 2 commits, want 1", n)603+ t.Fatalf("%d switch attempts for 2 commits, want 1", n)
582- }604+ }
583- i := 8605+ i := 8
584- waitFor(t, "compaction after the failed switch", func() bool {606+ i = applyUntil(t, db, i, "compaction after the failed switch", func() bool { return logged(logs, "compaction done") >= 1 })
585- if logged(logs, "compaction done") >= 1 {607+ if err := db.Close(bg); err != nil {
586- return true608+ t.Fatal(err)
587 }609 }
588- mustApply(t, db, i)610+ db = mustOpen(t, fx, Options{})
589- i++611+ mustCheck(t, db, i-1)
590- return false
591 })612 })
592- if err := db.Close(bg); err != nil {
593- t.Fatal(err)
594- }
595- db = mustOpen(t, fx, Options{})
596- mustCheck(t, db, i-1)
597 })613 })
598 }614 }
599 }615 }
@@ -613,35 +629,33 @@ func TestLogSwitchFailsDatabase(t *testing.T) {
613 }629 }
614 for _, c := range cases {630 for _, c := range cases {
615 t.Run(c.name, func(t *testing.T) {631 t.Run(c.name, func(t *testing.T) {
616- fx := faultfs.New()632+ synctest.Test(t, func(t *testing.T) {
617- ctx, logs := debugCtx()633+ fx := faultfs.New()
618- db := mustOpenCtx(t, ctx, fx, compactOpts())634+ ctx, logs := debugCtx()
619- for i := 1; i <= 5; i++ {635+ db := mustOpenCtx(t, ctx, fx, compactOpts())
620- mustApply(t, db, i)636+ for i := 1; i <= 5; i++ {
621- }637+ mustApply(t, db, i)
622- c.inject(fx)638+ }
623- mustApply(t, db, 6) // acknowledged: the failure comes after the reply639+ c.inject(fx)
624- waitFor(t, "failed state", func() bool {640+ mustApply(t, db, 6) // acknowledged: the failure comes after the reply
625- select {641+ settle() // the writer has seen the failure
626- case <-db.failed:642+ if !failedDB(db) {
627- return true643+ t.Fatal("the database did not fail")
628- default:644+ }
629- return false645+ var b Batch
646+ b.Put([]byte("later"), nil)
647+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrFailed) {
648+ t.Fatalf("Apply after the failure: %v", err)
649+ }
650+ if !strings.Contains(logs.String(), "database failed") {
651+ t.Errorf("failure not logged: %s", logs.String())
630 }652 }
653+ if err := db.Close(bg); !errors.Is(err, ErrFailed) {
654+ t.Fatalf("Close = %v, want the failure", err)
655+ }
656+ db = mustOpen(t, fx, Options{})
657+ mustCheck(t, db, 6)
631 })658 })
632- var b Batch
633- b.Put([]byte("later"), nil)
634- if _, err := db.Apply(bg, &b); !errors.Is(err, ErrFailed) {
635- t.Fatalf("Apply after the failure: %v", err)
636- }
637- if !strings.Contains(logs.String(), "database failed") {
638- t.Errorf("failure not logged: %s", logs.String())
639- }
640- if err := db.Close(bg); !errors.Is(err, ErrFailed) {
641- t.Fatalf("Close = %v, want the failure", err)
642- }
643- db = mustOpen(t, fx, Options{})
644- mustCheck(t, db, 6)
645 })659 })
646 }660 }
647 }661 }
@@ -652,79 +666,82 @@ func TestLogSwitchFailsDatabase(t *testing.T) {
652 // sync are gone; the database must fail rather than continue, and what is666 // sync are gone; the database must fail rather than continue, and what is
653 // left must open as a prefix.667 // left must open as a prefix.
654 func TestLogSwitchSyncFailureDropsUnsyncedData(t *testing.T) {668 func TestLogSwitchSyncFailureDropsUnsyncedData(t *testing.T) {
655- fx := faultfs.New()669+ synctest.Test(t, func(t *testing.T) {
656- fx.DropOnSyncFailure(true)670+ fx := faultfs.New()
657- opts := compactOpts()671+ fx.DropOnSyncFailure(true)
658- opts.NoSync = true672+ opts := compactOpts()
659- db := mustOpen(t, fx, opts)673+ opts.NoSync = true
660- for i := 1; i <= 5; i++ {674+ db := mustOpen(t, fx, opts)
661- mustApply(t, db, i)675+ for i := 1; i <= 5; i++ {
662- }676+ mustApply(t, db, i)
663- fx.FailSync(1, nil)677+ }
664- mustApply(t, db, 6)678+ fx.FailSync(1, nil)
665- waitFor(t, "failed state", func() bool {679+ mustApply(t, db, 6)
666- select {680+ settle() // the writer has seen the failure
667- case <-db.failed:681+ if !failedDB(db) {
668- return true682+ t.Fatal("the database did not fail")
669- default:683+ }
670- return false684+ if err := db.Close(bg); !errors.Is(err, ErrFailed) {
685+ t.Fatalf("Close = %v, want the failure", err)
686+ }
687+ db = mustOpen(t, fx, Options{})
688+ if n := mustCheck(t, db, 0); n != 0 {
689+ t.Fatalf("recovered %d batches, want 0 (all were dropped)", n)
671 }690 }
672 })691 })
673- if err := db.Close(bg); !errors.Is(err, ErrFailed) {
674- t.Fatalf("Close = %v, want the failure", err)
675- }
676- db = mustOpen(t, fx, Options{})
677- if n := mustCheck(t, db, 0); n != 0 {
678- t.Fatalf("recovered %d batches, want 0 (all were dropped)", n)
679- }
680 }692 }
681 693
682 // TestCloseWaitsForCompaction checks that Close does not cancel a running694 // TestCloseWaitsForCompaction checks that Close does not cancel a running
683 // compaction, waits for it, and leaves a directory that opens.695 // compaction, waits for it, and leaves a directory that opens.
684 func TestCloseWaitsForCompaction(t *testing.T) {696 func TestCloseWaitsForCompaction(t *testing.T) {
685- fx := faultfs.New()697+ synctest.Test(t, func(t *testing.T) {
686- started, release := make(chan struct{}), make(chan struct{})698+ fx := faultfs.New()
687- var once sync.Once699+ started, release := make(chan struct{}), make(chan struct{})
688- h := &hookFS{FS: fx, hook: func(op, path string) error {700+ var once sync.Once
689- if op == "open" && strings.Contains(path, snapshotPrefix) {701+ h := &hookFS{FS: fx, hook: func(op, path string) error {
690- once.Do(func() { close(started); <-release })702+ if op == "open" && strings.Contains(path, snapshotPrefix) {
691- }703+ once.Do(func() { close(started); <-release })
692- return nil704+ }
693- }}705+ return nil
694- ctx, logs := debugCtx()706+ }}
695- db := mustOpenCtx(t, ctx, h, compactOpts())707+ ctx, logs := debugCtx()
696- for i := 1; i <= 6; i++ {708+ db := mustOpenCtx(t, ctx, h, compactOpts())
697- mustApply(t, db, i)709+ for i := 1; i <= 6; i++ {
698- }710+ mustApply(t, db, i)
699- select {
700- case <-started:
701- case <-time.After(10 * time.Second):
702- t.Fatal("the compactor did not start")
703- }
704- closed := make(chan error, 1)
705- go func() { closed <- db.Close(bg) }()
706- select {
707- case err := <-closed:
708- t.Fatalf("Close returned during the compaction: %v", err)
709- case <-time.After(100 * time.Millisecond):
710- }
711- if _, err := db.Apply(bg, new(Batch)); !errors.Is(err, ErrClosed) {
712- t.Fatalf("Apply while closing: %v", err)
713- }
714- close(release)
715- select {
716- case err := <-closed:
717- if err != nil {
718- t.Fatalf("Close: %v", err)
719 }711 }
720- case <-time.After(10 * time.Second):712+ settle() // the compactor is parked in the hook
721- t.Fatal("Close did not return after the compaction was released")713+ select {
722- }714+ case <-started:
723- if logged(logs, "compaction done") != 1 || !settled(t, fx, 6) {715+ default:
724- t.Fatalf("the compaction did not complete: %s", logs.String())716+ t.Fatal("the compactor did not start")
725- }717+ }
726- db = mustOpen(t, fx, Options{})718+ closed := make(chan error, 1)
727- mustCheck(t, db, 6)719+ go func() { closed <- db.Close(bg) }()
720+ settle() // Close is blocked waiting for the compactor, which is parked in the hook
721+ select {
722+ case err := <-closed:
723+ t.Fatalf("Close returned during the compaction: %v", err)
724+ default:
725+ }
726+ if _, err := db.Apply(bg, new(Batch)); !errors.Is(err, ErrClosed) {
727+ t.Fatalf("Apply while closing: %v", err)
728+ }
729+ close(release)
730+ settle() // the compaction finishes and Close returns
731+ select {
732+ case err := <-closed:
733+ if err != nil {
734+ t.Fatalf("Close: %v", err)
735+ }
736+ default:
737+ t.Fatal("Close did not return after the compaction was released")
738+ }
739+ if logged(logs, "compaction done") != 1 || !settled(t, fx, 6) {
740+ t.Fatalf("the compaction did not complete: %s", logs.String())
741+ }
742+ db = mustOpen(t, fx, Options{})
743+ mustCheck(t, db, 6)
744+ })
728 }745 }
729 746
730 // TestCloseDuringCompaction (AC9) closes the database while writers run and747 // TestCloseDuringCompaction (AC9) closes the database while writers run and
M db.go +4 -6
@@ -228,19 +228,17 @@ func (db *DB) Apply(ctx context.Context, b *Batch) (uint64, error) {
228228 return 0, err
229229 }
230230
231- r := requestPool.Get().(*request)
232- r.b = b
231+ r := &request{b: b, reply: make(chan result, 1)}
233232 select {
234233 case db.reqCh <- r:
235234 case <-ctx.Done():
236- return 0, r.release(ctx.Err())
235+ return 0, ctx.Err()
237236 case <-db.closing:
238- return 0, r.release(db.stopped())
237+ return 0, db.stopped()
239238 case <-db.failed:
240- return 0, r.release(db.stopped())
239+ return 0, db.stopped()
241240 }
242241 res := <-r.reply // the writer always answers a request it has taken
243- _ = r.release(nil)
244242 return res.seq, res.err
245243 }
246244
@@ -228,19 +228,17 @@ func (db *DB) Apply(ctx context.Context, b *Batch) (uint64, error) {
228 return 0, err228 return 0, err
229 }229 }
230 230
231- r := requestPool.Get().(*request)231+ r := &request{b: b, reply: make(chan result, 1)}
232- r.b = b
233 select {232 select {
234 case db.reqCh <- r:233 case db.reqCh <- r:
235 case <-ctx.Done():234 case <-ctx.Done():
236- return 0, r.release(ctx.Err())235+ return 0, ctx.Err()
237 case <-db.closing:236 case <-db.closing:
238- return 0, r.release(db.stopped())237+ return 0, db.stopped()
239 case <-db.failed:238 case <-db.failed:
240- return 0, r.release(db.stopped())239+ return 0, db.stopped()
241 }240 }
242 res := <-r.reply // the writer always answers a request it has taken241 res := <-r.reply // the writer always answers a request it has taken
243- _ = r.release(nil)
244 return res.seq, res.err242 return res.seq, res.err
245 }243 }
246 244
M db_test.go +157 -138
@@ -16,6 +16,7 @@ import (
1616 "sync"
1717 "sync/atomic"
1818 "testing"
19+ "testing/synctest"
1920 "time"
2021
2122 "github.com/bnema/zerowrap"
@@ -354,7 +355,9 @@ func TestPreconditionsBeforeOps(t *testing.T) {
354355 // holdWriter applies a batch whose sync is slow and returns once the writer
355356 // is inside that sync, so batches applied next queue up and form one group
356357 // (GroupWindow 0 drains everything already queued). wait returns when the
357-// held batch is acknowledged; the sync delay stays set until then.
358+// held batch is acknowledged; the sync delay stays set until then. It must be
359+// called inside a synctest bubble: the writer sleeps in the sync on the fake
360+// clock, so no real time passes.
358361 func holdWriter(t *testing.T, db *DB, fx *faultfs.FS) (wait func()) {
359362 t.Helper()
360363 base := fx.Syncs()
@@ -366,8 +369,9 @@ func holdWriter(t *testing.T, db *DB, fx *faultfs.FS) (wait func()) {
366369 _, err := db.Apply(bg, &b)
367370 done <- err
368371 }()
369- for fx.Syncs() == base {
370- time.Sleep(time.Millisecond)
372+ synctest.Wait() // the writer is parked in the delayed sync
373+ if fx.Syncs() == base {
374+ t.Fatal("the writer did not reach the sync")
371375 }
372376 return func() {
373377 t.Helper()
@@ -381,28 +385,30 @@ func holdWriter(t *testing.T, db *DB, fx *faultfs.FS) (wait func()) {
381385 // TestGroupPreconditions checks that preconditions see earlier batches of
382386 // the same group: the second batch of the group must conflict.
383387 func TestGroupPreconditions(t *testing.T) {
384- fx := faultfs.New()
385- db := mustOpen(t, fx, Options{})
386- syncs := fx.Syncs()
387- wait := holdWriter(t, db, fx)
388- errs := make([]error, 2)
389- var wg sync.WaitGroup
390- for i := range errs {
391- wg.Go(func() {
392- var b Batch
393- b.RequireAbsent([]byte("k"))
394- b.Put([]byte("k"), []byte{byte('a' + i)})
395- _, errs[i] = db.Apply(bg, &b)
396- })
397- }
398- wg.Wait()
399- wait()
400- if (errs[0] == nil) == (errs[1] == nil) {
401- t.Fatalf("errors %v, want exactly one success", errs)
402- }
403- if got := fx.Syncs() - syncs; got != 2 {
404- t.Fatalf("%d syncs, want 2 (held batch, then one group)", got)
405- }
388+ synctest.Test(t, func(t *testing.T) {
389+ fx := faultfs.New()
390+ db := mustOpen(t, fx, Options{})
391+ syncs := fx.Syncs()
392+ wait := holdWriter(t, db, fx)
393+ errs := make([]error, 2)
394+ var wg sync.WaitGroup
395+ for i := range errs {
396+ wg.Go(func() {
397+ var b Batch
398+ b.RequireAbsent([]byte("k"))
399+ b.Put([]byte("k"), []byte{byte('a' + i)})
400+ _, errs[i] = db.Apply(bg, &b)
401+ })
402+ }
403+ wg.Wait()
404+ wait()
405+ if (errs[0] == nil) == (errs[1] == nil) {
406+ t.Fatalf("errors %v, want exactly one success", errs)
407+ }
408+ if got := fx.Syncs() - syncs; got != 2 {
409+ t.Fatalf("%d syncs, want 2 (held batch, then one group)", got)
410+ }
411+ })
406412 }
407413
408414 func TestEmptyAndConflictOnlyBatches(t *testing.T) {
@@ -462,25 +468,27 @@ func TestEmptyAndConflictOnlyBatches(t *testing.T) {
462468 // TestNoWriteForSkippedGroup checks that a group of only conflicts and empty
463469 // batches neither writes nor syncs.
464470 func TestNoWriteForSkippedGroup(t *testing.T) {
465- fx := faultfs.New()
466- db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
467- putKV(t, db, "a", "1")
468- writes, syncs := fx.Writes(), fx.Syncs()
469- var wg sync.WaitGroup
470- for range 4 {
471- wg.Go(func() {
472- var b Batch
473- b.RequireAbsent([]byte("a"))
474- b.Put([]byte("x"), nil)
475- if _, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) {
476- t.Errorf("err = %v, want ErrConflict", err)
477- }
478- })
479- }
480- wg.Wait()
481- if fx.Writes() != writes || fx.Syncs() != syncs {
482- t.Fatalf("writes %d→%d syncs %d→%d", writes, fx.Writes(), syncs, fx.Syncs())
483- }
471+ synctest.Test(t, func(t *testing.T) {
472+ fx := faultfs.New()
473+ db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
474+ putKV(t, db, "a", "1")
475+ writes, syncs := fx.Writes(), fx.Syncs()
476+ var wg sync.WaitGroup
477+ for range 4 {
478+ wg.Go(func() {
479+ var b Batch
480+ b.RequireAbsent([]byte("a"))
481+ b.Put([]byte("x"), nil)
482+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) {
483+ t.Errorf("err = %v, want ErrConflict", err)
484+ }
485+ })
486+ }
487+ wg.Wait()
488+ if fx.Writes() != writes || fx.Syncs() != syncs {
489+ t.Fatalf("writes %d→%d syncs %d→%d", writes, fx.Writes(), syncs, fx.Syncs())
490+ }
491+ })
484492 }
485493
486494 func TestGroupCommitSyncs(t *testing.T) {
@@ -527,33 +535,35 @@ func TestNoSync(t *testing.T) {
527535 // TestLargeBatches covers a batch above the 4 MiB group cap (a group of one)
528536 // and a burst of batches that does not fit in one group.
529537 func TestLargeBatches(t *testing.T) {
530- fx := faultfs.New()
531- db := mustOpen(t, fx, Options{GroupWindow: 50 * time.Millisecond})
532- big := strings.Repeat("B", 5<<20)
533- putKV(t, db, "big", big)
534-
535- const n = 6
536- chunk := strings.Repeat("c", 3<<19) // 1.5 MiB: three do not fit in 4 MiB
537- seqs := make([]uint64, n)
538- var wg sync.WaitGroup
539- for i := range n {
540- wg.Go(func() { seqs[i] = putKV(t, db, fmt.Sprintf("chunk%d", i), chunk) })
541- }
542- wg.Wait()
543- slices.Sort(seqs)
544- for i, s := range seqs {
545- if s != uint64(i+2) {
546- t.Fatalf("seqs = %v", seqs)
538+ synctest.Test(t, func(t *testing.T) {
539+ fx := faultfs.New()
540+ db := mustOpen(t, fx, Options{GroupWindow: 50 * time.Millisecond})
541+ big := strings.Repeat("B", 5<<20)
542+ putKV(t, db, "big", big)
543+
544+ const n = 6
545+ chunk := strings.Repeat("c", 3<<19) // 1.5 MiB: three do not fit in 4 MiB
546+ seqs := make([]uint64, n)
547+ var wg sync.WaitGroup
548+ for i := range n {
549+ wg.Go(func() { seqs[i] = putKV(t, db, fmt.Sprintf("chunk%d", i), chunk) })
547550 }
548- }
549- if err := db.Close(bg); err != nil {
550- t.Fatal(err)
551- }
552- s := mustOpen(t, fx, Options{}).Snapshot(bg)
553- mustGet(t, s, "big", big)
554- for i := range n {
555- mustGet(t, s, fmt.Sprintf("chunk%d", i), chunk)
556- }
551+ wg.Wait()
552+ slices.Sort(seqs)
553+ for i, s := range seqs {
554+ if s != uint64(i+2) {
555+ t.Fatalf("seqs = %v", seqs)
556+ }
557+ }
558+ if err := db.Close(bg); err != nil {
559+ t.Fatal(err)
560+ }
561+ s := mustOpen(t, fx, Options{}).Snapshot(bg)
562+ mustGet(t, s, "big", big)
563+ for i := range n {
564+ mustGet(t, s, fmt.Sprintf("chunk%d", i), chunk)
565+ }
566+ })
557567 }
558568
559569 func TestInvalidInputs(t *testing.T) {
@@ -675,58 +685,65 @@ func TestApplyAfterClose(t *testing.T) {
675685 // TestContextBeforeDequeue blocks the writer in a slow sync so that a second
676686 // batch cannot be dequeued, then cancels its context.
677687 func TestContextBeforeDequeue(t *testing.T) {
678- fx := faultfs.New()
679- db := mustOpen(t, fx, Options{})
688+ synctest.Test(t, func(t *testing.T) {
689+ fx := faultfs.New()
690+ db := mustOpen(t, fx, Options{})
680691
681- done := make(chan error, 1)
682- base := fx.Syncs()
683- fx.SetSyncDelay(300 * time.Millisecond)
684- go func() {
685- var b Batch
686- b.Put([]byte("first"), []byte("1"))
687- _, err := db.Apply(bg, &b)
688- done <- err
689- }()
690- for fx.Syncs() == base { // the writer is now inside the delayed sync
691- time.Sleep(time.Millisecond)
692- }
692+ done := make(chan error, 1)
693+ base := fx.Syncs()
694+ fx.SetSyncDelay(300 * time.Millisecond)
695+ go func() {
696+ var b Batch
697+ b.Put([]byte("first"), []byte("1"))
698+ _, err := db.Apply(bg, &b)
699+ done <- err
700+ }()
701+ synctest.Wait() // the writer is now inside the delayed sync
702+ if fx.Syncs() == base {
703+ t.Fatal("the writer did not reach the sync")
704+ }
705+ start := time.Now()
693706
694- ctx, cancel := context.WithCancel(bg)
695- second := make(chan error, 1)
696- go func() {
697- var b Batch
698- b.Put([]byte("second"), []byte("2"))
699- _, err := db.Apply(ctx, &b)
700- second <- err
701- }()
702- time.Sleep(50 * time.Millisecond) // let Apply reach its blocking select
703- cancel()
704- if err := <-second; !errors.Is(err, context.Canceled) {
705- t.Fatalf("Apply = %v, want context.Canceled", err)
706- }
707- if err := <-done; err != nil {
708- t.Fatalf("first Apply: %v", err)
709- }
710- fx.SetSyncDelay(0)
707+ ctx, cancel := context.WithCancel(bg)
708+ second := make(chan error, 1)
709+ go func() {
710+ var b Batch
711+ b.Put([]byte("second"), []byte("2"))
712+ _, err := db.Apply(ctx, &b)
713+ second <- err
714+ }()
715+ synctest.Wait() // Apply is blocked in its select: the writer cannot dequeue
716+ cancel()
717+ if err := <-second; !errors.Is(err, context.Canceled) {
718+ t.Fatalf("Apply = %v, want context.Canceled", err)
719+ }
720+ if d := time.Since(start); d != 0 {
721+ t.Fatalf("the cancelled Apply returned after %v, it must not wait for the sync", d)
722+ }
723+ if err := <-done; err != nil {
724+ t.Fatalf("first Apply: %v", err)
725+ }
726+ fx.SetSyncDelay(0)
711727
712- // An already cancelled context never reaches the writer either.
713- var b Batch
714- b.Put([]byte("third"), nil)
715- if _, err := db.Apply(ctx, &b); !errors.Is(err, context.Canceled) {
716- t.Fatalf("Apply with cancelled ctx = %v", err)
717- }
718- putKV(t, db, "fourth", "4") // the writer is still healthy
719- if err := db.Close(bg); err != nil {
720- t.Fatal(err)
721- }
722- s := mustOpen(t, fx, Options{}).Snapshot(bg)
723- mustGet(t, s, "first", "1")
724- mustGet(t, s, "fourth", "4")
725- mustMissing(t, s, "second")
726- mustMissing(t, s, "third")
727- if s.Seq() != 2 {
728- t.Fatalf("Seq = %d, want 2", s.Seq())
729- }
728+ // An already cancelled context never reaches the writer either.
729+ var b Batch
730+ b.Put([]byte("third"), nil)
731+ if _, err := db.Apply(ctx, &b); !errors.Is(err, context.Canceled) {
732+ t.Fatalf("Apply with cancelled ctx = %v", err)
733+ }
734+ putKV(t, db, "fourth", "4") // the writer is still healthy
735+ if err := db.Close(bg); err != nil {
736+ t.Fatal(err)
737+ }
738+ s := mustOpen(t, fx, Options{}).Snapshot(bg)
739+ mustGet(t, s, "first", "1")
740+ mustGet(t, s, "fourth", "4")
741+ mustMissing(t, s, "second")
742+ mustMissing(t, s, "third")
743+ if s.Seq() != 2 {
744+ t.Fatalf("Seq = %d, want 2", s.Seq())
745+ }
746+ })
730747 }
731748
732749 func TestCloseDuringApplies(t *testing.T) {
@@ -804,7 +821,7 @@ func TestInjectedFailures(t *testing.T) {
804821 for _, f := range failures {
805822 for _, powerLoss := range []bool{false, true} {
806823 t.Run(fmt.Sprintf("%s/powerloss=%v", f.name, powerLoss), func(t *testing.T) {
807- testInjectedFailure(t, f, powerLoss)
824+ synctest.Test(t, func(t *testing.T) { testInjectedFailure(t, f, powerLoss) })
808825 })
809826 }
810827 }
@@ -894,25 +911,27 @@ func testInjectedFailure(t *testing.T, f failure, powerLoss bool) {
894911 // TestFailedWriterAnswersOverflow checks that no Apply hangs when the group
895912 // that fails leaves a dequeued request behind.
896913 func TestFailedWriterAnswersOverflow(t *testing.T) {
897- fx := faultfs.New()
898- db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
899- fx.FailWrite(1, nil)
900- chunk := strings.Repeat("c", 3<<20) // two do not fit in one group
901- errs := make([]error, 4)
902- var wg sync.WaitGroup
903- for i := range errs {
904- wg.Go(func() {
905- var b Batch
906- b.Put(fmt.Appendf(nil, "k%d", i), []byte(chunk))
907- _, errs[i] = db.Apply(bg, &b)
908- })
909- }
910- wg.Wait()
911- for i, err := range errs {
912- if !errors.Is(err, ErrFailed) {
913- t.Errorf("batch %d: %v", i, err)
914+ synctest.Test(t, func(t *testing.T) {
915+ fx := faultfs.New()
916+ db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
917+ fx.FailWrite(1, nil)
918+ chunk := strings.Repeat("c", 3<<20) // two do not fit in one group
919+ errs := make([]error, 4)
920+ var wg sync.WaitGroup
921+ for i := range errs {
922+ wg.Go(func() {
923+ var b Batch
924+ b.Put(fmt.Appendf(nil, "k%d", i), []byte(chunk))
925+ _, errs[i] = db.Apply(bg, &b)
926+ })
914927 }
915- }
928+ wg.Wait()
929+ for i, err := range errs {
930+ if !errors.Is(err, ErrFailed) {
931+ t.Errorf("batch %d: %v", i, err)
932+ }
933+ }
934+ })
916935 }
917936
918937 type syncBuffer struct {
@@ -16,6 +16,7 @@ import (
16 "sync"16 "sync"
17 "sync/atomic"17 "sync/atomic"
18 "testing"18 "testing"
19+ "testing/synctest"
19 "time"20 "time"
20 21
21 "github.com/bnema/zerowrap"22 "github.com/bnema/zerowrap"
@@ -354,7 +355,9 @@ func TestPreconditionsBeforeOps(t *testing.T) {
354 // holdWriter applies a batch whose sync is slow and returns once the writer355 // holdWriter applies a batch whose sync is slow and returns once the writer
355 // is inside that sync, so batches applied next queue up and form one group356 // is inside that sync, so batches applied next queue up and form one group
356 // (GroupWindow 0 drains everything already queued). wait returns when the357 // (GroupWindow 0 drains everything already queued). wait returns when the
357-// held batch is acknowledged; the sync delay stays set until then.358+// held batch is acknowledged; the sync delay stays set until then. It must be
359+// called inside a synctest bubble: the writer sleeps in the sync on the fake
360+// clock, so no real time passes.
358 func holdWriter(t *testing.T, db *DB, fx *faultfs.FS) (wait func()) {361 func holdWriter(t *testing.T, db *DB, fx *faultfs.FS) (wait func()) {
359 t.Helper()362 t.Helper()
360 base := fx.Syncs()363 base := fx.Syncs()
@@ -366,8 +369,9 @@ func holdWriter(t *testing.T, db *DB, fx *faultfs.FS) (wait func()) {
366 _, err := db.Apply(bg, &b)369 _, err := db.Apply(bg, &b)
367 done <- err370 done <- err
368 }()371 }()
369- for fx.Syncs() == base {372+ synctest.Wait() // the writer is parked in the delayed sync
370- time.Sleep(time.Millisecond)373+ if fx.Syncs() == base {
374+ t.Fatal("the writer did not reach the sync")
371 }375 }
372 return func() {376 return func() {
373 t.Helper()377 t.Helper()
@@ -381,28 +385,30 @@ func holdWriter(t *testing.T, db *DB, fx *faultfs.FS) (wait func()) {
381 // TestGroupPreconditions checks that preconditions see earlier batches of385 // TestGroupPreconditions checks that preconditions see earlier batches of
382 // the same group: the second batch of the group must conflict.386 // the same group: the second batch of the group must conflict.
383 func TestGroupPreconditions(t *testing.T) {387 func TestGroupPreconditions(t *testing.T) {
384- fx := faultfs.New()388+ synctest.Test(t, func(t *testing.T) {
385- db := mustOpen(t, fx, Options{})389+ fx := faultfs.New()
386- syncs := fx.Syncs()390+ db := mustOpen(t, fx, Options{})
387- wait := holdWriter(t, db, fx)391+ syncs := fx.Syncs()
388- errs := make([]error, 2)392+ wait := holdWriter(t, db, fx)
389- var wg sync.WaitGroup393+ errs := make([]error, 2)
390- for i := range errs {394+ var wg sync.WaitGroup
391- wg.Go(func() {395+ for i := range errs {
392- var b Batch396+ wg.Go(func() {
393- b.RequireAbsent([]byte("k"))397+ var b Batch
394- b.Put([]byte("k"), []byte{byte('a' + i)})398+ b.RequireAbsent([]byte("k"))
395- _, errs[i] = db.Apply(bg, &b)399+ b.Put([]byte("k"), []byte{byte('a' + i)})
396- })400+ _, errs[i] = db.Apply(bg, &b)
397- }401+ })
398- wg.Wait()402+ }
399- wait()403+ wg.Wait()
400- if (errs[0] == nil) == (errs[1] == nil) {404+ wait()
401- t.Fatalf("errors %v, want exactly one success", errs)405+ if (errs[0] == nil) == (errs[1] == nil) {
402- }406+ t.Fatalf("errors %v, want exactly one success", errs)
403- if got := fx.Syncs() - syncs; got != 2 {407+ }
404- t.Fatalf("%d syncs, want 2 (held batch, then one group)", got)408+ if got := fx.Syncs() - syncs; got != 2 {
405- }409+ t.Fatalf("%d syncs, want 2 (held batch, then one group)", got)
410+ }
411+ })
406 }412 }
407 413
408 func TestEmptyAndConflictOnlyBatches(t *testing.T) {414 func TestEmptyAndConflictOnlyBatches(t *testing.T) {
@@ -462,25 +468,27 @@ func TestEmptyAndConflictOnlyBatches(t *testing.T) {
462 // TestNoWriteForSkippedGroup checks that a group of only conflicts and empty468 // TestNoWriteForSkippedGroup checks that a group of only conflicts and empty
463 // batches neither writes nor syncs.469 // batches neither writes nor syncs.
464 func TestNoWriteForSkippedGroup(t *testing.T) {470 func TestNoWriteForSkippedGroup(t *testing.T) {
465- fx := faultfs.New()471+ synctest.Test(t, func(t *testing.T) {
466- db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})472+ fx := faultfs.New()
467- putKV(t, db, "a", "1")473+ db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
468- writes, syncs := fx.Writes(), fx.Syncs()474+ putKV(t, db, "a", "1")
469- var wg sync.WaitGroup475+ writes, syncs := fx.Writes(), fx.Syncs()
470- for range 4 {476+ var wg sync.WaitGroup
471- wg.Go(func() {477+ for range 4 {
472- var b Batch478+ wg.Go(func() {
473- b.RequireAbsent([]byte("a"))479+ var b Batch
474- b.Put([]byte("x"), nil)480+ b.RequireAbsent([]byte("a"))
475- if _, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) {481+ b.Put([]byte("x"), nil)
476- t.Errorf("err = %v, want ErrConflict", err)482+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) {
477- }483+ t.Errorf("err = %v, want ErrConflict", err)
478- })484+ }
479- }485+ })
480- wg.Wait()486+ }
481- if fx.Writes() != writes || fx.Syncs() != syncs {487+ wg.Wait()
482- t.Fatalf("writes %d→%d syncs %d→%d", writes, fx.Writes(), syncs, fx.Syncs())488+ if fx.Writes() != writes || fx.Syncs() != syncs {
483- }489+ t.Fatalf("writes %d→%d syncs %d→%d", writes, fx.Writes(), syncs, fx.Syncs())
490+ }
491+ })
484 }492 }
485 493
486 func TestGroupCommitSyncs(t *testing.T) {494 func TestGroupCommitSyncs(t *testing.T) {
@@ -527,33 +535,35 @@ func TestNoSync(t *testing.T) {
527 // TestLargeBatches covers a batch above the 4 MiB group cap (a group of one)535 // TestLargeBatches covers a batch above the 4 MiB group cap (a group of one)
528 // and a burst of batches that does not fit in one group.536 // and a burst of batches that does not fit in one group.
529 func TestLargeBatches(t *testing.T) {537 func TestLargeBatches(t *testing.T) {
530- fx := faultfs.New()538+ synctest.Test(t, func(t *testing.T) {
531- db := mustOpen(t, fx, Options{GroupWindow: 50 * time.Millisecond})539+ fx := faultfs.New()
532- big := strings.Repeat("B", 5<<20)540+ db := mustOpen(t, fx, Options{GroupWindow: 50 * time.Millisecond})
533- putKV(t, db, "big", big)541+ big := strings.Repeat("B", 5<<20)
534-542+ putKV(t, db, "big", big)
535- const n = 6543+
536- chunk := strings.Repeat("c", 3<<19) // 1.5 MiB: three do not fit in 4 MiB544+ const n = 6
537- seqs := make([]uint64, n)545+ chunk := strings.Repeat("c", 3<<19) // 1.5 MiB: three do not fit in 4 MiB
538- var wg sync.WaitGroup546+ seqs := make([]uint64, n)
539- for i := range n {547+ var wg sync.WaitGroup
540- wg.Go(func() { seqs[i] = putKV(t, db, fmt.Sprintf("chunk%d", i), chunk) })548+ for i := range n {
541- }549+ wg.Go(func() { seqs[i] = putKV(t, db, fmt.Sprintf("chunk%d", i), chunk) })
542- wg.Wait()
543- slices.Sort(seqs)
544- for i, s := range seqs {
545- if s != uint64(i+2) {
546- t.Fatalf("seqs = %v", seqs)
547 }550 }
548- }551+ wg.Wait()
549- if err := db.Close(bg); err != nil {552+ slices.Sort(seqs)
550- t.Fatal(err)553+ for i, s := range seqs {
551- }554+ if s != uint64(i+2) {
552- s := mustOpen(t, fx, Options{}).Snapshot(bg)555+ t.Fatalf("seqs = %v", seqs)
553- mustGet(t, s, "big", big)556+ }
554- for i := range n {557+ }
555- mustGet(t, s, fmt.Sprintf("chunk%d", i), chunk)558+ if err := db.Close(bg); err != nil {
556- }559+ t.Fatal(err)
560+ }
561+ s := mustOpen(t, fx, Options{}).Snapshot(bg)
562+ mustGet(t, s, "big", big)
563+ for i := range n {
564+ mustGet(t, s, fmt.Sprintf("chunk%d", i), chunk)
565+ }
566+ })
557 }567 }
558 568
559 func TestInvalidInputs(t *testing.T) {569 func TestInvalidInputs(t *testing.T) {
@@ -675,58 +685,65 @@ func TestApplyAfterClose(t *testing.T) {
675 // TestContextBeforeDequeue blocks the writer in a slow sync so that a second685 // TestContextBeforeDequeue blocks the writer in a slow sync so that a second
676 // batch cannot be dequeued, then cancels its context.686 // batch cannot be dequeued, then cancels its context.
677 func TestContextBeforeDequeue(t *testing.T) {687 func TestContextBeforeDequeue(t *testing.T) {
678- fx := faultfs.New()688+ synctest.Test(t, func(t *testing.T) {
679- db := mustOpen(t, fx, Options{})689+ fx := faultfs.New()
690+ db := mustOpen(t, fx, Options{})
680 691
681- done := make(chan error, 1)692+ done := make(chan error, 1)
682- base := fx.Syncs()693+ base := fx.Syncs()
683- fx.SetSyncDelay(300 * time.Millisecond)694+ fx.SetSyncDelay(300 * time.Millisecond)
684- go func() {695+ go func() {
685- var b Batch696+ var b Batch
686- b.Put([]byte("first"), []byte("1"))697+ b.Put([]byte("first"), []byte("1"))
687- _, err := db.Apply(bg, &b)698+ _, err := db.Apply(bg, &b)
688- done <- err699+ done <- err
689- }()700+ }()
690- for fx.Syncs() == base { // the writer is now inside the delayed sync701+ synctest.Wait() // the writer is now inside the delayed sync
691- time.Sleep(time.Millisecond)702+ if fx.Syncs() == base {
692- }703+ t.Fatal("the writer did not reach the sync")
704+ }
705+ start := time.Now()
693 706
694- ctx, cancel := context.WithCancel(bg)707+ ctx, cancel := context.WithCancel(bg)
695- second := make(chan error, 1)708+ second := make(chan error, 1)
696- go func() {709+ go func() {
697- var b Batch710+ var b Batch
698- b.Put([]byte("second"), []byte("2"))711+ b.Put([]byte("second"), []byte("2"))
699- _, err := db.Apply(ctx, &b)712+ _, err := db.Apply(ctx, &b)
700- second <- err713+ second <- err
701- }()714+ }()
702- time.Sleep(50 * time.Millisecond) // let Apply reach its blocking select715+ synctest.Wait() // Apply is blocked in its select: the writer cannot dequeue
703- cancel()716+ cancel()
704- if err := <-second; !errors.Is(err, context.Canceled) {717+ if err := <-second; !errors.Is(err, context.Canceled) {
705- t.Fatalf("Apply = %v, want context.Canceled", err)718+ t.Fatalf("Apply = %v, want context.Canceled", err)
706- }719+ }
707- if err := <-done; err != nil {720+ if d := time.Since(start); d != 0 {
708- t.Fatalf("first Apply: %v", err)721+ t.Fatalf("the cancelled Apply returned after %v, it must not wait for the sync", d)
709- }722+ }
710- fx.SetSyncDelay(0)723+ if err := <-done; err != nil {
724+ t.Fatalf("first Apply: %v", err)
725+ }
726+ fx.SetSyncDelay(0)
711 727
712- // An already cancelled context never reaches the writer either.728+ // An already cancelled context never reaches the writer either.
713- var b Batch729+ var b Batch
714- b.Put([]byte("third"), nil)730+ b.Put([]byte("third"), nil)
715- if _, err := db.Apply(ctx, &b); !errors.Is(err, context.Canceled) {731+ if _, err := db.Apply(ctx, &b); !errors.Is(err, context.Canceled) {
716- t.Fatalf("Apply with cancelled ctx = %v", err)732+ t.Fatalf("Apply with cancelled ctx = %v", err)
717- }733+ }
718- putKV(t, db, "fourth", "4") // the writer is still healthy734+ putKV(t, db, "fourth", "4") // the writer is still healthy
719- if err := db.Close(bg); err != nil {735+ if err := db.Close(bg); err != nil {
720- t.Fatal(err)736+ t.Fatal(err)
721- }737+ }
722- s := mustOpen(t, fx, Options{}).Snapshot(bg)738+ s := mustOpen(t, fx, Options{}).Snapshot(bg)
723- mustGet(t, s, "first", "1")739+ mustGet(t, s, "first", "1")
724- mustGet(t, s, "fourth", "4")740+ mustGet(t, s, "fourth", "4")
725- mustMissing(t, s, "second")741+ mustMissing(t, s, "second")
726- mustMissing(t, s, "third")742+ mustMissing(t, s, "third")
727- if s.Seq() != 2 {743+ if s.Seq() != 2 {
728- t.Fatalf("Seq = %d, want 2", s.Seq())744+ t.Fatalf("Seq = %d, want 2", s.Seq())
729- }745+ }
746+ })
730 }747 }
731 748
732 func TestCloseDuringApplies(t *testing.T) {749 func TestCloseDuringApplies(t *testing.T) {
@@ -804,7 +821,7 @@ func TestInjectedFailures(t *testing.T) {
804 for _, f := range failures {821 for _, f := range failures {
805 for _, powerLoss := range []bool{false, true} {822 for _, powerLoss := range []bool{false, true} {
806 t.Run(fmt.Sprintf("%s/powerloss=%v", f.name, powerLoss), func(t *testing.T) {823 t.Run(fmt.Sprintf("%s/powerloss=%v", f.name, powerLoss), func(t *testing.T) {
807- testInjectedFailure(t, f, powerLoss)824+ synctest.Test(t, func(t *testing.T) { testInjectedFailure(t, f, powerLoss) })
808 })825 })
809 }826 }
810 }827 }
@@ -894,25 +911,27 @@ func testInjectedFailure(t *testing.T, f failure, powerLoss bool) {
894 // TestFailedWriterAnswersOverflow checks that no Apply hangs when the group911 // TestFailedWriterAnswersOverflow checks that no Apply hangs when the group
895 // that fails leaves a dequeued request behind.912 // that fails leaves a dequeued request behind.
896 func TestFailedWriterAnswersOverflow(t *testing.T) {913 func TestFailedWriterAnswersOverflow(t *testing.T) {
897- fx := faultfs.New()914+ synctest.Test(t, func(t *testing.T) {
898- db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})915+ fx := faultfs.New()
899- fx.FailWrite(1, nil)916+ db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
900- chunk := strings.Repeat("c", 3<<20) // two do not fit in one group917+ fx.FailWrite(1, nil)
901- errs := make([]error, 4)918+ chunk := strings.Repeat("c", 3<<20) // two do not fit in one group
902- var wg sync.WaitGroup919+ errs := make([]error, 4)
903- for i := range errs {920+ var wg sync.WaitGroup
904- wg.Go(func() {921+ for i := range errs {
905- var b Batch922+ wg.Go(func() {
906- b.Put(fmt.Appendf(nil, "k%d", i), []byte(chunk))923+ var b Batch
907- _, errs[i] = db.Apply(bg, &b)924+ b.Put(fmt.Appendf(nil, "k%d", i), []byte(chunk))
908- })925+ _, errs[i] = db.Apply(bg, &b)
909- }926+ })
910- wg.Wait()
911- for i, err := range errs {
912- if !errors.Is(err, ErrFailed) {
913- t.Errorf("batch %d: %v", i, err)
914 }927 }
915- }928+ wg.Wait()
929+ for i, err := range errs {
930+ if !errors.Is(err, ErrFailed) {
931+ t.Errorf("batch %d: %v", i, err)
932+ }
933+ }
934+ })
916 }935 }
917 936
918 type syncBuffer struct {937 type syncBuffer struct {
M internal/fsys/faultfs/faultfs_test.go +23 -10
@@ -10,6 +10,7 @@ import (
1010 "os"
1111 "sync"
1212 "testing"
13+ "testing/synctest"
1314 "time"
1415
1516 "bnema.dev/ermite-kv/internal/fsys"
@@ -140,16 +141,28 @@ func TestFailSyncAndCounters(t *testing.T) {
140141 }
141142
142143 func TestSyncDelay(t *testing.T) {
143- f := New()
144- h := create(t, f, "/a")
145- f.SetSyncDelay(20 * time.Millisecond)
146- start := time.Now()
147- if err := h.Sync(); err != nil {
148- t.Fatal(err)
149- }
150- if time.Since(start) < 20*time.Millisecond {
151- t.Fatal("sync returned before the delay")
152- }
144+ synctest.Test(t, func(t *testing.T) {
145+ f := New()
146+ h := create(t, f, "/a")
147+ f.SetSyncDelay(20 * time.Millisecond)
148+ for name, sync := range map[string]func() error{"Sync": h.Sync, "SyncAll": h.SyncAll} {
149+ start := time.Now()
150+ if err := sync(); err != nil {
151+ t.Fatal(err)
152+ }
153+ if got := time.Since(start); got != 20*time.Millisecond {
154+ t.Fatalf("%s took %v, want exactly 20ms", name, got)
155+ }
156+ }
157+ f.SetSyncDelay(0)
158+ start := time.Now()
159+ if err := h.Sync(); err != nil {
160+ t.Fatal(err)
161+ }
162+ if got := time.Since(start); got != 0 {
163+ t.Fatalf("Sync without delay took %v", got)
164+ }
165+ })
153166 }
154167
155168 func TestPowerLossDropsUnsyncedBytes(t *testing.T) {
@@ -10,6 +10,7 @@ import (
10 "os"10 "os"
11 "sync"11 "sync"
12 "testing"12 "testing"
13+ "testing/synctest"
13 "time"14 "time"
14 15
15 "bnema.dev/ermite-kv/internal/fsys"16 "bnema.dev/ermite-kv/internal/fsys"
@@ -140,16 +141,28 @@ func TestFailSyncAndCounters(t *testing.T) {
140 }141 }
141 142
142 func TestSyncDelay(t *testing.T) {143 func TestSyncDelay(t *testing.T) {
143- f := New()144+ synctest.Test(t, func(t *testing.T) {
144- h := create(t, f, "/a")145+ f := New()
145- f.SetSyncDelay(20 * time.Millisecond)146+ h := create(t, f, "/a")
146- start := time.Now()147+ f.SetSyncDelay(20 * time.Millisecond)
147- if err := h.Sync(); err != nil {148+ for name, sync := range map[string]func() error{"Sync": h.Sync, "SyncAll": h.SyncAll} {
148- t.Fatal(err)149+ start := time.Now()
149- }150+ if err := sync(); err != nil {
150- if time.Since(start) < 20*time.Millisecond {151+ t.Fatal(err)
151- t.Fatal("sync returned before the delay")152+ }
152- }153+ if got := time.Since(start); got != 20*time.Millisecond {
154+ t.Fatalf("%s took %v, want exactly 20ms", name, got)
155+ }
156+ }
157+ f.SetSyncDelay(0)
158+ start := time.Now()
159+ if err := h.Sync(); err != nil {
160+ t.Fatal(err)
161+ }
162+ if got := time.Since(start); got != 0 {
163+ t.Fatalf("Sync without delay took %v", got)
164+ }
165+ })
153 }166 }
154 167
155 func TestPowerLossDropsUnsyncedBytes(t *testing.T) {168 func TestPowerLossDropsUnsyncedBytes(t *testing.T) {
M writer.go +3 -14
@@ -5,7 +5,6 @@ package kv
55 import (
66 "fmt"
77 "io"
8- "sync"
98 "time"
109 )
1110
@@ -14,7 +13,9 @@ import (
1413 const maxGroupBytes = 4 << 20
1514
1615 // request is a batch waiting for the writer. The writer answers on reply
17-// exactly once.
16+// exactly once. Requests are allocated per Apply and never pooled: a pooled
17+// channel would be shared across testing/synctest bubbles, and receiving on a
18+// bubbled channel from outside its bubble is fatal.
1819 type request struct {
1920 b *Batch
2021 reply chan result // capacity 1
@@ -25,18 +26,6 @@ type result struct {
2526 err error
2627 }
2728
28-var requestPool = sync.Pool{New: func() any {
29- return &request{reply: make(chan result, 1)}
30-}}
31-
32-// release returns r to the pool (it must not be referenced by the writer)
33-// and passes err through.
34-func (r *request) release(err error) error {
35- r.b = nil
36- requestPool.Put(r)
37- return err
38-}
39-
4029 // writer is the single goroutine that owns the log file and builds the new
4130 // state. It exits on Close or when the database fails.
4231 func (db *DB) writer() {
@@ -5,7 +5,6 @@ package kv
5 import (5 import (
6 "fmt"6 "fmt"
7 "io"7 "io"
8- "sync"
9 "time"8 "time"
10 )9 )
11 10
@@ -14,7 +13,9 @@ import (
14 const maxGroupBytes = 4 << 2013 const maxGroupBytes = 4 << 20
15 14
16 // request is a batch waiting for the writer. The writer answers on reply15 // request is a batch waiting for the writer. The writer answers on reply
17-// exactly once.16+// exactly once. Requests are allocated per Apply and never pooled: a pooled
17+// channel would be shared across testing/synctest bubbles, and receiving on a
18+// bubbled channel from outside its bubble is fatal.
18 type request struct {19 type request struct {
19 b *Batch20 b *Batch
20 reply chan result // capacity 121 reply chan result // capacity 1
@@ -25,18 +26,6 @@ type result struct {
25 err error26 err error
26 }27 }
27 28
28-var requestPool = sync.Pool{New: func() any {
29- return &request{reply: make(chan result, 1)}
30-}}
31-
32-// release returns r to the pool (it must not be referenced by the writer)
33-// and passes err through.
34-func (r *request) release(err error) error {
35- r.b = nil
36- requestPool.Put(r)
37- return err
38-}
39-
40 // writer is the single goroutine that owns the log file and builds the new29 // writer is the single goroutine that owns the log file and builds the new
41 // state. It exits on Close or when the database fails.30 // state. It exits on Close or when the database fails.
42 func (db *DB) writer() {31 func (db *DB) writer() {

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