detached f83ddf7

feat(compact): switch logs and write snapshots in the background

After a group is acknowledged and the current log exceeds CompactAfter,
the writer syncs the log, creates log-<seq>, switches appends to it and
starts a compactor on the immutable snapshot. The compactor writes
snapshot-<seq> and removes older files; failures are logged and retried at
the next trigger. Close waits for a running compaction.
commit
f83ddf7bfbe2585713e7235e20086a565d6674b1
author
brice · · 2026-10-10 02:51 UTC
committer
brice · · 2026-10-10 02:52 UTC
parent
b799b92
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
4 files · +1145 -4
Unified
UnifiedSplit
A compact.go +163 -0
@@ -0,0 +1,163 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "errors"
7+ "fmt"
8+ "io"
9+ "io/fs"
10+ "path/filepath"
11+ "time"
12+)
13+
14+// maybeCompact starts a compaction once the current log holds more than
15+// Options.CompactAfter bytes after its header, unless one is already running.
16+// It runs on the writer goroutine after the group has been answered and
17+// reports false if the database failed.
18+func (db *DB) maybeCompact() bool {
19+ if db.opts.CompactAfter < 0 || db.compacting.Load() ||
20+ db.logSize-logHeaderSize <= db.opts.CompactAfter || db.logSize < db.retryAt {
21+ return true
22+ }
23+ return db.switchLog()
24+}
25+
26+// switchLog ends the current log at the published seq S and starts the
27+// compactor on the snapshot at S: SyncAll the current log, create log-S, make
28+// it the current log, then compact in the background. It never waits for the
29+// compactor, so the writer is not blocked by the snapshot size.
30+//
31+// Failure handling:
32+// - SyncAll of the current log fails: the database fails. With NoSync this
33+// is the first sync of the log's data, and after a failed fsync the state
34+// of the file is unknown, so the writer must not go on appending.
35+// - Creating log-S fails: a warning is logged and appends continue on the
36+// current log (nothing is lost, only compaction is postponed). The next
37+// attempt waits for another CompactAfter bytes, so that a persistent
38+// failure does not cost a sync per commit. If createFile renamed log-S
39+// before failing, it is removed: left behind, it would become the last
40+// log while records go on in the old one, and the next Open would find a
41+// last log that ends below the recovered seq (ErrCorrupt). If that removal
42+// fails too, the database fails, because the directory may no longer open.
43+//
44+// It reports false if the database failed.
45+func (db *DB) switchLog() bool {
46+ start := time.Now()
47+ snap := db.snap.Load() // the writer is the only publisher: this is the seq of the log
48+ seq := snap.Seq()
49+ if err := db.file.SyncAll(); err != nil {
50+ _ = db.fail(fmt.Errorf("sync log before switch: %w", err))
51+ return false
52+ }
53+ name := logName(seq)
54+ header := appendLogHeader(nil, seq)
55+ f, err := createFile(db.fx, db.dir, name, func(w io.Writer) error {
56+ _, err := w.Write(header)
57+ return err
58+ })
59+ if err != nil {
60+ db.log.Warn().Err(err).Uint64("seq", seq).Msg("log switch failed")
61+ if rerr := db.removeStrayLog(filepath.Join(db.dir, name)); rerr != nil {
62+ _ = db.fail(fmt.Errorf("remove incomplete log %s: %w", name, rerr))
63+ return false
64+ }
65+ db.retryAt = db.logSize + db.opts.CompactAfter
66+ return true
67+ }
68+
69+ from := db.logBase
70+ old := db.file
71+ db.file, db.logBase, db.logSize, db.retryAt = f, seq, logHeaderSize, 0
72+ if err := old.Close(); err != nil {
73+ db.log.Warn().Err(err).Uint64("base", from).Msg("closing the previous log")
74+ }
75+ db.log.Debug().Uint64("from", from).Uint64("seq", seq).Dur("duration", time.Since(start)).Msg("log switch")
76+
77+ db.compacting.Store(true)
78+ db.compactors.Go(func() {
79+ defer db.compacting.Store(false)
80+ db.compact(snap)
81+ })
82+ return true
83+}
84+
85+// removeStrayLog deletes a log file that createFile may have left in place
86+// when it failed, and makes the removal durable.
87+func (db *DB) removeStrayLog(path string) error {
88+ err := db.fx.Remove(path)
89+ if errors.Is(err, fs.ErrNotExist) {
90+ return nil
91+ }
92+ if err != nil {
93+ return err
94+ }
95+ return db.fx.SyncDir(db.dir)
96+}
97+
98+// compact writes snapshot-<seq> from snap and deletes the snapshots and logs
99+// older than it. A failure is logged and never fails the database: the next
100+// trigger starts again from a newer snapshot.
101+func (db *DB) compact(snap *Snapshot) {
102+ start := time.Now()
103+ seq := snap.Seq()
104+ size, err := db.writeSnapshotFile(snap)
105+ if err == nil {
106+ err = db.removeObsolete(seq)
107+ }
108+ if err != nil {
109+ db.log.Warn().Err(err).Uint64("seq", seq).Msg("compaction failed")
110+ return
111+ }
112+ db.log.Info().Uint64("seq", seq).Int("keys", snap.Len()).Int64("bytes", size).
113+ Dur("duration", time.Since(start)).Msg("compaction done")
114+}
115+
116+// writeSnapshotFile durably creates snapshot-<snap.Seq()> and returns its size.
117+func (db *DB) writeSnapshotFile(snap *Snapshot) (int64, error) {
118+ seq := snap.Seq()
119+ entries := func(yield func(key, value []byte, version uint64) bool) {
120+ for k, it := range snap.t.ascend(nil, nil) {
121+ if !yield(k, it.Value, it.Version) {
122+ return
123+ }
124+ }
125+ }
126+ f, err := createFile(db.fx, db.dir, snapshotName(seq), func(w io.Writer) error {
127+ return writeSnapshot(w, seq, uint64(snap.Len()), entries)
128+ })
129+ if err != nil {
130+ return 0, fmt.Errorf("write snapshot: %w", err)
131+ }
132+ size, _ := f.Size() // informational
133+ _ = f.Close() // the file is synced and renamed; nothing is pending
134+ return size, nil
135+}
136+
137+// removeObsolete deletes the snapshots and logs with a seq below seq, then
138+// syncs the directory. It tries every file and returns the first error. The
139+// files are removed only after snapshot-<seq> is durable, so a crash at any
140+// point leaves a recoverable directory.
141+func (db *DB) removeObsolete(seq uint64) error {
142+ names, err := db.fx.ReadDir(db.dir)
143+ if err != nil {
144+ return fmt.Errorf("list obsolete files: %w", err)
145+ }
146+ var first error
147+ for _, name := range names {
148+ n, ok := parseSeqName(name, logPrefix)
149+ if !ok {
150+ n, ok = parseSeqName(name, snapshotPrefix)
151+ }
152+ if !ok || n >= seq {
153+ continue
154+ }
155+ if err := db.fx.Remove(filepath.Join(db.dir, name)); err != nil && !errors.Is(err, fs.ErrNotExist) && first == nil {
156+ first = fmt.Errorf("remove %s: %w", name, err)
157+ }
158+ }
159+ if err := db.fx.SyncDir(db.dir); err != nil && first == nil {
160+ first = fmt.Errorf("sync directory: %w", err)
161+ }
162+ return first
163+}
@@ -0,0 +1,163 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "errors"
7+ "fmt"
8+ "io"
9+ "io/fs"
10+ "path/filepath"
11+ "time"
12+)
13+
14+// maybeCompact starts a compaction once the current log holds more than
15+// Options.CompactAfter bytes after its header, unless one is already running.
16+// It runs on the writer goroutine after the group has been answered and
17+// reports false if the database failed.
18+func (db *DB) maybeCompact() bool {
19+ if db.opts.CompactAfter < 0 || db.compacting.Load() ||
20+ db.logSize-logHeaderSize <= db.opts.CompactAfter || db.logSize < db.retryAt {
21+ return true
22+ }
23+ return db.switchLog()
24+}
25+
26+// switchLog ends the current log at the published seq S and starts the
27+// compactor on the snapshot at S: SyncAll the current log, create log-S, make
28+// it the current log, then compact in the background. It never waits for the
29+// compactor, so the writer is not blocked by the snapshot size.
30+//
31+// Failure handling:
32+// - SyncAll of the current log fails: the database fails. With NoSync this
33+// is the first sync of the log's data, and after a failed fsync the state
34+// of the file is unknown, so the writer must not go on appending.
35+// - Creating log-S fails: a warning is logged and appends continue on the
36+// current log (nothing is lost, only compaction is postponed). The next
37+// attempt waits for another CompactAfter bytes, so that a persistent
38+// failure does not cost a sync per commit. If createFile renamed log-S
39+// before failing, it is removed: left behind, it would become the last
40+// log while records go on in the old one, and the next Open would find a
41+// last log that ends below the recovered seq (ErrCorrupt). If that removal
42+// fails too, the database fails, because the directory may no longer open.
43+//
44+// It reports false if the database failed.
45+func (db *DB) switchLog() bool {
46+ start := time.Now()
47+ snap := db.snap.Load() // the writer is the only publisher: this is the seq of the log
48+ seq := snap.Seq()
49+ if err := db.file.SyncAll(); err != nil {
50+ _ = db.fail(fmt.Errorf("sync log before switch: %w", err))
51+ return false
52+ }
53+ name := logName(seq)
54+ header := appendLogHeader(nil, seq)
55+ f, err := createFile(db.fx, db.dir, name, func(w io.Writer) error {
56+ _, err := w.Write(header)
57+ return err
58+ })
59+ if err != nil {
60+ db.log.Warn().Err(err).Uint64("seq", seq).Msg("log switch failed")
61+ if rerr := db.removeStrayLog(filepath.Join(db.dir, name)); rerr != nil {
62+ _ = db.fail(fmt.Errorf("remove incomplete log %s: %w", name, rerr))
63+ return false
64+ }
65+ db.retryAt = db.logSize + db.opts.CompactAfter
66+ return true
67+ }
68+
69+ from := db.logBase
70+ old := db.file
71+ db.file, db.logBase, db.logSize, db.retryAt = f, seq, logHeaderSize, 0
72+ if err := old.Close(); err != nil {
73+ db.log.Warn().Err(err).Uint64("base", from).Msg("closing the previous log")
74+ }
75+ db.log.Debug().Uint64("from", from).Uint64("seq", seq).Dur("duration", time.Since(start)).Msg("log switch")
76+
77+ db.compacting.Store(true)
78+ db.compactors.Go(func() {
79+ defer db.compacting.Store(false)
80+ db.compact(snap)
81+ })
82+ return true
83+}
84+
85+// removeStrayLog deletes a log file that createFile may have left in place
86+// when it failed, and makes the removal durable.
87+func (db *DB) removeStrayLog(path string) error {
88+ err := db.fx.Remove(path)
89+ if errors.Is(err, fs.ErrNotExist) {
90+ return nil
91+ }
92+ if err != nil {
93+ return err
94+ }
95+ return db.fx.SyncDir(db.dir)
96+}
97+
98+// compact writes snapshot-<seq> from snap and deletes the snapshots and logs
99+// older than it. A failure is logged and never fails the database: the next
100+// trigger starts again from a newer snapshot.
101+func (db *DB) compact(snap *Snapshot) {
102+ start := time.Now()
103+ seq := snap.Seq()
104+ size, err := db.writeSnapshotFile(snap)
105+ if err == nil {
106+ err = db.removeObsolete(seq)
107+ }
108+ if err != nil {
109+ db.log.Warn().Err(err).Uint64("seq", seq).Msg("compaction failed")
110+ return
111+ }
112+ db.log.Info().Uint64("seq", seq).Int("keys", snap.Len()).Int64("bytes", size).
113+ Dur("duration", time.Since(start)).Msg("compaction done")
114+}
115+
116+// writeSnapshotFile durably creates snapshot-<snap.Seq()> and returns its size.
117+func (db *DB) writeSnapshotFile(snap *Snapshot) (int64, error) {
118+ seq := snap.Seq()
119+ entries := func(yield func(key, value []byte, version uint64) bool) {
120+ for k, it := range snap.t.ascend(nil, nil) {
121+ if !yield(k, it.Value, it.Version) {
122+ return
123+ }
124+ }
125+ }
126+ f, err := createFile(db.fx, db.dir, snapshotName(seq), func(w io.Writer) error {
127+ return writeSnapshot(w, seq, uint64(snap.Len()), entries)
128+ })
129+ if err != nil {
130+ return 0, fmt.Errorf("write snapshot: %w", err)
131+ }
132+ size, _ := f.Size() // informational
133+ _ = f.Close() // the file is synced and renamed; nothing is pending
134+ return size, nil
135+}
136+
137+// removeObsolete deletes the snapshots and logs with a seq below seq, then
138+// syncs the directory. It tries every file and returns the first error. The
139+// files are removed only after snapshot-<seq> is durable, so a crash at any
140+// point leaves a recoverable directory.
141+func (db *DB) removeObsolete(seq uint64) error {
142+ names, err := db.fx.ReadDir(db.dir)
143+ if err != nil {
144+ return fmt.Errorf("list obsolete files: %w", err)
145+ }
146+ var first error
147+ for _, name := range names {
148+ n, ok := parseSeqName(name, logPrefix)
149+ if !ok {
150+ n, ok = parseSeqName(name, snapshotPrefix)
151+ }
152+ if !ok || n >= seq {
153+ continue
154+ }
155+ if err := db.fx.Remove(filepath.Join(db.dir, name)); err != nil && !errors.Is(err, fs.ErrNotExist) && first == nil {
156+ first = fmt.Errorf("remove %s: %w", name, err)
157+ }
158+ }
159+ if err := db.fx.SyncDir(db.dir); err != nil && first == nil {
160+ first = fmt.Errorf("sync directory: %w", err)
161+ }
162+ return first
163+}
A compact_test.go +973 -0
@@ -0,0 +1,973 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
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+ "time"
18+
19+ "github.com/bnema/zerowrap"
20+
21+ "bnema.dev/ermite-kv/internal/fsys"
22+ "bnema.dev/ermite-kv/internal/fsys/faultfs"
23+)
24+
25+var errCrash = errors.New("test: power was cut")
26+
27+// hookFS wraps an FS and intercepts the operations that change files:
28+// hook (if set) may block or fail them, and a power loss can be cut at a
29+// given step. With serial set, operations run one at a time under mu, so a
30+// cut is atomic with respect to every other operation and nothing touches the
31+// file system afterwards: once cut, every operation fails with errCrash, like
32+// a process that no longer exists.
33+type hookFS struct {
34+ fsys.FS
35+ ff *faultfs.FS // underlying FS, needed by cut
36+ serial bool // run operations under mu
37+ hook func(op, path string) error // optional; set before use
38+
39+ mu sync.Mutex // guards the fields below when serial
40+ armed bool // record the operations in trace
41+ trace []string
42+ crashAt int // cut the power before the crashAt-th traced operation (0: never)
43+ plo faultfs.PowerLossOptions
44+ dead bool
45+}
46+
47+func (h *hookFS) run(op, path string, fn func() error) error {
48+ if h.serial {
49+ h.mu.Lock()
50+ defer h.mu.Unlock()
51+ if h.dead {
52+ return errCrash
53+ }
54+ if h.armed {
55+ h.trace = append(h.trace, op+" "+filepath.Base(path))
56+ if len(h.trace) == h.crashAt {
57+ h.cutLocked(h.plo)
58+ return errCrash
59+ }
60+ }
61+ }
62+ if h.hook != nil {
63+ if err := h.hook(op, path); err != nil {
64+ return err
65+ }
66+ }
67+ return fn()
68+}
69+
70+func (h *hookFS) arm() {
71+ h.mu.Lock()
72+ h.armed = true
73+ h.mu.Unlock()
74+}
75+
76+func (h *hookFS) isDead() bool {
77+ h.mu.Lock()
78+ defer h.mu.Unlock()
79+ return h.dead
80+}
81+
82+func (h *hookFS) steps() []string {
83+ h.mu.Lock()
84+ defer h.mu.Unlock()
85+ return slices.Clone(h.trace)
86+}
87+
88+// cut cuts the power: the unsynced state is dropped as plo says and every
89+// later operation fails.
90+func (h *hookFS) cut(plo faultfs.PowerLossOptions) {
91+ h.mu.Lock()
92+ defer h.mu.Unlock()
93+ if !h.dead {
94+ h.cutLocked(plo)
95+ }
96+}
97+
98+func (h *hookFS) cutLocked(plo faultfs.PowerLossOptions) {
99+ h.dead = true
100+ h.ff.PowerLoss(plo)
101+}
102+
103+func (h *hookFS) OpenFile(name string, flag int, perm os.FileMode) (fsys.File, error) {
104+ var f fsys.File
105+ err := h.run("open", name, func() (err error) {
106+ f, err = h.FS.OpenFile(name, flag, perm)
107+ return err
108+ })
109+ if err != nil {
110+ return nil, err
111+ }
112+ return &hookFile{File: f, fs: h, path: name}, nil
113+}
114+
115+func (h *hookFS) Rename(oldpath, newpath string) error {
116+ return h.run("rename", newpath, func() error { return h.FS.Rename(oldpath, newpath) })
117+}
118+
119+func (h *hookFS) Remove(name string) error {
120+ return h.run("remove", name, func() error { return h.FS.Remove(name) })
121+}
122+
123+func (h *hookFS) SyncDir(dir string) error {
124+ return h.run("syncdir", dir, func() error { return h.FS.SyncDir(dir) })
125+}
126+
127+type hookFile struct {
128+ fsys.File
129+ fs *hookFS
130+ path string
131+}
132+
133+func (f *hookFile) Write(p []byte) (n int, err error) {
134+ err = f.fs.run("write", f.path, func() (err error) {
135+ n, err = f.File.Write(p)
136+ return err
137+ })
138+ return n, err
139+}
140+
141+func (f *hookFile) Sync() error { return f.fs.run("sync", f.path, f.File.Sync) }
142+func (f *hookFile) SyncAll() error { return f.fs.run("syncall", f.path, f.File.SyncAll) }
143+
144+func (f *hookFile) Truncate(size int64) error {
145+ return f.fs.run("truncate", f.path, func() error { return f.File.Truncate(size) })
146+}
147+
148+// The sequential workload: batch i (which gets seq i) puts one key.
149+const seqPad = "................................"
150+
151+func seqKey(i int) []byte { return fmt.Appendf(nil, "k%06d", i) }
152+func seqValue(i int) []byte { return fmt.Appendf(nil, "v%06d%s", i, seqPad) }
153+
154+// recSize is the encoded size of the log record of one sequential batch; it
155+// does not depend on i. Options.CompactAfter = n*recSize makes the (n+1)-th
156+// record of a log trigger a compaction.
157+var recSize = int64(len(appendRecord(nil, 1, []op{{kind: opPut, key: seqKey(1), value: seqValue(1)}})))
158+
159+// compactOpts compacts when the 6th record of a log is written.
160+func compactOpts() Options { return Options{CompactAfter: 5 * recSize} }
161+
162+// applySeq applies batch i and checks that it gets seq i.
163+func applySeq(db *DB, i int) error {
164+ var b Batch
165+ b.Put(seqKey(i), seqValue(i))
166+ seq, err := db.Apply(bg, &b)
167+ if err == nil && seq != uint64(i) {
168+ err = fmt.Errorf("batch %d got seq %d", i, seq)
169+ }
170+ return err
171+}
172+
173+func mustApply(t testing.TB, db *DB, i int) {
174+ t.Helper()
175+ if err := applySeq(db, i); err != nil {
176+ t.Fatalf("apply %d: %v", i, err)
177+ }
178+}
179+
180+// barrier returns once the writer has finished everything it did for the
181+// batches applied so far, including a log switch: an empty batch is only
182+// taken after the previous loop iteration, which ends with the compaction
183+// trigger.
184+func barrier(t testing.TB, db *DB) {
185+ t.Helper()
186+ if _, err := db.Apply(bg, new(Batch)); err != nil {
187+ t.Fatalf("barrier: %v", err)
188+ }
189+}
190+
191+// checkSequential checks that s holds exactly the keys of batches 1..N for
192+// some N ≥ lastAcked, with the right values and versions, and returns N.
193+func checkSequential(s *Snapshot, lastAcked int) (int, error) {
194+ n := s.Len()
195+ if s.Seq() != uint64(n) {
196+ return n, fmt.Errorf("seq %d but %d keys", s.Seq(), n)
197+ }
198+ if n < lastAcked {
199+ return n, fmt.Errorf("acknowledged batch %d lost: recovered %d batches", lastAcked, n)
200+ }
201+ for i := 1; i <= n; i++ {
202+ it, ok := s.Get(bg, seqKey(i))
203+ if !ok || string(it.Value) != string(seqValue(i)) || it.Version != uint64(i) {
204+ return n, fmt.Errorf("key %d: %q at version %d, present %v", i, it.Value, it.Version, ok)
205+ }
206+ }
207+ return n, nil
208+}
209+
210+func mustCheck(t testing.TB, db *DB, lastAcked int) int {
211+ t.Helper()
212+ n, err := checkSequential(db.Snapshot(bg), lastAcked)
213+ if err != nil {
214+ t.Fatal(err)
215+ }
216+ return n
217+}
218+
219+// listFiles returns the seqs of the snapshot and log files in testDir.
220+func listFiles(t testing.TB, fx fsys.FS) (snaps, logs []uint64) {
221+ t.Helper()
222+ names, err := fx.ReadDir(testDir)
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ for _, name := range names {
227+ if s, ok := parseSeqName(name, snapshotPrefix); ok {
228+ snaps = append(snaps, s)
229+ } else if b, ok := parseSeqName(name, logPrefix); ok {
230+ logs = append(logs, b)
231+ }
232+ }
233+ return snaps, logs
234+}
235+
236+// settled reports whether the directory holds exactly snapshot-S and log-S.
237+func settled(t testing.TB, fx fsys.FS, s uint64) bool {
238+ snaps, logs := listFiles(t, fx)
239+ return slices.Equal(snaps, []uint64{s}) && slices.Equal(logs, []uint64{s})
240+}
241+
242+func waitFor(t testing.TB, what string, cond func() bool) {
243+ 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)
248+ }
249+ time.Sleep(time.Millisecond)
250+ }
251+}
252+
253+// debugCtx returns a context whose logger writes JSON at debug level to the
254+// returned buffer.
255+func debugCtx() (context.Context, *syncBuffer) {
256+ buf := new(syncBuffer)
257+ return zerowrap.WithCtx(bg, zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: buf})), buf
258+}
259+
260+func logged(buf *syncBuffer, msg string) int {
261+ return strings.Count(buf.String(), `"message":"`+msg+`"`)
262+}
263+
264+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+ }
277+
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())
290+ }
291+ }
292+
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+ }
302+}
303+
304+func TestCompactDisabled(t *testing.T) {
305+ fx := faultfs.New()
306+ db := mustOpen(t, fx, Options{CompactAfter: -1})
307+ for i := 1; i <= 50; i++ {
308+ mustApply(t, db, i)
309+ }
310+ barrier(t, db)
311+ if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
312+ t.Fatalf("snapshots %v, logs %v", snaps, files)
313+ }
314+}
315+
316+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+ }
336+
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+ }
349+
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
357+ }
358+ mustApply(t, db, i)
359+ i++
360+ return false
361+ })
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+}
375+
376+func TestCompactRemovesOldFiles(t *testing.T) {
377+ fx := faultfs.New()
378+ db := mustOpen(t, fx, compactOpts())
379+ const total = 200
380+ for i := 1; i <= total; i++ {
381+ mustApply(t, db, i)
382+ }
383+ if err := db.Close(bg); err != nil { // waits for the last compaction
384+ t.Fatal(err)
385+ }
386+ snaps, files := listFiles(t, fx)
387+ if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] == 0 {
388+ t.Fatalf("snapshots %v, logs %v: want only the newest pair", snaps, files)
389+ }
390+ names, _ := fx.ReadDir(testDir)
391+ for _, n := range names {
392+ if strings.HasSuffix(n, tmpSuffix) {
393+ t.Errorf("leftover %s", n)
394+ }
395+ }
396+ db = mustOpen(t, fx, compactOpts())
397+ if n := mustCheck(t, db, total); n != total {
398+ t.Fatalf("recovered %d batches, want %d", n, total)
399+ }
400+ // A reopened database keeps compacting.
401+ for i := total + 1; i <= 2*total; i++ {
402+ mustApply(t, db, i)
403+ }
404+ if err := db.Close(bg); err != nil {
405+ t.Fatal(err)
406+ }
407+ db = mustOpen(t, fx, Options{})
408+ mustCheck(t, db, 2*total)
409+}
410+
411+// TestCompactOS runs compaction on the real file system.
412+func TestCompactOS(t *testing.T) {
413+ dir := filepath.Join(t.TempDir(), "db")
414+ ctx, logs := debugCtx()
415+ db, err := Open(ctx, dir, compactOpts())
416+ if err != nil {
417+ t.Fatal(err)
418+ }
419+ for i := 1; i <= 100; i++ {
420+ mustApply(t, db, i)
421+ }
422+ if err := db.Close(bg); err != nil {
423+ t.Fatal(err)
424+ }
425+ if logged(logs, "compaction done") == 0 {
426+ t.Fatal("no compaction on the OS file system")
427+ }
428+ db, err = Open(bg, dir, Options{})
429+ if err != nil {
430+ t.Fatal(err)
431+ }
432+ defer func() { _ = db.Close(bg) }()
433+ mustCheck(t, db, 100)
434+}
435+
436+// compactorFaults fail one step of the compactor, which runs after the log
437+// switch. The injection happens before batch 6 is applied, so the numbering
438+// counts from the group write: Write 1 is the group, 2 the header of log-6, 3
439+// the snapshot; the syncs count Sync and SyncAll (1 the group, 2 the log
440+// before the switch, 3 log-6, 4 the snapshot); SyncDir 1 is log-6, 2 the
441+// snapshot, 3 the cleanup; Rename 1 is log-6, 2 the snapshot; OpenFile 1 is
442+// log-6, 2 the snapshot; Remove 1 is log-0.
443+var compactorFaults = []struct {
444+ name string
445+ inject func(fx *faultfs.FS)
446+}{
447+ {"open snapshot", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpOpenFile, 2, nil) }},
448+ {"write snapshot", func(fx *faultfs.FS) { fx.FailWrite(3, nil) }},
449+ {"short write snapshot", func(fx *faultfs.FS) { fx.ShortWrite(3, 10) }},
450+ {"sync snapshot", func(fx *faultfs.FS) { fx.FailSync(4, nil) }},
451+ {"rename snapshot", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpRename, 2, nil) }},
452+ {"syncdir snapshot", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpSyncDir, 2, nil) }},
453+ {"remove old log", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpRemove, 1, nil) }},
454+ {"syncdir cleanup", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpSyncDir, 3, nil) }},
455+}
456+
457+// TestCompactorFailure checks that a failing compactor is logged, leaves the
458+// database usable and a recoverable directory, and that the next trigger
459+// compacts.
460+func TestCompactorFailure(t *testing.T) {
461+ for _, f := range compactorFaults {
462+ 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+ }
479+
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
488+ }
489+ mustApply(t, db, i)
490+ i++
491+ return false
492+ })
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+ })
503+ }
504+}
505+
506+// TestCompactorRetryWithoutReopen checks the retry in the same process.
507+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
521+ }
522+ mustApply(t, db, i)
523+ i++
524+ return false
525+ })
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+}
534+
535+// switchFaults fail the creation of log-6 in the writer. The numbering is
536+// the one of compactorFaults.
537+var switchFaults = []struct {
538+ name string
539+ inject func(fx *faultfs.FS)
540+}{
541+ {"open", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpOpenFile, 1, nil) }},
542+ {"write", func(fx *faultfs.FS) { fx.FailWrite(2, nil) }},
543+ {"sync", func(fx *faultfs.FS) { fx.FailSync(3, nil) }},
544+ {"rename", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpRename, 1, nil) }},
545+ {"syncdir", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpSyncDir, 1, nil) }},
546+}
547+
548+// TestLogSwitchFailure checks that failing to create the new log only
549+// postpones the compaction: the database goes on appending to the current
550+// log and no log-6 is left behind.
551+func TestLogSwitchFailure(t *testing.T) {
552+ for _, f := range switchFaults {
553+ 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)
574+ }
575+ }
576+
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
587+ }
588+ mustApply(t, db, i)
589+ i++
590+ return false
591+ })
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+ })
598+ }
599+}
600+
601+// TestLogSwitchFailsDatabase covers the failures that must fail the database:
602+// the sync of the current log, and a stray log-S that cannot be removed.
603+func TestLogSwitchFailsDatabase(t *testing.T) {
604+ cases := []struct {
605+ name string
606+ inject func(fx *faultfs.FS)
607+ }{
608+ {"syncall of the current log", func(fx *faultfs.FS) { fx.FailSync(2, nil) }},
609+ {"syncdir then remove", func(fx *faultfs.FS) {
610+ fx.FailOp(faultfs.OpSyncDir, 1, nil)
611+ fx.FailOp(faultfs.OpRemove, 1, nil)
612+ }},
613+ }
614+ for _, c := range cases {
615+ 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
630+ }
631+ })
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+ })
646+ }
647+}
648+
649+// TestLogSwitchSyncFailureDropsUnsyncedData is the fsyncgate case with
650+// NoSync: the SyncAll before the switch is the first sync of the log. When it
651+// fails and the kernel drops the pages, the batches acknowledged without a
652+// sync are gone; the database must fail rather than continue, and what is
653+// left must open as a prefix.
654+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
671+ }
672+ })
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+}
681+
682+// TestCloseWaitsForCompaction checks that Close does not cancel a running
683+// compaction, waits for it, and leaves a directory that opens.
684+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)
719+ }
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)
728+}
729+
730+// TestCloseDuringCompaction (AC9) closes the database while writers run and
731+// compactions are frequent: Close returns, and the next Open finds every
732+// acknowledged batch.
733+func TestCloseDuringCompaction(t *testing.T) {
734+ var switches atomic.Int64
735+ for i := range short(12, 4) {
736+ t.Run(fmt.Sprint(i), func(t *testing.T) {
737+ fx := faultfs.New()
738+ fx.SetSyncDelay(50 * time.Microsecond)
739+ ctx, logs := debugCtx()
740+ db := mustOpenCtx(t, ctx, fx, Options{CompactAfter: 2 * recSize})
741+ const writers = 16
742+ acked := make([]int, writers) // highest acknowledged n+1
743+ var wg sync.WaitGroup
744+ for g := range writers {
745+ wg.Go(func() {
746+ for n := 0; ; n++ {
747+ var b Batch
748+ b.Put(fmt.Appendf(nil, "g%02d", g), seqValue(n))
749+ b.Put(fmt.Appendf(nil, "p%02d/%06d", g, n), nil)
750+ _, err := db.Apply(bg, &b)
751+ if errors.Is(err, ErrClosed) {
752+ return
753+ }
754+ if err != nil {
755+ t.Errorf("writer %d: %v", g, err)
756+ return
757+ }
758+ acked[g] = n + 1
759+ }
760+ })
761+ }
762+ time.Sleep(time.Duration(5+3*i) * time.Millisecond)
763+ closed := make(chan error, 1)
764+ go func() { closed <- db.Close(bg) }()
765+ select {
766+ case err := <-closed:
767+ if err != nil {
768+ t.Errorf("Close: %v", err)
769+ }
770+ case <-time.After(20 * time.Second):
771+ t.Fatal("Close did not return")
772+ }
773+ wg.Wait()
774+ switches.Add(int64(logged(logs, "log switch")))
775+
776+ fx.SetSyncDelay(0)
777+ s := mustOpen(t, fx, Options{}).Snapshot(bg)
778+ for g, n := range acked {
779+ it, ok := s.Get(bg, fmt.Appendf(nil, "g%02d", g))
780+ if n == 0 {
781+ continue
782+ }
783+ // The batch after the last acknowledged one may have been
784+ // committed by the final group.
785+ if !ok || string(it.Value) != string(seqValue(n-1)) && string(it.Value) != string(seqValue(n)) {
786+ t.Errorf("writer %d: acked %d batches, recovered %q (%v)", g, n, it.Value, ok)
787+ }
788+ if _, ok := s.Get(bg, fmt.Appendf(nil, "p%02d/%06d", g, n-1)); !ok {
789+ t.Errorf("writer %d: acked batch %d lost", g, n-1)
790+ }
791+ }
792+ })
793+ }
794+ if switches.Load() == 0 {
795+ t.Error("no log switch happened: the test did not exercise compaction")
796+ }
797+}
798+
799+// compactSteps runs a workload of three rounds of six batches (each round
800+// ends with a compaction) on a fresh file system and cuts the power before
801+// the crashAt-th file operation after Open (0 and out of range: after the
802+// workload). It then reopens the database and reports a violation of AC2.
803+// The workload is sequential and waits for each compaction, so the sequence
804+// of operations is the same in every run.
805+func compactSteps(t *testing.T, plo faultfs.PowerLossOptions, crashAt int) (steps []string, err error) {
806+ ff := faultfs.New()
807+ h := &hookFS{FS: ff, ff: ff, serial: true, crashAt: crashAt, plo: plo}
808+ db, oerr := open(testCtx(t), h, testDir, compactOpts())
809+ if oerr != nil {
810+ return nil, fmt.Errorf("open: %w", oerr)
811+ }
812+ h.arm()
813+ lastAcked := 0
814+workload:
815+ for round := 1; round <= 3; round++ {
816+ for k := 1; k <= 6; k++ {
817+ i := (round-1)*6 + k
818+ if applySeq(db, i) != nil {
819+ break workload
820+ }
821+ lastAcked = i
822+ }
823+ // Wait for the compaction of this round.
824+ deadline := time.Now().Add(10 * time.Second)
825+ for !h.isDead() && (!settled(t, ff, uint64(6*round)) || db.compacting.Load()) {
826+ if time.Now().After(deadline) {
827+ return h.steps(), errors.New("timeout waiting for the compaction")
828+ }
829+ time.Sleep(100 * time.Microsecond)
830+ }
831+ }
832+ steps = h.steps()
833+ _ = db.Close(bg)
834+ h.cut(plo) // no-op if the power was already cut
835+
836+ for range 2 {
837+ db2, err := open(testCtx(t), ff, testDir, Options{})
838+ if err != nil {
839+ return steps, fmt.Errorf("reopen: %w", err)
840+ }
841+ n, err := checkSequential(db2.Snapshot(bg), lastAcked)
842+ if err == nil {
843+ // The recovered log accepts new batches.
844+ err = applySeq(db2, n+1)
845+ lastAcked = n + 1
846+ }
847+ if cerr := db2.Close(bg); err == nil {
848+ err = cerr
849+ }
850+ if err != nil {
851+ return steps, err
852+ }
853+ }
854+ return steps, nil
855+}
856+
857+// TestCompactPowerLossAtEachStep cuts the power before each file operation
858+// of three compactions in turn, in every power-loss mode, and checks AC2:
859+// the reopened database holds every acknowledged batch and is a prefix.
860+func TestCompactPowerLossAtEachStep(t *testing.T) {
861+ modes := []struct {
862+ name string
863+ plo faultfs.PowerLossOptions
864+ }{
865+ {"drop", faultfs.PowerLossOptions{}},
866+ {"zerofill", faultfs.PowerLossOptions{ZeroFill: true}},
867+ {"keeppages", faultfs.PowerLossOptions{KeepPages: true, PageSize: 64}},
868+ {"keeppages+zerofill", faultfs.PowerLossOptions{KeepPages: true, ZeroFill: true, PageSize: 64}},
869+ }
870+ base := powerLossBase(t)
871+ for _, m := range modes {
872+ t.Run(m.name, func(t *testing.T) {
873+ plo := m.plo
874+ plo.Rand = rand.New(rand.NewPCG(base, 7))
875+ steps, err := compactSteps(t, plo, 0)
876+ if err != nil {
877+ t.Fatalf("without a cut: %v", err)
878+ }
879+ t.Logf("%d file operations in the workload", len(steps))
880+ for step := 1; step <= len(steps); step++ {
881+ plo.Rand = rand.New(rand.NewPCG(base+uint64(step), 7))
882+ if _, err := compactSteps(t, plo, step); err != nil {
883+ t.Fatalf("power cut before step %d (%s), seed %d (ERMITE_KV_SEED=%d reproduces it): %v",
884+ step, steps[step-1], base, base, err)
885+ }
886+ }
887+ })
888+ }
889+}
890+
891+// TestCompactPowerLoss cuts the power at a random time during a workload
892+// that compacts all the time, then checks AC2. With NoSync the acknowledged
893+// batches may be lost, but the directory must open and be a prefix; only the
894+// modes that cannot make Open fail with ErrCorrupt are used (see
895+// Options.NoSync).
896+func TestCompactPowerLoss(t *testing.T) {
897+ modes := []struct {
898+ name string
899+ nosync bool
900+ plo faultfs.PowerLossOptions
901+ }{
902+ {"drop", false, faultfs.PowerLossOptions{}},
903+ {"zerofill", false, faultfs.PowerLossOptions{ZeroFill: true}},
904+ {"keeppages", false, faultfs.PowerLossOptions{KeepPages: true}},
905+ {"keeppages+zerofill", false, faultfs.PowerLossOptions{KeepPages: true, ZeroFill: true}},
906+ {"nosync drop", true, faultfs.PowerLossOptions{}},
907+ {"nosync zerofill", true, faultfs.PowerLossOptions{ZeroFill: true}},
908+ }
909+ base := powerLossBase(t)
910+ iterations := short(100, 10)
911+ for _, m := range modes {
912+ t.Run(m.name, func(t *testing.T) {
913+ for i := range iterations {
914+ seed := base + uint64(i)
915+ if err := runCompactPowerLoss(t, seed, m.plo, m.nosync); err != nil {
916+ t.Fatalf("seed %d (ERMITE_KV_SEED=%d reproduces it): %v", seed, seed, err)
917+ }
918+ }
919+ })
920+ }
921+}
922+
923+func runCompactPowerLoss(t *testing.T, seed uint64, plo faultfs.PowerLossOptions, nosync bool) error {
924+ rng := rand.New(rand.NewPCG(seed, 0x9e3779b97f4a7c15))
925+ ff := faultfs.New()
926+ h := &hookFS{FS: ff, ff: ff, serial: true}
927+ opts := Options{NoSync: nosync, CompactAfter: int64(2+rng.IntN(10)) * recSize}
928+ db, err := open(testCtx(t), h, testDir, opts)
929+ if err != nil {
930+ return fmt.Errorf("open: %w", err)
931+ }
932+ ff.SetSyncDelay(time.Duration(rng.IntN(100)) * time.Microsecond)
933+ var lastAcked atomic.Int64
934+ done := make(chan struct{})
935+ go func() {
936+ defer close(done)
937+ for i := 1; applySeq(db, i) == nil; i++ {
938+ lastAcked.Store(int64(i))
939+ }
940+ }()
941+ time.Sleep(time.Duration(rng.IntN(6000)) * time.Microsecond)
942+ plo.Rand = rand.New(rand.NewPCG(seed, 7))
943+ if plo.KeepPages {
944+ plo.PageSize = 64
945+ }
946+ h.cut(plo)
947+ <-done
948+ _ = db.Close(bg)
949+ ff.SetSyncDelay(0)
950+
951+ floor := int(lastAcked.Load())
952+ if nosync {
953+ floor = 0
954+ }
955+ for range 2 {
956+ db, err = open(testCtx(t), ff, testDir, Options{})
957+ if err != nil {
958+ return fmt.Errorf("reopen: %w", err)
959+ }
960+ n, err := checkSequential(db.Snapshot(bg), floor)
961+ if err == nil {
962+ err = applySeq(db, n+1) // appends to the right log
963+ floor = n + 1
964+ }
965+ if cerr := db.Close(bg); err == nil {
966+ err = cerr
967+ }
968+ if err != nil {
969+ return err
970+ }
971+ }
972+ return nil
973+}
@@ -0,0 +1,973 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
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+ "time"
18+
19+ "github.com/bnema/zerowrap"
20+
21+ "bnema.dev/ermite-kv/internal/fsys"
22+ "bnema.dev/ermite-kv/internal/fsys/faultfs"
23+)
24+
25+var errCrash = errors.New("test: power was cut")
26+
27+// hookFS wraps an FS and intercepts the operations that change files:
28+// hook (if set) may block or fail them, and a power loss can be cut at a
29+// given step. With serial set, operations run one at a time under mu, so a
30+// cut is atomic with respect to every other operation and nothing touches the
31+// file system afterwards: once cut, every operation fails with errCrash, like
32+// a process that no longer exists.
33+type hookFS struct {
34+ fsys.FS
35+ ff *faultfs.FS // underlying FS, needed by cut
36+ serial bool // run operations under mu
37+ hook func(op, path string) error // optional; set before use
38+
39+ mu sync.Mutex // guards the fields below when serial
40+ armed bool // record the operations in trace
41+ trace []string
42+ crashAt int // cut the power before the crashAt-th traced operation (0: never)
43+ plo faultfs.PowerLossOptions
44+ dead bool
45+}
46+
47+func (h *hookFS) run(op, path string, fn func() error) error {
48+ if h.serial {
49+ h.mu.Lock()
50+ defer h.mu.Unlock()
51+ if h.dead {
52+ return errCrash
53+ }
54+ if h.armed {
55+ h.trace = append(h.trace, op+" "+filepath.Base(path))
56+ if len(h.trace) == h.crashAt {
57+ h.cutLocked(h.plo)
58+ return errCrash
59+ }
60+ }
61+ }
62+ if h.hook != nil {
63+ if err := h.hook(op, path); err != nil {
64+ return err
65+ }
66+ }
67+ return fn()
68+}
69+
70+func (h *hookFS) arm() {
71+ h.mu.Lock()
72+ h.armed = true
73+ h.mu.Unlock()
74+}
75+
76+func (h *hookFS) isDead() bool {
77+ h.mu.Lock()
78+ defer h.mu.Unlock()
79+ return h.dead
80+}
81+
82+func (h *hookFS) steps() []string {
83+ h.mu.Lock()
84+ defer h.mu.Unlock()
85+ return slices.Clone(h.trace)
86+}
87+
88+// cut cuts the power: the unsynced state is dropped as plo says and every
89+// later operation fails.
90+func (h *hookFS) cut(plo faultfs.PowerLossOptions) {
91+ h.mu.Lock()
92+ defer h.mu.Unlock()
93+ if !h.dead {
94+ h.cutLocked(plo)
95+ }
96+}
97+
98+func (h *hookFS) cutLocked(plo faultfs.PowerLossOptions) {
99+ h.dead = true
100+ h.ff.PowerLoss(plo)
101+}
102+
103+func (h *hookFS) OpenFile(name string, flag int, perm os.FileMode) (fsys.File, error) {
104+ var f fsys.File
105+ err := h.run("open", name, func() (err error) {
106+ f, err = h.FS.OpenFile(name, flag, perm)
107+ return err
108+ })
109+ if err != nil {
110+ return nil, err
111+ }
112+ return &hookFile{File: f, fs: h, path: name}, nil
113+}
114+
115+func (h *hookFS) Rename(oldpath, newpath string) error {
116+ return h.run("rename", newpath, func() error { return h.FS.Rename(oldpath, newpath) })
117+}
118+
119+func (h *hookFS) Remove(name string) error {
120+ return h.run("remove", name, func() error { return h.FS.Remove(name) })
121+}
122+
123+func (h *hookFS) SyncDir(dir string) error {
124+ return h.run("syncdir", dir, func() error { return h.FS.SyncDir(dir) })
125+}
126+
127+type hookFile struct {
128+ fsys.File
129+ fs *hookFS
130+ path string
131+}
132+
133+func (f *hookFile) Write(p []byte) (n int, err error) {
134+ err = f.fs.run("write", f.path, func() (err error) {
135+ n, err = f.File.Write(p)
136+ return err
137+ })
138+ return n, err
139+}
140+
141+func (f *hookFile) Sync() error { return f.fs.run("sync", f.path, f.File.Sync) }
142+func (f *hookFile) SyncAll() error { return f.fs.run("syncall", f.path, f.File.SyncAll) }
143+
144+func (f *hookFile) Truncate(size int64) error {
145+ return f.fs.run("truncate", f.path, func() error { return f.File.Truncate(size) })
146+}
147+
148+// The sequential workload: batch i (which gets seq i) puts one key.
149+const seqPad = "................................"
150+
151+func seqKey(i int) []byte { return fmt.Appendf(nil, "k%06d", i) }
152+func seqValue(i int) []byte { return fmt.Appendf(nil, "v%06d%s", i, seqPad) }
153+
154+// recSize is the encoded size of the log record of one sequential batch; it
155+// does not depend on i. Options.CompactAfter = n*recSize makes the (n+1)-th
156+// record of a log trigger a compaction.
157+var recSize = int64(len(appendRecord(nil, 1, []op{{kind: opPut, key: seqKey(1), value: seqValue(1)}})))
158+
159+// compactOpts compacts when the 6th record of a log is written.
160+func compactOpts() Options { return Options{CompactAfter: 5 * recSize} }
161+
162+// applySeq applies batch i and checks that it gets seq i.
163+func applySeq(db *DB, i int) error {
164+ var b Batch
165+ b.Put(seqKey(i), seqValue(i))
166+ seq, err := db.Apply(bg, &b)
167+ if err == nil && seq != uint64(i) {
168+ err = fmt.Errorf("batch %d got seq %d", i, seq)
169+ }
170+ return err
171+}
172+
173+func mustApply(t testing.TB, db *DB, i int) {
174+ t.Helper()
175+ if err := applySeq(db, i); err != nil {
176+ t.Fatalf("apply %d: %v", i, err)
177+ }
178+}
179+
180+// barrier returns once the writer has finished everything it did for the
181+// batches applied so far, including a log switch: an empty batch is only
182+// taken after the previous loop iteration, which ends with the compaction
183+// trigger.
184+func barrier(t testing.TB, db *DB) {
185+ t.Helper()
186+ if _, err := db.Apply(bg, new(Batch)); err != nil {
187+ t.Fatalf("barrier: %v", err)
188+ }
189+}
190+
191+// checkSequential checks that s holds exactly the keys of batches 1..N for
192+// some N ≥ lastAcked, with the right values and versions, and returns N.
193+func checkSequential(s *Snapshot, lastAcked int) (int, error) {
194+ n := s.Len()
195+ if s.Seq() != uint64(n) {
196+ return n, fmt.Errorf("seq %d but %d keys", s.Seq(), n)
197+ }
198+ if n < lastAcked {
199+ return n, fmt.Errorf("acknowledged batch %d lost: recovered %d batches", lastAcked, n)
200+ }
201+ for i := 1; i <= n; i++ {
202+ it, ok := s.Get(bg, seqKey(i))
203+ if !ok || string(it.Value) != string(seqValue(i)) || it.Version != uint64(i) {
204+ return n, fmt.Errorf("key %d: %q at version %d, present %v", i, it.Value, it.Version, ok)
205+ }
206+ }
207+ return n, nil
208+}
209+
210+func mustCheck(t testing.TB, db *DB, lastAcked int) int {
211+ t.Helper()
212+ n, err := checkSequential(db.Snapshot(bg), lastAcked)
213+ if err != nil {
214+ t.Fatal(err)
215+ }
216+ return n
217+}
218+
219+// listFiles returns the seqs of the snapshot and log files in testDir.
220+func listFiles(t testing.TB, fx fsys.FS) (snaps, logs []uint64) {
221+ t.Helper()
222+ names, err := fx.ReadDir(testDir)
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ for _, name := range names {
227+ if s, ok := parseSeqName(name, snapshotPrefix); ok {
228+ snaps = append(snaps, s)
229+ } else if b, ok := parseSeqName(name, logPrefix); ok {
230+ logs = append(logs, b)
231+ }
232+ }
233+ return snaps, logs
234+}
235+
236+// settled reports whether the directory holds exactly snapshot-S and log-S.
237+func settled(t testing.TB, fx fsys.FS, s uint64) bool {
238+ snaps, logs := listFiles(t, fx)
239+ return slices.Equal(snaps, []uint64{s}) && slices.Equal(logs, []uint64{s})
240+}
241+
242+func waitFor(t testing.TB, what string, cond func() bool) {
243+ 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)
248+ }
249+ time.Sleep(time.Millisecond)
250+ }
251+}
252+
253+// debugCtx returns a context whose logger writes JSON at debug level to the
254+// returned buffer.
255+func debugCtx() (context.Context, *syncBuffer) {
256+ buf := new(syncBuffer)
257+ return zerowrap.WithCtx(bg, zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: buf})), buf
258+}
259+
260+func logged(buf *syncBuffer, msg string) int {
261+ return strings.Count(buf.String(), `"message":"`+msg+`"`)
262+}
263+
264+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+ }
277+
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())
290+ }
291+ }
292+
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+ }
302+}
303+
304+func TestCompactDisabled(t *testing.T) {
305+ fx := faultfs.New()
306+ db := mustOpen(t, fx, Options{CompactAfter: -1})
307+ for i := 1; i <= 50; i++ {
308+ mustApply(t, db, i)
309+ }
310+ barrier(t, db)
311+ if snaps, files := listFiles(t, fx); len(snaps) != 0 || !slices.Equal(files, []uint64{0}) {
312+ t.Fatalf("snapshots %v, logs %v", snaps, files)
313+ }
314+}
315+
316+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+ }
336+
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+ }
349+
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
357+ }
358+ mustApply(t, db, i)
359+ i++
360+ return false
361+ })
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+}
375+
376+func TestCompactRemovesOldFiles(t *testing.T) {
377+ fx := faultfs.New()
378+ db := mustOpen(t, fx, compactOpts())
379+ const total = 200
380+ for i := 1; i <= total; i++ {
381+ mustApply(t, db, i)
382+ }
383+ if err := db.Close(bg); err != nil { // waits for the last compaction
384+ t.Fatal(err)
385+ }
386+ snaps, files := listFiles(t, fx)
387+ if len(snaps) != 1 || !slices.Equal(files, snaps) || snaps[0] == 0 {
388+ t.Fatalf("snapshots %v, logs %v: want only the newest pair", snaps, files)
389+ }
390+ names, _ := fx.ReadDir(testDir)
391+ for _, n := range names {
392+ if strings.HasSuffix(n, tmpSuffix) {
393+ t.Errorf("leftover %s", n)
394+ }
395+ }
396+ db = mustOpen(t, fx, compactOpts())
397+ if n := mustCheck(t, db, total); n != total {
398+ t.Fatalf("recovered %d batches, want %d", n, total)
399+ }
400+ // A reopened database keeps compacting.
401+ for i := total + 1; i <= 2*total; i++ {
402+ mustApply(t, db, i)
403+ }
404+ if err := db.Close(bg); err != nil {
405+ t.Fatal(err)
406+ }
407+ db = mustOpen(t, fx, Options{})
408+ mustCheck(t, db, 2*total)
409+}
410+
411+// TestCompactOS runs compaction on the real file system.
412+func TestCompactOS(t *testing.T) {
413+ dir := filepath.Join(t.TempDir(), "db")
414+ ctx, logs := debugCtx()
415+ db, err := Open(ctx, dir, compactOpts())
416+ if err != nil {
417+ t.Fatal(err)
418+ }
419+ for i := 1; i <= 100; i++ {
420+ mustApply(t, db, i)
421+ }
422+ if err := db.Close(bg); err != nil {
423+ t.Fatal(err)
424+ }
425+ if logged(logs, "compaction done") == 0 {
426+ t.Fatal("no compaction on the OS file system")
427+ }
428+ db, err = Open(bg, dir, Options{})
429+ if err != nil {
430+ t.Fatal(err)
431+ }
432+ defer func() { _ = db.Close(bg) }()
433+ mustCheck(t, db, 100)
434+}
435+
436+// compactorFaults fail one step of the compactor, which runs after the log
437+// switch. The injection happens before batch 6 is applied, so the numbering
438+// counts from the group write: Write 1 is the group, 2 the header of log-6, 3
439+// the snapshot; the syncs count Sync and SyncAll (1 the group, 2 the log
440+// before the switch, 3 log-6, 4 the snapshot); SyncDir 1 is log-6, 2 the
441+// snapshot, 3 the cleanup; Rename 1 is log-6, 2 the snapshot; OpenFile 1 is
442+// log-6, 2 the snapshot; Remove 1 is log-0.
443+var compactorFaults = []struct {
444+ name string
445+ inject func(fx *faultfs.FS)
446+}{
447+ {"open snapshot", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpOpenFile, 2, nil) }},
448+ {"write snapshot", func(fx *faultfs.FS) { fx.FailWrite(3, nil) }},
449+ {"short write snapshot", func(fx *faultfs.FS) { fx.ShortWrite(3, 10) }},
450+ {"sync snapshot", func(fx *faultfs.FS) { fx.FailSync(4, nil) }},
451+ {"rename snapshot", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpRename, 2, nil) }},
452+ {"syncdir snapshot", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpSyncDir, 2, nil) }},
453+ {"remove old log", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpRemove, 1, nil) }},
454+ {"syncdir cleanup", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpSyncDir, 3, nil) }},
455+}
456+
457+// TestCompactorFailure checks that a failing compactor is logged, leaves the
458+// database usable and a recoverable directory, and that the next trigger
459+// compacts.
460+func TestCompactorFailure(t *testing.T) {
461+ for _, f := range compactorFaults {
462+ 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+ }
479+
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
488+ }
489+ mustApply(t, db, i)
490+ i++
491+ return false
492+ })
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+ })
503+ }
504+}
505+
506+// TestCompactorRetryWithoutReopen checks the retry in the same process.
507+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
521+ }
522+ mustApply(t, db, i)
523+ i++
524+ return false
525+ })
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+}
534+
535+// switchFaults fail the creation of log-6 in the writer. The numbering is
536+// the one of compactorFaults.
537+var switchFaults = []struct {
538+ name string
539+ inject func(fx *faultfs.FS)
540+}{
541+ {"open", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpOpenFile, 1, nil) }},
542+ {"write", func(fx *faultfs.FS) { fx.FailWrite(2, nil) }},
543+ {"sync", func(fx *faultfs.FS) { fx.FailSync(3, nil) }},
544+ {"rename", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpRename, 1, nil) }},
545+ {"syncdir", func(fx *faultfs.FS) { fx.FailOp(faultfs.OpSyncDir, 1, nil) }},
546+}
547+
548+// TestLogSwitchFailure checks that failing to create the new log only
549+// postpones the compaction: the database goes on appending to the current
550+// log and no log-6 is left behind.
551+func TestLogSwitchFailure(t *testing.T) {
552+ for _, f := range switchFaults {
553+ 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)
574+ }
575+ }
576+
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
587+ }
588+ mustApply(t, db, i)
589+ i++
590+ return false
591+ })
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+ })
598+ }
599+}
600+
601+// TestLogSwitchFailsDatabase covers the failures that must fail the database:
602+// the sync of the current log, and a stray log-S that cannot be removed.
603+func TestLogSwitchFailsDatabase(t *testing.T) {
604+ cases := []struct {
605+ name string
606+ inject func(fx *faultfs.FS)
607+ }{
608+ {"syncall of the current log", func(fx *faultfs.FS) { fx.FailSync(2, nil) }},
609+ {"syncdir then remove", func(fx *faultfs.FS) {
610+ fx.FailOp(faultfs.OpSyncDir, 1, nil)
611+ fx.FailOp(faultfs.OpRemove, 1, nil)
612+ }},
613+ }
614+ for _, c := range cases {
615+ 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
630+ }
631+ })
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+ })
646+ }
647+}
648+
649+// TestLogSwitchSyncFailureDropsUnsyncedData is the fsyncgate case with
650+// NoSync: the SyncAll before the switch is the first sync of the log. When it
651+// fails and the kernel drops the pages, the batches acknowledged without a
652+// sync are gone; the database must fail rather than continue, and what is
653+// left must open as a prefix.
654+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
671+ }
672+ })
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+}
681+
682+// TestCloseWaitsForCompaction checks that Close does not cancel a running
683+// compaction, waits for it, and leaves a directory that opens.
684+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)
719+ }
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)
728+}
729+
730+// TestCloseDuringCompaction (AC9) closes the database while writers run and
731+// compactions are frequent: Close returns, and the next Open finds every
732+// acknowledged batch.
733+func TestCloseDuringCompaction(t *testing.T) {
734+ var switches atomic.Int64
735+ for i := range short(12, 4) {
736+ t.Run(fmt.Sprint(i), func(t *testing.T) {
737+ fx := faultfs.New()
738+ fx.SetSyncDelay(50 * time.Microsecond)
739+ ctx, logs := debugCtx()
740+ db := mustOpenCtx(t, ctx, fx, Options{CompactAfter: 2 * recSize})
741+ const writers = 16
742+ acked := make([]int, writers) // highest acknowledged n+1
743+ var wg sync.WaitGroup
744+ for g := range writers {
745+ wg.Go(func() {
746+ for n := 0; ; n++ {
747+ var b Batch
748+ b.Put(fmt.Appendf(nil, "g%02d", g), seqValue(n))
749+ b.Put(fmt.Appendf(nil, "p%02d/%06d", g, n), nil)
750+ _, err := db.Apply(bg, &b)
751+ if errors.Is(err, ErrClosed) {
752+ return
753+ }
754+ if err != nil {
755+ t.Errorf("writer %d: %v", g, err)
756+ return
757+ }
758+ acked[g] = n + 1
759+ }
760+ })
761+ }
762+ time.Sleep(time.Duration(5+3*i) * time.Millisecond)
763+ closed := make(chan error, 1)
764+ go func() { closed <- db.Close(bg) }()
765+ select {
766+ case err := <-closed:
767+ if err != nil {
768+ t.Errorf("Close: %v", err)
769+ }
770+ case <-time.After(20 * time.Second):
771+ t.Fatal("Close did not return")
772+ }
773+ wg.Wait()
774+ switches.Add(int64(logged(logs, "log switch")))
775+
776+ fx.SetSyncDelay(0)
777+ s := mustOpen(t, fx, Options{}).Snapshot(bg)
778+ for g, n := range acked {
779+ it, ok := s.Get(bg, fmt.Appendf(nil, "g%02d", g))
780+ if n == 0 {
781+ continue
782+ }
783+ // The batch after the last acknowledged one may have been
784+ // committed by the final group.
785+ if !ok || string(it.Value) != string(seqValue(n-1)) && string(it.Value) != string(seqValue(n)) {
786+ t.Errorf("writer %d: acked %d batches, recovered %q (%v)", g, n, it.Value, ok)
787+ }
788+ if _, ok := s.Get(bg, fmt.Appendf(nil, "p%02d/%06d", g, n-1)); !ok {
789+ t.Errorf("writer %d: acked batch %d lost", g, n-1)
790+ }
791+ }
792+ })
793+ }
794+ if switches.Load() == 0 {
795+ t.Error("no log switch happened: the test did not exercise compaction")
796+ }
797+}
798+
799+// compactSteps runs a workload of three rounds of six batches (each round
800+// ends with a compaction) on a fresh file system and cuts the power before
801+// the crashAt-th file operation after Open (0 and out of range: after the
802+// workload). It then reopens the database and reports a violation of AC2.
803+// The workload is sequential and waits for each compaction, so the sequence
804+// of operations is the same in every run.
805+func compactSteps(t *testing.T, plo faultfs.PowerLossOptions, crashAt int) (steps []string, err error) {
806+ ff := faultfs.New()
807+ h := &hookFS{FS: ff, ff: ff, serial: true, crashAt: crashAt, plo: plo}
808+ db, oerr := open(testCtx(t), h, testDir, compactOpts())
809+ if oerr != nil {
810+ return nil, fmt.Errorf("open: %w", oerr)
811+ }
812+ h.arm()
813+ lastAcked := 0
814+workload:
815+ for round := 1; round <= 3; round++ {
816+ for k := 1; k <= 6; k++ {
817+ i := (round-1)*6 + k
818+ if applySeq(db, i) != nil {
819+ break workload
820+ }
821+ lastAcked = i
822+ }
823+ // Wait for the compaction of this round.
824+ deadline := time.Now().Add(10 * time.Second)
825+ for !h.isDead() && (!settled(t, ff, uint64(6*round)) || db.compacting.Load()) {
826+ if time.Now().After(deadline) {
827+ return h.steps(), errors.New("timeout waiting for the compaction")
828+ }
829+ time.Sleep(100 * time.Microsecond)
830+ }
831+ }
832+ steps = h.steps()
833+ _ = db.Close(bg)
834+ h.cut(plo) // no-op if the power was already cut
835+
836+ for range 2 {
837+ db2, err := open(testCtx(t), ff, testDir, Options{})
838+ if err != nil {
839+ return steps, fmt.Errorf("reopen: %w", err)
840+ }
841+ n, err := checkSequential(db2.Snapshot(bg), lastAcked)
842+ if err == nil {
843+ // The recovered log accepts new batches.
844+ err = applySeq(db2, n+1)
845+ lastAcked = n + 1
846+ }
847+ if cerr := db2.Close(bg); err == nil {
848+ err = cerr
849+ }
850+ if err != nil {
851+ return steps, err
852+ }
853+ }
854+ return steps, nil
855+}
856+
857+// TestCompactPowerLossAtEachStep cuts the power before each file operation
858+// of three compactions in turn, in every power-loss mode, and checks AC2:
859+// the reopened database holds every acknowledged batch and is a prefix.
860+func TestCompactPowerLossAtEachStep(t *testing.T) {
861+ modes := []struct {
862+ name string
863+ plo faultfs.PowerLossOptions
864+ }{
865+ {"drop", faultfs.PowerLossOptions{}},
866+ {"zerofill", faultfs.PowerLossOptions{ZeroFill: true}},
867+ {"keeppages", faultfs.PowerLossOptions{KeepPages: true, PageSize: 64}},
868+ {"keeppages+zerofill", faultfs.PowerLossOptions{KeepPages: true, ZeroFill: true, PageSize: 64}},
869+ }
870+ base := powerLossBase(t)
871+ for _, m := range modes {
872+ t.Run(m.name, func(t *testing.T) {
873+ plo := m.plo
874+ plo.Rand = rand.New(rand.NewPCG(base, 7))
875+ steps, err := compactSteps(t, plo, 0)
876+ if err != nil {
877+ t.Fatalf("without a cut: %v", err)
878+ }
879+ t.Logf("%d file operations in the workload", len(steps))
880+ for step := 1; step <= len(steps); step++ {
881+ plo.Rand = rand.New(rand.NewPCG(base+uint64(step), 7))
882+ if _, err := compactSteps(t, plo, step); err != nil {
883+ t.Fatalf("power cut before step %d (%s), seed %d (ERMITE_KV_SEED=%d reproduces it): %v",
884+ step, steps[step-1], base, base, err)
885+ }
886+ }
887+ })
888+ }
889+}
890+
891+// TestCompactPowerLoss cuts the power at a random time during a workload
892+// that compacts all the time, then checks AC2. With NoSync the acknowledged
893+// batches may be lost, but the directory must open and be a prefix; only the
894+// modes that cannot make Open fail with ErrCorrupt are used (see
895+// Options.NoSync).
896+func TestCompactPowerLoss(t *testing.T) {
897+ modes := []struct {
898+ name string
899+ nosync bool
900+ plo faultfs.PowerLossOptions
901+ }{
902+ {"drop", false, faultfs.PowerLossOptions{}},
903+ {"zerofill", false, faultfs.PowerLossOptions{ZeroFill: true}},
904+ {"keeppages", false, faultfs.PowerLossOptions{KeepPages: true}},
905+ {"keeppages+zerofill", false, faultfs.PowerLossOptions{KeepPages: true, ZeroFill: true}},
906+ {"nosync drop", true, faultfs.PowerLossOptions{}},
907+ {"nosync zerofill", true, faultfs.PowerLossOptions{ZeroFill: true}},
908+ }
909+ base := powerLossBase(t)
910+ iterations := short(100, 10)
911+ for _, m := range modes {
912+ t.Run(m.name, func(t *testing.T) {
913+ for i := range iterations {
914+ seed := base + uint64(i)
915+ if err := runCompactPowerLoss(t, seed, m.plo, m.nosync); err != nil {
916+ t.Fatalf("seed %d (ERMITE_KV_SEED=%d reproduces it): %v", seed, seed, err)
917+ }
918+ }
919+ })
920+ }
921+}
922+
923+func runCompactPowerLoss(t *testing.T, seed uint64, plo faultfs.PowerLossOptions, nosync bool) error {
924+ rng := rand.New(rand.NewPCG(seed, 0x9e3779b97f4a7c15))
925+ ff := faultfs.New()
926+ h := &hookFS{FS: ff, ff: ff, serial: true}
927+ opts := Options{NoSync: nosync, CompactAfter: int64(2+rng.IntN(10)) * recSize}
928+ db, err := open(testCtx(t), h, testDir, opts)
929+ if err != nil {
930+ return fmt.Errorf("open: %w", err)
931+ }
932+ ff.SetSyncDelay(time.Duration(rng.IntN(100)) * time.Microsecond)
933+ var lastAcked atomic.Int64
934+ done := make(chan struct{})
935+ go func() {
936+ defer close(done)
937+ for i := 1; applySeq(db, i) == nil; i++ {
938+ lastAcked.Store(int64(i))
939+ }
940+ }()
941+ time.Sleep(time.Duration(rng.IntN(6000)) * time.Microsecond)
942+ plo.Rand = rand.New(rand.NewPCG(seed, 7))
943+ if plo.KeepPages {
944+ plo.PageSize = 64
945+ }
946+ h.cut(plo)
947+ <-done
948+ _ = db.Close(bg)
949+ ff.SetSyncDelay(0)
950+
951+ floor := int(lastAcked.Load())
952+ if nosync {
953+ floor = 0
954+ }
955+ for range 2 {
956+ db, err = open(testCtx(t), ff, testDir, Options{})
957+ if err != nil {
958+ return fmt.Errorf("reopen: %w", err)
959+ }
960+ n, err := checkSequential(db.Snapshot(bg), floor)
961+ if err == nil {
962+ err = applySeq(db, n+1) // appends to the right log
963+ floor = n + 1
964+ }
965+ if cerr := db.Close(bg); err == nil {
966+ err = cerr
967+ }
968+ if err != nil {
969+ return err
970+ }
971+ }
972+ return nil
973+}
M db.go +6 -2
@@ -83,10 +83,14 @@ type DB struct {
8383
8484 lock io.Closer
8585
86+ compacting atomic.Bool // a compactor goroutine is running
87+ compactors sync.WaitGroup // started by the writer only; Close waits after the writer exits
88+
8689 // Owned by the writer goroutine (see writer.go).
8790 file fsys.File
8891 logBase uint64 // seq the current log starts after
8992 logSize int64 // bytes in the current log, header included
93+ retryAt int64 // do not try a log switch again before logSize reaches this
9094 buf []byte // group encoding buffer
9195 group []*request
9296 }
@@ -167,8 +171,8 @@ func (db *DB) Close(ctx context.Context) error {
167171 var err error
168172 db.closeOnce.Do(func() {
169173 close(db.closing)
170- <-db.done // the writer finishes its group and exits
171- // P1.D: wait here for an in-flight compaction.
174+ <-db.done // the writer finishes its group and exits
175+ db.compactors.Wait() // an in-flight compaction is not cancelled
172176 select {
173177 case <-db.failed:
174178 err = db.failErr
@@ -83,10 +83,14 @@ type DB struct {
83 83
84 lock io.Closer84 lock io.Closer
85 85
86+ compacting atomic.Bool // a compactor goroutine is running
87+ compactors sync.WaitGroup // started by the writer only; Close waits after the writer exits
88+
86 // Owned by the writer goroutine (see writer.go).89 // Owned by the writer goroutine (see writer.go).
87 file fsys.File90 file fsys.File
88 logBase uint64 // seq the current log starts after91 logBase uint64 // seq the current log starts after
89 logSize int64 // bytes in the current log, header included92 logSize int64 // bytes in the current log, header included
93+ retryAt int64 // do not try a log switch again before logSize reaches this
90 buf []byte // group encoding buffer94 buf []byte // group encoding buffer
91 group []*request95 group []*request
92 }96 }
@@ -167,8 +171,8 @@ func (db *DB) Close(ctx context.Context) error {
167 var err error171 var err error
168 db.closeOnce.Do(func() {172 db.closeOnce.Do(func() {
169 close(db.closing)173 close(db.closing)
170- <-db.done // the writer finishes its group and exits174+ <-db.done // the writer finishes its group and exits
171- // P1.D: wait here for an in-flight compaction.175+ db.compactors.Wait() // an in-flight compaction is not cancelled
172 select {176 select {
173 case <-db.failed:177 case <-db.failed:
174 err = db.failErr178 err = db.failErr
M writer.go +3 -2
@@ -72,8 +72,9 @@ func (db *DB) writer() {
7272 if !db.commit() {
7373 return
7474 }
75- // P1.D hook: after replying, trigger compaction here once
76- // db.logSize exceeds db.opts.CompactAfter (if it is positive).
75+ if !db.maybeCompact() {
76+ return
77+ }
7778 }
7879 }
7980
@@ -72,8 +72,9 @@ func (db *DB) writer() {
72 if !db.commit() {72 if !db.commit() {
73 return73 return
74 }74 }
75- // P1.D hook: after replying, trigger compaction here once75+ if !db.maybeCompact() {
76- // db.logSize exceeds db.opts.CompactAfter (if it is positive).76+ return
77+ }
77 }78 }
78 }79 }
79 80

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