1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
6 "errors"
7 "fmt"
8 "io"
9 "os"
10 "path/filepath"
11 "slices"
12 "strconv"
13 "strings"
14
15 "github.com/bnema/zerowrap"
16
17 "bnema.dev/ermite-kv/internal/fsys"
18)
19
20const (
21 lockName = "LOCK"
22 snapshotPrefix = "snapshot-"
23 logPrefix = "log-"
24 tmpSuffix = ".tmp"
25)
26
27func snapshotName(seq uint64) string { return fmt.Sprintf("%s%016x", snapshotPrefix, seq) }
28func logName(base uint64) string { return fmt.Sprintf("%s%016x", logPrefix, base) }
29
30// parseSeqName parses "<prefix><seq:016x>"; ok is false for any other name.
31func parseSeqName(name, prefix string) (seq uint64, ok bool) {
32 if !strings.HasPrefix(name, prefix) {
33 return 0, false
34 }
35 seq, err := strconv.ParseUint(name[len(prefix):], 16, 64)
36 if err != nil || fmt.Sprintf("%s%016x", prefix, seq) != name {
37 return 0, false
38 }
39 return seq, true
40}
41
42// recovered is the state handed to the DB by recoverDir.
43type recovered struct {
44 snapSeq uint64 // seq of the loaded snapshot, 0 if none
45 lastSeq uint64 // seq of the last applied record (snapSeq if none)
46
47 // log is the last log file, opened read-write with appending writes. It
48 // is at logBase and holds logSize bytes (header included) after recovery.
49 // The caller owns it.
50 log fsys.File
51 logBase uint64
52 logSize int64
53
54 // lock holds the directory lock; closing it releases the lock. The
55 // caller owns it.
56 lock io.Closer
57}
58
59// recoverDir opens the database directory and rebuilds its state: recovery
60// rule 1 (directory, lock, *.tmp files) and rules 2–7. It does not depend on
61// the in-memory tree.
62//
63// load is called once per snapshot entry in ascending key order, only after
64// the whole snapshot file has been validated. apply is called for each batch
65// newer than the snapshot, in seq order (a record that straddles the snapshot
66// seq contributes only its newer batches). The ops passed to apply are freshly
67// allocated and the callee may retain them but must not modify them. load
68// receives slices of one shared read-only buffer: it must copy what it keeps.
69// Either callback may be nil. A zero logger discards logs.
70//
71// On error nothing stays open or locked. ErrLocked and ErrCorrupt are
72// returned wrapped.
73func recoverDir(fx fsys.FS, dir string, logger zerowrap.Logger,
74 load func(key, value []byte, version uint64),
75 apply func(seq uint64, ops []op),
76) (rec recovered, err error) {
77 dir = filepath.Clean(dir)
78 if err := fx.MkdirAll(dir, 0o700); err != nil {
79 return rec, err
80 }
81 if err := fx.SyncDir(dir); err != nil {
82 return rec, err
83 }
84 if parent := filepath.Dir(dir); parent != dir {
85 if err := fx.SyncDir(parent); err != nil {
86 return rec, err
87 }
88 }
89 lock, err := fx.Lock(filepath.Join(dir, lockName))
90 if errors.Is(err, fsys.ErrLocked) {
91 return rec, fmt.Errorf("%w: %s", ErrLocked, dir)
92 }
93 if err != nil {
94 return rec, err
95 }
96 rec.lock = lock
97 defer func() {
98 if err != nil {
99 if rec.log != nil {
100 _ = rec.log.Close()
101 }
102 _ = lock.Close()
103 rec = recovered{}
104 }
105 }()
106
107 names, err := fx.ReadDir(dir)
108 if err != nil {
109 return rec, err
110 }
111 var snaps, logs []uint64
112 for _, name := range names {
113 if strings.HasSuffix(name, tmpSuffix) {
114 if err := fx.Remove(filepath.Join(dir, name)); err != nil && !errors.Is(err, os.ErrNotExist) {
115 return rec, err
116 }
117 } else if seq, ok := parseSeqName(name, snapshotPrefix); ok {
118 snaps = append(snaps, seq)
119 } else if base, ok := parseSeqName(name, logPrefix); ok {
120 logs = append(logs, base)
121 }
122 }
123 if err := fx.SyncDir(dir); err != nil {
124 return rec, err
125 }
126
127 // Rule 2: load the highest-seq snapshot that validates.
128 slices.Sort(snaps)
129 for i := len(snaps) - 1; i >= 0; i-- {
130 path := filepath.Join(dir, snapshotName(snaps[i]))
131 data, err := readFile(fx, path)
132 if err != nil {
133 return rec, err
134 }
135 view, err := openSnapshot(data)
136 if err == nil && view.seq != snaps[i] {
137 err = fmt.Errorf("%w: snapshot seq %d does not match its name", ErrCorrupt, view.seq)
138 }
139 if err != nil {
140 logger.Warn().Err(err).Str("file", path).Msg("ignoring invalid snapshot")
141 continue
142 }
143 if load != nil {
144 view.each(load)
145 }
146 rec.snapSeq = view.seq
147 break
148 }
149 rec.lastSeq = rec.snapSeq
150
151 // Rules 3–6: replay the logs in base order.
152 slices.Sort(logs)
153 for i, base := range logs {
154 last := i == len(logs)-1
155 path := filepath.Join(dir, logName(base))
156 if !last {
157 data, err := readFile(fx, path)
158 if err != nil {
159 return rec, err
160 }
161 if _, err := replayLog(nil, data, path, base, &rec, logger, apply); err != nil {
162 return rec, err
163 }
164 continue
165 }
166 f, err := fx.OpenFile(path, os.O_RDWR, 0)
167 if err != nil {
168 return rec, err
169 }
170 data, err := readAll(f)
171 if err == nil {
172 rec.logSize, err = replayLog(f, data, path, base, &rec, logger, apply)
173 }
174 if err != nil {
175 _ = f.Close()
176 return rec, err
177 }
178 rec.log, rec.logBase = f, base
179 }
180
181 // Rule 7: no log file at all.
182 if rec.log == nil {
183 name := logName(rec.snapSeq)
184 header := appendLogHeader(nil, rec.snapSeq)
185 f, err := createFile(fx, dir, name, func(w io.Writer) error {
186 _, err := w.Write(header)
187 return err
188 })
189 if err != nil {
190 return rec, err
191 }
192 rec.log, rec.logBase, rec.logSize = f, rec.snapSeq, logHeaderSize
193 }
194 return rec, nil
195}
196
197// replayLog applies the records of the log file contents data (rules 3–6)
198// and updates rec.lastSeq. f is nil for a log that is not the last one; for
199// the last log it is used to truncate a torn tail. It returns the size of the
200// log after recovery.
201func replayLog(f fsys.File, data []byte, path string, base uint64, rec *recovered,
202 logger zerowrap.Logger, apply func(seq uint64, ops []op),
203) (int64, error) {
204 last := f != nil
205 corrupt := func(off int, format string, args ...any) error {
206 return fmt.Errorf("%w: %s at offset %d: %s", ErrCorrupt, path, off, fmt.Sprintf(format, args...))
207 }
208 hdrBase, err := decodeLogHeader(data)
209 if err != nil {
210 return 0, fmt.Errorf("%s: %w", path, err)
211 }
212 if hdrBase != base {
213 return 0, corrupt(0, "header base %d does not match the file name", hdrBase)
214 }
215 if base > rec.lastSeq {
216 return 0, corrupt(0, "base %d is above the last recovered seq %d", base, rec.lastSeq)
217 }
218
219 expected := base + 1 // seq of the first batch of the next record
220 off := logHeaderSize
221 for off < len(data) {
222 seq, batches, n, err := decodeRecord(data[off:])
223 if err != nil {
224 if !last {
225 return 0, corrupt(off, "%v", err)
226 }
227 // Rule 6: a readable record after the damaged one proves that
228 // the damaged one was synced: acknowledged data is damaged.
229 if total, ok := peekRecord(data[off:]); ok && off+total < len(data) {
230 if _, _, _, nerr := decodeRecord(data[off+total:]); nerr == nil {
231 return 0, corrupt(off, "%v, followed by a valid record", err)
232 }
233 }
234 if err := f.Truncate(int64(off)); err != nil {
235 return 0, err
236 }
237 if err := f.SyncAll(); err != nil {
238 return 0, err
239 }
240 logger.Warn().Str("file", path).Int("offset", off).Int("bytes", len(data)-off).
241 Msg("dropped torn log tail")
242 data = data[:off]
243 break
244 }
245 if seq != expected {
246 return 0, corrupt(off, "record seq %d, want %d", seq, expected)
247 }
248 for i, ops := range batches {
249 s := seq + uint64(i)
250 if s <= rec.snapSeq {
251 continue
252 }
253 if s != rec.lastSeq+1 {
254 return 0, corrupt(off, "batch seq %d, want %d", s, rec.lastSeq+1)
255 }
256 if apply != nil {
257 apply(s, ownOps(ops))
258 }
259 rec.lastSeq = s
260 }
261 expected += uint64(len(batches))
262 off += n
263 }
264 if last && expected-1 < rec.lastSeq {
265 // Cannot happen unless records acknowledged before the snapshot
266 // are missing: appending would create a seq gap.
267 return 0, corrupt(len(data), "log ends at seq %d, want %d", expected-1, rec.lastSeq)
268 }
269 return int64(len(data)), nil
270}
271
272// readFile returns the whole content of the file at path.
273func readFile(fx fsys.FS, path string) ([]byte, error) {
274 f, err := fx.OpenFile(path, os.O_RDONLY, 0)
275 if err != nil {
276 return nil, err
277 }
278 data, err := readAll(f)
279 if cerr := f.Close(); err == nil {
280 err = cerr
281 }
282 return data, err
283}
284
285// readAll returns the whole content of f.
286func readAll(f fsys.File) ([]byte, error) {
287 size, err := f.Size()
288 if err != nil {
289 return nil, err
290 }
291 data := make([]byte, size)
292 n, err := f.ReadAt(data, 0)
293 if n == len(data) {
294 return data, nil // io.EOF is allowed when the read ends at EOF
295 }
296 if err == nil || errors.Is(err, io.EOF) {
297 err = io.ErrUnexpectedEOF
298 }
299 return nil, err
300}
301
302// createFile durably creates dir/name with the content produced by write:
303// name.tmp is written and fsynced, renamed to name, then dir is synced. The
304// returned file is open for appending (and reading) at its end. On error the
305// temporary file is removed on a best-effort basis.
306func createFile(fx fsys.FS, dir, name string, write func(w io.Writer) error) (fsys.File, error) {
307 tmp := filepath.Join(dir, name+tmpSuffix)
308 f, err := fx.OpenFile(tmp, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0o600)
309 if err != nil {
310 return nil, err
311 }
312 fail := func(err error) (fsys.File, error) {
313 _ = f.Close()
314 _ = fx.Remove(tmp)
315 return nil, err
316 }
317 if err := write(f); err != nil {
318 return fail(err)
319 }
320 if err := f.SyncAll(); err != nil {
321 return fail(err)
322 }
323 if err := fx.Rename(tmp, filepath.Join(dir, name)); err != nil {
324 return fail(err)
325 }
326 if err := fx.SyncDir(dir); err != nil {
327 _ = f.Close()
328 return nil, err
329 }
330 return f, nil
331}