detached b496766

feat(db): add Open, Close, batches and group-commit writer

commit
b496766f8ae7f52fd9abcdeded2c85c1bd5a5087
author
brice · · 2026-10-10 02:23 UTC
parent
60f7034
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
5 files · +1741 -0
A batch.go +110 -0
@@ -0,0 +1,110 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+// Batch is an ordered set of mutations and preconditions that Apply commits
6+// atomically. The zero value is ready to use. A Batch must not be modified
7+// while Apply runs and is not safe for concurrent use.
8+//
9+// Batch methods do not return errors: the batch records its first error and
10+// Apply returns it.
11+type Batch struct {
12+ ops []op
13+ pre []precond
14+ opsBytes int // encoded size of ops
15+ err error
16+}
17+
18+// precond is a precondition checked by the writer before any op.
19+type precond struct {
20+ key []byte
21+ absent bool // the key must not exist
22+ version uint64 // otherwise the key must exist at this version
23+}
24+
25+// Put sets key to value. It copies key and value.
26+func (b *Batch) Put(key, value []byte) {
27+ if !b.validKey(key) {
28+ return
29+ }
30+ // One allocation holds both; the tree retains the slices, so they are
31+ // never reused (not even by Reset).
32+ buf := make([]byte, len(key)+len(value))
33+ n := copy(buf, key)
34+ copy(buf[n:], value)
35+ b.add(op{kind: opPut, key: buf[:n:n], value: buf[n:len(buf):len(buf)]})
36+}
37+
38+// Delete removes key. It copies key. Deleting a missing key is not an error.
39+func (b *Batch) Delete(key []byte) {
40+ if !b.validKey(key) {
41+ return
42+ }
43+ b.add(op{kind: opDelete, key: append([]byte(nil), key...)})
44+}
45+
46+// RequireAbsent adds the precondition that key does not exist. If it does
47+// not hold when the batch is committed, Apply returns ErrConflict.
48+func (b *Batch) RequireAbsent(key []byte) {
49+ if !b.validKey(key) {
50+ return
51+ }
52+ b.pre = append(b.pre, precond{key: append([]byte(nil), key...), absent: true})
53+}
54+
55+// RequireVersion adds the precondition that key exists at version v, the seq
56+// of the batch that last wrote it. v must be greater than zero, otherwise the
57+// batch records ErrInvalidKey.
58+func (b *Batch) RequireVersion(key []byte, v uint64) {
59+ if !b.validKey(key) {
60+ return
61+ }
62+ if v == 0 {
63+ b.fail(ErrInvalidKey)
64+ return
65+ }
66+ b.pre = append(b.pre, precond{key: append([]byte(nil), key...), version: v})
67+}
68+
69+// Len returns the number of ops and preconditions.
70+func (b *Batch) Len() int { return len(b.ops) + len(b.pre) }
71+
72+// Reset empties the batch, clears its error and keeps allocated capacity for
73+// the op list.
74+func (b *Batch) Reset() {
75+ clear(b.ops)
76+ b.ops = b.ops[:0]
77+ clear(b.pre)
78+ b.pre = b.pre[:0]
79+ b.opsBytes = 0
80+ b.err = nil
81+}
82+
83+// payload returns the encoded record payload size; 0 for a batch without
84+// ops, which writes nothing.
85+func (b *Batch) payload() int {
86+ if len(b.ops) == 0 {
87+ return 0
88+ }
89+ return recordMinBody + b.opsBytes
90+}
91+
92+func (b *Batch) add(o op) {
93+ b.ops = append(b.ops, o)
94+ b.opsBytes += opSize(o)
95+}
96+
97+func (b *Batch) fail(err error) {
98+ if b.err == nil {
99+ b.err = err
100+ }
101+}
102+
103+// validKey records ErrInvalidKey and reports false for an invalid key.
104+func (b *Batch) validKey(key []byte) bool {
105+ if len(key) == 0 || len(key) > maxKeyLen {
106+ b.fail(ErrInvalidKey)
107+ return false
108+ }
109+ return b.err == nil
110+}
@@ -0,0 +1,110 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+// Batch is an ordered set of mutations and preconditions that Apply commits
6+// atomically. The zero value is ready to use. A Batch must not be modified
7+// while Apply runs and is not safe for concurrent use.
8+//
9+// Batch methods do not return errors: the batch records its first error and
10+// Apply returns it.
11+type Batch struct {
12+ ops []op
13+ pre []precond
14+ opsBytes int // encoded size of ops
15+ err error
16+}
17+
18+// precond is a precondition checked by the writer before any op.
19+type precond struct {
20+ key []byte
21+ absent bool // the key must not exist
22+ version uint64 // otherwise the key must exist at this version
23+}
24+
25+// Put sets key to value. It copies key and value.
26+func (b *Batch) Put(key, value []byte) {
27+ if !b.validKey(key) {
28+ return
29+ }
30+ // One allocation holds both; the tree retains the slices, so they are
31+ // never reused (not even by Reset).
32+ buf := make([]byte, len(key)+len(value))
33+ n := copy(buf, key)
34+ copy(buf[n:], value)
35+ b.add(op{kind: opPut, key: buf[:n:n], value: buf[n:len(buf):len(buf)]})
36+}
37+
38+// Delete removes key. It copies key. Deleting a missing key is not an error.
39+func (b *Batch) Delete(key []byte) {
40+ if !b.validKey(key) {
41+ return
42+ }
43+ b.add(op{kind: opDelete, key: append([]byte(nil), key...)})
44+}
45+
46+// RequireAbsent adds the precondition that key does not exist. If it does
47+// not hold when the batch is committed, Apply returns ErrConflict.
48+func (b *Batch) RequireAbsent(key []byte) {
49+ if !b.validKey(key) {
50+ return
51+ }
52+ b.pre = append(b.pre, precond{key: append([]byte(nil), key...), absent: true})
53+}
54+
55+// RequireVersion adds the precondition that key exists at version v, the seq
56+// of the batch that last wrote it. v must be greater than zero, otherwise the
57+// batch records ErrInvalidKey.
58+func (b *Batch) RequireVersion(key []byte, v uint64) {
59+ if !b.validKey(key) {
60+ return
61+ }
62+ if v == 0 {
63+ b.fail(ErrInvalidKey)
64+ return
65+ }
66+ b.pre = append(b.pre, precond{key: append([]byte(nil), key...), version: v})
67+}
68+
69+// Len returns the number of ops and preconditions.
70+func (b *Batch) Len() int { return len(b.ops) + len(b.pre) }
71+
72+// Reset empties the batch, clears its error and keeps allocated capacity for
73+// the op list.
74+func (b *Batch) Reset() {
75+ clear(b.ops)
76+ b.ops = b.ops[:0]
77+ clear(b.pre)
78+ b.pre = b.pre[:0]
79+ b.opsBytes = 0
80+ b.err = nil
81+}
82+
83+// payload returns the encoded record payload size; 0 for a batch without
84+// ops, which writes nothing.
85+func (b *Batch) payload() int {
86+ if len(b.ops) == 0 {
87+ return 0
88+ }
89+ return recordMinBody + b.opsBytes
90+}
91+
92+func (b *Batch) add(o op) {
93+ b.ops = append(b.ops, o)
94+ b.opsBytes += opSize(o)
95+}
96+
97+func (b *Batch) fail(err error) {
98+ if b.err == nil {
99+ b.err = err
100+ }
101+}
102+
103+// validKey records ErrInvalidKey and reports false for an invalid key.
104+func (b *Batch) validKey(key []byte) bool {
105+ if len(key) == 0 || len(key) > maxKeyLen {
106+ b.fail(ErrInvalidKey)
107+ return false
108+ }
109+ return b.err == nil
110+}
A batch_test.go +83 -0
@@ -0,0 +1,83 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bytes"
7+ "errors"
8+ "strings"
9+ "testing"
10+)
11+
12+func TestBatchCopiesInputs(t *testing.T) {
13+ var b Batch
14+ key, val := []byte("key"), []byte("value")
15+ b.Put(key, val)
16+ b.Delete(key)
17+ b.RequireAbsent(key)
18+ b.RequireVersion(key, 3)
19+ key[0], val[0] = 'X', 'X'
20+ if got := b.ops[0]; string(got.key) != "key" || string(got.value) != "value" {
21+ t.Fatalf("put aliases its inputs: %q=%q", got.key, got.value)
22+ }
23+ if got := b.ops[1]; string(got.key) != "key" {
24+ t.Fatalf("delete aliases its key: %q", got.key)
25+ }
26+ for i, p := range b.pre {
27+ if string(p.key) != "key" {
28+ t.Fatalf("precondition %d aliases its key: %q", i, p.key)
29+ }
30+ }
31+ if b.Len() != 4 {
32+ t.Fatalf("Len = %d, want 4", b.Len())
33+ }
34+}
35+
36+func TestBatchErrors(t *testing.T) {
37+ long := bytes.Repeat([]byte("k"), maxKeyLen+1)
38+ cases := map[string]func(b *Batch){
39+ "put empty": func(b *Batch) { b.Put(nil, []byte("v")) },
40+ "put long": func(b *Batch) { b.Put(long, nil) },
41+ "delete empty": func(b *Batch) { b.Delete([]byte{}) },
42+ "absent empty": func(b *Batch) { b.RequireAbsent(nil) },
43+ "version empty": func(b *Batch) { b.RequireVersion(nil, 1) },
44+ "version zero": func(b *Batch) { b.RequireVersion([]byte("k"), 0) },
45+ "after valid op": func(b *Batch) { b.Put([]byte("a"), nil); b.Put(nil, nil); b.Put([]byte("b"), nil) },
46+ }
47+ for name, fn := range cases {
48+ t.Run(name, func(t *testing.T) {
49+ var b Batch
50+ fn(&b)
51+ if !errors.Is(b.err, ErrInvalidKey) {
52+ t.Fatalf("err = %v, want ErrInvalidKey", b.err)
53+ }
54+ b.Reset()
55+ if b.err != nil || b.Len() != 0 || b.payload() != 0 {
56+ t.Fatalf("Reset left err=%v len=%d payload=%d", b.err, b.Len(), b.payload())
57+ }
58+ })
59+ }
60+
61+ var b Batch
62+ b.Put(bytes.Repeat([]byte("k"), maxKeyLen), nil) // exactly the limit
63+ if b.err != nil {
64+ t.Fatalf("64 KiB key rejected: %v", b.err)
65+ }
66+}
67+
68+func TestBatchPayload(t *testing.T) {
69+ var b Batch
70+ if b.payload() != 0 {
71+ t.Fatal("empty batch has a payload")
72+ }
73+ b.RequireAbsent([]byte("k"))
74+ if b.payload() != 0 {
75+ t.Fatal("precondition-only batch has a payload")
76+ }
77+ b.Put([]byte("k"), []byte(strings.Repeat("v", 200)))
78+ b.Delete([]byte("kk"))
79+ want := len(appendRecord(nil, 1, b.ops)) - recordFrame
80+ if got := b.payload(); got != want {
81+ t.Fatalf("payload = %d, encoded record payload = %d", got, want)
82+ }
83+}
@@ -0,0 +1,83 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bytes"
7+ "errors"
8+ "strings"
9+ "testing"
10+)
11+
12+func TestBatchCopiesInputs(t *testing.T) {
13+ var b Batch
14+ key, val := []byte("key"), []byte("value")
15+ b.Put(key, val)
16+ b.Delete(key)
17+ b.RequireAbsent(key)
18+ b.RequireVersion(key, 3)
19+ key[0], val[0] = 'X', 'X'
20+ if got := b.ops[0]; string(got.key) != "key" || string(got.value) != "value" {
21+ t.Fatalf("put aliases its inputs: %q=%q", got.key, got.value)
22+ }
23+ if got := b.ops[1]; string(got.key) != "key" {
24+ t.Fatalf("delete aliases its key: %q", got.key)
25+ }
26+ for i, p := range b.pre {
27+ if string(p.key) != "key" {
28+ t.Fatalf("precondition %d aliases its key: %q", i, p.key)
29+ }
30+ }
31+ if b.Len() != 4 {
32+ t.Fatalf("Len = %d, want 4", b.Len())
33+ }
34+}
35+
36+func TestBatchErrors(t *testing.T) {
37+ long := bytes.Repeat([]byte("k"), maxKeyLen+1)
38+ cases := map[string]func(b *Batch){
39+ "put empty": func(b *Batch) { b.Put(nil, []byte("v")) },
40+ "put long": func(b *Batch) { b.Put(long, nil) },
41+ "delete empty": func(b *Batch) { b.Delete([]byte{}) },
42+ "absent empty": func(b *Batch) { b.RequireAbsent(nil) },
43+ "version empty": func(b *Batch) { b.RequireVersion(nil, 1) },
44+ "version zero": func(b *Batch) { b.RequireVersion([]byte("k"), 0) },
45+ "after valid op": func(b *Batch) { b.Put([]byte("a"), nil); b.Put(nil, nil); b.Put([]byte("b"), nil) },
46+ }
47+ for name, fn := range cases {
48+ t.Run(name, func(t *testing.T) {
49+ var b Batch
50+ fn(&b)
51+ if !errors.Is(b.err, ErrInvalidKey) {
52+ t.Fatalf("err = %v, want ErrInvalidKey", b.err)
53+ }
54+ b.Reset()
55+ if b.err != nil || b.Len() != 0 || b.payload() != 0 {
56+ t.Fatalf("Reset left err=%v len=%d payload=%d", b.err, b.Len(), b.payload())
57+ }
58+ })
59+ }
60+
61+ var b Batch
62+ b.Put(bytes.Repeat([]byte("k"), maxKeyLen), nil) // exactly the limit
63+ if b.err != nil {
64+ t.Fatalf("64 KiB key rejected: %v", b.err)
65+ }
66+}
67+
68+func TestBatchPayload(t *testing.T) {
69+ var b Batch
70+ if b.payload() != 0 {
71+ t.Fatal("empty batch has a payload")
72+ }
73+ b.RequireAbsent([]byte("k"))
74+ if b.payload() != 0 {
75+ t.Fatal("precondition-only batch has a payload")
76+ }
77+ b.Put([]byte("k"), []byte(strings.Repeat("v", 200)))
78+ b.Delete([]byte("kk"))
79+ want := len(appendRecord(nil, 1, b.ops)) - recordFrame
80+ if got := b.payload(); got != want {
81+ t.Fatalf("payload = %d, encoded record payload = %d", got, want)
82+ }
83+}
A db.go +224 -0
@@ -0,0 +1,224 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "context"
7+ "fmt"
8+ "io"
9+ "log/slog"
10+ "sync"
11+ "sync/atomic"
12+ "time"
13+
14+ "bnema.dev/ermite-kv/internal/fsys"
15+)
16+
17+// Defaults for zero Options fields.
18+const (
19+ defaultMaxBatchBytes = 16 << 20
20+ defaultCompactAfter = 64 << 20
21+)
22+
23+// Options configures Open. The zero value of every field means "default";
24+// Options{} is safe and durable.
25+type Options struct {
26+ // NoSync skips fdatasync on commits. For tests and benchmarks only: an
27+ // acknowledged batch may be lost on power failure.
28+ NoSync bool
29+ // MaxBatchBytes is the encoded batch limit. 0 means 16 MiB; it must be
30+ // at most 64 MiB (the format limit).
31+ MaxBatchBytes int
32+ // CompactAfter is the number of bytes in the current log that trigger
33+ // compaction. 0 means 64 MiB; a negative value disables compaction.
34+ CompactAfter int64
35+ // GroupWindow is the extra time the writer waits to grow a commit
36+ // group. 0 means it only drains what is already queued.
37+ GroupWindow time.Duration
38+ // Logger receives warnings and errors. nil discards them.
39+ Logger *slog.Logger
40+}
41+
42+// resolve validates o and fills in the defaults.
43+func (o Options) resolve() (Options, error) {
44+ switch {
45+ case o.MaxBatchBytes < 0 || o.MaxBatchBytes > maxRecord:
46+ return o, fmt.Errorf("kv: invalid MaxBatchBytes %d: must be in [0, %d]", o.MaxBatchBytes, maxRecord)
47+ case o.GroupWindow < 0:
48+ return o, fmt.Errorf("kv: invalid GroupWindow %v: must not be negative", o.GroupWindow)
49+ }
50+ if o.MaxBatchBytes == 0 {
51+ o.MaxBatchBytes = defaultMaxBatchBytes
52+ }
53+ if o.CompactAfter == 0 {
54+ o.CompactAfter = defaultCompactAfter
55+ }
56+ if o.Logger == nil {
57+ o.Logger = slog.New(slog.DiscardHandler)
58+ }
59+ return o, nil
60+}
61+
62+// DB is an embedded, durable, ordered key-value store. All methods are safe
63+// for concurrent use.
64+type DB struct {
65+ fx fsys.FS
66+ opts Options
67+ snap atomic.Pointer[Snapshot] // last published state
68+
69+ reqCh chan *request // never closed
70+ closing chan struct{} // closed by Close
71+ done chan struct{} // closed when the writer has exited
72+
73+ failed chan struct{} // closed when the database enters the failed state
74+ failOnce sync.Once
75+ failErr error // first cause, wrapping ErrFailed; set before failed is closed
76+
77+ closeOnce sync.Once
78+ closeErr error
79+
80+ lock io.Closer
81+
82+ // Owned by the writer goroutine (see writer.go).
83+ log fsys.File
84+ logBase uint64 // seq the current log starts after
85+ logSize int64 // bytes in the current log, header included
86+ buf []byte // group encoding buffer
87+ group []*request
88+}
89+
90+// Open opens the database in dir, creating it (mode 0700) if needed. It
91+// takes an exclusive lock on the directory, returning ErrLocked if another
92+// DB holds it, and recovers the state from the snapshot and logs. Invalid
93+// options are an error.
94+func Open(dir string, opts Options) (*DB, error) {
95+ return open(fsys.OS(), dir, opts)
96+}
97+
98+func open(fx fsys.FS, dir string, opts Options) (*DB, error) {
99+ opts, err := opts.resolve()
100+ if err != nil {
101+ return nil, err
102+ }
103+ bld := tree{}.builder()
104+ rec, err := recoverDir(fx, dir, opts.Logger,
105+ func(key, value []byte, version uint64) {
106+ // Snapshot entries alias one shared file buffer: copy them so
107+ // the tree does not keep the whole buffer alive.
108+ buf := make([]byte, len(key)+len(value))
109+ n := copy(buf, key)
110+ copy(buf[n:], value)
111+ bld.put(buf[:n:n], buf[n:len(buf):len(buf)], version)
112+ },
113+ func(seq uint64, ops []op) {
114+ for _, o := range ops {
115+ if o.kind == opPut {
116+ bld.put(o.key, o.value, seq)
117+ } else {
118+ bld.delete(o.key)
119+ }
120+ }
121+ })
122+ if err != nil {
123+ return nil, err
124+ }
125+ db := &DB{
126+ fx: fx,
127+ opts: opts,
128+ reqCh: make(chan *request),
129+ closing: make(chan struct{}),
130+ done: make(chan struct{}),
131+ failed: make(chan struct{}),
132+ lock: rec.lock,
133+ log: rec.log,
134+ logBase: rec.logBase,
135+ logSize: rec.logSize,
136+ }
137+ db.snap.Store(newSnapshot(bld.freeze(), rec.lastSeq))
138+ go db.writer()
139+ return db, nil
140+}
141+
142+// Close stops the database: requests not yet taken by the writer get
143+// ErrClosed, the group in progress is committed and acknowledged, then the
144+// log is closed and the directory lock released. Snapshots stay readable.
145+//
146+// Close returns the first error met while stopping, including the cause of
147+// the failed state if the database failed, or nil. It is idempotent: later
148+// calls return nil.
149+func (db *DB) Close() error {
150+ var err error
151+ db.closeOnce.Do(func() {
152+ close(db.closing)
153+ <-db.done // the writer finishes its group and exits
154+ // P1.D: wait here for an in-flight compaction.
155+ select {
156+ case <-db.failed:
157+ err = db.failErr
158+ default:
159+ }
160+ if cerr := db.log.Close(); err == nil {
161+ err = cerr
162+ }
163+ if cerr := db.lock.Close(); err == nil {
164+ err = cerr
165+ }
166+ })
167+ return err
168+}
169+
170+// Snapshot returns the current immutable state. It is lock-free and O(1), and
171+// remains usable after Close.
172+func (db *DB) Snapshot() *Snapshot { return db.snap.Load() }
173+
174+// Get returns the item stored under key in the current snapshot.
175+func (db *DB) Get(key []byte) (Item, bool) { return db.snap.Load().Get(key) }
176+
177+// Apply commits b atomically and returns its seq once it is durable (unless
178+// Options.NoSync) and visible in Snapshot.
179+//
180+// It returns the first error recorded by the batch (ErrInvalidKey), then
181+// ErrBatchTooLarge if the encoded batch exceeds Options.MaxBatchBytes, both
182+// without touching the database. ErrConflict (seq 0) means a precondition did
183+// not hold at commit time. A batch without puts and deletes writes nothing,
184+// consumes no seq and returns the current seq.
185+//
186+// If ctx ends before the writer takes the batch, Apply returns ctx.Err() and
187+// the batch is not applied. Once the writer has taken it, ctx is ignored and
188+// Apply waits for the outcome. ErrClosed is returned after Close. ErrFailed
189+// (wrapping the I/O error) means the database failed: for the batch that was
190+// being committed its outcome is unknown, since its bytes may be on disk and
191+// replayed by the next Open.
192+func (db *DB) Apply(ctx context.Context, b *Batch) (uint64, error) {
193+ if b.err != nil {
194+ return 0, b.err
195+ }
196+ if b.payload() > db.opts.MaxBatchBytes {
197+ return 0, ErrBatchTooLarge
198+ }
199+ select {
200+ case <-db.closing:
201+ return 0, ErrClosed
202+ case <-db.failed:
203+ return 0, db.failErr
204+ default:
205+ }
206+ if err := ctx.Err(); err != nil {
207+ return 0, err
208+ }
209+
210+ r := requestPool.Get().(*request)
211+ r.b = b
212+ select {
213+ case db.reqCh <- r:
214+ case <-ctx.Done():
215+ return 0, r.release(ctx.Err())
216+ case <-db.closing:
217+ return 0, r.release(ErrClosed)
218+ case <-db.failed:
219+ return 0, r.release(db.failErr)
220+ }
221+ res := <-r.reply // the writer always answers a request it has taken
222+ r.release(nil)
223+ return res.seq, res.err
224+}
@@ -0,0 +1,224 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "context"
7+ "fmt"
8+ "io"
9+ "log/slog"
10+ "sync"
11+ "sync/atomic"
12+ "time"
13+
14+ "bnema.dev/ermite-kv/internal/fsys"
15+)
16+
17+// Defaults for zero Options fields.
18+const (
19+ defaultMaxBatchBytes = 16 << 20
20+ defaultCompactAfter = 64 << 20
21+)
22+
23+// Options configures Open. The zero value of every field means "default";
24+// Options{} is safe and durable.
25+type Options struct {
26+ // NoSync skips fdatasync on commits. For tests and benchmarks only: an
27+ // acknowledged batch may be lost on power failure.
28+ NoSync bool
29+ // MaxBatchBytes is the encoded batch limit. 0 means 16 MiB; it must be
30+ // at most 64 MiB (the format limit).
31+ MaxBatchBytes int
32+ // CompactAfter is the number of bytes in the current log that trigger
33+ // compaction. 0 means 64 MiB; a negative value disables compaction.
34+ CompactAfter int64
35+ // GroupWindow is the extra time the writer waits to grow a commit
36+ // group. 0 means it only drains what is already queued.
37+ GroupWindow time.Duration
38+ // Logger receives warnings and errors. nil discards them.
39+ Logger *slog.Logger
40+}
41+
42+// resolve validates o and fills in the defaults.
43+func (o Options) resolve() (Options, error) {
44+ switch {
45+ case o.MaxBatchBytes < 0 || o.MaxBatchBytes > maxRecord:
46+ return o, fmt.Errorf("kv: invalid MaxBatchBytes %d: must be in [0, %d]", o.MaxBatchBytes, maxRecord)
47+ case o.GroupWindow < 0:
48+ return o, fmt.Errorf("kv: invalid GroupWindow %v: must not be negative", o.GroupWindow)
49+ }
50+ if o.MaxBatchBytes == 0 {
51+ o.MaxBatchBytes = defaultMaxBatchBytes
52+ }
53+ if o.CompactAfter == 0 {
54+ o.CompactAfter = defaultCompactAfter
55+ }
56+ if o.Logger == nil {
57+ o.Logger = slog.New(slog.DiscardHandler)
58+ }
59+ return o, nil
60+}
61+
62+// DB is an embedded, durable, ordered key-value store. All methods are safe
63+// for concurrent use.
64+type DB struct {
65+ fx fsys.FS
66+ opts Options
67+ snap atomic.Pointer[Snapshot] // last published state
68+
69+ reqCh chan *request // never closed
70+ closing chan struct{} // closed by Close
71+ done chan struct{} // closed when the writer has exited
72+
73+ failed chan struct{} // closed when the database enters the failed state
74+ failOnce sync.Once
75+ failErr error // first cause, wrapping ErrFailed; set before failed is closed
76+
77+ closeOnce sync.Once
78+ closeErr error
79+
80+ lock io.Closer
81+
82+ // Owned by the writer goroutine (see writer.go).
83+ log fsys.File
84+ logBase uint64 // seq the current log starts after
85+ logSize int64 // bytes in the current log, header included
86+ buf []byte // group encoding buffer
87+ group []*request
88+}
89+
90+// Open opens the database in dir, creating it (mode 0700) if needed. It
91+// takes an exclusive lock on the directory, returning ErrLocked if another
92+// DB holds it, and recovers the state from the snapshot and logs. Invalid
93+// options are an error.
94+func Open(dir string, opts Options) (*DB, error) {
95+ return open(fsys.OS(), dir, opts)
96+}
97+
98+func open(fx fsys.FS, dir string, opts Options) (*DB, error) {
99+ opts, err := opts.resolve()
100+ if err != nil {
101+ return nil, err
102+ }
103+ bld := tree{}.builder()
104+ rec, err := recoverDir(fx, dir, opts.Logger,
105+ func(key, value []byte, version uint64) {
106+ // Snapshot entries alias one shared file buffer: copy them so
107+ // the tree does not keep the whole buffer alive.
108+ buf := make([]byte, len(key)+len(value))
109+ n := copy(buf, key)
110+ copy(buf[n:], value)
111+ bld.put(buf[:n:n], buf[n:len(buf):len(buf)], version)
112+ },
113+ func(seq uint64, ops []op) {
114+ for _, o := range ops {
115+ if o.kind == opPut {
116+ bld.put(o.key, o.value, seq)
117+ } else {
118+ bld.delete(o.key)
119+ }
120+ }
121+ })
122+ if err != nil {
123+ return nil, err
124+ }
125+ db := &DB{
126+ fx: fx,
127+ opts: opts,
128+ reqCh: make(chan *request),
129+ closing: make(chan struct{}),
130+ done: make(chan struct{}),
131+ failed: make(chan struct{}),
132+ lock: rec.lock,
133+ log: rec.log,
134+ logBase: rec.logBase,
135+ logSize: rec.logSize,
136+ }
137+ db.snap.Store(newSnapshot(bld.freeze(), rec.lastSeq))
138+ go db.writer()
139+ return db, nil
140+}
141+
142+// Close stops the database: requests not yet taken by the writer get
143+// ErrClosed, the group in progress is committed and acknowledged, then the
144+// log is closed and the directory lock released. Snapshots stay readable.
145+//
146+// Close returns the first error met while stopping, including the cause of
147+// the failed state if the database failed, or nil. It is idempotent: later
148+// calls return nil.
149+func (db *DB) Close() error {
150+ var err error
151+ db.closeOnce.Do(func() {
152+ close(db.closing)
153+ <-db.done // the writer finishes its group and exits
154+ // P1.D: wait here for an in-flight compaction.
155+ select {
156+ case <-db.failed:
157+ err = db.failErr
158+ default:
159+ }
160+ if cerr := db.log.Close(); err == nil {
161+ err = cerr
162+ }
163+ if cerr := db.lock.Close(); err == nil {
164+ err = cerr
165+ }
166+ })
167+ return err
168+}
169+
170+// Snapshot returns the current immutable state. It is lock-free and O(1), and
171+// remains usable after Close.
172+func (db *DB) Snapshot() *Snapshot { return db.snap.Load() }
173+
174+// Get returns the item stored under key in the current snapshot.
175+func (db *DB) Get(key []byte) (Item, bool) { return db.snap.Load().Get(key) }
176+
177+// Apply commits b atomically and returns its seq once it is durable (unless
178+// Options.NoSync) and visible in Snapshot.
179+//
180+// It returns the first error recorded by the batch (ErrInvalidKey), then
181+// ErrBatchTooLarge if the encoded batch exceeds Options.MaxBatchBytes, both
182+// without touching the database. ErrConflict (seq 0) means a precondition did
183+// not hold at commit time. A batch without puts and deletes writes nothing,
184+// consumes no seq and returns the current seq.
185+//
186+// If ctx ends before the writer takes the batch, Apply returns ctx.Err() and
187+// the batch is not applied. Once the writer has taken it, ctx is ignored and
188+// Apply waits for the outcome. ErrClosed is returned after Close. ErrFailed
189+// (wrapping the I/O error) means the database failed: for the batch that was
190+// being committed its outcome is unknown, since its bytes may be on disk and
191+// replayed by the next Open.
192+func (db *DB) Apply(ctx context.Context, b *Batch) (uint64, error) {
193+ if b.err != nil {
194+ return 0, b.err
195+ }
196+ if b.payload() > db.opts.MaxBatchBytes {
197+ return 0, ErrBatchTooLarge
198+ }
199+ select {
200+ case <-db.closing:
201+ return 0, ErrClosed
202+ case <-db.failed:
203+ return 0, db.failErr
204+ default:
205+ }
206+ if err := ctx.Err(); err != nil {
207+ return 0, err
208+ }
209+
210+ r := requestPool.Get().(*request)
211+ r.b = b
212+ select {
213+ case db.reqCh <- r:
214+ case <-ctx.Done():
215+ return 0, r.release(ctx.Err())
216+ case <-db.closing:
217+ return 0, r.release(ErrClosed)
218+ case <-db.failed:
219+ return 0, r.release(db.failErr)
220+ }
221+ res := <-r.reply // the writer always answers a request it has taken
222+ r.release(nil)
223+ return res.seq, res.err
224+}
A db_test.go +1112 -0
@@ -0,0 +1,1112 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bytes"
7+ "context"
8+ "errors"
9+ "fmt"
10+ "log/slog"
11+ "math/rand/v2"
12+ "os"
13+ "path/filepath"
14+ "slices"
15+ "strconv"
16+ "strings"
17+ "sync"
18+ "sync/atomic"
19+ "testing"
20+ "time"
21+
22+ "bnema.dev/ermite-kv/internal/fsys"
23+ "bnema.dev/ermite-kv/internal/fsys/faultfs"
24+)
25+
26+var bg = context.Background()
27+
28+// mustOpen opens a database on fx and closes it when the test ends.
29+func mustOpen(t testing.TB, fx fsys.FS, opts Options) *DB {
30+ t.Helper()
31+ db, err := open(fx, testDir, opts)
32+ if err != nil {
33+ t.Fatalf("open: %v", err)
34+ }
35+ t.Cleanup(func() { _ = db.Close() })
36+ return db
37+}
38+
39+// putKV applies a single put and returns its seq.
40+func putKV(t testing.TB, db *DB, key, value string) uint64 {
41+ t.Helper()
42+ var b Batch
43+ b.Put([]byte(key), []byte(value))
44+ seq, err := db.Apply(bg, &b)
45+ if err != nil {
46+ t.Fatalf("put %q: %v", key, err)
47+ }
48+ return seq
49+}
50+
51+func mustGet(t testing.TB, s *Snapshot, key, want string) Item {
52+ t.Helper()
53+ it, ok := s.Get([]byte(key))
54+ if !ok || string(it.Value) != want {
55+ t.Fatalf("Get(%q) = %q, %v; want %q", key, it.Value, ok, want)
56+ }
57+ return it
58+}
59+
60+func mustMissing(t testing.TB, s *Snapshot, key string) {
61+ t.Helper()
62+ if it, ok := s.Get([]byte(key)); ok {
63+ t.Fatalf("Get(%q) = %q, want missing", key, it.Value)
64+ }
65+}
66+
67+func short(full, quick int) int {
68+ if testing.Short() {
69+ return quick
70+ }
71+ return full
72+}
73+
74+func TestReopenFaultfs(t *testing.T) {
75+ fx := faultfs.New()
76+ db := mustOpen(t, fx, Options{})
77+ if s := db.Snapshot(); s.Seq() != 0 || s.Len() != 0 {
78+ t.Fatalf("new db: seq %d len %d", s.Seq(), s.Len())
79+ }
80+ for i := range 50 {
81+ if seq := putKV(t, db, fmt.Sprintf("k%02d", i), fmt.Sprintf("v%d", i)); seq != uint64(i+1) {
82+ t.Fatalf("seq = %d, want %d", seq, i+1)
83+ }
84+ }
85+ var b Batch
86+ b.Delete([]byte("k00"))
87+ b.Put([]byte("k01"), []byte("again"))
88+ b.Put([]byte("k01"), []byte("last"))
89+ if seq, err := db.Apply(bg, &b); err != nil || seq != 51 {
90+ t.Fatalf("Apply = %d, %v", seq, err)
91+ }
92+ if err := db.Close(); err != nil {
93+ t.Fatalf("Close: %v", err)
94+ }
95+ if err := db.Close(); err != nil {
96+ t.Fatalf("second Close: %v", err)
97+ }
98+
99+ db = mustOpen(t, fx, Options{})
100+ s := db.Snapshot()
101+ if s.Seq() != 51 || s.Len() != 49 {
102+ t.Fatalf("reopened: seq %d len %d, want 51 and 49", s.Seq(), s.Len())
103+ }
104+ mustMissing(t, s, "k00")
105+ if it := mustGet(t, s, "k01", "last"); it.Version != 51 {
106+ t.Fatalf("k01 version %d, want 51", it.Version)
107+ }
108+ if it := mustGet(t, s, "k02", "v2"); it.Version != 3 {
109+ t.Fatalf("k02 version %d, want 3", it.Version)
110+ }
111+ if seq := putKV(t, db, "next", "x"); seq != 52 {
112+ t.Fatalf("seq after reopen = %d, want 52", seq)
113+ }
114+}
115+
116+func TestReopenOS(t *testing.T) {
117+ dir := filepath.Join(t.TempDir(), "sub", "db")
118+ db, err := Open(dir, Options{})
119+ if err != nil {
120+ t.Fatal(err)
121+ }
122+ for i := range 100 {
123+ putKV(t, db, fmt.Sprintf("k%03d", i), strings.Repeat("v", i))
124+ }
125+ if err := db.Close(); err != nil {
126+ t.Fatal(err)
127+ }
128+ if fi, err := os.Stat(dir); err != nil || fi.Mode().Perm() != 0o700 {
129+ t.Fatalf("dir stat: %v, %v", fi, err)
130+ }
131+
132+ db, err = Open(dir, Options{})
133+ if err != nil {
134+ t.Fatal(err)
135+ }
136+ defer db.Close()
137+ s := db.Snapshot()
138+ if s.Seq() != 100 || s.Len() != 100 {
139+ t.Fatalf("seq %d len %d, want 100", s.Seq(), s.Len())
140+ }
141+ for i := range 100 {
142+ mustGet(t, s, fmt.Sprintf("k%03d", i), strings.Repeat("v", i))
143+ }
144+ if got := db.Snapshot(); got != s {
145+ t.Fatal("Snapshot changed without a write")
146+ }
147+ if it, ok := db.Get([]byte("k005")); !ok || it.Version != 6 {
148+ t.Fatalf("DB.Get = %v, %v", it, ok)
149+ }
150+}
151+
152+// TestOpenFromSnapshot covers loading a snapshot file (written by the future
153+// compactor) and appending to the log created next to it.
154+func TestOpenFromSnapshot(t *testing.T) {
155+ var buf bytes.Buffer
156+ entries := func(yield func(k, v []byte, ver uint64) bool) {
157+ _ = yield([]byte("a"), []byte("1"), 3) && yield([]byte("b"), nil, 7) && yield([]byte("c"), []byte("3"), 9)
158+ }
159+ if err := writeSnapshot(&buf, 9, 3, entries); err != nil {
160+ t.Fatal(err)
161+ }
162+ fx := faultfs.New()
163+ fx.WriteFile(filepath.Join(testDir, snapshotName(9)), buf.Bytes())
164+
165+ db := mustOpen(t, fx, Options{})
166+ s := db.Snapshot()
167+ if s.Seq() != 9 || s.Len() != 3 {
168+ t.Fatalf("seq %d len %d", s.Seq(), s.Len())
169+ }
170+ if it := mustGet(t, s, "b", ""); it.Version != 7 {
171+ t.Fatalf("b = %+v", it)
172+ }
173+ if seq := putKV(t, db, "d", "4"); seq != 10 {
174+ t.Fatalf("seq = %d, want 10", seq)
175+ }
176+ if err := db.Close(); err != nil {
177+ t.Fatal(err)
178+ }
179+ db = mustOpen(t, fx, Options{})
180+ mustGet(t, db.Snapshot(), "d", "4")
181+ mustGet(t, db.Snapshot(), "c", "3")
182+ if db.Snapshot().Seq() != 10 {
183+ t.Fatalf("seq = %d", db.Snapshot().Seq())
184+ }
185+}
186+
187+func TestConcurrentWriters(t *testing.T) {
188+ fx := faultfs.New()
189+ db := mustOpen(t, fx, Options{})
190+ const writers = 64
191+ per := short(50, 10)
192+
193+ seqs := make([][]uint64, writers)
194+ var wg sync.WaitGroup
195+ for g := range writers {
196+ wg.Add(1)
197+ go func() {
198+ defer wg.Done()
199+ for n := range per {
200+ key := fmt.Sprintf("w%02d/%03d", g, n)
201+ var b Batch
202+ b.Put([]byte(key), []byte(key))
203+ seq, err := db.Apply(bg, &b)
204+ if err != nil {
205+ t.Errorf("writer %d: %v", g, err)
206+ return
207+ }
208+ if it, ok := db.Get([]byte(key)); !ok || it.Version < seq {
209+ t.Errorf("writer %d: no read-your-writes for %s at seq %d", g, key, seq)
210+ return
211+ }
212+ seqs[g] = append(seqs[g], seq)
213+ }
214+ }()
215+ }
216+ wg.Wait()
217+
218+ var all []uint64
219+ for g, s := range seqs {
220+ if !slices.IsSorted(s) || len(slices.Compact(slices.Clone(s))) != len(s) {
221+ t.Fatalf("writer %d seqs not strictly increasing: %v", g, s)
222+ }
223+ all = append(all, s...)
224+ }
225+ slices.Sort(all)
226+ for i, s := range all {
227+ if s != uint64(i+1) {
228+ t.Fatalf("seq %d at rank %d: seqs are not 1..%d", s, i, len(all))
229+ }
230+ }
231+ if got := db.Snapshot().Seq(); got != writers*uint64(per) {
232+ t.Fatalf("Seq = %d, want %d", got, writers*per)
233+ }
234+ if got := db.Snapshot().Len(); got != writers*per {
235+ t.Fatalf("Len = %d, want %d", got, writers*per)
236+ }
237+}
238+
239+func TestRequireAbsentRace(t *testing.T) {
240+ for _, window := range []time.Duration{0, 5 * time.Millisecond} {
241+ t.Run(window.String(), func(t *testing.T) {
242+ fx := faultfs.New()
243+ db := mustOpen(t, fx, Options{GroupWindow: window})
244+ const n = 64
245+ var wins atomic.Int32
246+ winner := make([]bool, n)
247+ var wg sync.WaitGroup
248+ start := make(chan struct{})
249+ for g := range n {
250+ wg.Add(1)
251+ go func() {
252+ defer wg.Done()
253+ var b Batch
254+ b.RequireAbsent([]byte("lock"))
255+ b.Put([]byte("lock"), []byte(strconv.Itoa(g)))
256+ <-start
257+ seq, err := db.Apply(bg, &b)
258+ switch {
259+ case err == nil && seq > 0:
260+ wins.Add(1)
261+ winner[g] = true
262+ case errors.Is(err, ErrConflict) && seq == 0:
263+ default:
264+ t.Errorf("writer %d: seq %d err %v", g, seq, err)
265+ }
266+ }()
267+ }
268+ close(start)
269+ wg.Wait()
270+ if wins.Load() != 1 {
271+ t.Fatalf("%d winners, want 1", wins.Load())
272+ }
273+ w := slices.Index(winner, true)
274+ mustGet(t, db.Snapshot(), "lock", strconv.Itoa(w))
275+ if db.Snapshot().Seq() != 1 {
276+ t.Fatalf("Seq = %d, want 1 (conflicts consume no seq)", db.Snapshot().Seq())
277+ }
278+ })
279+ }
280+}
281+
282+func TestRequireVersion(t *testing.T) {
283+ db := mustOpen(t, faultfs.New(), Options{})
284+ v1 := putKV(t, db, "k", "a")
285+ cas := func(key string, v uint64, val string) (uint64, error) {
286+ var b Batch
287+ b.RequireVersion([]byte(key), v)
288+ b.Put([]byte(key), []byte(val))
289+ return db.Apply(bg, &b)
290+ }
291+ v2, err := cas("k", v1, "b")
292+ if err != nil || v2 != v1+1 {
293+ t.Fatalf("cas at current version: %d, %v", v2, err)
294+ }
295+ if seq, err := cas("k", v1, "c"); !errors.Is(err, ErrConflict) || seq != 0 {
296+ t.Fatalf("cas at stale version: %d, %v", seq, err)
297+ }
298+ if seq, err := cas("missing", 1, "c"); !errors.Is(err, ErrConflict) || seq != 0 {
299+ t.Fatalf("cas on missing key: %d, %v", seq, err)
300+ }
301+ mustGet(t, db.Snapshot(), "k", "b")
302+ mustMissing(t, db.Snapshot(), "missing")
303+ if got := db.Snapshot().Seq(); got != v2 {
304+ t.Fatalf("Seq = %d, want %d", got, v2)
305+ }
306+}
307+
308+// TestPreconditionsBeforeOps checks that all preconditions are evaluated
309+// before any op of the same batch and that a conflict contributes nothing.
310+func TestPreconditionsBeforeOps(t *testing.T) {
311+ db := mustOpen(t, faultfs.New(), Options{})
312+ putKV(t, db, "exists", "1")
313+
314+ var b Batch
315+ b.Put([]byte("new"), []byte("x"))
316+ b.RequireAbsent([]byte("new")) // holds: checked before the put
317+ if _, err := db.Apply(bg, &b); err != nil {
318+ t.Fatalf("Apply: %v", err)
319+ }
320+
321+ b.Reset()
322+ b.Delete([]byte("exists"))
323+ b.Put([]byte("other"), []byte("x"))
324+ b.RequireAbsent([]byte("never"))
325+ b.RequireAbsent([]byte("exists")) // fails
326+ before := db.Snapshot()
327+ if seq, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) || seq != 0 {
328+ t.Fatalf("Apply = %d, %v; want ErrConflict", seq, err)
329+ }
330+ if db.Snapshot() != before {
331+ t.Fatal("a conflicting batch published a snapshot")
332+ }
333+ mustGet(t, db.Snapshot(), "exists", "1")
334+ mustMissing(t, db.Snapshot(), "other")
335+}
336+
337+// TestGroupPreconditions checks that preconditions see earlier batches of
338+// the same group: with a group window, the second batch must conflict.
339+func TestGroupPreconditions(t *testing.T) {
340+ fx := faultfs.New()
341+ db := mustOpen(t, fx, Options{GroupWindow: 200 * time.Millisecond})
342+ syncs := fx.Syncs()
343+ errs := make([]error, 2)
344+ var wg sync.WaitGroup
345+ for i := range errs {
346+ wg.Add(1)
347+ go func() {
348+ defer wg.Done()
349+ var b Batch
350+ b.RequireAbsent([]byte("k"))
351+ b.Put([]byte("k"), []byte{byte('a' + i)})
352+ _, errs[i] = db.Apply(bg, &b)
353+ }()
354+ }
355+ wg.Wait()
356+ if (errs[0] == nil) == (errs[1] == nil) {
357+ t.Fatalf("errors %v, want exactly one success", errs)
358+ }
359+ if fx.Syncs() != syncs+1 {
360+ t.Fatalf("%d syncs for the group, want 1", fx.Syncs()-syncs)
361+ }
362+}
363+
364+func TestEmptyAndConflictOnlyBatches(t *testing.T) {
365+ fx := faultfs.New()
366+ db := mustOpen(t, fx, Options{})
367+ putKV(t, db, "a", "1")
368+ putKV(t, db, "b", "2")
369+ writes, syncs := fx.Writes(), fx.Syncs()
370+ snap := db.Snapshot()
371+
372+ var empty Batch
373+ if seq, err := db.Apply(bg, &empty); err != nil || seq != 2 {
374+ t.Fatalf("empty batch: %d, %v; want current seq 2", seq, err)
375+ }
376+ var pre Batch
377+ pre.RequireVersion([]byte("a"), 1)
378+ pre.RequireAbsent([]byte("zzz"))
379+ if seq, err := db.Apply(bg, &pre); err != nil || seq != 2 {
380+ t.Fatalf("holding precondition-only batch: %d, %v; want seq 2", seq, err)
381+ }
382+ var bad Batch
383+ bad.RequireAbsent([]byte("a"))
384+ if seq, err := db.Apply(bg, &bad); !errors.Is(err, ErrConflict) || seq != 0 {
385+ t.Fatalf("failing precondition-only batch: %d, %v", seq, err)
386+ }
387+ var conflict Batch
388+ conflict.RequireAbsent([]byte("a"))
389+ conflict.Put([]byte("c"), nil)
390+ if _, err := db.Apply(bg, &conflict); !errors.Is(err, ErrConflict) {
391+ t.Fatalf("conflict: %v", err)
392+ }
393+ var del Batch // deleting a missing key still writes a record
394+ del.Delete([]byte("nope"))
395+ if seq, err := db.Apply(bg, &del); err != nil || seq != 3 {
396+ t.Fatalf("delete of a missing key: %d, %v; want seq 3", seq, err)
397+ }
398+
399+ // Only the last batch touched the log.
400+ if got := fx.Writes() - writes; got != 1 {
401+ t.Fatalf("%d writes, want 1", got)
402+ }
403+ if got := fx.Syncs() - syncs; got != 1 {
404+ t.Fatalf("%d syncs, want 1", got)
405+ }
406+ if db.Snapshot().Len() != snap.Len() || db.Snapshot().Seq() != 3 {
407+ t.Fatalf("snapshot after: seq %d len %d", db.Snapshot().Seq(), db.Snapshot().Len())
408+ }
409+ if err := db.Close(); err != nil {
410+ t.Fatal(err)
411+ }
412+ db = mustOpen(t, fx, Options{})
413+ if db.Snapshot().Seq() != 3 || db.Snapshot().Len() != 2 {
414+ t.Fatalf("reopened: seq %d len %d", db.Snapshot().Seq(), db.Snapshot().Len())
415+ }
416+}
417+
418+// TestNoWriteForSkippedGroup checks that a group of only conflicts and empty
419+// batches neither writes nor syncs.
420+func TestNoWriteForSkippedGroup(t *testing.T) {
421+ fx := faultfs.New()
422+ db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
423+ putKV(t, db, "a", "1")
424+ writes, syncs := fx.Writes(), fx.Syncs()
425+ var wg sync.WaitGroup
426+ for range 4 {
427+ wg.Go(func() {
428+ var b Batch
429+ b.RequireAbsent([]byte("a"))
430+ b.Put([]byte("x"), nil)
431+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) {
432+ t.Errorf("err = %v, want ErrConflict", err)
433+ }
434+ })
435+ }
436+ wg.Wait()
437+ if fx.Writes() != writes || fx.Syncs() != syncs {
438+ t.Fatalf("writes %d→%d syncs %d→%d", writes, fx.Writes(), syncs, fx.Syncs())
439+ }
440+}
441+
442+func TestGroupCommitSyncs(t *testing.T) {
443+ fx := faultfs.New()
444+ db := mustOpen(t, fx, Options{})
445+ fx.SetSyncDelay(time.Millisecond)
446+ base := fx.Syncs()
447+ const writers = 64
448+ per := short(100, 16)
449+ var wg sync.WaitGroup
450+ for g := range writers {
451+ wg.Go(func() {
452+ for n := range per {
453+ putKV(t, db, fmt.Sprintf("g%02d/%03d", g, n), "v")
454+ }
455+ })
456+ }
457+ wg.Wait()
458+ total := writers * per
459+ syncs := fx.Syncs() - base
460+ t.Logf("%d batches, %d syncs", total, syncs)
461+ if syncs >= total/4 {
462+ t.Fatalf("%d syncs for %d batches: no group commit", syncs, total)
463+ }
464+ if got := db.Snapshot().Len(); got != total {
465+ t.Fatalf("Len = %d, want %d", got, total)
466+ }
467+}
468+
469+func TestNoSync(t *testing.T) {
470+ fx := faultfs.New()
471+ db := mustOpen(t, fx, Options{NoSync: true})
472+ base := fx.Syncs()
473+ putKV(t, db, "a", "1")
474+ if fx.Syncs() != base {
475+ t.Fatal("NoSync still syncs")
476+ }
477+ if err := db.Close(); err != nil {
478+ t.Fatal(err)
479+ }
480+ mustGet(t, mustOpen(t, fx, Options{}).Snapshot(), "a", "1")
481+}
482+
483+// TestLargeBatches covers a batch above the 4 MiB group cap (a group of one)
484+// and a burst of batches that does not fit in one group.
485+func TestLargeBatches(t *testing.T) {
486+ fx := faultfs.New()
487+ db := mustOpen(t, fx, Options{GroupWindow: 50 * time.Millisecond})
488+ big := strings.Repeat("B", 5<<20)
489+ putKV(t, db, "big", big)
490+
491+ const n = 6
492+ chunk := strings.Repeat("c", 3<<19) // 1.5 MiB: three do not fit in 4 MiB
493+ seqs := make([]uint64, n)
494+ var wg sync.WaitGroup
495+ for i := range n {
496+ wg.Go(func() { seqs[i] = putKV(t, db, fmt.Sprintf("chunk%d", i), chunk) })
497+ }
498+ wg.Wait()
499+ slices.Sort(seqs)
500+ for i, s := range seqs {
501+ if s != uint64(i+2) {
502+ t.Fatalf("seqs = %v", seqs)
503+ }
504+ }
505+ if err := db.Close(); err != nil {
506+ t.Fatal(err)
507+ }
508+ s := mustOpen(t, fx, Options{}).Snapshot()
509+ mustGet(t, s, "big", big)
510+ for i := range n {
511+ mustGet(t, s, fmt.Sprintf("chunk%d", i), chunk)
512+ }
513+}
514+
515+func TestInvalidInputs(t *testing.T) {
516+ fx := faultfs.New()
517+ db := mustOpen(t, fx, Options{MaxBatchBytes: 100})
518+
519+ var b Batch
520+ b.Put(nil, []byte("v"))
521+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrInvalidKey) {
522+ t.Fatalf("empty key: %v", err)
523+ }
524+ b.Reset()
525+ b.Put(bytes.Repeat([]byte("k"), maxKeyLen+1), nil)
526+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrInvalidKey) {
527+ t.Fatalf("long key: %v", err)
528+ }
529+ b.Reset()
530+ b.RequireVersion([]byte("k"), 0)
531+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrInvalidKey) {
532+ t.Fatalf("version 0: %v", err)
533+ }
534+
535+ // Payload = 12 (seq, count) + 1 (kind) + 1 (klen) + 1 (key) + 1 (vlen) + v.
536+ b.Reset()
537+ b.Put([]byte("k"), make([]byte, 85))
538+ if seq, err := db.Apply(bg, &b); !errors.Is(err, ErrBatchTooLarge) || seq != 0 {
539+ t.Fatalf("101-byte payload: %d, %v", seq, err)
540+ }
541+ b.Reset()
542+ b.Put([]byte("k"), make([]byte, 84))
543+ if _, err := db.Apply(bg, &b); err != nil {
544+ t.Fatalf("100-byte payload: %v", err)
545+ }
546+ if db.Snapshot().Seq() != 1 {
547+ t.Fatalf("Seq = %d: rejected batches reached the log", db.Snapshot().Seq())
548+ }
549+}
550+
551+func TestOptionValidation(t *testing.T) {
552+ bad := []Options{
553+ {MaxBatchBytes: maxRecord + 1},
554+ {MaxBatchBytes: -1},
555+ {GroupWindow: -time.Second},
556+ }
557+ for _, o := range bad {
558+ fx := faultfs.New()
559+ if db, err := open(fx, testDir, o); err == nil {
560+ _ = db.Close()
561+ t.Errorf("open(%+v) succeeded", o)
562+ }
563+ if _, err := fx.ReadDir(testDir); err == nil {
564+ t.Errorf("open(%+v) created the directory", o)
565+ }
566+ }
567+ if _, err := Open(t.TempDir(), Options{MaxBatchBytes: 65 << 20}); err == nil {
568+ t.Error("Open with MaxBatchBytes > 64 MiB succeeded")
569+ }
570+ for _, o := range []Options{{MaxBatchBytes: maxRecord}, {CompactAfter: -1}, {NoSync: true, CompactAfter: 1}} {
571+ db, err := open(faultfs.New(), testDir, o)
572+ if err != nil {
573+ t.Errorf("open(%+v): %v", o, err)
574+ continue
575+ }
576+ _ = db.Close()
577+ }
578+}
579+
580+func TestLocked(t *testing.T) {
581+ t.Run("faultfs", func(t *testing.T) {
582+ fx := faultfs.New()
583+ db := mustOpen(t, fx, Options{})
584+ if _, err := open(fx, testDir, Options{}); !errors.Is(err, ErrLocked) {
585+ t.Fatalf("second open: %v, want ErrLocked", err)
586+ }
587+ if err := db.Close(); err != nil {
588+ t.Fatal(err)
589+ }
590+ mustOpen(t, fx, Options{})
591+ })
592+ t.Run("os", func(t *testing.T) {
593+ dir := t.TempDir()
594+ db, err := Open(dir, Options{})
595+ if err != nil {
596+ t.Fatal(err)
597+ }
598+ if _, err := Open(dir, Options{}); !errors.Is(err, ErrLocked) {
599+ t.Fatalf("second Open: %v, want ErrLocked", err)
600+ }
601+ putKV(t, db, "still", "works")
602+ if err := db.Close(); err != nil {
603+ t.Fatal(err)
604+ }
605+ db, err = Open(dir, Options{})
606+ if err != nil {
607+ t.Fatalf("Open after Close: %v", err)
608+ }
609+ db.Close()
610+ })
611+}
612+
613+func TestApplyAfterClose(t *testing.T) {
614+ db := mustOpen(t, faultfs.New(), Options{})
615+ putKV(t, db, "a", "1")
616+ if err := db.Close(); err != nil {
617+ t.Fatal(err)
618+ }
619+ var b Batch
620+ b.Put([]byte("b"), nil)
621+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrClosed) {
622+ t.Fatalf("Apply after Close: %v", err)
623+ }
624+ mustGet(t, db.Snapshot(), "a", "1") // reads keep working
625+ if err := db.Close(); err != nil {
626+ t.Fatalf("second Close: %v", err)
627+ }
628+}
629+
630+// TestContextBeforeDequeue blocks the writer in a slow sync so that a second
631+// batch cannot be dequeued, then cancels its context.
632+func TestContextBeforeDequeue(t *testing.T) {
633+ fx := faultfs.New()
634+ db := mustOpen(t, fx, Options{})
635+
636+ done := make(chan error, 1)
637+ base := fx.Syncs()
638+ fx.SetSyncDelay(300 * time.Millisecond)
639+ go func() {
640+ var b Batch
641+ b.Put([]byte("first"), []byte("1"))
642+ _, err := db.Apply(bg, &b)
643+ done <- err
644+ }()
645+ for fx.Syncs() == base { // the writer is now inside the delayed sync
646+ time.Sleep(time.Millisecond)
647+ }
648+
649+ ctx, cancel := context.WithCancel(bg)
650+ second := make(chan error, 1)
651+ go func() {
652+ var b Batch
653+ b.Put([]byte("second"), []byte("2"))
654+ _, err := db.Apply(ctx, &b)
655+ second <- err
656+ }()
657+ time.Sleep(10 * time.Millisecond)
658+ cancel()
659+ if err := <-second; !errors.Is(err, context.Canceled) {
660+ t.Fatalf("Apply = %v, want context.Canceled", err)
661+ }
662+ if err := <-done; err != nil {
663+ t.Fatalf("first Apply: %v", err)
664+ }
665+ fx.SetSyncDelay(0)
666+
667+ // An already cancelled context never reaches the writer either.
668+ var b Batch
669+ b.Put([]byte("third"), nil)
670+ if _, err := db.Apply(ctx, &b); !errors.Is(err, context.Canceled) {
671+ t.Fatalf("Apply with cancelled ctx = %v", err)
672+ }
673+ putKV(t, db, "fourth", "4") // the writer is still healthy
674+ if err := db.Close(); err != nil {
675+ t.Fatal(err)
676+ }
677+ s := mustOpen(t, fx, Options{}).Snapshot()
678+ mustGet(t, s, "first", "1")
679+ mustGet(t, s, "fourth", "4")
680+ mustMissing(t, s, "second")
681+ mustMissing(t, s, "third")
682+ if s.Seq() != 2 {
683+ t.Fatalf("Seq = %d, want 2", s.Seq())
684+ }
685+}
686+
687+func TestCloseDuringApplies(t *testing.T) {
688+ for i := range short(10, 3) {
689+ t.Run(strconv.Itoa(i), func(t *testing.T) {
690+ fx := faultfs.New()
691+ fx.SetSyncDelay(100 * time.Microsecond)
692+ db := mustOpen(t, fx, Options{})
693+ const writers = 32
694+ acked := make([]uint64, writers) // highest acknowledged n+1
695+ var wg sync.WaitGroup
696+ for g := range writers {
697+ wg.Go(func() {
698+ for n := 0; ; n++ {
699+ var b Batch
700+ b.Put(fmt.Appendf(nil, "g%02d", g), strconv.AppendInt(nil, int64(n), 10))
701+ _, err := db.Apply(bg, &b)
702+ if errors.Is(err, ErrClosed) {
703+ return
704+ }
705+ if err != nil {
706+ t.Errorf("writer %d: %v", g, err)
707+ return
708+ }
709+ acked[g] = uint64(n) + 1
710+ }
711+ })
712+ }
713+ time.Sleep(time.Duration(2+i) * time.Millisecond)
714+ closed := make(chan error, 1)
715+ go func() { closed <- db.Close() }()
716+ select {
717+ case err := <-closed:
718+ if err != nil {
719+ t.Errorf("Close: %v", err)
720+ }
721+ case <-time.After(10 * time.Second):
722+ t.Fatal("Close did not return")
723+ }
724+ wg.Wait()
725+
726+ fx.SetSyncDelay(0)
727+ s := mustOpen(t, fx, Options{}).Snapshot()
728+ for g, n := range acked {
729+ if n == 0 {
730+ continue
731+ }
732+ it, ok := s.Get(fmt.Appendf(nil, "g%02d", g))
733+ got, _ := strconv.Atoi(string(it.Value))
734+ // The batch after the last acknowledged one may have been
735+ // committed by the final group.
736+ if !ok || uint64(got)+1 < n || uint64(got) > n {
737+ t.Errorf("writer %d: acked %d batches, recovered n=%d (%v)", g, n, got, ok)
738+ }
739+ }
740+ })
741+ }
742+}
743+
744+// failure is one injected I/O fault and what a plain reopen (no power loss)
745+// must find of the failed group.
746+type failure struct {
747+ name string
748+ inject func(fx *faultfs.FS)
749+ kept bool // the group's bytes reach the file even though the call failed
750+ cause error
751+}
752+
753+func TestInjectedFailures(t *testing.T) {
754+ failures := []failure{
755+ {"write", func(fx *faultfs.FS) { fx.FailWrite(1, nil) }, false, faultfs.ErrInjected},
756+ {"short write", func(fx *faultfs.FS) { fx.ShortWrite(1, 10) }, false, nil},
757+ {"sync", func(fx *faultfs.FS) { fx.FailSync(1, nil) }, true, faultfs.ErrInjected},
758+ }
759+ for _, f := range failures {
760+ for _, powerLoss := range []bool{false, true} {
761+ t.Run(fmt.Sprintf("%s/powerloss=%v", f.name, powerLoss), func(t *testing.T) {
762+ testInjectedFailure(t, f, powerLoss)
763+ })
764+ }
765+ }
766+}
767+
768+func testInjectedFailure(t *testing.T, f failure, powerLoss bool) {
769+ fx := faultfs.New()
770+ logger, logs := captureLogger()
771+ db := mustOpen(t, fx, Options{GroupWindow: 150 * time.Millisecond, Logger: logger})
772+ putKV(t, db, "before", "1")
773+ f.inject(fx)
774+
775+ // Three batches form the failed group (the window gathers them).
776+ const group = 3
777+ errs := make([]error, group)
778+ var wg sync.WaitGroup
779+ for i := range group {
780+ wg.Go(func() {
781+ var b Batch
782+ b.Put([]byte(fmt.Sprintf("failed%d", i)), []byte("x"))
783+ _, errs[i] = db.Apply(bg, &b)
784+ })
785+ }
786+ wg.Wait()
787+ for i, err := range errs {
788+ if !errors.Is(err, ErrFailed) {
789+ t.Fatalf("batch %d: %v, want ErrFailed", i, err)
790+ }
791+ if f.cause != nil && !errors.Is(err, f.cause) {
792+ t.Fatalf("batch %d: %v does not wrap %v", i, err, f.cause)
793+ }
794+ }
795+ if !strings.Contains(logs.String(), "database failed") {
796+ t.Errorf("failure not logged: %q", logs.String())
797+ }
798+
799+ // Every later Apply fails at once; reads serve the last snapshot.
800+ for range 3 {
801+ var b Batch
802+ b.Put([]byte("later"), nil)
803+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrFailed) {
804+ t.Fatalf("later Apply: %v, want ErrFailed", err)
805+ }
806+ }
807+ var empty Batch
808+ if _, err := db.Apply(bg, &empty); !errors.Is(err, ErrFailed) {
809+ t.Fatalf("empty Apply on a failed db: %v", err)
810+ }
811+ s := db.Snapshot()
812+ if s.Seq() != 1 || s.Len() != 1 {
813+ t.Fatalf("snapshot after failure: seq %d len %d", s.Seq(), s.Len())
814+ }
815+ mustGet(t, s, "before", "1")
816+ mustMissing(t, s, "failed0")
817+ if err := db.Close(); !errors.Is(err, ErrFailed) {
818+ t.Fatalf("Close = %v, want the failure", err)
819+ }
820+ if err := db.Close(); err != nil {
821+ t.Fatalf("second Close = %v", err)
822+ }
823+
824+ if powerLoss {
825+ fx.PowerLoss(faultfs.PowerLossOptions{})
826+ }
827+ s = mustOpen(t, fx, Options{}).Snapshot()
828+ mustGet(t, s, "before", "1")
829+ present := 0
830+ for i := range group {
831+ if _, ok := s.Get(fmt.Appendf(nil, "failed%d", i)); ok {
832+ present++
833+ }
834+ }
835+ want := 0
836+ if f.kept && !powerLoss {
837+ want = group
838+ }
839+ if present != want {
840+ t.Fatalf("%d of %d failed batches recovered, want %d", present, group, want)
841+ }
842+ if s.Seq() != uint64(1+present) {
843+ t.Fatalf("Seq = %d, want %d", s.Seq(), 1+present)
844+ }
845+}
846+
847+// TestFailedWriterAnswersOverflow checks that no Apply hangs when the group
848+// that fails leaves a dequeued request behind.
849+func TestFailedWriterAnswersOverflow(t *testing.T) {
850+ fx := faultfs.New()
851+ db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
852+ fx.FailWrite(1, nil)
853+ chunk := strings.Repeat("c", 3<<20) // two do not fit in one group
854+ errs := make([]error, 4)
855+ var wg sync.WaitGroup
856+ for i := range errs {
857+ wg.Go(func() {
858+ var b Batch
859+ b.Put(fmt.Appendf(nil, "k%d", i), []byte(chunk))
860+ _, errs[i] = db.Apply(bg, &b)
861+ })
862+ }
863+ wg.Wait()
864+ for i, err := range errs {
865+ if !errors.Is(err, ErrFailed) {
866+ t.Errorf("batch %d: %v", i, err)
867+ }
868+ }
869+}
870+
871+// captureLogger returns a logger writing to a buffer.
872+func captureLogger() (*slog.Logger, *syncBuffer) {
873+ var sb syncBuffer
874+ return slog.New(slog.NewTextHandler(&sb, nil)), &sb
875+}
876+
877+type syncBuffer struct {
878+ mu sync.Mutex
879+ b strings.Builder
880+}
881+
882+func (s *syncBuffer) Write(p []byte) (int, error) {
883+ s.mu.Lock()
884+ defer s.mu.Unlock()
885+ return s.b.Write(p)
886+}
887+
888+func (s *syncBuffer) String() string {
889+ s.mu.Lock()
890+ defer s.mu.Unlock()
891+ return s.b.String()
892+}
893+
894+// powerLossBase returns the first seed; ERMITE_KV_SEED reproduces a run.
895+func powerLossBase(t *testing.T) uint64 {
896+ if v := os.Getenv("ERMITE_KV_SEED"); v != "" {
897+ seed, err := strconv.ParseUint(v, 10, 64)
898+ if err != nil {
899+ t.Fatalf("ERMITE_KV_SEED: %v", err)
900+ }
901+ return seed
902+ }
903+ return uint64(time.Now().UnixNano())
904+}
905+
906+// TestPowerLoss cuts the power at a random point while writers run, then
907+// checks AC2: every acknowledged batch is present and the recovered state is
908+// exactly a prefix of the applied batches.
909+//
910+// Every writer g applies batches n = 0, 1, … that set cur/g = n and
911+// all/g/n, and delete all/g/(n-1). The recovered state therefore says how
912+// many batches of each writer were applied, and the seq must equal their sum.
913+// (PowerLossOptions.KeepPages is not exercised: see the open question about
914+// recovery rule 6 under out-of-order page writeback.)
915+func TestPowerLoss(t *testing.T) {
916+ modes := []struct {
917+ name string
918+ opts faultfs.PowerLossOptions
919+ }{
920+ {"drop", faultfs.PowerLossOptions{}},
921+ {"zerofill", faultfs.PowerLossOptions{ZeroFill: true}},
922+ }
923+ base := powerLossBase(t)
924+ iterations := short(150, 10)
925+ for _, m := range modes {
926+ t.Run(m.name, func(t *testing.T) {
927+ for i := range iterations {
928+ seed := base + uint64(i)
929+ if !runPowerLoss(t, seed, m.opts) {
930+ t.Fatalf("failed with seed %d (ERMITE_KV_SEED=%d reproduces it)", seed, seed)
931+ }
932+ }
933+ })
934+ }
935+}
936+
937+func runPowerLoss(t *testing.T, seed uint64, plo faultfs.PowerLossOptions) (ok bool) {
938+ rng := rand.New(rand.NewPCG(seed, 0x9e3779b97f4a7c15))
939+ fx := faultfs.New()
940+ db, err := open(fx, testDir, Options{GroupWindow: time.Duration(rng.IntN(3)) * 100 * time.Microsecond})
941+ if err != nil {
942+ t.Errorf("seed %d: open: %v", seed, err)
943+ return false
944+ }
945+ fx.SetSyncDelay(time.Duration(rng.IntN(300)) * time.Microsecond)
946+ writers := 1 + rng.IntN(8)
947+ acked := make([]int, writers) // last acknowledged n, -1 if none
948+ ackSeq := make([]uint64, writers)
949+ var wg sync.WaitGroup
950+ for g := range writers {
951+ acked[g] = -1
952+ wg.Go(func() {
953+ for n := 0; ; n++ {
954+ var b Batch
955+ b.Put(fmt.Appendf(nil, "cur/%d", g), strconv.AppendInt(nil, int64(n), 10))
956+ b.Put(fmt.Appendf(nil, "all/%d/%d", g, n), nil)
957+ if n > 0 {
958+ b.Delete(fmt.Appendf(nil, "all/%d/%d", g, n-1))
959+ }
960+ seq, err := db.Apply(bg, &b)
961+ if err != nil {
962+ return
963+ }
964+ acked[g], ackSeq[g] = n, seq
965+ }
966+ })
967+ }
968+ time.Sleep(time.Duration(rng.IntN(4000)) * time.Microsecond)
969+ fx.PowerLoss(plo)
970+ wg.Wait()
971+ _ = db.Close()
972+ fx.SetSyncDelay(0)
973+
974+ db, err = open(fx, testDir, Options{})
975+ if err != nil {
976+ t.Errorf("seed %d: reopen after power loss: %v", seed, err)
977+ return false
978+ }
979+ defer db.Close()
980+ if !checkPowerLossState(t, seed, db.Snapshot(), acked, ackSeq) {
981+ return false
982+ }
983+
984+ // The recovered log accepts new records and survives another reopen.
985+ seq := db.Snapshot().Seq()
986+ if got := putKV(t, db, "after", "x"); got != seq+1 {
987+ t.Errorf("seed %d: seq after recovery = %d, want %d", seed, got, seq+1)
988+ return false
989+ }
990+ if err := db.Close(); err != nil {
991+ t.Errorf("seed %d: Close: %v", seed, err)
992+ return false
993+ }
994+ db, err = open(fx, testDir, Options{})
995+ if err != nil {
996+ t.Errorf("seed %d: second reopen: %v", seed, err)
997+ return false
998+ }
999+ defer db.Close()
1000+ if db.Snapshot().Seq() != seq+1 {
1001+ t.Errorf("seed %d: seq after second reopen = %d, want %d", seed, db.Snapshot().Seq(), seq+1)
1002+ return false
1003+ }
1004+ return true
1005+}
1006+
1007+func checkPowerLossState(t *testing.T, seed uint64, s *Snapshot, acked []int, ackSeq []uint64) bool {
1008+ fail := func(format string, args ...any) bool {
1009+ t.Errorf("seed %d: "+format, append([]any{seed}, args...)...)
1010+ return false
1011+ }
1012+ var applied uint64
1013+ keys := 0
1014+ for g := range acked {
1015+ it, ok := s.Get(fmt.Appendf(nil, "cur/%d", g))
1016+ last := -1
1017+ if ok {
1018+ n, err := strconv.Atoi(string(it.Value))
1019+ if err != nil {
1020+ return fail("writer %d: bad cur value %q", g, it.Value)
1021+ }
1022+ last = n
1023+ }
1024+ if last < acked[g] {
1025+ return fail("writer %d: acknowledged batch %d lost, recovered %d", g, acked[g], last)
1026+ }
1027+ if last == acked[g] && last >= 0 && it.Version != ackSeq[g] {
1028+ return fail("writer %d: batch %d acknowledged at seq %d, recovered at %d", g, last, ackSeq[g], it.Version)
1029+ }
1030+ if !ok {
1031+ continue
1032+ }
1033+ cur, ok := s.Get(fmt.Appendf(nil, "all/%d/%d", g, last))
1034+ if !ok || cur.Version != it.Version {
1035+ return fail("writer %d: batch %d is partially applied (%v, %v)", g, last, cur, ok)
1036+ }
1037+ count := 0
1038+ for range s.Prefix(fmt.Appendf(nil, "all/%d/", g)) {
1039+ count++
1040+ }
1041+ if count != 1 {
1042+ return fail("writer %d: %d all/ keys, want 1", g, count)
1043+ }
1044+ applied += uint64(last) + 1
1045+ keys += 2
1046+ }
1047+ if s.Seq() != applied {
1048+ return fail("Seq = %d but %d batches are visible: not a prefix", s.Seq(), applied)
1049+ }
1050+ if s.Len() != keys {
1051+ return fail("Len = %d, want %d", s.Len(), keys)
1052+ }
1053+ return true
1054+}
1055+
1056+// countFS counts the fdatasync calls of the files it opens.
1057+type countFS struct {
1058+ fsys.FS
1059+ syncs atomic.Int64
1060+}
1061+
1062+func (c *countFS) OpenFile(name string, flag int, perm os.FileMode) (fsys.File, error) {
1063+ f, err := c.FS.OpenFile(name, flag, perm)
1064+ if err != nil {
1065+ return nil, err
1066+ }
1067+ return countFile{f, &c.syncs}, nil
1068+}
1069+
1070+type countFile struct {
1071+ fsys.File
1072+ n *atomic.Int64
1073+}
1074+
1075+func (f countFile) Sync() error {
1076+ f.n.Add(1)
1077+ return f.File.Sync()
1078+}
1079+
1080+// BenchmarkApply measures group-commit throughput with parallel writers on a
1081+// real directory (NoSync=false). The syncs are counted by a thin wrapper
1082+// around the OS file system, so the numbers are real fdatasync calls.
1083+func BenchmarkApply(b *testing.B) {
1084+ fx := &countFS{FS: fsys.OS()}
1085+ db, err := open(fx, filepath.Join(b.TempDir(), "db"), Options{})
1086+ if err != nil {
1087+ b.Fatal(err)
1088+ }
1089+ defer db.Close()
1090+ b.SetParallelism(16)
1091+ value := bytes.Repeat([]byte("v"), 100)
1092+ var next atomic.Uint64
1093+ b.ReportAllocs()
1094+ b.ResetTimer()
1095+ b.RunParallel(func(pb *testing.PB) {
1096+ var batch Batch
1097+ for pb.Next() {
1098+ batch.Reset()
1099+ batch.Put(fmt.Appendf(nil, "key-%d", next.Add(1)), value)
1100+ if _, err := db.Apply(bg, &batch); err != nil {
1101+ b.Error(err)
1102+ return
1103+ }
1104+ }
1105+ })
1106+ b.StopTimer()
1107+ secs := b.Elapsed().Seconds()
1108+ syncs := float64(fx.syncs.Load())
1109+ b.ReportMetric(float64(b.N)/secs, "batches/s")
1110+ b.ReportMetric(syncs/secs, "syncs/s")
1111+ b.ReportMetric(float64(b.N)/syncs, "batches/sync")
1112+}
@@ -0,0 +1,1112 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "bytes"
7+ "context"
8+ "errors"
9+ "fmt"
10+ "log/slog"
11+ "math/rand/v2"
12+ "os"
13+ "path/filepath"
14+ "slices"
15+ "strconv"
16+ "strings"
17+ "sync"
18+ "sync/atomic"
19+ "testing"
20+ "time"
21+
22+ "bnema.dev/ermite-kv/internal/fsys"
23+ "bnema.dev/ermite-kv/internal/fsys/faultfs"
24+)
25+
26+var bg = context.Background()
27+
28+// mustOpen opens a database on fx and closes it when the test ends.
29+func mustOpen(t testing.TB, fx fsys.FS, opts Options) *DB {
30+ t.Helper()
31+ db, err := open(fx, testDir, opts)
32+ if err != nil {
33+ t.Fatalf("open: %v", err)
34+ }
35+ t.Cleanup(func() { _ = db.Close() })
36+ return db
37+}
38+
39+// putKV applies a single put and returns its seq.
40+func putKV(t testing.TB, db *DB, key, value string) uint64 {
41+ t.Helper()
42+ var b Batch
43+ b.Put([]byte(key), []byte(value))
44+ seq, err := db.Apply(bg, &b)
45+ if err != nil {
46+ t.Fatalf("put %q: %v", key, err)
47+ }
48+ return seq
49+}
50+
51+func mustGet(t testing.TB, s *Snapshot, key, want string) Item {
52+ t.Helper()
53+ it, ok := s.Get([]byte(key))
54+ if !ok || string(it.Value) != want {
55+ t.Fatalf("Get(%q) = %q, %v; want %q", key, it.Value, ok, want)
56+ }
57+ return it
58+}
59+
60+func mustMissing(t testing.TB, s *Snapshot, key string) {
61+ t.Helper()
62+ if it, ok := s.Get([]byte(key)); ok {
63+ t.Fatalf("Get(%q) = %q, want missing", key, it.Value)
64+ }
65+}
66+
67+func short(full, quick int) int {
68+ if testing.Short() {
69+ return quick
70+ }
71+ return full
72+}
73+
74+func TestReopenFaultfs(t *testing.T) {
75+ fx := faultfs.New()
76+ db := mustOpen(t, fx, Options{})
77+ if s := db.Snapshot(); s.Seq() != 0 || s.Len() != 0 {
78+ t.Fatalf("new db: seq %d len %d", s.Seq(), s.Len())
79+ }
80+ for i := range 50 {
81+ if seq := putKV(t, db, fmt.Sprintf("k%02d", i), fmt.Sprintf("v%d", i)); seq != uint64(i+1) {
82+ t.Fatalf("seq = %d, want %d", seq, i+1)
83+ }
84+ }
85+ var b Batch
86+ b.Delete([]byte("k00"))
87+ b.Put([]byte("k01"), []byte("again"))
88+ b.Put([]byte("k01"), []byte("last"))
89+ if seq, err := db.Apply(bg, &b); err != nil || seq != 51 {
90+ t.Fatalf("Apply = %d, %v", seq, err)
91+ }
92+ if err := db.Close(); err != nil {
93+ t.Fatalf("Close: %v", err)
94+ }
95+ if err := db.Close(); err != nil {
96+ t.Fatalf("second Close: %v", err)
97+ }
98+
99+ db = mustOpen(t, fx, Options{})
100+ s := db.Snapshot()
101+ if s.Seq() != 51 || s.Len() != 49 {
102+ t.Fatalf("reopened: seq %d len %d, want 51 and 49", s.Seq(), s.Len())
103+ }
104+ mustMissing(t, s, "k00")
105+ if it := mustGet(t, s, "k01", "last"); it.Version != 51 {
106+ t.Fatalf("k01 version %d, want 51", it.Version)
107+ }
108+ if it := mustGet(t, s, "k02", "v2"); it.Version != 3 {
109+ t.Fatalf("k02 version %d, want 3", it.Version)
110+ }
111+ if seq := putKV(t, db, "next", "x"); seq != 52 {
112+ t.Fatalf("seq after reopen = %d, want 52", seq)
113+ }
114+}
115+
116+func TestReopenOS(t *testing.T) {
117+ dir := filepath.Join(t.TempDir(), "sub", "db")
118+ db, err := Open(dir, Options{})
119+ if err != nil {
120+ t.Fatal(err)
121+ }
122+ for i := range 100 {
123+ putKV(t, db, fmt.Sprintf("k%03d", i), strings.Repeat("v", i))
124+ }
125+ if err := db.Close(); err != nil {
126+ t.Fatal(err)
127+ }
128+ if fi, err := os.Stat(dir); err != nil || fi.Mode().Perm() != 0o700 {
129+ t.Fatalf("dir stat: %v, %v", fi, err)
130+ }
131+
132+ db, err = Open(dir, Options{})
133+ if err != nil {
134+ t.Fatal(err)
135+ }
136+ defer db.Close()
137+ s := db.Snapshot()
138+ if s.Seq() != 100 || s.Len() != 100 {
139+ t.Fatalf("seq %d len %d, want 100", s.Seq(), s.Len())
140+ }
141+ for i := range 100 {
142+ mustGet(t, s, fmt.Sprintf("k%03d", i), strings.Repeat("v", i))
143+ }
144+ if got := db.Snapshot(); got != s {
145+ t.Fatal("Snapshot changed without a write")
146+ }
147+ if it, ok := db.Get([]byte("k005")); !ok || it.Version != 6 {
148+ t.Fatalf("DB.Get = %v, %v", it, ok)
149+ }
150+}
151+
152+// TestOpenFromSnapshot covers loading a snapshot file (written by the future
153+// compactor) and appending to the log created next to it.
154+func TestOpenFromSnapshot(t *testing.T) {
155+ var buf bytes.Buffer
156+ entries := func(yield func(k, v []byte, ver uint64) bool) {
157+ _ = yield([]byte("a"), []byte("1"), 3) && yield([]byte("b"), nil, 7) && yield([]byte("c"), []byte("3"), 9)
158+ }
159+ if err := writeSnapshot(&buf, 9, 3, entries); err != nil {
160+ t.Fatal(err)
161+ }
162+ fx := faultfs.New()
163+ fx.WriteFile(filepath.Join(testDir, snapshotName(9)), buf.Bytes())
164+
165+ db := mustOpen(t, fx, Options{})
166+ s := db.Snapshot()
167+ if s.Seq() != 9 || s.Len() != 3 {
168+ t.Fatalf("seq %d len %d", s.Seq(), s.Len())
169+ }
170+ if it := mustGet(t, s, "b", ""); it.Version != 7 {
171+ t.Fatalf("b = %+v", it)
172+ }
173+ if seq := putKV(t, db, "d", "4"); seq != 10 {
174+ t.Fatalf("seq = %d, want 10", seq)
175+ }
176+ if err := db.Close(); err != nil {
177+ t.Fatal(err)
178+ }
179+ db = mustOpen(t, fx, Options{})
180+ mustGet(t, db.Snapshot(), "d", "4")
181+ mustGet(t, db.Snapshot(), "c", "3")
182+ if db.Snapshot().Seq() != 10 {
183+ t.Fatalf("seq = %d", db.Snapshot().Seq())
184+ }
185+}
186+
187+func TestConcurrentWriters(t *testing.T) {
188+ fx := faultfs.New()
189+ db := mustOpen(t, fx, Options{})
190+ const writers = 64
191+ per := short(50, 10)
192+
193+ seqs := make([][]uint64, writers)
194+ var wg sync.WaitGroup
195+ for g := range writers {
196+ wg.Add(1)
197+ go func() {
198+ defer wg.Done()
199+ for n := range per {
200+ key := fmt.Sprintf("w%02d/%03d", g, n)
201+ var b Batch
202+ b.Put([]byte(key), []byte(key))
203+ seq, err := db.Apply(bg, &b)
204+ if err != nil {
205+ t.Errorf("writer %d: %v", g, err)
206+ return
207+ }
208+ if it, ok := db.Get([]byte(key)); !ok || it.Version < seq {
209+ t.Errorf("writer %d: no read-your-writes for %s at seq %d", g, key, seq)
210+ return
211+ }
212+ seqs[g] = append(seqs[g], seq)
213+ }
214+ }()
215+ }
216+ wg.Wait()
217+
218+ var all []uint64
219+ for g, s := range seqs {
220+ if !slices.IsSorted(s) || len(slices.Compact(slices.Clone(s))) != len(s) {
221+ t.Fatalf("writer %d seqs not strictly increasing: %v", g, s)
222+ }
223+ all = append(all, s...)
224+ }
225+ slices.Sort(all)
226+ for i, s := range all {
227+ if s != uint64(i+1) {
228+ t.Fatalf("seq %d at rank %d: seqs are not 1..%d", s, i, len(all))
229+ }
230+ }
231+ if got := db.Snapshot().Seq(); got != writers*uint64(per) {
232+ t.Fatalf("Seq = %d, want %d", got, writers*per)
233+ }
234+ if got := db.Snapshot().Len(); got != writers*per {
235+ t.Fatalf("Len = %d, want %d", got, writers*per)
236+ }
237+}
238+
239+func TestRequireAbsentRace(t *testing.T) {
240+ for _, window := range []time.Duration{0, 5 * time.Millisecond} {
241+ t.Run(window.String(), func(t *testing.T) {
242+ fx := faultfs.New()
243+ db := mustOpen(t, fx, Options{GroupWindow: window})
244+ const n = 64
245+ var wins atomic.Int32
246+ winner := make([]bool, n)
247+ var wg sync.WaitGroup
248+ start := make(chan struct{})
249+ for g := range n {
250+ wg.Add(1)
251+ go func() {
252+ defer wg.Done()
253+ var b Batch
254+ b.RequireAbsent([]byte("lock"))
255+ b.Put([]byte("lock"), []byte(strconv.Itoa(g)))
256+ <-start
257+ seq, err := db.Apply(bg, &b)
258+ switch {
259+ case err == nil && seq > 0:
260+ wins.Add(1)
261+ winner[g] = true
262+ case errors.Is(err, ErrConflict) && seq == 0:
263+ default:
264+ t.Errorf("writer %d: seq %d err %v", g, seq, err)
265+ }
266+ }()
267+ }
268+ close(start)
269+ wg.Wait()
270+ if wins.Load() != 1 {
271+ t.Fatalf("%d winners, want 1", wins.Load())
272+ }
273+ w := slices.Index(winner, true)
274+ mustGet(t, db.Snapshot(), "lock", strconv.Itoa(w))
275+ if db.Snapshot().Seq() != 1 {
276+ t.Fatalf("Seq = %d, want 1 (conflicts consume no seq)", db.Snapshot().Seq())
277+ }
278+ })
279+ }
280+}
281+
282+func TestRequireVersion(t *testing.T) {
283+ db := mustOpen(t, faultfs.New(), Options{})
284+ v1 := putKV(t, db, "k", "a")
285+ cas := func(key string, v uint64, val string) (uint64, error) {
286+ var b Batch
287+ b.RequireVersion([]byte(key), v)
288+ b.Put([]byte(key), []byte(val))
289+ return db.Apply(bg, &b)
290+ }
291+ v2, err := cas("k", v1, "b")
292+ if err != nil || v2 != v1+1 {
293+ t.Fatalf("cas at current version: %d, %v", v2, err)
294+ }
295+ if seq, err := cas("k", v1, "c"); !errors.Is(err, ErrConflict) || seq != 0 {
296+ t.Fatalf("cas at stale version: %d, %v", seq, err)
297+ }
298+ if seq, err := cas("missing", 1, "c"); !errors.Is(err, ErrConflict) || seq != 0 {
299+ t.Fatalf("cas on missing key: %d, %v", seq, err)
300+ }
301+ mustGet(t, db.Snapshot(), "k", "b")
302+ mustMissing(t, db.Snapshot(), "missing")
303+ if got := db.Snapshot().Seq(); got != v2 {
304+ t.Fatalf("Seq = %d, want %d", got, v2)
305+ }
306+}
307+
308+// TestPreconditionsBeforeOps checks that all preconditions are evaluated
309+// before any op of the same batch and that a conflict contributes nothing.
310+func TestPreconditionsBeforeOps(t *testing.T) {
311+ db := mustOpen(t, faultfs.New(), Options{})
312+ putKV(t, db, "exists", "1")
313+
314+ var b Batch
315+ b.Put([]byte("new"), []byte("x"))
316+ b.RequireAbsent([]byte("new")) // holds: checked before the put
317+ if _, err := db.Apply(bg, &b); err != nil {
318+ t.Fatalf("Apply: %v", err)
319+ }
320+
321+ b.Reset()
322+ b.Delete([]byte("exists"))
323+ b.Put([]byte("other"), []byte("x"))
324+ b.RequireAbsent([]byte("never"))
325+ b.RequireAbsent([]byte("exists")) // fails
326+ before := db.Snapshot()
327+ if seq, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) || seq != 0 {
328+ t.Fatalf("Apply = %d, %v; want ErrConflict", seq, err)
329+ }
330+ if db.Snapshot() != before {
331+ t.Fatal("a conflicting batch published a snapshot")
332+ }
333+ mustGet(t, db.Snapshot(), "exists", "1")
334+ mustMissing(t, db.Snapshot(), "other")
335+}
336+
337+// TestGroupPreconditions checks that preconditions see earlier batches of
338+// the same group: with a group window, the second batch must conflict.
339+func TestGroupPreconditions(t *testing.T) {
340+ fx := faultfs.New()
341+ db := mustOpen(t, fx, Options{GroupWindow: 200 * time.Millisecond})
342+ syncs := fx.Syncs()
343+ errs := make([]error, 2)
344+ var wg sync.WaitGroup
345+ for i := range errs {
346+ wg.Add(1)
347+ go func() {
348+ defer wg.Done()
349+ var b Batch
350+ b.RequireAbsent([]byte("k"))
351+ b.Put([]byte("k"), []byte{byte('a' + i)})
352+ _, errs[i] = db.Apply(bg, &b)
353+ }()
354+ }
355+ wg.Wait()
356+ if (errs[0] == nil) == (errs[1] == nil) {
357+ t.Fatalf("errors %v, want exactly one success", errs)
358+ }
359+ if fx.Syncs() != syncs+1 {
360+ t.Fatalf("%d syncs for the group, want 1", fx.Syncs()-syncs)
361+ }
362+}
363+
364+func TestEmptyAndConflictOnlyBatches(t *testing.T) {
365+ fx := faultfs.New()
366+ db := mustOpen(t, fx, Options{})
367+ putKV(t, db, "a", "1")
368+ putKV(t, db, "b", "2")
369+ writes, syncs := fx.Writes(), fx.Syncs()
370+ snap := db.Snapshot()
371+
372+ var empty Batch
373+ if seq, err := db.Apply(bg, &empty); err != nil || seq != 2 {
374+ t.Fatalf("empty batch: %d, %v; want current seq 2", seq, err)
375+ }
376+ var pre Batch
377+ pre.RequireVersion([]byte("a"), 1)
378+ pre.RequireAbsent([]byte("zzz"))
379+ if seq, err := db.Apply(bg, &pre); err != nil || seq != 2 {
380+ t.Fatalf("holding precondition-only batch: %d, %v; want seq 2", seq, err)
381+ }
382+ var bad Batch
383+ bad.RequireAbsent([]byte("a"))
384+ if seq, err := db.Apply(bg, &bad); !errors.Is(err, ErrConflict) || seq != 0 {
385+ t.Fatalf("failing precondition-only batch: %d, %v", seq, err)
386+ }
387+ var conflict Batch
388+ conflict.RequireAbsent([]byte("a"))
389+ conflict.Put([]byte("c"), nil)
390+ if _, err := db.Apply(bg, &conflict); !errors.Is(err, ErrConflict) {
391+ t.Fatalf("conflict: %v", err)
392+ }
393+ var del Batch // deleting a missing key still writes a record
394+ del.Delete([]byte("nope"))
395+ if seq, err := db.Apply(bg, &del); err != nil || seq != 3 {
396+ t.Fatalf("delete of a missing key: %d, %v; want seq 3", seq, err)
397+ }
398+
399+ // Only the last batch touched the log.
400+ if got := fx.Writes() - writes; got != 1 {
401+ t.Fatalf("%d writes, want 1", got)
402+ }
403+ if got := fx.Syncs() - syncs; got != 1 {
404+ t.Fatalf("%d syncs, want 1", got)
405+ }
406+ if db.Snapshot().Len() != snap.Len() || db.Snapshot().Seq() != 3 {
407+ t.Fatalf("snapshot after: seq %d len %d", db.Snapshot().Seq(), db.Snapshot().Len())
408+ }
409+ if err := db.Close(); err != nil {
410+ t.Fatal(err)
411+ }
412+ db = mustOpen(t, fx, Options{})
413+ if db.Snapshot().Seq() != 3 || db.Snapshot().Len() != 2 {
414+ t.Fatalf("reopened: seq %d len %d", db.Snapshot().Seq(), db.Snapshot().Len())
415+ }
416+}
417+
418+// TestNoWriteForSkippedGroup checks that a group of only conflicts and empty
419+// batches neither writes nor syncs.
420+func TestNoWriteForSkippedGroup(t *testing.T) {
421+ fx := faultfs.New()
422+ db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
423+ putKV(t, db, "a", "1")
424+ writes, syncs := fx.Writes(), fx.Syncs()
425+ var wg sync.WaitGroup
426+ for range 4 {
427+ wg.Go(func() {
428+ var b Batch
429+ b.RequireAbsent([]byte("a"))
430+ b.Put([]byte("x"), nil)
431+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) {
432+ t.Errorf("err = %v, want ErrConflict", err)
433+ }
434+ })
435+ }
436+ wg.Wait()
437+ if fx.Writes() != writes || fx.Syncs() != syncs {
438+ t.Fatalf("writes %d→%d syncs %d→%d", writes, fx.Writes(), syncs, fx.Syncs())
439+ }
440+}
441+
442+func TestGroupCommitSyncs(t *testing.T) {
443+ fx := faultfs.New()
444+ db := mustOpen(t, fx, Options{})
445+ fx.SetSyncDelay(time.Millisecond)
446+ base := fx.Syncs()
447+ const writers = 64
448+ per := short(100, 16)
449+ var wg sync.WaitGroup
450+ for g := range writers {
451+ wg.Go(func() {
452+ for n := range per {
453+ putKV(t, db, fmt.Sprintf("g%02d/%03d", g, n), "v")
454+ }
455+ })
456+ }
457+ wg.Wait()
458+ total := writers * per
459+ syncs := fx.Syncs() - base
460+ t.Logf("%d batches, %d syncs", total, syncs)
461+ if syncs >= total/4 {
462+ t.Fatalf("%d syncs for %d batches: no group commit", syncs, total)
463+ }
464+ if got := db.Snapshot().Len(); got != total {
465+ t.Fatalf("Len = %d, want %d", got, total)
466+ }
467+}
468+
469+func TestNoSync(t *testing.T) {
470+ fx := faultfs.New()
471+ db := mustOpen(t, fx, Options{NoSync: true})
472+ base := fx.Syncs()
473+ putKV(t, db, "a", "1")
474+ if fx.Syncs() != base {
475+ t.Fatal("NoSync still syncs")
476+ }
477+ if err := db.Close(); err != nil {
478+ t.Fatal(err)
479+ }
480+ mustGet(t, mustOpen(t, fx, Options{}).Snapshot(), "a", "1")
481+}
482+
483+// TestLargeBatches covers a batch above the 4 MiB group cap (a group of one)
484+// and a burst of batches that does not fit in one group.
485+func TestLargeBatches(t *testing.T) {
486+ fx := faultfs.New()
487+ db := mustOpen(t, fx, Options{GroupWindow: 50 * time.Millisecond})
488+ big := strings.Repeat("B", 5<<20)
489+ putKV(t, db, "big", big)
490+
491+ const n = 6
492+ chunk := strings.Repeat("c", 3<<19) // 1.5 MiB: three do not fit in 4 MiB
493+ seqs := make([]uint64, n)
494+ var wg sync.WaitGroup
495+ for i := range n {
496+ wg.Go(func() { seqs[i] = putKV(t, db, fmt.Sprintf("chunk%d", i), chunk) })
497+ }
498+ wg.Wait()
499+ slices.Sort(seqs)
500+ for i, s := range seqs {
501+ if s != uint64(i+2) {
502+ t.Fatalf("seqs = %v", seqs)
503+ }
504+ }
505+ if err := db.Close(); err != nil {
506+ t.Fatal(err)
507+ }
508+ s := mustOpen(t, fx, Options{}).Snapshot()
509+ mustGet(t, s, "big", big)
510+ for i := range n {
511+ mustGet(t, s, fmt.Sprintf("chunk%d", i), chunk)
512+ }
513+}
514+
515+func TestInvalidInputs(t *testing.T) {
516+ fx := faultfs.New()
517+ db := mustOpen(t, fx, Options{MaxBatchBytes: 100})
518+
519+ var b Batch
520+ b.Put(nil, []byte("v"))
521+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrInvalidKey) {
522+ t.Fatalf("empty key: %v", err)
523+ }
524+ b.Reset()
525+ b.Put(bytes.Repeat([]byte("k"), maxKeyLen+1), nil)
526+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrInvalidKey) {
527+ t.Fatalf("long key: %v", err)
528+ }
529+ b.Reset()
530+ b.RequireVersion([]byte("k"), 0)
531+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrInvalidKey) {
532+ t.Fatalf("version 0: %v", err)
533+ }
534+
535+ // Payload = 12 (seq, count) + 1 (kind) + 1 (klen) + 1 (key) + 1 (vlen) + v.
536+ b.Reset()
537+ b.Put([]byte("k"), make([]byte, 85))
538+ if seq, err := db.Apply(bg, &b); !errors.Is(err, ErrBatchTooLarge) || seq != 0 {
539+ t.Fatalf("101-byte payload: %d, %v", seq, err)
540+ }
541+ b.Reset()
542+ b.Put([]byte("k"), make([]byte, 84))
543+ if _, err := db.Apply(bg, &b); err != nil {
544+ t.Fatalf("100-byte payload: %v", err)
545+ }
546+ if db.Snapshot().Seq() != 1 {
547+ t.Fatalf("Seq = %d: rejected batches reached the log", db.Snapshot().Seq())
548+ }
549+}
550+
551+func TestOptionValidation(t *testing.T) {
552+ bad := []Options{
553+ {MaxBatchBytes: maxRecord + 1},
554+ {MaxBatchBytes: -1},
555+ {GroupWindow: -time.Second},
556+ }
557+ for _, o := range bad {
558+ fx := faultfs.New()
559+ if db, err := open(fx, testDir, o); err == nil {
560+ _ = db.Close()
561+ t.Errorf("open(%+v) succeeded", o)
562+ }
563+ if _, err := fx.ReadDir(testDir); err == nil {
564+ t.Errorf("open(%+v) created the directory", o)
565+ }
566+ }
567+ if _, err := Open(t.TempDir(), Options{MaxBatchBytes: 65 << 20}); err == nil {
568+ t.Error("Open with MaxBatchBytes > 64 MiB succeeded")
569+ }
570+ for _, o := range []Options{{MaxBatchBytes: maxRecord}, {CompactAfter: -1}, {NoSync: true, CompactAfter: 1}} {
571+ db, err := open(faultfs.New(), testDir, o)
572+ if err != nil {
573+ t.Errorf("open(%+v): %v", o, err)
574+ continue
575+ }
576+ _ = db.Close()
577+ }
578+}
579+
580+func TestLocked(t *testing.T) {
581+ t.Run("faultfs", func(t *testing.T) {
582+ fx := faultfs.New()
583+ db := mustOpen(t, fx, Options{})
584+ if _, err := open(fx, testDir, Options{}); !errors.Is(err, ErrLocked) {
585+ t.Fatalf("second open: %v, want ErrLocked", err)
586+ }
587+ if err := db.Close(); err != nil {
588+ t.Fatal(err)
589+ }
590+ mustOpen(t, fx, Options{})
591+ })
592+ t.Run("os", func(t *testing.T) {
593+ dir := t.TempDir()
594+ db, err := Open(dir, Options{})
595+ if err != nil {
596+ t.Fatal(err)
597+ }
598+ if _, err := Open(dir, Options{}); !errors.Is(err, ErrLocked) {
599+ t.Fatalf("second Open: %v, want ErrLocked", err)
600+ }
601+ putKV(t, db, "still", "works")
602+ if err := db.Close(); err != nil {
603+ t.Fatal(err)
604+ }
605+ db, err = Open(dir, Options{})
606+ if err != nil {
607+ t.Fatalf("Open after Close: %v", err)
608+ }
609+ db.Close()
610+ })
611+}
612+
613+func TestApplyAfterClose(t *testing.T) {
614+ db := mustOpen(t, faultfs.New(), Options{})
615+ putKV(t, db, "a", "1")
616+ if err := db.Close(); err != nil {
617+ t.Fatal(err)
618+ }
619+ var b Batch
620+ b.Put([]byte("b"), nil)
621+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrClosed) {
622+ t.Fatalf("Apply after Close: %v", err)
623+ }
624+ mustGet(t, db.Snapshot(), "a", "1") // reads keep working
625+ if err := db.Close(); err != nil {
626+ t.Fatalf("second Close: %v", err)
627+ }
628+}
629+
630+// TestContextBeforeDequeue blocks the writer in a slow sync so that a second
631+// batch cannot be dequeued, then cancels its context.
632+func TestContextBeforeDequeue(t *testing.T) {
633+ fx := faultfs.New()
634+ db := mustOpen(t, fx, Options{})
635+
636+ done := make(chan error, 1)
637+ base := fx.Syncs()
638+ fx.SetSyncDelay(300 * time.Millisecond)
639+ go func() {
640+ var b Batch
641+ b.Put([]byte("first"), []byte("1"))
642+ _, err := db.Apply(bg, &b)
643+ done <- err
644+ }()
645+ for fx.Syncs() == base { // the writer is now inside the delayed sync
646+ time.Sleep(time.Millisecond)
647+ }
648+
649+ ctx, cancel := context.WithCancel(bg)
650+ second := make(chan error, 1)
651+ go func() {
652+ var b Batch
653+ b.Put([]byte("second"), []byte("2"))
654+ _, err := db.Apply(ctx, &b)
655+ second <- err
656+ }()
657+ time.Sleep(10 * time.Millisecond)
658+ cancel()
659+ if err := <-second; !errors.Is(err, context.Canceled) {
660+ t.Fatalf("Apply = %v, want context.Canceled", err)
661+ }
662+ if err := <-done; err != nil {
663+ t.Fatalf("first Apply: %v", err)
664+ }
665+ fx.SetSyncDelay(0)
666+
667+ // An already cancelled context never reaches the writer either.
668+ var b Batch
669+ b.Put([]byte("third"), nil)
670+ if _, err := db.Apply(ctx, &b); !errors.Is(err, context.Canceled) {
671+ t.Fatalf("Apply with cancelled ctx = %v", err)
672+ }
673+ putKV(t, db, "fourth", "4") // the writer is still healthy
674+ if err := db.Close(); err != nil {
675+ t.Fatal(err)
676+ }
677+ s := mustOpen(t, fx, Options{}).Snapshot()
678+ mustGet(t, s, "first", "1")
679+ mustGet(t, s, "fourth", "4")
680+ mustMissing(t, s, "second")
681+ mustMissing(t, s, "third")
682+ if s.Seq() != 2 {
683+ t.Fatalf("Seq = %d, want 2", s.Seq())
684+ }
685+}
686+
687+func TestCloseDuringApplies(t *testing.T) {
688+ for i := range short(10, 3) {
689+ t.Run(strconv.Itoa(i), func(t *testing.T) {
690+ fx := faultfs.New()
691+ fx.SetSyncDelay(100 * time.Microsecond)
692+ db := mustOpen(t, fx, Options{})
693+ const writers = 32
694+ acked := make([]uint64, writers) // highest acknowledged n+1
695+ var wg sync.WaitGroup
696+ for g := range writers {
697+ wg.Go(func() {
698+ for n := 0; ; n++ {
699+ var b Batch
700+ b.Put(fmt.Appendf(nil, "g%02d", g), strconv.AppendInt(nil, int64(n), 10))
701+ _, err := db.Apply(bg, &b)
702+ if errors.Is(err, ErrClosed) {
703+ return
704+ }
705+ if err != nil {
706+ t.Errorf("writer %d: %v", g, err)
707+ return
708+ }
709+ acked[g] = uint64(n) + 1
710+ }
711+ })
712+ }
713+ time.Sleep(time.Duration(2+i) * time.Millisecond)
714+ closed := make(chan error, 1)
715+ go func() { closed <- db.Close() }()
716+ select {
717+ case err := <-closed:
718+ if err != nil {
719+ t.Errorf("Close: %v", err)
720+ }
721+ case <-time.After(10 * time.Second):
722+ t.Fatal("Close did not return")
723+ }
724+ wg.Wait()
725+
726+ fx.SetSyncDelay(0)
727+ s := mustOpen(t, fx, Options{}).Snapshot()
728+ for g, n := range acked {
729+ if n == 0 {
730+ continue
731+ }
732+ it, ok := s.Get(fmt.Appendf(nil, "g%02d", g))
733+ got, _ := strconv.Atoi(string(it.Value))
734+ // The batch after the last acknowledged one may have been
735+ // committed by the final group.
736+ if !ok || uint64(got)+1 < n || uint64(got) > n {
737+ t.Errorf("writer %d: acked %d batches, recovered n=%d (%v)", g, n, got, ok)
738+ }
739+ }
740+ })
741+ }
742+}
743+
744+// failure is one injected I/O fault and what a plain reopen (no power loss)
745+// must find of the failed group.
746+type failure struct {
747+ name string
748+ inject func(fx *faultfs.FS)
749+ kept bool // the group's bytes reach the file even though the call failed
750+ cause error
751+}
752+
753+func TestInjectedFailures(t *testing.T) {
754+ failures := []failure{
755+ {"write", func(fx *faultfs.FS) { fx.FailWrite(1, nil) }, false, faultfs.ErrInjected},
756+ {"short write", func(fx *faultfs.FS) { fx.ShortWrite(1, 10) }, false, nil},
757+ {"sync", func(fx *faultfs.FS) { fx.FailSync(1, nil) }, true, faultfs.ErrInjected},
758+ }
759+ for _, f := range failures {
760+ for _, powerLoss := range []bool{false, true} {
761+ t.Run(fmt.Sprintf("%s/powerloss=%v", f.name, powerLoss), func(t *testing.T) {
762+ testInjectedFailure(t, f, powerLoss)
763+ })
764+ }
765+ }
766+}
767+
768+func testInjectedFailure(t *testing.T, f failure, powerLoss bool) {
769+ fx := faultfs.New()
770+ logger, logs := captureLogger()
771+ db := mustOpen(t, fx, Options{GroupWindow: 150 * time.Millisecond, Logger: logger})
772+ putKV(t, db, "before", "1")
773+ f.inject(fx)
774+
775+ // Three batches form the failed group (the window gathers them).
776+ const group = 3
777+ errs := make([]error, group)
778+ var wg sync.WaitGroup
779+ for i := range group {
780+ wg.Go(func() {
781+ var b Batch
782+ b.Put([]byte(fmt.Sprintf("failed%d", i)), []byte("x"))
783+ _, errs[i] = db.Apply(bg, &b)
784+ })
785+ }
786+ wg.Wait()
787+ for i, err := range errs {
788+ if !errors.Is(err, ErrFailed) {
789+ t.Fatalf("batch %d: %v, want ErrFailed", i, err)
790+ }
791+ if f.cause != nil && !errors.Is(err, f.cause) {
792+ t.Fatalf("batch %d: %v does not wrap %v", i, err, f.cause)
793+ }
794+ }
795+ if !strings.Contains(logs.String(), "database failed") {
796+ t.Errorf("failure not logged: %q", logs.String())
797+ }
798+
799+ // Every later Apply fails at once; reads serve the last snapshot.
800+ for range 3 {
801+ var b Batch
802+ b.Put([]byte("later"), nil)
803+ if _, err := db.Apply(bg, &b); !errors.Is(err, ErrFailed) {
804+ t.Fatalf("later Apply: %v, want ErrFailed", err)
805+ }
806+ }
807+ var empty Batch
808+ if _, err := db.Apply(bg, &empty); !errors.Is(err, ErrFailed) {
809+ t.Fatalf("empty Apply on a failed db: %v", err)
810+ }
811+ s := db.Snapshot()
812+ if s.Seq() != 1 || s.Len() != 1 {
813+ t.Fatalf("snapshot after failure: seq %d len %d", s.Seq(), s.Len())
814+ }
815+ mustGet(t, s, "before", "1")
816+ mustMissing(t, s, "failed0")
817+ if err := db.Close(); !errors.Is(err, ErrFailed) {
818+ t.Fatalf("Close = %v, want the failure", err)
819+ }
820+ if err := db.Close(); err != nil {
821+ t.Fatalf("second Close = %v", err)
822+ }
823+
824+ if powerLoss {
825+ fx.PowerLoss(faultfs.PowerLossOptions{})
826+ }
827+ s = mustOpen(t, fx, Options{}).Snapshot()
828+ mustGet(t, s, "before", "1")
829+ present := 0
830+ for i := range group {
831+ if _, ok := s.Get(fmt.Appendf(nil, "failed%d", i)); ok {
832+ present++
833+ }
834+ }
835+ want := 0
836+ if f.kept && !powerLoss {
837+ want = group
838+ }
839+ if present != want {
840+ t.Fatalf("%d of %d failed batches recovered, want %d", present, group, want)
841+ }
842+ if s.Seq() != uint64(1+present) {
843+ t.Fatalf("Seq = %d, want %d", s.Seq(), 1+present)
844+ }
845+}
846+
847+// TestFailedWriterAnswersOverflow checks that no Apply hangs when the group
848+// that fails leaves a dequeued request behind.
849+func TestFailedWriterAnswersOverflow(t *testing.T) {
850+ fx := faultfs.New()
851+ db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
852+ fx.FailWrite(1, nil)
853+ chunk := strings.Repeat("c", 3<<20) // two do not fit in one group
854+ errs := make([]error, 4)
855+ var wg sync.WaitGroup
856+ for i := range errs {
857+ wg.Go(func() {
858+ var b Batch
859+ b.Put(fmt.Appendf(nil, "k%d", i), []byte(chunk))
860+ _, errs[i] = db.Apply(bg, &b)
861+ })
862+ }
863+ wg.Wait()
864+ for i, err := range errs {
865+ if !errors.Is(err, ErrFailed) {
866+ t.Errorf("batch %d: %v", i, err)
867+ }
868+ }
869+}
870+
871+// captureLogger returns a logger writing to a buffer.
872+func captureLogger() (*slog.Logger, *syncBuffer) {
873+ var sb syncBuffer
874+ return slog.New(slog.NewTextHandler(&sb, nil)), &sb
875+}
876+
877+type syncBuffer struct {
878+ mu sync.Mutex
879+ b strings.Builder
880+}
881+
882+func (s *syncBuffer) Write(p []byte) (int, error) {
883+ s.mu.Lock()
884+ defer s.mu.Unlock()
885+ return s.b.Write(p)
886+}
887+
888+func (s *syncBuffer) String() string {
889+ s.mu.Lock()
890+ defer s.mu.Unlock()
891+ return s.b.String()
892+}
893+
894+// powerLossBase returns the first seed; ERMITE_KV_SEED reproduces a run.
895+func powerLossBase(t *testing.T) uint64 {
896+ if v := os.Getenv("ERMITE_KV_SEED"); v != "" {
897+ seed, err := strconv.ParseUint(v, 10, 64)
898+ if err != nil {
899+ t.Fatalf("ERMITE_KV_SEED: %v", err)
900+ }
901+ return seed
902+ }
903+ return uint64(time.Now().UnixNano())
904+}
905+
906+// TestPowerLoss cuts the power at a random point while writers run, then
907+// checks AC2: every acknowledged batch is present and the recovered state is
908+// exactly a prefix of the applied batches.
909+//
910+// Every writer g applies batches n = 0, 1, … that set cur/g = n and
911+// all/g/n, and delete all/g/(n-1). The recovered state therefore says how
912+// many batches of each writer were applied, and the seq must equal their sum.
913+// (PowerLossOptions.KeepPages is not exercised: see the open question about
914+// recovery rule 6 under out-of-order page writeback.)
915+func TestPowerLoss(t *testing.T) {
916+ modes := []struct {
917+ name string
918+ opts faultfs.PowerLossOptions
919+ }{
920+ {"drop", faultfs.PowerLossOptions{}},
921+ {"zerofill", faultfs.PowerLossOptions{ZeroFill: true}},
922+ }
923+ base := powerLossBase(t)
924+ iterations := short(150, 10)
925+ for _, m := range modes {
926+ t.Run(m.name, func(t *testing.T) {
927+ for i := range iterations {
928+ seed := base + uint64(i)
929+ if !runPowerLoss(t, seed, m.opts) {
930+ t.Fatalf("failed with seed %d (ERMITE_KV_SEED=%d reproduces it)", seed, seed)
931+ }
932+ }
933+ })
934+ }
935+}
936+
937+func runPowerLoss(t *testing.T, seed uint64, plo faultfs.PowerLossOptions) (ok bool) {
938+ rng := rand.New(rand.NewPCG(seed, 0x9e3779b97f4a7c15))
939+ fx := faultfs.New()
940+ db, err := open(fx, testDir, Options{GroupWindow: time.Duration(rng.IntN(3)) * 100 * time.Microsecond})
941+ if err != nil {
942+ t.Errorf("seed %d: open: %v", seed, err)
943+ return false
944+ }
945+ fx.SetSyncDelay(time.Duration(rng.IntN(300)) * time.Microsecond)
946+ writers := 1 + rng.IntN(8)
947+ acked := make([]int, writers) // last acknowledged n, -1 if none
948+ ackSeq := make([]uint64, writers)
949+ var wg sync.WaitGroup
950+ for g := range writers {
951+ acked[g] = -1
952+ wg.Go(func() {
953+ for n := 0; ; n++ {
954+ var b Batch
955+ b.Put(fmt.Appendf(nil, "cur/%d", g), strconv.AppendInt(nil, int64(n), 10))
956+ b.Put(fmt.Appendf(nil, "all/%d/%d", g, n), nil)
957+ if n > 0 {
958+ b.Delete(fmt.Appendf(nil, "all/%d/%d", g, n-1))
959+ }
960+ seq, err := db.Apply(bg, &b)
961+ if err != nil {
962+ return
963+ }
964+ acked[g], ackSeq[g] = n, seq
965+ }
966+ })
967+ }
968+ time.Sleep(time.Duration(rng.IntN(4000)) * time.Microsecond)
969+ fx.PowerLoss(plo)
970+ wg.Wait()
971+ _ = db.Close()
972+ fx.SetSyncDelay(0)
973+
974+ db, err = open(fx, testDir, Options{})
975+ if err != nil {
976+ t.Errorf("seed %d: reopen after power loss: %v", seed, err)
977+ return false
978+ }
979+ defer db.Close()
980+ if !checkPowerLossState(t, seed, db.Snapshot(), acked, ackSeq) {
981+ return false
982+ }
983+
984+ // The recovered log accepts new records and survives another reopen.
985+ seq := db.Snapshot().Seq()
986+ if got := putKV(t, db, "after", "x"); got != seq+1 {
987+ t.Errorf("seed %d: seq after recovery = %d, want %d", seed, got, seq+1)
988+ return false
989+ }
990+ if err := db.Close(); err != nil {
991+ t.Errorf("seed %d: Close: %v", seed, err)
992+ return false
993+ }
994+ db, err = open(fx, testDir, Options{})
995+ if err != nil {
996+ t.Errorf("seed %d: second reopen: %v", seed, err)
997+ return false
998+ }
999+ defer db.Close()
1000+ if db.Snapshot().Seq() != seq+1 {
1001+ t.Errorf("seed %d: seq after second reopen = %d, want %d", seed, db.Snapshot().Seq(), seq+1)
1002+ return false
1003+ }
1004+ return true
1005+}
1006+
1007+func checkPowerLossState(t *testing.T, seed uint64, s *Snapshot, acked []int, ackSeq []uint64) bool {
1008+ fail := func(format string, args ...any) bool {
1009+ t.Errorf("seed %d: "+format, append([]any{seed}, args...)...)
1010+ return false
1011+ }
1012+ var applied uint64
1013+ keys := 0
1014+ for g := range acked {
1015+ it, ok := s.Get(fmt.Appendf(nil, "cur/%d", g))
1016+ last := -1
1017+ if ok {
1018+ n, err := strconv.Atoi(string(it.Value))
1019+ if err != nil {
1020+ return fail("writer %d: bad cur value %q", g, it.Value)
1021+ }
1022+ last = n
1023+ }
1024+ if last < acked[g] {
1025+ return fail("writer %d: acknowledged batch %d lost, recovered %d", g, acked[g], last)
1026+ }
1027+ if last == acked[g] && last >= 0 && it.Version != ackSeq[g] {
1028+ return fail("writer %d: batch %d acknowledged at seq %d, recovered at %d", g, last, ackSeq[g], it.Version)
1029+ }
1030+ if !ok {
1031+ continue
1032+ }
1033+ cur, ok := s.Get(fmt.Appendf(nil, "all/%d/%d", g, last))
1034+ if !ok || cur.Version != it.Version {
1035+ return fail("writer %d: batch %d is partially applied (%v, %v)", g, last, cur, ok)
1036+ }
1037+ count := 0
1038+ for range s.Prefix(fmt.Appendf(nil, "all/%d/", g)) {
1039+ count++
1040+ }
1041+ if count != 1 {
1042+ return fail("writer %d: %d all/ keys, want 1", g, count)
1043+ }
1044+ applied += uint64(last) + 1
1045+ keys += 2
1046+ }
1047+ if s.Seq() != applied {
1048+ return fail("Seq = %d but %d batches are visible: not a prefix", s.Seq(), applied)
1049+ }
1050+ if s.Len() != keys {
1051+ return fail("Len = %d, want %d", s.Len(), keys)
1052+ }
1053+ return true
1054+}
1055+
1056+// countFS counts the fdatasync calls of the files it opens.
1057+type countFS struct {
1058+ fsys.FS
1059+ syncs atomic.Int64
1060+}
1061+
1062+func (c *countFS) OpenFile(name string, flag int, perm os.FileMode) (fsys.File, error) {
1063+ f, err := c.FS.OpenFile(name, flag, perm)
1064+ if err != nil {
1065+ return nil, err
1066+ }
1067+ return countFile{f, &c.syncs}, nil
1068+}
1069+
1070+type countFile struct {
1071+ fsys.File
1072+ n *atomic.Int64
1073+}
1074+
1075+func (f countFile) Sync() error {
1076+ f.n.Add(1)
1077+ return f.File.Sync()
1078+}
1079+
1080+// BenchmarkApply measures group-commit throughput with parallel writers on a
1081+// real directory (NoSync=false). The syncs are counted by a thin wrapper
1082+// around the OS file system, so the numbers are real fdatasync calls.
1083+func BenchmarkApply(b *testing.B) {
1084+ fx := &countFS{FS: fsys.OS()}
1085+ db, err := open(fx, filepath.Join(b.TempDir(), "db"), Options{})
1086+ if err != nil {
1087+ b.Fatal(err)
1088+ }
1089+ defer db.Close()
1090+ b.SetParallelism(16)
1091+ value := bytes.Repeat([]byte("v"), 100)
1092+ var next atomic.Uint64
1093+ b.ReportAllocs()
1094+ b.ResetTimer()
1095+ b.RunParallel(func(pb *testing.PB) {
1096+ var batch Batch
1097+ for pb.Next() {
1098+ batch.Reset()
1099+ batch.Put(fmt.Appendf(nil, "key-%d", next.Add(1)), value)
1100+ if _, err := db.Apply(bg, &batch); err != nil {
1101+ b.Error(err)
1102+ return
1103+ }
1104+ }
1105+ })
1106+ b.StopTimer()
1107+ secs := b.Elapsed().Seconds()
1108+ syncs := float64(fx.syncs.Load())
1109+ b.ReportMetric(float64(b.N)/secs, "batches/s")
1110+ b.ReportMetric(syncs/secs, "syncs/s")
1111+ b.ReportMetric(float64(b.N)/syncs, "batches/sync")
1112+}
A writer.go +212 -0
@@ -0,0 +1,212 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "fmt"
7+ "io"
8+ "sync"
9+ "time"
10+)
11+
12+// maxGroupBytes caps the encoded payload of one commit group. A single batch
13+// larger than this forms a group of one.
14+const maxGroupBytes = 4 << 20
15+
16+// request is a batch waiting for the writer. The writer answers on reply
17+// exactly once.
18+type request struct {
19+ b *Batch
20+ reply chan result // capacity 1
21+}
22+
23+type result struct {
24+ seq uint64
25+ err error
26+}
27+
28+var requestPool = sync.Pool{New: func() any {
29+ return &request{reply: make(chan result, 1)}
30+}}
31+
32+// release returns r to the pool (it must not be referenced by the writer)
33+// and passes err through.
34+func (r *request) release(err error) error {
35+ r.b = nil
36+ requestPool.Put(r)
37+ return err
38+}
39+
40+// writer is the single goroutine that owns the log file and builds the new
41+// state. It exits on Close or when the database fails.
42+func (db *DB) writer() {
43+ defer close(db.done)
44+ var pending *request // dequeued but did not fit in the previous group
45+ defer func() {
46+ // A request taken but never grouped must still be answered.
47+ if pending != nil {
48+ err := ErrClosed
49+ select {
50+ case <-db.failed:
51+ err = db.failErr
52+ default:
53+ }
54+ pending.reply <- result{err: err}
55+ }
56+ }()
57+ for {
58+ // Closing takes priority over queued requests.
59+ select {
60+ case <-db.closing:
61+ return
62+ default:
63+ }
64+ first := pending
65+ pending = nil
66+ if first == nil {
67+ select {
68+ case <-db.closing:
69+ return
70+ case first = <-db.reqCh:
71+ }
72+ }
73+ pending = db.collect(first)
74+ if !db.commit() {
75+ return
76+ }
77+ // P1.D hook: after replying, trigger compaction here once
78+ // db.logSize exceeds db.opts.CompactAfter (if it is positive).
79+ }
80+}
81+
82+// collect fills db.group starting with first: it drains the queued requests
83+// and, with GroupWindow > 0, waits for more until the window expires or the
84+// database is closing. It returns a request that was taken but did not fit
85+// under maxGroupBytes.
86+func (db *DB) collect(first *request) (overflow *request) {
87+ db.group = append(db.group[:0], first)
88+ size := first.b.payload()
89+ var expired <-chan time.Time
90+ if w := db.opts.GroupWindow; w > 0 {
91+ t := time.NewTimer(w)
92+ defer t.Stop()
93+ expired = t.C
94+ }
95+ for {
96+ var r *request
97+ if expired == nil {
98+ select {
99+ case r = <-db.reqCh:
100+ default:
101+ return nil
102+ }
103+ } else {
104+ select {
105+ case r = <-db.reqCh:
106+ case <-expired:
107+ return nil
108+ case <-db.closing:
109+ return nil
110+ }
111+ }
112+ if p := r.b.payload(); size+p <= maxGroupBytes {
113+ db.group = append(db.group, r)
114+ size += p
115+ } else {
116+ return r
117+ }
118+ }
119+}
120+
121+// commit applies db.group to a fresh builder, writes and syncs the records,
122+// publishes the new state and answers every request. It reports false if the
123+// database failed.
124+func (db *DB) commit() bool {
125+ defer clear(db.group)
126+ cur := db.snap.Load()
127+ bld := cur.t.builder()
128+ seq := cur.seq
129+ buf := db.buf[:0]
130+ results := make([]result, len(db.group))
131+
132+ for i, r := range db.group {
133+ b := r.b
134+ if !checkPreconditions(bld, b.pre) {
135+ results[i] = result{err: ErrConflict}
136+ continue
137+ }
138+ if len(b.ops) == 0 {
139+ results[i] = result{seq: seq}
140+ continue
141+ }
142+ seq++
143+ for _, o := range b.ops {
144+ if o.kind == opPut {
145+ bld.put(o.key, o.value, seq)
146+ } else {
147+ bld.delete(o.key)
148+ }
149+ }
150+ buf = appendRecord(buf, seq, b.ops)
151+ results[i] = result{seq: seq}
152+ }
153+ db.buf = buf[:0]
154+ if cap(db.buf) > 2*maxGroupBytes {
155+ db.buf = nil // do not pin the memory of one huge batch
156+ }
157+
158+ if len(buf) > 0 {
159+ if err := db.persist(buf); err != nil {
160+ err = db.fail(err)
161+ for _, r := range db.group {
162+ r.reply <- result{err: err}
163+ }
164+ return false
165+ }
166+ db.logSize += int64(len(buf))
167+ db.snap.Store(newSnapshot(bld.freeze(), seq))
168+ }
169+ for i, r := range db.group {
170+ r.reply <- results[i]
171+ }
172+ return true
173+}
174+
175+// persist appends buf to the log and syncs it (unless NoSync).
176+func (db *DB) persist(buf []byte) error {
177+ n, err := db.log.Write(buf)
178+ if err == nil && n < len(buf) {
179+ err = io.ErrShortWrite
180+ }
181+ if err != nil {
182+ return fmt.Errorf("write log: %w", err)
183+ }
184+ if !db.opts.NoSync {
185+ if err := db.log.Sync(); err != nil {
186+ return fmt.Errorf("sync log: %w", err)
187+ }
188+ }
189+ return nil
190+}
191+
192+// fail puts the database in the failed state, keeping the first cause, and
193+// returns the error that requests receive.
194+func (db *DB) fail(cause error) error {
195+ db.failOnce.Do(func() {
196+ db.failErr = fmt.Errorf("%w: %w", ErrFailed, cause)
197+ db.opts.Logger.Error("kv: database failed", "err", cause)
198+ close(db.failed)
199+ })
200+ return db.failErr
201+}
202+
203+// checkPreconditions reports whether all preconditions hold on bld.
204+func checkPreconditions(bld *builder, pre []precond) bool {
205+ for _, p := range pre {
206+ item, ok := bld.get(p.key)
207+ if p.absent && ok || !p.absent && (!ok || item.Version != p.version) {
208+ return false
209+ }
210+ }
211+ return true
212+}
@@ -0,0 +1,212 @@
1+// SPDX-License-Identifier: MIT
2+
3+package kv
4+
5+import (
6+ "fmt"
7+ "io"
8+ "sync"
9+ "time"
10+)
11+
12+// maxGroupBytes caps the encoded payload of one commit group. A single batch
13+// larger than this forms a group of one.
14+const maxGroupBytes = 4 << 20
15+
16+// request is a batch waiting for the writer. The writer answers on reply
17+// exactly once.
18+type request struct {
19+ b *Batch
20+ reply chan result // capacity 1
21+}
22+
23+type result struct {
24+ seq uint64
25+ err error
26+}
27+
28+var requestPool = sync.Pool{New: func() any {
29+ return &request{reply: make(chan result, 1)}
30+}}
31+
32+// release returns r to the pool (it must not be referenced by the writer)
33+// and passes err through.
34+func (r *request) release(err error) error {
35+ r.b = nil
36+ requestPool.Put(r)
37+ return err
38+}
39+
40+// writer is the single goroutine that owns the log file and builds the new
41+// state. It exits on Close or when the database fails.
42+func (db *DB) writer() {
43+ defer close(db.done)
44+ var pending *request // dequeued but did not fit in the previous group
45+ defer func() {
46+ // A request taken but never grouped must still be answered.
47+ if pending != nil {
48+ err := ErrClosed
49+ select {
50+ case <-db.failed:
51+ err = db.failErr
52+ default:
53+ }
54+ pending.reply <- result{err: err}
55+ }
56+ }()
57+ for {
58+ // Closing takes priority over queued requests.
59+ select {
60+ case <-db.closing:
61+ return
62+ default:
63+ }
64+ first := pending
65+ pending = nil
66+ if first == nil {
67+ select {
68+ case <-db.closing:
69+ return
70+ case first = <-db.reqCh:
71+ }
72+ }
73+ pending = db.collect(first)
74+ if !db.commit() {
75+ return
76+ }
77+ // P1.D hook: after replying, trigger compaction here once
78+ // db.logSize exceeds db.opts.CompactAfter (if it is positive).
79+ }
80+}
81+
82+// collect fills db.group starting with first: it drains the queued requests
83+// and, with GroupWindow > 0, waits for more until the window expires or the
84+// database is closing. It returns a request that was taken but did not fit
85+// under maxGroupBytes.
86+func (db *DB) collect(first *request) (overflow *request) {
87+ db.group = append(db.group[:0], first)
88+ size := first.b.payload()
89+ var expired <-chan time.Time
90+ if w := db.opts.GroupWindow; w > 0 {
91+ t := time.NewTimer(w)
92+ defer t.Stop()
93+ expired = t.C
94+ }
95+ for {
96+ var r *request
97+ if expired == nil {
98+ select {
99+ case r = <-db.reqCh:
100+ default:
101+ return nil
102+ }
103+ } else {
104+ select {
105+ case r = <-db.reqCh:
106+ case <-expired:
107+ return nil
108+ case <-db.closing:
109+ return nil
110+ }
111+ }
112+ if p := r.b.payload(); size+p <= maxGroupBytes {
113+ db.group = append(db.group, r)
114+ size += p
115+ } else {
116+ return r
117+ }
118+ }
119+}
120+
121+// commit applies db.group to a fresh builder, writes and syncs the records,
122+// publishes the new state and answers every request. It reports false if the
123+// database failed.
124+func (db *DB) commit() bool {
125+ defer clear(db.group)
126+ cur := db.snap.Load()
127+ bld := cur.t.builder()
128+ seq := cur.seq
129+ buf := db.buf[:0]
130+ results := make([]result, len(db.group))
131+
132+ for i, r := range db.group {
133+ b := r.b
134+ if !checkPreconditions(bld, b.pre) {
135+ results[i] = result{err: ErrConflict}
136+ continue
137+ }
138+ if len(b.ops) == 0 {
139+ results[i] = result{seq: seq}
140+ continue
141+ }
142+ seq++
143+ for _, o := range b.ops {
144+ if o.kind == opPut {
145+ bld.put(o.key, o.value, seq)
146+ } else {
147+ bld.delete(o.key)
148+ }
149+ }
150+ buf = appendRecord(buf, seq, b.ops)
151+ results[i] = result{seq: seq}
152+ }
153+ db.buf = buf[:0]
154+ if cap(db.buf) > 2*maxGroupBytes {
155+ db.buf = nil // do not pin the memory of one huge batch
156+ }
157+
158+ if len(buf) > 0 {
159+ if err := db.persist(buf); err != nil {
160+ err = db.fail(err)
161+ for _, r := range db.group {
162+ r.reply <- result{err: err}
163+ }
164+ return false
165+ }
166+ db.logSize += int64(len(buf))
167+ db.snap.Store(newSnapshot(bld.freeze(), seq))
168+ }
169+ for i, r := range db.group {
170+ r.reply <- results[i]
171+ }
172+ return true
173+}
174+
175+// persist appends buf to the log and syncs it (unless NoSync).
176+func (db *DB) persist(buf []byte) error {
177+ n, err := db.log.Write(buf)
178+ if err == nil && n < len(buf) {
179+ err = io.ErrShortWrite
180+ }
181+ if err != nil {
182+ return fmt.Errorf("write log: %w", err)
183+ }
184+ if !db.opts.NoSync {
185+ if err := db.log.Sync(); err != nil {
186+ return fmt.Errorf("sync log: %w", err)
187+ }
188+ }
189+ return nil
190+}
191+
192+// fail puts the database in the failed state, keeping the first cause, and
193+// returns the error that requests receive.
194+func (db *DB) fail(cause error) error {
195+ db.failOnce.Do(func() {
196+ db.failErr = fmt.Errorf("%w: %w", ErrFailed, cause)
197+ db.opts.Logger.Error("kv: database failed", "err", cause)
198+ close(db.failed)
199+ })
200+ return db.failErr
201+}
202+
203+// checkPreconditions reports whether all preconditions hold on bld.
204+func checkPreconditions(bld *builder, pre []precond) bool {
205+ for _, p := range pre {
206+ item, ok := bld.get(p.key)
207+ if p.absent && ok || !p.absent && (!ok || item.Version != p.version) {
208+ return false
209+ }
210+ }
211+ return true
212+}

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