1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
6 "context"
7 "fmt"
8 "io"
9 "sync"
10 "sync/atomic"
11 "time"
12
13 "github.com/bnema/zerowrap"
14
15 "bnema.dev/ermite-kv/internal/fsys"
16)
17
18// Defaults for zero Options fields.
19const (
20 defaultMaxBatchBytes = 16 << 20
21 defaultCompactAfter = 64 << 20
22
23 // maxBatchLimit is the largest encoded batch that fits in a record
24 // together with the record head (seq and batch count).
25 maxBatchLimit = maxRecord - recordHead
26)
27
28// Options configures Open. The zero value of every field means "default";
29// Options{} is safe and durable.
30type Options struct {
31 // NoSync skips fdatasync on commits. For tests and benchmarks only: an
32 // acknowledged batch may be lost on power failure, and a power loss can
33 // leave a log that Open rejects with ErrCorrupt (the pages of one
34 // unsynced group may reach the disk out of order, which recovery cannot
35 // tell from damage to acknowledged data).
36 NoSync bool
37 // MaxBatchBytes is the encoded batch limit (op count and ops). 0 means
38 // 16 MiB; it must be at most 64 MiB minus 12 bytes (the format limit).
39 MaxBatchBytes int
40 // CompactAfter is the number of bytes in the current log that trigger
41 // compaction. 0 means 64 MiB; a negative value disables compaction.
42 CompactAfter int64
43 // GroupWindow is the extra time the writer waits to grow a commit
44 // group. 0 means it only drains what is already queued.
45 GroupWindow time.Duration
46}
47
48// resolve validates o and fills in the defaults.
49func (o Options) resolve() (Options, error) {
50 switch {
51 case o.MaxBatchBytes < 0 || o.MaxBatchBytes > maxBatchLimit:
52 return o, fmt.Errorf("kv: invalid MaxBatchBytes %d: must be in [0, %d]", o.MaxBatchBytes, maxBatchLimit)
53 case o.GroupWindow < 0:
54 return o, fmt.Errorf("kv: invalid GroupWindow %v: must not be negative", o.GroupWindow)
55 }
56 if o.MaxBatchBytes == 0 {
57 o.MaxBatchBytes = defaultMaxBatchBytes
58 }
59 if o.CompactAfter == 0 {
60 o.CompactAfter = defaultCompactAfter
61 }
62 return o, nil
63}
64
65// DB is an embedded, durable, ordered key-value store. All methods are safe
66// for concurrent use.
67type DB struct {
68 fx fsys.FS
69 dir string
70 opts Options
71 log zerowrap.Logger // captured at Open; used by the writer
72 snap atomic.Pointer[Snapshot] // last published state
73
74 reqCh chan *request // never closed
75 closing chan struct{} // closed by Close
76 done chan struct{} // closed when the writer has exited
77
78 failed chan struct{} // closed when the database enters the failed state
79 failOnce sync.Once
80 failErr error // first cause, wrapping ErrFailed; set before failed is closed
81
82 closeOnce sync.Once
83
84 lock io.Closer
85
86 compacting atomic.Bool // a compactor goroutine is running
87 compactors sync.WaitGroup // started by the writer only; Close waits after the writer exits
88
89 // Owned by the writer goroutine (see writer.go).
90 file fsys.File
91 logBase uint64 // seq the current log starts after
92 logSize int64 // bytes in the current log, header included
93 retryAt int64 // do not try a log switch again before logSize reaches this
94 buf []byte // group encoding buffer
95 group []*request
96}
97
98// Open opens the database in dir, creating it (mode 0700) if needed. It
99// takes an exclusive lock on the directory, returning ErrLocked if another
100// DB holds it, and recovers the state from the snapshot and logs. Invalid
101// options are an error.
102//
103// The logger of ctx (zerowrap.FromCtx), enriched with the component and the
104// directory, is used for the life of the DB; ctx is not used for anything
105// else and does not bound the life of the DB.
106func Open(ctx context.Context, dir string, opts Options) (*DB, error) {
107 return open(ctx, fsys.OS(), dir, opts)
108}
109
110func open(ctx context.Context, fx fsys.FS, dir string, opts Options) (*DB, error) {
111 opts, err := opts.resolve()
112 if err != nil {
113 return nil, err
114 }
115 start := time.Now()
116 log := zerowrap.FromCtxWithFields(ctx, map[string]any{"component": "ermite-kv", "dir": dir})
117 bld := tree{}.builder()
118 rec, err := recoverDir(fx, dir, log,
119 func(key, value []byte, version uint64) {
120 // Snapshot entries alias one shared file buffer: copy them so
121 // the tree does not keep the whole buffer alive.
122 buf := make([]byte, len(key)+len(value))
123 n := copy(buf, key)
124 copy(buf[n:], value)
125 bld.put(buf[:n:n], buf[n:len(buf):len(buf)], version)
126 },
127 func(seq uint64, ops []op) {
128 for _, o := range ops {
129 if o.kind == opPut {
130 bld.put(o.key, o.value, seq)
131 } else {
132 bld.delete(o.key)
133 }
134 }
135 })
136 if err != nil {
137 return nil, err
138 }
139 db := &DB{
140 fx: fx,
141 dir: dir,
142 opts: opts,
143 log: log,
144 reqCh: make(chan *request),
145 closing: make(chan struct{}),
146 done: make(chan struct{}),
147 failed: make(chan struct{}),
148 lock: rec.lock,
149 file: rec.log,
150 logBase: rec.logBase,
151 logSize: rec.logSize,
152 }
153 t := bld.freeze()
154 db.snap.Store(newSnapshot(t, rec.lastSeq))
155 go db.writer()
156 log.Info().Uint64("seq", rec.lastSeq).Int("keys", t.len()).Uint64("snapshot_seq", rec.snapSeq).
157 Dur("duration", time.Since(start)).Msg("opened")
158 return db, nil
159}
160
161// Close stops the database: requests not yet taken by the writer get
162// ErrClosed, the group in progress is committed and acknowledged, then the
163// log is closed and the directory lock released. Snapshots stay readable.
164// Close always completes: ctx is only used for the logger of the closing
165// message.
166//
167// Close returns the first error met while stopping, including the cause of
168// the failed state if the database failed, or nil. It is idempotent: later
169// calls return nil.
170func (db *DB) Close(ctx context.Context) error {
171 var err error
172 db.closeOnce.Do(func() {
173 close(db.closing)
174 <-db.done // the writer finishes its group and exits
175 db.compactors.Wait() // an in-flight compaction is not cancelled
176 select {
177 case <-db.failed:
178 err = db.failErr
179 default:
180 }
181 if cerr := db.file.Close(); err == nil {
182 err = cerr
183 }
184 if cerr := db.lock.Close(); err == nil {
185 err = cerr
186 }
187 log := zerowrap.FromCtxWithFields(ctx, map[string]any{"component": "ermite-kv", "dir": db.dir})
188 log.Info().Uint64("seq", db.snap.Load().Seq()).Err(err).Msg("closed")
189 })
190 return err
191}
192
193// Snapshot returns the current immutable state. It is lock-free and O(1), and
194// remains usable after Close. Like every read, it ignores ctx.
195func (db *DB) Snapshot(_ context.Context) *Snapshot { return db.snap.Load() }
196
197// Get returns the item stored under key in the current snapshot.
198func (db *DB) Get(ctx context.Context, key []byte) (Item, bool) {
199 return db.snap.Load().Get(ctx, key)
200}
201
202// Apply commits b atomically and returns its seq once it is durable (unless
203// Options.NoSync) and visible in Snapshot.
204//
205// It returns the first error recorded by the batch (ErrInvalidKey), then
206// ErrBatchTooLarge if the encoded batch exceeds Options.MaxBatchBytes, both
207// without touching the database. ErrConflict (seq 0) means a precondition did
208// not hold at commit time. A batch without puts and deletes writes nothing,
209// consumes no seq and returns the current seq.
210//
211// If ctx ends before the writer takes the batch, Apply returns ctx.Err() and
212// the batch is not applied. Once the writer has taken it, ctx is ignored and
213// Apply waits for the outcome. ErrClosed is returned after Close. ErrFailed
214// (wrapping the I/O error) means the database failed: for the batch that was
215// being committed its outcome is unknown, since its bytes may be on disk and
216// replayed by the next Open.
217func (db *DB) Apply(ctx context.Context, b *Batch) (uint64, error) {
218 if b.err != nil {
219 return 0, b.err
220 }
221 if b.encodedSize() > db.opts.MaxBatchBytes {
222 return 0, ErrBatchTooLarge
223 }
224 if err := db.stopped(); err != nil {
225 return 0, err
226 }
227 if err := ctx.Err(); err != nil {
228 return 0, err
229 }
230
231 r := &request{b: b, reply: make(chan result, 1)}
232 select {
233 case db.reqCh <- r:
234 case <-ctx.Done():
235 return 0, ctx.Err()
236 case <-db.closing:
237 return 0, db.stopped()
238 case <-db.failed:
239 return 0, db.stopped()
240 }
241 res := <-r.reply // the writer always answers a request it has taken
242 return res.seq, res.err
243}
244
245// Err reports why the database stopped: ErrClosed after Close, ErrFailed
246// (wrapped as Apply returns it) after a fatal write error, nil while it runs.
247// It does not touch the writer.
248func (db *DB) Err() error { return db.stopped() }
249
250// stopped returns the failure (which wins over Close, so callers see a
251// stable error), ErrClosed, or nil while the database accepts batches.
252func (db *DB) stopped() error {
253 select {
254 case <-db.failed:
255 return db.failErr
256 default:
257 }
258 select {
259 case <-db.closing:
260 return ErrClosed
261 default:
262 return nil
263 }
264}