1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
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.
18func (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.
45func (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.
87func (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.
101func (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.
117func (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.
141func (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}