detached b62a1d4

feat(recover): add directory recovery over fsys

commit
b62a1d44e120122213c217f710dac9cf6b7e73f6
author
brice · · 2026-10-10 01:52 UTC
committer
brice · · 2026-10-10 01:57 UTC
parent
15cd1c5
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +895 -0
Unified
UnifiedSplit
A recover.go +330 -0
@@ -0,0 +1,330 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "errors"
7+ "fmt"
8+ "io"
9+ "log/slog"
10+ "os"
11+ "path/filepath"
12+ "slices"
13+ "strconv"
14+ "strings"
15+
16+ "bnema.dev/ermite-kv/internal/fsys"
17+)
18+
19+const (
20+ lockName = "LOCK"
21+ snapshotPrefix = "snapshot-"
22+ logPrefix = "log-"
23+ tmpSuffix = ".tmp"
24+)
25+
26+func snapshotName(seq uint64) string { return fmt.Sprintf("%s%016x", snapshotPrefix, seq) }
27+func logName(base uint64) string { return fmt.Sprintf("%s%016x", logPrefix, base) }
28+
29+// parseSeqName parses "<prefix><seq:016x>"; ok is false for any other name.
30+func parseSeqName(name, prefix string) (seq uint64, ok bool) {
31+ if !strings.HasPrefix(name, prefix) {
32+ return 0, false
33+ }
34+ seq, err := strconv.ParseUint(name[len(prefix):], 16, 64)
35+ if err != nil || fmt.Sprintf("%s%016x", prefix, seq) != name {
36+ return 0, false
37+ }
38+ return seq, true
39+}
40+
41+// recovered is the state handed to the DB by recoverDir.
42+type recovered struct {
43+ snapSeq uint64 // seq of the loaded snapshot, 0 if none
44+ lastSeq uint64 // seq of the last applied record (snapSeq if none)
45+
46+ // log is the last log file, opened read-write with appending writes. It
47+ // is at logBase and holds logSize bytes (header included) after recovery.
48+ // The caller owns it.
49+ log fsys.File
50+ logBase uint64
51+ logSize int64
52+
53+ // lock holds the directory lock; closing it releases the lock. The
54+ // caller owns it.
55+ lock io.Closer
56+}
57+
58+// recoverDir opens the database directory and rebuilds its state: recovery
59+// rule 1 (directory, lock, *.tmp files) and rules 2–7. It does not depend on
60+// the in-memory tree.
61+//
62+// load is called once per snapshot entry in ascending key order, only after
63+// the whole snapshot file has been validated. apply is called for each log
64+// record newer than the snapshot, in seq order. Both receive freshly
65+// allocated slices that the callee may retain but must not modify; recovery
66+// never reuses them (snapshot entries share one read-only buffer, log ops are
67+// copied per record). Either callback may be nil. A nil logger discards logs.
68+//
69+// On error nothing stays open or locked. ErrLocked and ErrCorrupt are
70+// returned wrapped.
71+func recoverDir(fx fsys.FS, dir string, logger *slog.Logger,
72+ load func(key, value []byte, version uint64),
73+ apply func(seq uint64, ops []op),
74+) (rec recovered, err error) {
75+ if logger == nil {
76+ logger = slog.New(slog.DiscardHandler)
77+ }
78+ dir = filepath.Clean(dir)
79+ if err := fx.MkdirAll(dir, 0o700); err != nil {
80+ return rec, err
81+ }
82+ if err := fx.SyncDir(dir); err != nil {
83+ return rec, err
84+ }
85+ if parent := filepath.Dir(dir); parent != dir {
86+ if err := fx.SyncDir(parent); err != nil {
87+ return rec, err
88+ }
89+ }
90+ lock, err := fx.Lock(filepath.Join(dir, lockName))
91+ if errors.Is(err, fsys.ErrLocked) {
92+ return rec, fmt.Errorf("%w: %s", ErrLocked, dir)
93+ }
94+ if err != nil {
95+ return rec, err
96+ }
97+ rec.lock = lock
98+ defer func() {
99+ if err != nil {
100+ if rec.log != nil {
101+ _ = rec.log.Close()
102+ }
103+ _ = lock.Close()
104+ rec = recovered{}
105+ }
106+ }()
107+
108+ names, err := fx.ReadDir(dir)
109+ if err != nil {
110+ return rec, err
111+ }
112+ var snaps, logs []uint64
113+ for _, name := range names {
114+ if strings.HasSuffix(name, tmpSuffix) {
115+ if err := fx.Remove(filepath.Join(dir, name)); err != nil && !errors.Is(err, os.ErrNotExist) {
116+ return rec, err
117+ }
118+ } else if seq, ok := parseSeqName(name, snapshotPrefix); ok {
119+ snaps = append(snaps, seq)
120+ } else if base, ok := parseSeqName(name, logPrefix); ok {
121+ logs = append(logs, base)
122+ }
123+ }
124+ if err := fx.SyncDir(dir); err != nil {
125+ return rec, err
126+ }
127+
128+ // Rule 2: load the highest-seq snapshot that validates.
129+ slices.Sort(snaps)
130+ for i := len(snaps) - 1; i >= 0; i-- {
131+ path := filepath.Join(dir, snapshotName(snaps[i]))
132+ data, err := readFile(fx, path)
133+ if err != nil {
134+ return rec, err
135+ }
136+ view, err := openSnapshot(data)
137+ if err == nil && view.seq != snaps[i] {
138+ err = fmt.Errorf("%w: snapshot seq %d does not match its name", ErrCorrupt, view.seq)
139+ }
140+ if err != nil {
141+ logger.Warn("kv: ignoring invalid snapshot", "file", path, "err", err)
142+ continue
143+ }
144+ if load != nil {
145+ view.each(load)
146+ }
147+ rec.snapSeq = view.seq
148+ break
149+ }
150+ rec.lastSeq = rec.snapSeq
151+
152+ // Rules 3–6: replay the logs in base order.
153+ slices.Sort(logs)
154+ for i, base := range logs {
155+ last := i == len(logs)-1
156+ path := filepath.Join(dir, logName(base))
157+ if !last {
158+ data, err := readFile(fx, path)
159+ if err != nil {
160+ return rec, err
161+ }
162+ if _, err := replayLog(nil, data, path, base, &rec, logger, apply); err != nil {
163+ return rec, err
164+ }
165+ continue
166+ }
167+ f, err := fx.OpenFile(path, os.O_RDWR, 0)
168+ if err != nil {
169+ return rec, err
170+ }
171+ data, err := readAll(f)
172+ if err == nil {
173+ rec.logSize, err = replayLog(f, data, path, base, &rec, logger, apply)
174+ }
175+ if err != nil {
176+ _ = f.Close()
177+ return rec, err
178+ }
179+ rec.log, rec.logBase = f, base
180+ }
181+
182+ // Rule 7: no log file at all.
183+ if rec.log == nil {
184+ name := logName(rec.snapSeq)
185+ header := appendLogHeader(nil, rec.snapSeq)
186+ f, err := createFile(fx, dir, name, func(w io.Writer) error {
187+ _, err := w.Write(header)
188+ return err
189+ })
190+ if err != nil {
191+ return rec, err
192+ }
193+ rec.log, rec.logBase, rec.logSize = f, rec.snapSeq, logHeaderSize
194+ }
195+ return rec, nil
196+}
197+
198+// replayLog applies the records of the log file contents data (rules 3–6)
199+// and updates rec.lastSeq. f is nil for a log that is not the last one; for
200+// the last log it is used to truncate a torn tail. It returns the size of the
201+// log after recovery.
202+func replayLog(f fsys.File, data []byte, path string, base uint64, rec *recovered,
203+ logger *slog.Logger, apply func(seq uint64, ops []op),
204+) (int64, error) {
205+ last := f != nil
206+ corrupt := func(off int, format string, args ...any) error {
207+ return fmt.Errorf("%w: %s at offset %d: %s", ErrCorrupt, path, off, fmt.Sprintf(format, args...))
208+ }
209+ hdrBase, err := decodeLogHeader(data)
210+ if err != nil {
211+ return 0, fmt.Errorf("%s: %w", path, err)
212+ }
213+ if hdrBase != base {
214+ return 0, corrupt(0, "header base %d does not match the file name", hdrBase)
215+ }
216+ if base > rec.lastSeq {
217+ return 0, corrupt(0, "base %d is above the last recovered seq %d", base, rec.lastSeq)
218+ }
219+
220+ var ops []op
221+ expected := base + 1 // seq of the next record of this file
222+ off := logHeaderSize
223+ for off < len(data) {
224+ seq, decoded, n, err := decodeRecord(data[off:], ops)
225+ if err != nil {
226+ if !last {
227+ return 0, corrupt(off, "%v", err)
228+ }
229+ // Rule 6: a readable next record means acknowledged data
230+ // lies beyond the damage.
231+ if total, ok := peekRecord(data[off:]); ok && off+total < len(data) {
232+ nseq, _, _, nerr := decodeRecord(data[off+total:], nil)
233+ if nerr == nil && nseq == expected+1 {
234+ return 0, corrupt(off, "%v, followed by a valid record", err)
235+ }
236+ }
237+ if err := f.Truncate(int64(off)); err != nil {
238+ return 0, err
239+ }
240+ if err := f.SyncAll(); err != nil {
241+ return 0, err
242+ }
243+ logger.Warn("kv: dropped torn log tail", "file", path, "offset", off, "bytes", len(data)-off)
244+ data = data[:off]
245+ break
246+ }
247+ ops = decoded[:0]
248+ if seq != expected {
249+ return 0, corrupt(off, "record seq %d, want %d", seq, expected)
250+ }
251+ if seq > rec.snapSeq {
252+ if seq != rec.lastSeq+1 {
253+ return 0, corrupt(off, "record seq %d, want %d", seq, rec.lastSeq+1)
254+ }
255+ if apply != nil {
256+ apply(seq, ownOps(decoded))
257+ }
258+ rec.lastSeq = seq
259+ }
260+ expected++
261+ off += n
262+ }
263+ if last && expected-1 < rec.lastSeq {
264+ // Cannot happen unless records acknowledged before the snapshot
265+ // are missing: appending would create a seq gap.
266+ return 0, corrupt(len(data), "log ends at seq %d, want %d", expected-1, rec.lastSeq)
267+ }
268+ return int64(len(data)), nil
269+}
270+
271+// readFile returns the whole content of the file at path.
272+func readFile(fx fsys.FS, path string) ([]byte, error) {
273+ f, err := fx.OpenFile(path, os.O_RDONLY, 0)
274+ if err != nil {
275+ return nil, err
276+ }
277+ data, err := readAll(f)
278+ if cerr := f.Close(); err == nil {
279+ err = cerr
280+ }
281+ return data, err
282+}
283+
284+// readAll returns the whole content of f.
285+func readAll(f fsys.File) ([]byte, error) {
286+ size, err := f.Size()
287+ if err != nil {
288+ return nil, err
289+ }
290+ data := make([]byte, size)
291+ n, err := f.ReadAt(data, 0)
292+ if n == len(data) {
293+ return data, nil // io.EOF is allowed when the read ends at EOF
294+ }
295+ if err == nil || errors.Is(err, io.EOF) {
296+ err = io.ErrUnexpectedEOF
297+ }
298+ return nil, err
299+}
300+
301+// createFile durably creates dir/name with the content produced by write:
302+// name.tmp is written and fsynced, renamed to name, then dir is synced. The
303+// returned file is open for appending (and reading) at its end. On error the
304+// temporary file is removed on a best-effort basis.
305+func createFile(fx fsys.FS, dir, name string, write func(w io.Writer) error) (fsys.File, error) {
306+ tmp := filepath.Join(dir, name+tmpSuffix)
307+ f, err := fx.OpenFile(tmp, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0o600)
308+ if err != nil {
309+ return nil, err
310+ }
311+ fail := func(err error) (fsys.File, error) {
312+ _ = f.Close()
313+ _ = fx.Remove(tmp)
314+ return nil, err
315+ }
316+ if err := write(f); err != nil {
317+ return fail(err)
318+ }
319+ if err := f.SyncAll(); err != nil {
320+ return fail(err)
321+ }
322+ if err := fx.Rename(tmp, filepath.Join(dir, name)); err != nil {
323+ return fail(err)
324+ }
325+ if err := fx.SyncDir(dir); err != nil {
326+ _ = f.Close()
327+ return nil, err
328+ }
329+ return f, nil
330+}
@@ -0,0 +1,330 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "errors"
7+ "fmt"
8+ "io"
9+ "log/slog"
10+ "os"
11+ "path/filepath"
12+ "slices"
13+ "strconv"
14+ "strings"
15+
16+ "bnema.dev/ermite-kv/internal/fsys"
17+)
18+
19+const (
20+ lockName = "LOCK"
21+ snapshotPrefix = "snapshot-"
22+ logPrefix = "log-"
23+ tmpSuffix = ".tmp"
24+)
25+
26+func snapshotName(seq uint64) string { return fmt.Sprintf("%s%016x", snapshotPrefix, seq) }
27+func logName(base uint64) string { return fmt.Sprintf("%s%016x", logPrefix, base) }
28+
29+// parseSeqName parses "<prefix><seq:016x>"; ok is false for any other name.
30+func parseSeqName(name, prefix string) (seq uint64, ok bool) {
31+ if !strings.HasPrefix(name, prefix) {
32+ return 0, false
33+ }
34+ seq, err := strconv.ParseUint(name[len(prefix):], 16, 64)
35+ if err != nil || fmt.Sprintf("%s%016x", prefix, seq) != name {
36+ return 0, false
37+ }
38+ return seq, true
39+}
40+
41+// recovered is the state handed to the DB by recoverDir.
42+type recovered struct {
43+ snapSeq uint64 // seq of the loaded snapshot, 0 if none
44+ lastSeq uint64 // seq of the last applied record (snapSeq if none)
45+
46+ // log is the last log file, opened read-write with appending writes. It
47+ // is at logBase and holds logSize bytes (header included) after recovery.
48+ // The caller owns it.
49+ log fsys.File
50+ logBase uint64
51+ logSize int64
52+
53+ // lock holds the directory lock; closing it releases the lock. The
54+ // caller owns it.
55+ lock io.Closer
56+}
57+
58+// recoverDir opens the database directory and rebuilds its state: recovery
59+// rule 1 (directory, lock, *.tmp files) and rules 2–7. It does not depend on
60+// the in-memory tree.
61+//
62+// load is called once per snapshot entry in ascending key order, only after
63+// the whole snapshot file has been validated. apply is called for each log
64+// record newer than the snapshot, in seq order. Both receive freshly
65+// allocated slices that the callee may retain but must not modify; recovery
66+// never reuses them (snapshot entries share one read-only buffer, log ops are
67+// copied per record). Either callback may be nil. A nil logger discards logs.
68+//
69+// On error nothing stays open or locked. ErrLocked and ErrCorrupt are
70+// returned wrapped.
71+func recoverDir(fx fsys.FS, dir string, logger *slog.Logger,
72+ load func(key, value []byte, version uint64),
73+ apply func(seq uint64, ops []op),
74+) (rec recovered, err error) {
75+ if logger == nil {
76+ logger = slog.New(slog.DiscardHandler)
77+ }
78+ dir = filepath.Clean(dir)
79+ if err := fx.MkdirAll(dir, 0o700); err != nil {
80+ return rec, err
81+ }
82+ if err := fx.SyncDir(dir); err != nil {
83+ return rec, err
84+ }
85+ if parent := filepath.Dir(dir); parent != dir {
86+ if err := fx.SyncDir(parent); err != nil {
87+ return rec, err
88+ }
89+ }
90+ lock, err := fx.Lock(filepath.Join(dir, lockName))
91+ if errors.Is(err, fsys.ErrLocked) {
92+ return rec, fmt.Errorf("%w: %s", ErrLocked, dir)
93+ }
94+ if err != nil {
95+ return rec, err
96+ }
97+ rec.lock = lock
98+ defer func() {
99+ if err != nil {
100+ if rec.log != nil {
101+ _ = rec.log.Close()
102+ }
103+ _ = lock.Close()
104+ rec = recovered{}
105+ }
106+ }()
107+
108+ names, err := fx.ReadDir(dir)
109+ if err != nil {
110+ return rec, err
111+ }
112+ var snaps, logs []uint64
113+ for _, name := range names {
114+ if strings.HasSuffix(name, tmpSuffix) {
115+ if err := fx.Remove(filepath.Join(dir, name)); err != nil && !errors.Is(err, os.ErrNotExist) {
116+ return rec, err
117+ }
118+ } else if seq, ok := parseSeqName(name, snapshotPrefix); ok {
119+ snaps = append(snaps, seq)
120+ } else if base, ok := parseSeqName(name, logPrefix); ok {
121+ logs = append(logs, base)
122+ }
123+ }
124+ if err := fx.SyncDir(dir); err != nil {
125+ return rec, err
126+ }
127+
128+ // Rule 2: load the highest-seq snapshot that validates.
129+ slices.Sort(snaps)
130+ for i := len(snaps) - 1; i >= 0; i-- {
131+ path := filepath.Join(dir, snapshotName(snaps[i]))
132+ data, err := readFile(fx, path)
133+ if err != nil {
134+ return rec, err
135+ }
136+ view, err := openSnapshot(data)
137+ if err == nil && view.seq != snaps[i] {
138+ err = fmt.Errorf("%w: snapshot seq %d does not match its name", ErrCorrupt, view.seq)
139+ }
140+ if err != nil {
141+ logger.Warn("kv: ignoring invalid snapshot", "file", path, "err", err)
142+ continue
143+ }
144+ if load != nil {
145+ view.each(load)
146+ }
147+ rec.snapSeq = view.seq
148+ break
149+ }
150+ rec.lastSeq = rec.snapSeq
151+
152+ // Rules 3–6: replay the logs in base order.
153+ slices.Sort(logs)
154+ for i, base := range logs {
155+ last := i == len(logs)-1
156+ path := filepath.Join(dir, logName(base))
157+ if !last {
158+ data, err := readFile(fx, path)
159+ if err != nil {
160+ return rec, err
161+ }
162+ if _, err := replayLog(nil, data, path, base, &rec, logger, apply); err != nil {
163+ return rec, err
164+ }
165+ continue
166+ }
167+ f, err := fx.OpenFile(path, os.O_RDWR, 0)
168+ if err != nil {
169+ return rec, err
170+ }
171+ data, err := readAll(f)
172+ if err == nil {
173+ rec.logSize, err = replayLog(f, data, path, base, &rec, logger, apply)
174+ }
175+ if err != nil {
176+ _ = f.Close()
177+ return rec, err
178+ }
179+ rec.log, rec.logBase = f, base
180+ }
181+
182+ // Rule 7: no log file at all.
183+ if rec.log == nil {
184+ name := logName(rec.snapSeq)
185+ header := appendLogHeader(nil, rec.snapSeq)
186+ f, err := createFile(fx, dir, name, func(w io.Writer) error {
187+ _, err := w.Write(header)
188+ return err
189+ })
190+ if err != nil {
191+ return rec, err
192+ }
193+ rec.log, rec.logBase, rec.logSize = f, rec.snapSeq, logHeaderSize
194+ }
195+ return rec, nil
196+}
197+
198+// replayLog applies the records of the log file contents data (rules 3–6)
199+// and updates rec.lastSeq. f is nil for a log that is not the last one; for
200+// the last log it is used to truncate a torn tail. It returns the size of the
201+// log after recovery.
202+func replayLog(f fsys.File, data []byte, path string, base uint64, rec *recovered,
203+ logger *slog.Logger, apply func(seq uint64, ops []op),
204+) (int64, error) {
205+ last := f != nil
206+ corrupt := func(off int, format string, args ...any) error {
207+ return fmt.Errorf("%w: %s at offset %d: %s", ErrCorrupt, path, off, fmt.Sprintf(format, args...))
208+ }
209+ hdrBase, err := decodeLogHeader(data)
210+ if err != nil {
211+ return 0, fmt.Errorf("%s: %w", path, err)
212+ }
213+ if hdrBase != base {
214+ return 0, corrupt(0, "header base %d does not match the file name", hdrBase)
215+ }
216+ if base > rec.lastSeq {
217+ return 0, corrupt(0, "base %d is above the last recovered seq %d", base, rec.lastSeq)
218+ }
219+
220+ var ops []op
221+ expected := base + 1 // seq of the next record of this file
222+ off := logHeaderSize
223+ for off < len(data) {
224+ seq, decoded, n, err := decodeRecord(data[off:], ops)
225+ if err != nil {
226+ if !last {
227+ return 0, corrupt(off, "%v", err)
228+ }
229+ // Rule 6: a readable next record means acknowledged data
230+ // lies beyond the damage.
231+ if total, ok := peekRecord(data[off:]); ok && off+total < len(data) {
232+ nseq, _, _, nerr := decodeRecord(data[off+total:], nil)
233+ if nerr == nil && nseq == expected+1 {
234+ return 0, corrupt(off, "%v, followed by a valid record", err)
235+ }
236+ }
237+ if err := f.Truncate(int64(off)); err != nil {
238+ return 0, err
239+ }
240+ if err := f.SyncAll(); err != nil {
241+ return 0, err
242+ }
243+ logger.Warn("kv: dropped torn log tail", "file", path, "offset", off, "bytes", len(data)-off)
244+ data = data[:off]
245+ break
246+ }
247+ ops = decoded[:0]
248+ if seq != expected {
249+ return 0, corrupt(off, "record seq %d, want %d", seq, expected)
250+ }
251+ if seq > rec.snapSeq {
252+ if seq != rec.lastSeq+1 {
253+ return 0, corrupt(off, "record seq %d, want %d", seq, rec.lastSeq+1)
254+ }
255+ if apply != nil {
256+ apply(seq, ownOps(decoded))
257+ }
258+ rec.lastSeq = seq
259+ }
260+ expected++
261+ off += n
262+ }
263+ if last && expected-1 < rec.lastSeq {
264+ // Cannot happen unless records acknowledged before the snapshot
265+ // are missing: appending would create a seq gap.
266+ return 0, corrupt(len(data), "log ends at seq %d, want %d", expected-1, rec.lastSeq)
267+ }
268+ return int64(len(data)), nil
269+}
270+
271+// readFile returns the whole content of the file at path.
272+func readFile(fx fsys.FS, path string) ([]byte, error) {
273+ f, err := fx.OpenFile(path, os.O_RDONLY, 0)
274+ if err != nil {
275+ return nil, err
276+ }
277+ data, err := readAll(f)
278+ if cerr := f.Close(); err == nil {
279+ err = cerr
280+ }
281+ return data, err
282+}
283+
284+// readAll returns the whole content of f.
285+func readAll(f fsys.File) ([]byte, error) {
286+ size, err := f.Size()
287+ if err != nil {
288+ return nil, err
289+ }
290+ data := make([]byte, size)
291+ n, err := f.ReadAt(data, 0)
292+ if n == len(data) {
293+ return data, nil // io.EOF is allowed when the read ends at EOF
294+ }
295+ if err == nil || errors.Is(err, io.EOF) {
296+ err = io.ErrUnexpectedEOF
297+ }
298+ return nil, err
299+}
300+
301+// createFile durably creates dir/name with the content produced by write:
302+// name.tmp is written and fsynced, renamed to name, then dir is synced. The
303+// returned file is open for appending (and reading) at its end. On error the
304+// temporary file is removed on a best-effort basis.
305+func createFile(fx fsys.FS, dir, name string, write func(w io.Writer) error) (fsys.File, error) {
306+ tmp := filepath.Join(dir, name+tmpSuffix)
307+ f, err := fx.OpenFile(tmp, os.O_RDWR|os.O_CREATE|os.O_TRUNC, 0o600)
308+ if err != nil {
309+ return nil, err
310+ }
311+ fail := func(err error) (fsys.File, error) {
312+ _ = f.Close()
313+ _ = fx.Remove(tmp)
314+ return nil, err
315+ }
316+ if err := write(f); err != nil {
317+ return fail(err)
318+ }
319+ if err := f.SyncAll(); err != nil {
320+ return fail(err)
321+ }
322+ if err := fx.Rename(tmp, filepath.Join(dir, name)); err != nil {
323+ return fail(err)
324+ }
325+ if err := fx.SyncDir(dir); err != nil {
326+ _ = f.Close()
327+ return nil, err
328+ }
329+ return f, nil
330+}
A recover_test.go +565 -0
@@ -0,0 +1,565 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bytes"
7+ "errors"
8+ "fmt"
9+ "io"
10+ "log/slog"
11+ "path/filepath"
12+ "strings"
13+ "testing"
14+
15+ "bnema.dev/ermite-kv/internal/fsys"
16+ "bnema.dev/ermite-kv/internal/fsys/faultfs"
17+)
18+
19+const testDir = "/db"
20+
21+// state is the replayed content of a database.
22+type state struct {
23+ m map[string]string
24+ versions map[string]uint64
25+ seqs []uint64 // seqs passed to apply, in order
26+}
27+
28+func (s *state) load(k, v []byte, ver uint64) {
29+ s.m[string(k)] = string(v)
30+ s.versions[string(k)] = ver
31+}
32+
33+func (s *state) apply(seq uint64, ops []op) {
34+ s.seqs = append(s.seqs, seq)
35+ for _, o := range ops {
36+ switch o.kind {
37+ case opPut:
38+ s.m[string(o.key)] = string(o.value)
39+ s.versions[string(o.key)] = seq
40+ case opDelete:
41+ delete(s.m, string(o.key))
42+ delete(s.versions, string(o.key))
43+ }
44+ }
45+}
46+
47+func recoverState(t *testing.T, fx fsys.FS, logger *slog.Logger) (*state, recovered, error) {
48+ t.Helper()
49+ s := &state{m: map[string]string{}, versions: map[string]uint64{}}
50+ rec, err := recoverDir(fx, testDir, logger, s.load, s.apply)
51+ return s, rec, err
52+}
53+
54+// mustRecover recovers, checks the invariants of a successful recovery and
55+// releases the lock and the log file.
56+func mustRecover(t *testing.T, fx fsys.FS) (*state, recovered) {
57+ t.Helper()
58+ s, rec, err := recoverState(t, fx, nil)
59+ if err != nil {
60+ t.Fatalf("recoverDir: %v", err)
61+ }
62+ size, err := rec.log.Size()
63+ if err != nil || size != rec.logSize {
64+ t.Fatalf("log size = %d (err %v), recovered.logSize = %d", size, err, rec.logSize)
65+ }
66+ t.Cleanup(func() { _ = rec.log.Close(); _ = rec.lock.Close() })
67+ return s, rec
68+}
69+
70+// release closes the log file and the lock of a recovery result.
71+func release(rec recovered) {
72+ _ = rec.log.Close()
73+ _ = rec.lock.Close()
74+}
75+
76+func putOp(k, v string) op { return op{kind: opPut, key: []byte(k), value: []byte(v)} }
77+func delOp(k string) op { return op{kind: opDelete, key: []byte(k)} }
78+
79+// logBytes builds a log file: header for base, then one record per batch with
80+// consecutive seqs starting at base+1.
81+func logBytes(base uint64, batches ...[]op) []byte {
82+ b := appendLogHeader(nil, base)
83+ for i, ops := range batches {
84+ b = appendRecord(b, base+1+uint64(i), ops)
85+ }
86+ return b
87+}
88+
89+func put(fx *faultfs.FS, name string, data []byte) { fx.WriteFile(filepath.Join(testDir, name), data) }
90+
91+func snapshotBytes(t *testing.T, seq uint64, entries ...snapEntry) []byte {
92+ t.Helper()
93+ return encodeSnapshot(t, seq, entries)
94+}
95+
96+func entry(k, v string, ver uint64) snapEntry { return snapEntry{[]byte(k), []byte(v), ver} }
97+
98+func wantState(t *testing.T, s *state, m map[string]string) {
99+ t.Helper()
100+ if len(s.m) != len(m) {
101+ t.Fatalf("state = %v, want %v", s.m, m)
102+ }
103+ for k, v := range m {
104+ if got, ok := s.m[k]; !ok || got != v {
105+ t.Fatalf("state = %v, want %v", s.m, m)
106+ }
107+ }
108+}
109+
110+func TestRecoverEmptyDirCreatesLog(t *testing.T) {
111+ fx := faultfs.New()
112+ s, rec := mustRecover(t, fx)
113+ wantState(t, s, nil)
114+ if rec.snapSeq != 0 || rec.lastSeq != 0 || rec.logBase != 0 || rec.logSize != logHeaderSize {
115+ t.Fatalf("rec = %+v", rec)
116+ }
117+ // The new log must survive a power loss: file creation rules.
118+ _ = rec.log.Close()
119+ _ = rec.lock.Close()
120+ fx.PowerLoss(faultfs.PowerLossOptions{})
121+ names, err := fx.ReadDir(testDir)
122+ if err != nil {
123+ t.Fatal(err)
124+ }
125+ if got := strings.Join(names, ","); got != "LOCK,"+logName(0) && got != logName(0) {
126+ t.Fatalf("after power loss: %v", names)
127+ }
128+ data, _ := fx.ReadFile(filepath.Join(testDir, logName(0)))
129+ if !bytes.Equal(data, appendLogHeader(nil, 0)) {
130+ t.Fatalf("log content %x", data)
131+ }
132+}
133+
134+func TestRecoverLockedAndRelease(t *testing.T) {
135+ fx := faultfs.New()
136+ _, rec := mustRecover(t, fx)
137+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrLocked) {
138+ t.Fatalf("second recover: err = %v, want ErrLocked", err)
139+ }
140+ _ = rec.log.Close()
141+ _ = rec.lock.Close()
142+ _, rec2, err := recoverState(t, fx, nil)
143+ if err != nil {
144+ t.Fatalf("recover after release: %v", err)
145+ }
146+ _ = rec2.log.Close()
147+ _ = rec2.lock.Close()
148+}
149+
150+func TestRecoverReleasesLockOnError(t *testing.T) {
151+ fx := faultfs.New()
152+ put(fx, logName(0), []byte("garbage"))
153+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
154+ t.Fatalf("err = %v, want ErrCorrupt", err)
155+ }
156+ l, err := fx.Lock(filepath.Join(testDir, lockName))
157+ if err != nil {
158+ t.Fatalf("lock still held after failure: %v", err)
159+ }
160+ _ = l.Close()
161+}
162+
163+func TestRecoverCleanLog(t *testing.T) {
164+ fx := faultfs.New()
165+ data := logBytes(0,
166+ []op{putOp("a", "1"), putOp("b", "2")},
167+ []op{delOp("a"), putOp("c", "3")},
168+ []op{putOp("b", "4")},
169+ )
170+ put(fx, logName(0), data)
171+ s, rec := mustRecover(t, fx)
172+ wantState(t, s, map[string]string{"b": "4", "c": "3"})
173+ if got := fmt.Sprint(s.seqs); got != "[1 2 3]" {
174+ t.Fatalf("applied seqs %s", got)
175+ }
176+ if s.versions["b"] != 3 || s.versions["c"] != 2 {
177+ t.Fatalf("versions = %v", s.versions)
178+ }
179+ if rec.snapSeq != 0 || rec.lastSeq != 3 || rec.logBase != 0 || rec.logSize != int64(len(data)) {
180+ t.Fatalf("rec = %+v", rec)
181+ }
182+ // Appends go to the end of the recovered log.
183+ if _, err := rec.log.Write([]byte("x")); err != nil {
184+ t.Fatal(err)
185+ }
186+ if size, _ := rec.log.Size(); size != int64(len(data))+1 {
187+ t.Fatalf("size after append = %d", size)
188+ }
189+}
190+
191+func TestRecoverDeletesTmpFiles(t *testing.T) {
192+ fx := faultfs.New()
193+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
194+ put(fx, snapshotName(5)+tmpSuffix, []byte("partial"))
195+ put(fx, logName(5)+tmpSuffix, []byte("partial"))
196+ s, _ := mustRecover(t, fx)
197+ wantState(t, s, map[string]string{"a": "1"})
198+ names, _ := fx.ReadDir(testDir)
199+ for _, n := range names {
200+ if strings.HasSuffix(n, tmpSuffix) {
201+ t.Fatalf("tmp file %s left", n)
202+ }
203+ }
204+ // Deletion is durable.
205+ fx.PowerLoss(faultfs.PowerLossOptions{})
206+ names, _ = fx.ReadDir(testDir)
207+ for _, n := range names {
208+ if strings.HasSuffix(n, tmpSuffix) {
209+ t.Fatalf("tmp file %s back after power loss", n)
210+ }
211+ }
212+}
213+
214+func TestRecoverTornTail(t *testing.T) {
215+ good := logBytes(0, []op{putOp("a", "1")}, []op{putOp("b", "2")})
216+ next := appendRecord(nil, 3, []op{putOp("c", "3")})
217+ cases := map[string][]byte{
218+ "short frame": next[:3],
219+ "short record": next[:len(next)-2],
220+ "zero-filled tail": make([]byte, 100),
221+ "garbage": []byte("this is not a record at all, just junk bytes"),
222+ "bad crc": append(append([]byte(nil), next[:len(next)-1]...), next[len(next)-1]^1),
223+ "huge length": {0xff, 0xff, 0xff, 0x7f, 1, 2, 3, 4, 5, 6, 7, 8},
224+ "one byte": {0x01},
225+ }
226+ for name, tail := range cases {
227+ t.Run(name, func(t *testing.T) {
228+ fx := faultfs.New()
229+ put(fx, logName(0), append(append([]byte(nil), good...), tail...))
230+ var logs bytes.Buffer
231+ s, rec, err := recoverState(t, fx, slog.New(slog.NewTextHandler(&logs, nil)))
232+ if err != nil {
233+ t.Fatal(err)
234+ }
235+ defer release(rec)
236+ wantState(t, s, map[string]string{"a": "1", "b": "2"})
237+ if rec.lastSeq != 2 || rec.logSize != int64(len(good)) {
238+ t.Fatalf("rec = %+v, want size %d", rec, len(good))
239+ }
240+ if got, _ := fx.ReadFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, good) {
241+ t.Fatalf("log not truncated: %d bytes, want %d", len(got), len(good))
242+ }
243+ if got, _ := fx.DurableFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, good) {
244+ t.Fatal("truncation not synced")
245+ }
246+ if !strings.Contains(logs.String(), "dropped torn log tail") {
247+ t.Fatalf("no warning logged: %q", logs.String())
248+ }
249+ })
250+ }
251+}
252+
253+func TestRecoverTornTailOnlyHeaderSurvives(t *testing.T) {
254+ fx := faultfs.New()
255+ put(fx, logName(0), append(logBytes(0), make([]byte, 7)...))
256+ s, rec := mustRecover(t, fx)
257+ wantState(t, s, nil)
258+ if rec.logSize != logHeaderSize {
259+ t.Fatalf("logSize = %d", rec.logSize)
260+ }
261+}
262+
263+func TestRecoverDamageFollowedByValidRecord(t *testing.T) {
264+ r1 := appendRecord(nil, 1, []op{putOp("a", "1")})
265+ r2 := appendRecord(nil, 2, []op{putOp("b", "2")})
266+ r3 := appendRecord(nil, 3, []op{putOp("c", "3")})
267+ flip := func(b []byte, i int) []byte {
268+ c := append([]byte(nil), b...)
269+ c[i] ^= 0xff
270+ return c
271+ }
272+ hdr := appendLogHeader(nil, 0)
273+ cat := func(parts ...[]byte) []byte { return bytes.Join(parts, nil) }
274+ cases := map[string][]byte{
275+ "bad crc in middle record": cat(hdr, r1, flip(r2, len(r2)-1), r3),
276+ "bad payload in first record": cat(hdr, flip(r1, 9), r2),
277+ "bad checksum field": cat(hdr, r1, flip(r2, 4), r3),
278+ "damaged record before last": cat(hdr, flip(r1, len(r1)-1), r2, r3),
279+ "record with wrong count": cat(hdr, rawRecord(rawPayload(1, 5, opDelete, 1, 'k')), r2),
280+ "corrupt but len looks intact": cat(hdr, r1, flip(r2, 12), r3),
281+ }
282+ for name, data := range cases {
283+ t.Run(name, func(t *testing.T) {
284+ fx := faultfs.New()
285+ put(fx, logName(0), data)
286+ _, _, err := recoverState(t, fx, nil)
287+ if !errors.Is(err, ErrCorrupt) {
288+ t.Fatalf("err = %v, want ErrCorrupt", err)
289+ }
290+ // The file must be left untouched.
291+ if got, _ := fx.ReadFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, data) {
292+ t.Fatal("corrupt log was modified")
293+ }
294+ })
295+ }
296+}
297+
298+func TestRecoverDamageInOlderLog(t *testing.T) {
299+ old := logBytes(0, []op{putOp("a", "1")}, []op{putOp("b", "2")})
300+ cases := map[string][]byte{
301+ "torn tail": old[:len(old)-3],
302+ "bad crc": append(append([]byte(nil), old[:len(old)-1]...), old[len(old)-1]^1),
303+ "zero tail": append(append([]byte(nil), old...), make([]byte, 30)...),
304+ "extra bytes": append(append([]byte(nil), old...), 1, 2, 3),
305+ }
306+ for name, data := range cases {
307+ t.Run(name, func(t *testing.T) {
308+ fx := faultfs.New()
309+ put(fx, logName(0), data)
310+ put(fx, logName(2), logBytes(2, []op{putOp("c", "3")}))
311+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
312+ t.Fatalf("err = %v, want ErrCorrupt", err)
313+ }
314+ })
315+ }
316+}
317+
318+func TestRecoverBadLogHeader(t *testing.T) {
319+ good := logBytes(0, []op{putOp("a", "1")})
320+ flip := func(i int) []byte {
321+ c := append([]byte(nil), good...)
322+ c[i] ^= 1
323+ return c
324+ }
325+ cases := map[string]struct {
326+ name string
327+ data []byte
328+ }{
329+ "empty file": {logName(0), nil},
330+ "short header": {logName(0), good[:10]},
331+ "bad magic": {logName(0), flip(0)},
332+ "bad crc": {logName(0), flip(18)},
333+ "flipped base": {logName(0), flip(9)},
334+ "base != name": {logName(7), good},
335+ "all zero header": {logName(0), make([]byte, logHeaderSize)},
336+ }
337+ for name, c := range cases {
338+ t.Run(name, func(t *testing.T) {
339+ fx := faultfs.New()
340+ put(fx, c.name, c.data)
341+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
342+ t.Fatalf("err = %v, want ErrCorrupt", err)
343+ }
344+ })
345+ }
346+ t.Run("bad header in older log", func(t *testing.T) {
347+ fx := faultfs.New()
348+ put(fx, logName(0), good[:10])
349+ put(fx, logName(1), logBytes(1, []op{putOp("b", "2")}))
350+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
351+ t.Fatalf("err = %v, want ErrCorrupt", err)
352+ }
353+ })
354+}
355+
356+func TestRecoverSeqGap(t *testing.T) {
357+ cases := map[string]func(fx *faultfs.FS){
358+ "gap inside a log": func(fx *faultfs.FS) {
359+ b := appendLogHeader(nil, 0)
360+ b = appendRecord(b, 1, []op{putOp("a", "1")})
361+ b = appendRecord(b, 3, []op{putOp("b", "2")})
362+ put(fx, logName(0), b)
363+ },
364+ "first record not base+1": func(fx *faultfs.FS) {
365+ b := appendRecord(appendLogHeader(nil, 0), 2, []op{putOp("a", "1")})
366+ put(fx, logName(0), b)
367+ },
368+ "gap between logs": func(fx *faultfs.FS) {
369+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
370+ put(fx, logName(2), logBytes(2, []op{putOp("c", "3")}))
371+ },
372+ "first log starts after seq 0": func(fx *faultfs.FS) {
373+ put(fx, logName(4), logBytes(4, []op{putOp("a", "1")}))
374+ },
375+ "log after snapshot has a gap": func(fx *faultfs.FS) {
376+ put(fx, snapshotName(3), snapshotBytes(t, 3, entry("a", "1", 3)))
377+ put(fx, logName(5), logBytes(5, []op{putOp("b", "2")}))
378+ },
379+ "logs end before snapshot": func(fx *faultfs.FS) {
380+ put(fx, snapshotName(5), snapshotBytes(t, 5, entry("a", "1", 3)))
381+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
382+ },
383+ "duplicate seq": func(fx *faultfs.FS) {
384+ b := appendLogHeader(nil, 0)
385+ b = appendRecord(b, 1, []op{putOp("a", "1")})
386+ b = appendRecord(b, 1, []op{putOp("b", "2")})
387+ put(fx, logName(0), b)
388+ },
389+ }
390+ for name, setup := range cases {
391+ t.Run(name, func(t *testing.T) {
392+ fx := faultfs.New()
393+ setup(fx)
394+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
395+ t.Fatalf("err = %v, want ErrCorrupt", err)
396+ }
397+ })
398+ }
399+}
400+
401+func TestRecoverSnapshotThenLogs(t *testing.T) {
402+ fx := faultfs.New()
403+ put(fx, snapshotName(3), snapshotBytes(t, 3, entry("a", "1", 1), entry("b", "2", 3)))
404+ // log-0 still holds records already in the snapshot; log-3 continues it.
405+ put(fx, logName(0), logBytes(0, []op{putOp("a", "old")}, []op{putOp("x", "gone")}, []op{putOp("b", "2")}))
406+ put(fx, logName(3), logBytes(3, []op{putOp("c", "3")}, []op{delOp("a")}))
407+ s, rec := mustRecover(t, fx)
408+ wantState(t, s, map[string]string{"b": "2", "c": "3"})
409+ if got := fmt.Sprint(s.seqs); got != "[4 5]" {
410+ t.Fatalf("applied seqs %s", got)
411+ }
412+ if s.versions["b"] != 3 || s.versions["c"] != 4 {
413+ t.Fatalf("versions = %v", s.versions)
414+ }
415+ if rec.snapSeq != 3 || rec.lastSeq != 5 || rec.logBase != 3 {
416+ t.Fatalf("rec = %+v", rec)
417+ }
418+}
419+
420+func TestRecoverSnapshotWithoutLogCreatesLog(t *testing.T) {
421+ fx := faultfs.New()
422+ put(fx, snapshotName(7), snapshotBytes(t, 7, entry("a", "1", 7)))
423+ s, rec := mustRecover(t, fx)
424+ wantState(t, s, map[string]string{"a": "1"})
425+ if rec.snapSeq != 7 || rec.lastSeq != 7 || rec.logBase != 7 || rec.logSize != logHeaderSize {
426+ t.Fatalf("rec = %+v", rec)
427+ }
428+ if _, ok := fx.ReadFile(filepath.Join(testDir, logName(7))); !ok {
429+ t.Fatal("log-7 not created")
430+ }
431+}
432+
433+func TestRecoverInvalidNewestSnapshotFallsBack(t *testing.T) {
434+ good := snapshotBytes(t, 2, entry("a", "1", 1), entry("b", "2", 2))
435+ newer := snapshotBytes(t, 4, entry("a", "1", 1), entry("z", "9", 4))
436+ truncated := newer[:len(newer)-2]
437+ flipped := append([]byte(nil), newer...)
438+ flipped[30] ^= 1
439+ wrongName := snapshotBytes(t, 3, entry("a", "1", 1))
440+ for name, bad := range map[string][]byte{"truncated": truncated, "flipped": flipped, "empty": nil, "seq/name mismatch": wrongName} {
441+ t.Run(name, func(t *testing.T) {
442+ fx := faultfs.New()
443+ put(fx, snapshotName(2), good)
444+ put(fx, snapshotName(4), bad)
445+ put(fx, logName(2), logBytes(2, []op{putOp("c", "3")}, []op{putOp("d", "4")}, []op{delOp("a")}))
446+ var logs bytes.Buffer
447+ s, rec, err := recoverState(t, fx, slog.New(slog.NewTextHandler(&logs, nil)))
448+ if err != nil {
449+ t.Fatal(err)
450+ }
451+ defer release(rec)
452+ wantState(t, s, map[string]string{"b": "2", "c": "3", "d": "4"})
453+ if rec.snapSeq != 2 || rec.lastSeq != 5 {
454+ t.Fatalf("rec = %+v", rec)
455+ }
456+ if !strings.Contains(logs.String(), "invalid snapshot") {
457+ t.Fatalf("fallback not logged: %q", logs.String())
458+ }
459+ })
460+ }
461+}
462+
463+func TestRecoverInvalidSnapshotNoOtherFallsBackToEmpty(t *testing.T) {
464+ fx := faultfs.New()
465+ put(fx, snapshotName(3), []byte("junk"))
466+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
467+ s, rec := mustRecover(t, fx)
468+ wantState(t, s, map[string]string{"a": "1"})
469+ if rec.snapSeq != 0 || rec.lastSeq != 1 {
470+ t.Fatalf("rec = %+v", rec)
471+ }
472+}
473+
474+func TestRecoverNoSnapshotStartsAtZero(t *testing.T) {
475+ fx := faultfs.New()
476+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
477+ put(fx, logName(1), logBytes(1, []op{putOp("b", "2")}))
478+ put(fx, "unrelated.txt", []byte("ignored"))
479+ put(fx, "log-xyz", []byte("ignored"))
480+ s, rec := mustRecover(t, fx)
481+ wantState(t, s, map[string]string{"a": "1", "b": "2"})
482+ if rec.snapSeq != 0 || rec.lastSeq != 2 || rec.logBase != 1 {
483+ t.Fatalf("rec = %+v", rec)
484+ }
485+}
486+
487+func TestRecoverSnapshotLoadOrderAndOwnership(t *testing.T) {
488+ fx := faultfs.New()
489+ put(fx, snapshotName(1), snapshotBytes(t, 1, entry("a", "1", 1), entry("b", "2", 1), entry("c", "3", 1)))
490+ var keys []string
491+ var held [][]byte
492+ _, err := recoverDir(fx, testDir, nil, func(k, v []byte, ver uint64) {
493+ keys = append(keys, string(k))
494+ held = append(held, k, v)
495+ }, nil)
496+ if err != nil {
497+ t.Fatal(err)
498+ }
499+ if strings.Join(keys, "") != "abc" {
500+ t.Fatalf("keys = %v", keys)
501+ }
502+ var all []byte
503+ for _, h := range held {
504+ all = append(all, h...)
505+ }
506+ if string(all) != "a1b2c3" {
507+ t.Fatalf("retained slices changed: %q", all)
508+ }
509+}
510+
511+func TestRecoverPowerLossAfterRecovery(t *testing.T) {
512+ // A fresh directory survives a power loss right after recovery and
513+ // reopens cleanly with the same state.
514+ fx := faultfs.New()
515+ _, rec := mustRecover(t, fx)
516+ _ = rec.log.Close()
517+ _ = rec.lock.Close()
518+ fx.PowerLoss(faultfs.PowerLossOptions{})
519+ s, rec2 := mustRecover(t, fx)
520+ wantState(t, s, nil)
521+ if rec2.logBase != 0 {
522+ t.Fatalf("rec = %+v", rec2)
523+ }
524+}
525+
526+func TestRecoverOSFilesystem(t *testing.T) {
527+ dir := filepath.Join(t.TempDir(), "nested", "db")
528+ data := logBytes(0, []op{putOp("a", "1")}, []op{putOp("b", "2")})
529+ torn := append(append([]byte(nil), data...), 9, 9, 9)
530+
531+ rec, err := recoverDir(fsys.OS(), dir, nil, nil, nil)
532+ if err != nil {
533+ t.Fatal(err)
534+ }
535+ if _, err := recoverDir(fsys.OS(), dir, nil, nil, nil); !errors.Is(err, ErrLocked) {
536+ t.Fatalf("second recover: err = %v, want ErrLocked", err)
537+ }
538+ if _, err := rec.log.Write(data[logHeaderSize:]); err != nil {
539+ t.Fatal(err)
540+ }
541+ if _, err := rec.log.Write(torn[len(data):]); err != nil {
542+ t.Fatal(err)
543+ }
544+ if err := rec.log.Sync(); err != nil {
545+ t.Fatal(err)
546+ }
547+ _ = rec.log.Close()
548+ _ = rec.lock.Close()
549+
550+ s := &state{m: map[string]string{}, versions: map[string]uint64{}}
551+ rec, err = recoverDir(fsys.OS(), dir, nil, s.load, s.apply)
552+ if err != nil {
553+ t.Fatal(err)
554+ }
555+ defer release(rec)
556+ wantState(t, s, map[string]string{"a": "1", "b": "2"})
557+ if rec.lastSeq != 2 || rec.logSize != int64(len(data)) {
558+ t.Fatalf("rec = %+v", rec)
559+ }
560+ buf := make([]byte, len(data)+10)
561+ n, err := rec.log.ReadAt(buf, 0)
562+ if n != len(data) || (err != nil && !errors.Is(err, io.EOF)) {
563+ t.Fatalf("ReadAt = %d, %v: tail not truncated", n, err)
564+ }
565+}
@@ -0,0 +1,565 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bytes"
7+ "errors"
8+ "fmt"
9+ "io"
10+ "log/slog"
11+ "path/filepath"
12+ "strings"
13+ "testing"
14+
15+ "bnema.dev/ermite-kv/internal/fsys"
16+ "bnema.dev/ermite-kv/internal/fsys/faultfs"
17+)
18+
19+const testDir = "/db"
20+
21+// state is the replayed content of a database.
22+type state struct {
23+ m map[string]string
24+ versions map[string]uint64
25+ seqs []uint64 // seqs passed to apply, in order
26+}
27+
28+func (s *state) load(k, v []byte, ver uint64) {
29+ s.m[string(k)] = string(v)
30+ s.versions[string(k)] = ver
31+}
32+
33+func (s *state) apply(seq uint64, ops []op) {
34+ s.seqs = append(s.seqs, seq)
35+ for _, o := range ops {
36+ switch o.kind {
37+ case opPut:
38+ s.m[string(o.key)] = string(o.value)
39+ s.versions[string(o.key)] = seq
40+ case opDelete:
41+ delete(s.m, string(o.key))
42+ delete(s.versions, string(o.key))
43+ }
44+ }
45+}
46+
47+func recoverState(t *testing.T, fx fsys.FS, logger *slog.Logger) (*state, recovered, error) {
48+ t.Helper()
49+ s := &state{m: map[string]string{}, versions: map[string]uint64{}}
50+ rec, err := recoverDir(fx, testDir, logger, s.load, s.apply)
51+ return s, rec, err
52+}
53+
54+// mustRecover recovers, checks the invariants of a successful recovery and
55+// releases the lock and the log file.
56+func mustRecover(t *testing.T, fx fsys.FS) (*state, recovered) {
57+ t.Helper()
58+ s, rec, err := recoverState(t, fx, nil)
59+ if err != nil {
60+ t.Fatalf("recoverDir: %v", err)
61+ }
62+ size, err := rec.log.Size()
63+ if err != nil || size != rec.logSize {
64+ t.Fatalf("log size = %d (err %v), recovered.logSize = %d", size, err, rec.logSize)
65+ }
66+ t.Cleanup(func() { _ = rec.log.Close(); _ = rec.lock.Close() })
67+ return s, rec
68+}
69+
70+// release closes the log file and the lock of a recovery result.
71+func release(rec recovered) {
72+ _ = rec.log.Close()
73+ _ = rec.lock.Close()
74+}
75+
76+func putOp(k, v string) op { return op{kind: opPut, key: []byte(k), value: []byte(v)} }
77+func delOp(k string) op { return op{kind: opDelete, key: []byte(k)} }
78+
79+// logBytes builds a log file: header for base, then one record per batch with
80+// consecutive seqs starting at base+1.
81+func logBytes(base uint64, batches ...[]op) []byte {
82+ b := appendLogHeader(nil, base)
83+ for i, ops := range batches {
84+ b = appendRecord(b, base+1+uint64(i), ops)
85+ }
86+ return b
87+}
88+
89+func put(fx *faultfs.FS, name string, data []byte) { fx.WriteFile(filepath.Join(testDir, name), data) }
90+
91+func snapshotBytes(t *testing.T, seq uint64, entries ...snapEntry) []byte {
92+ t.Helper()
93+ return encodeSnapshot(t, seq, entries)
94+}
95+
96+func entry(k, v string, ver uint64) snapEntry { return snapEntry{[]byte(k), []byte(v), ver} }
97+
98+func wantState(t *testing.T, s *state, m map[string]string) {
99+ t.Helper()
100+ if len(s.m) != len(m) {
101+ t.Fatalf("state = %v, want %v", s.m, m)
102+ }
103+ for k, v := range m {
104+ if got, ok := s.m[k]; !ok || got != v {
105+ t.Fatalf("state = %v, want %v", s.m, m)
106+ }
107+ }
108+}
109+
110+func TestRecoverEmptyDirCreatesLog(t *testing.T) {
111+ fx := faultfs.New()
112+ s, rec := mustRecover(t, fx)
113+ wantState(t, s, nil)
114+ if rec.snapSeq != 0 || rec.lastSeq != 0 || rec.logBase != 0 || rec.logSize != logHeaderSize {
115+ t.Fatalf("rec = %+v", rec)
116+ }
117+ // The new log must survive a power loss: file creation rules.
118+ _ = rec.log.Close()
119+ _ = rec.lock.Close()
120+ fx.PowerLoss(faultfs.PowerLossOptions{})
121+ names, err := fx.ReadDir(testDir)
122+ if err != nil {
123+ t.Fatal(err)
124+ }
125+ if got := strings.Join(names, ","); got != "LOCK,"+logName(0) && got != logName(0) {
126+ t.Fatalf("after power loss: %v", names)
127+ }
128+ data, _ := fx.ReadFile(filepath.Join(testDir, logName(0)))
129+ if !bytes.Equal(data, appendLogHeader(nil, 0)) {
130+ t.Fatalf("log content %x", data)
131+ }
132+}
133+
134+func TestRecoverLockedAndRelease(t *testing.T) {
135+ fx := faultfs.New()
136+ _, rec := mustRecover(t, fx)
137+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrLocked) {
138+ t.Fatalf("second recover: err = %v, want ErrLocked", err)
139+ }
140+ _ = rec.log.Close()
141+ _ = rec.lock.Close()
142+ _, rec2, err := recoverState(t, fx, nil)
143+ if err != nil {
144+ t.Fatalf("recover after release: %v", err)
145+ }
146+ _ = rec2.log.Close()
147+ _ = rec2.lock.Close()
148+}
149+
150+func TestRecoverReleasesLockOnError(t *testing.T) {
151+ fx := faultfs.New()
152+ put(fx, logName(0), []byte("garbage"))
153+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
154+ t.Fatalf("err = %v, want ErrCorrupt", err)
155+ }
156+ l, err := fx.Lock(filepath.Join(testDir, lockName))
157+ if err != nil {
158+ t.Fatalf("lock still held after failure: %v", err)
159+ }
160+ _ = l.Close()
161+}
162+
163+func TestRecoverCleanLog(t *testing.T) {
164+ fx := faultfs.New()
165+ data := logBytes(0,
166+ []op{putOp("a", "1"), putOp("b", "2")},
167+ []op{delOp("a"), putOp("c", "3")},
168+ []op{putOp("b", "4")},
169+ )
170+ put(fx, logName(0), data)
171+ s, rec := mustRecover(t, fx)
172+ wantState(t, s, map[string]string{"b": "4", "c": "3"})
173+ if got := fmt.Sprint(s.seqs); got != "[1 2 3]" {
174+ t.Fatalf("applied seqs %s", got)
175+ }
176+ if s.versions["b"] != 3 || s.versions["c"] != 2 {
177+ t.Fatalf("versions = %v", s.versions)
178+ }
179+ if rec.snapSeq != 0 || rec.lastSeq != 3 || rec.logBase != 0 || rec.logSize != int64(len(data)) {
180+ t.Fatalf("rec = %+v", rec)
181+ }
182+ // Appends go to the end of the recovered log.
183+ if _, err := rec.log.Write([]byte("x")); err != nil {
184+ t.Fatal(err)
185+ }
186+ if size, _ := rec.log.Size(); size != int64(len(data))+1 {
187+ t.Fatalf("size after append = %d", size)
188+ }
189+}
190+
191+func TestRecoverDeletesTmpFiles(t *testing.T) {
192+ fx := faultfs.New()
193+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
194+ put(fx, snapshotName(5)+tmpSuffix, []byte("partial"))
195+ put(fx, logName(5)+tmpSuffix, []byte("partial"))
196+ s, _ := mustRecover(t, fx)
197+ wantState(t, s, map[string]string{"a": "1"})
198+ names, _ := fx.ReadDir(testDir)
199+ for _, n := range names {
200+ if strings.HasSuffix(n, tmpSuffix) {
201+ t.Fatalf("tmp file %s left", n)
202+ }
203+ }
204+ // Deletion is durable.
205+ fx.PowerLoss(faultfs.PowerLossOptions{})
206+ names, _ = fx.ReadDir(testDir)
207+ for _, n := range names {
208+ if strings.HasSuffix(n, tmpSuffix) {
209+ t.Fatalf("tmp file %s back after power loss", n)
210+ }
211+ }
212+}
213+
214+func TestRecoverTornTail(t *testing.T) {
215+ good := logBytes(0, []op{putOp("a", "1")}, []op{putOp("b", "2")})
216+ next := appendRecord(nil, 3, []op{putOp("c", "3")})
217+ cases := map[string][]byte{
218+ "short frame": next[:3],
219+ "short record": next[:len(next)-2],
220+ "zero-filled tail": make([]byte, 100),
221+ "garbage": []byte("this is not a record at all, just junk bytes"),
222+ "bad crc": append(append([]byte(nil), next[:len(next)-1]...), next[len(next)-1]^1),
223+ "huge length": {0xff, 0xff, 0xff, 0x7f, 1, 2, 3, 4, 5, 6, 7, 8},
224+ "one byte": {0x01},
225+ }
226+ for name, tail := range cases {
227+ t.Run(name, func(t *testing.T) {
228+ fx := faultfs.New()
229+ put(fx, logName(0), append(append([]byte(nil), good...), tail...))
230+ var logs bytes.Buffer
231+ s, rec, err := recoverState(t, fx, slog.New(slog.NewTextHandler(&logs, nil)))
232+ if err != nil {
233+ t.Fatal(err)
234+ }
235+ defer release(rec)
236+ wantState(t, s, map[string]string{"a": "1", "b": "2"})
237+ if rec.lastSeq != 2 || rec.logSize != int64(len(good)) {
238+ t.Fatalf("rec = %+v, want size %d", rec, len(good))
239+ }
240+ if got, _ := fx.ReadFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, good) {
241+ t.Fatalf("log not truncated: %d bytes, want %d", len(got), len(good))
242+ }
243+ if got, _ := fx.DurableFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, good) {
244+ t.Fatal("truncation not synced")
245+ }
246+ if !strings.Contains(logs.String(), "dropped torn log tail") {
247+ t.Fatalf("no warning logged: %q", logs.String())
248+ }
249+ })
250+ }
251+}
252+
253+func TestRecoverTornTailOnlyHeaderSurvives(t *testing.T) {
254+ fx := faultfs.New()
255+ put(fx, logName(0), append(logBytes(0), make([]byte, 7)...))
256+ s, rec := mustRecover(t, fx)
257+ wantState(t, s, nil)
258+ if rec.logSize != logHeaderSize {
259+ t.Fatalf("logSize = %d", rec.logSize)
260+ }
261+}
262+
263+func TestRecoverDamageFollowedByValidRecord(t *testing.T) {
264+ r1 := appendRecord(nil, 1, []op{putOp("a", "1")})
265+ r2 := appendRecord(nil, 2, []op{putOp("b", "2")})
266+ r3 := appendRecord(nil, 3, []op{putOp("c", "3")})
267+ flip := func(b []byte, i int) []byte {
268+ c := append([]byte(nil), b...)
269+ c[i] ^= 0xff
270+ return c
271+ }
272+ hdr := appendLogHeader(nil, 0)
273+ cat := func(parts ...[]byte) []byte { return bytes.Join(parts, nil) }
274+ cases := map[string][]byte{
275+ "bad crc in middle record": cat(hdr, r1, flip(r2, len(r2)-1), r3),
276+ "bad payload in first record": cat(hdr, flip(r1, 9), r2),
277+ "bad checksum field": cat(hdr, r1, flip(r2, 4), r3),
278+ "damaged record before last": cat(hdr, flip(r1, len(r1)-1), r2, r3),
279+ "record with wrong count": cat(hdr, rawRecord(rawPayload(1, 5, opDelete, 1, 'k')), r2),
280+ "corrupt but len looks intact": cat(hdr, r1, flip(r2, 12), r3),
281+ }
282+ for name, data := range cases {
283+ t.Run(name, func(t *testing.T) {
284+ fx := faultfs.New()
285+ put(fx, logName(0), data)
286+ _, _, err := recoverState(t, fx, nil)
287+ if !errors.Is(err, ErrCorrupt) {
288+ t.Fatalf("err = %v, want ErrCorrupt", err)
289+ }
290+ // The file must be left untouched.
291+ if got, _ := fx.ReadFile(filepath.Join(testDir, logName(0))); !bytes.Equal(got, data) {
292+ t.Fatal("corrupt log was modified")
293+ }
294+ })
295+ }
296+}
297+
298+func TestRecoverDamageInOlderLog(t *testing.T) {
299+ old := logBytes(0, []op{putOp("a", "1")}, []op{putOp("b", "2")})
300+ cases := map[string][]byte{
301+ "torn tail": old[:len(old)-3],
302+ "bad crc": append(append([]byte(nil), old[:len(old)-1]...), old[len(old)-1]^1),
303+ "zero tail": append(append([]byte(nil), old...), make([]byte, 30)...),
304+ "extra bytes": append(append([]byte(nil), old...), 1, 2, 3),
305+ }
306+ for name, data := range cases {
307+ t.Run(name, func(t *testing.T) {
308+ fx := faultfs.New()
309+ put(fx, logName(0), data)
310+ put(fx, logName(2), logBytes(2, []op{putOp("c", "3")}))
311+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
312+ t.Fatalf("err = %v, want ErrCorrupt", err)
313+ }
314+ })
315+ }
316+}
317+
318+func TestRecoverBadLogHeader(t *testing.T) {
319+ good := logBytes(0, []op{putOp("a", "1")})
320+ flip := func(i int) []byte {
321+ c := append([]byte(nil), good...)
322+ c[i] ^= 1
323+ return c
324+ }
325+ cases := map[string]struct {
326+ name string
327+ data []byte
328+ }{
329+ "empty file": {logName(0), nil},
330+ "short header": {logName(0), good[:10]},
331+ "bad magic": {logName(0), flip(0)},
332+ "bad crc": {logName(0), flip(18)},
333+ "flipped base": {logName(0), flip(9)},
334+ "base != name": {logName(7), good},
335+ "all zero header": {logName(0), make([]byte, logHeaderSize)},
336+ }
337+ for name, c := range cases {
338+ t.Run(name, func(t *testing.T) {
339+ fx := faultfs.New()
340+ put(fx, c.name, c.data)
341+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
342+ t.Fatalf("err = %v, want ErrCorrupt", err)
343+ }
344+ })
345+ }
346+ t.Run("bad header in older log", func(t *testing.T) {
347+ fx := faultfs.New()
348+ put(fx, logName(0), good[:10])
349+ put(fx, logName(1), logBytes(1, []op{putOp("b", "2")}))
350+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
351+ t.Fatalf("err = %v, want ErrCorrupt", err)
352+ }
353+ })
354+}
355+
356+func TestRecoverSeqGap(t *testing.T) {
357+ cases := map[string]func(fx *faultfs.FS){
358+ "gap inside a log": func(fx *faultfs.FS) {
359+ b := appendLogHeader(nil, 0)
360+ b = appendRecord(b, 1, []op{putOp("a", "1")})
361+ b = appendRecord(b, 3, []op{putOp("b", "2")})
362+ put(fx, logName(0), b)
363+ },
364+ "first record not base+1": func(fx *faultfs.FS) {
365+ b := appendRecord(appendLogHeader(nil, 0), 2, []op{putOp("a", "1")})
366+ put(fx, logName(0), b)
367+ },
368+ "gap between logs": func(fx *faultfs.FS) {
369+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
370+ put(fx, logName(2), logBytes(2, []op{putOp("c", "3")}))
371+ },
372+ "first log starts after seq 0": func(fx *faultfs.FS) {
373+ put(fx, logName(4), logBytes(4, []op{putOp("a", "1")}))
374+ },
375+ "log after snapshot has a gap": func(fx *faultfs.FS) {
376+ put(fx, snapshotName(3), snapshotBytes(t, 3, entry("a", "1", 3)))
377+ put(fx, logName(5), logBytes(5, []op{putOp("b", "2")}))
378+ },
379+ "logs end before snapshot": func(fx *faultfs.FS) {
380+ put(fx, snapshotName(5), snapshotBytes(t, 5, entry("a", "1", 3)))
381+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
382+ },
383+ "duplicate seq": func(fx *faultfs.FS) {
384+ b := appendLogHeader(nil, 0)
385+ b = appendRecord(b, 1, []op{putOp("a", "1")})
386+ b = appendRecord(b, 1, []op{putOp("b", "2")})
387+ put(fx, logName(0), b)
388+ },
389+ }
390+ for name, setup := range cases {
391+ t.Run(name, func(t *testing.T) {
392+ fx := faultfs.New()
393+ setup(fx)
394+ if _, _, err := recoverState(t, fx, nil); !errors.Is(err, ErrCorrupt) {
395+ t.Fatalf("err = %v, want ErrCorrupt", err)
396+ }
397+ })
398+ }
399+}
400+
401+func TestRecoverSnapshotThenLogs(t *testing.T) {
402+ fx := faultfs.New()
403+ put(fx, snapshotName(3), snapshotBytes(t, 3, entry("a", "1", 1), entry("b", "2", 3)))
404+ // log-0 still holds records already in the snapshot; log-3 continues it.
405+ put(fx, logName(0), logBytes(0, []op{putOp("a", "old")}, []op{putOp("x", "gone")}, []op{putOp("b", "2")}))
406+ put(fx, logName(3), logBytes(3, []op{putOp("c", "3")}, []op{delOp("a")}))
407+ s, rec := mustRecover(t, fx)
408+ wantState(t, s, map[string]string{"b": "2", "c": "3"})
409+ if got := fmt.Sprint(s.seqs); got != "[4 5]" {
410+ t.Fatalf("applied seqs %s", got)
411+ }
412+ if s.versions["b"] != 3 || s.versions["c"] != 4 {
413+ t.Fatalf("versions = %v", s.versions)
414+ }
415+ if rec.snapSeq != 3 || rec.lastSeq != 5 || rec.logBase != 3 {
416+ t.Fatalf("rec = %+v", rec)
417+ }
418+}
419+
420+func TestRecoverSnapshotWithoutLogCreatesLog(t *testing.T) {
421+ fx := faultfs.New()
422+ put(fx, snapshotName(7), snapshotBytes(t, 7, entry("a", "1", 7)))
423+ s, rec := mustRecover(t, fx)
424+ wantState(t, s, map[string]string{"a": "1"})
425+ if rec.snapSeq != 7 || rec.lastSeq != 7 || rec.logBase != 7 || rec.logSize != logHeaderSize {
426+ t.Fatalf("rec = %+v", rec)
427+ }
428+ if _, ok := fx.ReadFile(filepath.Join(testDir, logName(7))); !ok {
429+ t.Fatal("log-7 not created")
430+ }
431+}
432+
433+func TestRecoverInvalidNewestSnapshotFallsBack(t *testing.T) {
434+ good := snapshotBytes(t, 2, entry("a", "1", 1), entry("b", "2", 2))
435+ newer := snapshotBytes(t, 4, entry("a", "1", 1), entry("z", "9", 4))
436+ truncated := newer[:len(newer)-2]
437+ flipped := append([]byte(nil), newer...)
438+ flipped[30] ^= 1
439+ wrongName := snapshotBytes(t, 3, entry("a", "1", 1))
440+ for name, bad := range map[string][]byte{"truncated": truncated, "flipped": flipped, "empty": nil, "seq/name mismatch": wrongName} {
441+ t.Run(name, func(t *testing.T) {
442+ fx := faultfs.New()
443+ put(fx, snapshotName(2), good)
444+ put(fx, snapshotName(4), bad)
445+ put(fx, logName(2), logBytes(2, []op{putOp("c", "3")}, []op{putOp("d", "4")}, []op{delOp("a")}))
446+ var logs bytes.Buffer
447+ s, rec, err := recoverState(t, fx, slog.New(slog.NewTextHandler(&logs, nil)))
448+ if err != nil {
449+ t.Fatal(err)
450+ }
451+ defer release(rec)
452+ wantState(t, s, map[string]string{"b": "2", "c": "3", "d": "4"})
453+ if rec.snapSeq != 2 || rec.lastSeq != 5 {
454+ t.Fatalf("rec = %+v", rec)
455+ }
456+ if !strings.Contains(logs.String(), "invalid snapshot") {
457+ t.Fatalf("fallback not logged: %q", logs.String())
458+ }
459+ })
460+ }
461+}
462+
463+func TestRecoverInvalidSnapshotNoOtherFallsBackToEmpty(t *testing.T) {
464+ fx := faultfs.New()
465+ put(fx, snapshotName(3), []byte("junk"))
466+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
467+ s, rec := mustRecover(t, fx)
468+ wantState(t, s, map[string]string{"a": "1"})
469+ if rec.snapSeq != 0 || rec.lastSeq != 1 {
470+ t.Fatalf("rec = %+v", rec)
471+ }
472+}
473+
474+func TestRecoverNoSnapshotStartsAtZero(t *testing.T) {
475+ fx := faultfs.New()
476+ put(fx, logName(0), logBytes(0, []op{putOp("a", "1")}))
477+ put(fx, logName(1), logBytes(1, []op{putOp("b", "2")}))
478+ put(fx, "unrelated.txt", []byte("ignored"))
479+ put(fx, "log-xyz", []byte("ignored"))
480+ s, rec := mustRecover(t, fx)
481+ wantState(t, s, map[string]string{"a": "1", "b": "2"})
482+ if rec.snapSeq != 0 || rec.lastSeq != 2 || rec.logBase != 1 {
483+ t.Fatalf("rec = %+v", rec)
484+ }
485+}
486+
487+func TestRecoverSnapshotLoadOrderAndOwnership(t *testing.T) {
488+ fx := faultfs.New()
489+ put(fx, snapshotName(1), snapshotBytes(t, 1, entry("a", "1", 1), entry("b", "2", 1), entry("c", "3", 1)))
490+ var keys []string
491+ var held [][]byte
492+ _, err := recoverDir(fx, testDir, nil, func(k, v []byte, ver uint64) {
493+ keys = append(keys, string(k))
494+ held = append(held, k, v)
495+ }, nil)
496+ if err != nil {
497+ t.Fatal(err)
498+ }
499+ if strings.Join(keys, "") != "abc" {
500+ t.Fatalf("keys = %v", keys)
501+ }
502+ var all []byte
503+ for _, h := range held {
504+ all = append(all, h...)
505+ }
506+ if string(all) != "a1b2c3" {
507+ t.Fatalf("retained slices changed: %q", all)
508+ }
509+}
510+
511+func TestRecoverPowerLossAfterRecovery(t *testing.T) {
512+ // A fresh directory survives a power loss right after recovery and
513+ // reopens cleanly with the same state.
514+ fx := faultfs.New()
515+ _, rec := mustRecover(t, fx)
516+ _ = rec.log.Close()
517+ _ = rec.lock.Close()
518+ fx.PowerLoss(faultfs.PowerLossOptions{})
519+ s, rec2 := mustRecover(t, fx)
520+ wantState(t, s, nil)
521+ if rec2.logBase != 0 {
522+ t.Fatalf("rec = %+v", rec2)
523+ }
524+}
525+
526+func TestRecoverOSFilesystem(t *testing.T) {
527+ dir := filepath.Join(t.TempDir(), "nested", "db")
528+ data := logBytes(0, []op{putOp("a", "1")}, []op{putOp("b", "2")})
529+ torn := append(append([]byte(nil), data...), 9, 9, 9)
530+
531+ rec, err := recoverDir(fsys.OS(), dir, nil, nil, nil)
532+ if err != nil {
533+ t.Fatal(err)
534+ }
535+ if _, err := recoverDir(fsys.OS(), dir, nil, nil, nil); !errors.Is(err, ErrLocked) {
536+ t.Fatalf("second recover: err = %v, want ErrLocked", err)
537+ }
538+ if _, err := rec.log.Write(data[logHeaderSize:]); err != nil {
539+ t.Fatal(err)
540+ }
541+ if _, err := rec.log.Write(torn[len(data):]); err != nil {
542+ t.Fatal(err)
543+ }
544+ if err := rec.log.Sync(); err != nil {
545+ t.Fatal(err)
546+ }
547+ _ = rec.log.Close()
548+ _ = rec.lock.Close()
549+
550+ s := &state{m: map[string]string{}, versions: map[string]uint64{}}
551+ rec, err = recoverDir(fsys.OS(), dir, nil, s.load, s.apply)
552+ if err != nil {
553+ t.Fatal(err)
554+ }
555+ defer release(rec)
556+ wantState(t, s, map[string]string{"a": "1", "b": "2"})
557+ if rec.lastSeq != 2 || rec.logSize != int64(len(data)) {
558+ t.Fatalf("rec = %+v", rec)
559+ }
560+ buf := make([]byte, len(data)+10)
561+ n, err := rec.log.ReadAt(buf, 0)
562+ if n != len(data) || (err != nil && !errors.Is(err, io.EOF)) {
563+ t.Fatalf("ReadAt = %d, %v: tail not truncated", n, err)
564+ }
565+}

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