1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
6 "bytes"
7 "context"
8 "errors"
9 "fmt"
10 "math/rand/v2"
11 "os"
12 "path/filepath"
13 "slices"
14 "strconv"
15 "strings"
16 "sync"
17 "sync/atomic"
18 "testing"
19 "testing/synctest"
20 "time"
21
22 "github.com/bnema/zerowrap"
23
24 "bnema.dev/ermite-kv/internal/fsys"
25 "bnema.dev/ermite-kv/internal/fsys/faultfs"
26)
27
28var bg = context.Background()
29
30// testCtx returns a context carrying a console logger when
31// ERMITE_KV_LOG_LEVEL is set (the library itself never reads the
32// environment), and context.Background() (logging disabled) otherwise.
33func testCtx(t testing.TB) context.Context {
34 level := os.Getenv("ERMITE_KV_LOG_LEVEL")
35 if level == "" {
36 return bg
37 }
38 return zerowrap.WithCtx(bg, zerowrap.New(zerowrap.Config{Level: level, Format: "console", Output: t.Output()}))
39}
40
41// mustOpen opens a database on fx and closes it when the test ends.
42func mustOpen(t testing.TB, fx fsys.FS, opts Options) *DB {
43 t.Helper()
44 return mustOpenCtx(t, testCtx(t), fx, opts)
45}
46
47func mustOpenCtx(t testing.TB, ctx context.Context, fx fsys.FS, opts Options) *DB {
48 t.Helper()
49 db, err := open(ctx, fx, testDir, opts)
50 if err != nil {
51 t.Fatalf("open: %v", err)
52 }
53 t.Cleanup(func() { _ = db.Close(bg) })
54 return db
55}
56
57// putKV applies a single put and returns its seq.
58func putKV(t testing.TB, db *DB, key, value string) uint64 {
59 t.Helper()
60 var b Batch
61 b.Put([]byte(key), []byte(value))
62 seq, err := db.Apply(bg, &b)
63 if err != nil {
64 t.Fatalf("put %q: %v", key, err)
65 }
66 return seq
67}
68
69func mustGet(t testing.TB, s *Snapshot, key, want string) Item {
70 t.Helper()
71 it, ok := s.Get(bg, []byte(key))
72 if !ok || string(it.Value) != want {
73 t.Fatalf("Get(%q) = %q, %v; want %q", key, it.Value, ok, want)
74 }
75 return it
76}
77
78func mustMissing(t testing.TB, s *Snapshot, key string) {
79 t.Helper()
80 if it, ok := s.Get(bg, []byte(key)); ok {
81 t.Fatalf("Get(%q) = %q, want missing", key, it.Value)
82 }
83}
84
85func short(full, quick int) int {
86 if testing.Short() {
87 return quick
88 }
89 return full
90}
91
92func TestReopenFaultfs(t *testing.T) {
93 fx := faultfs.New()
94 db := mustOpen(t, fx, Options{})
95 if s := db.Snapshot(bg); s.Seq() != 0 || s.Len() != 0 {
96 t.Fatalf("new db: seq %d len %d", s.Seq(), s.Len())
97 }
98 for i := range 50 {
99 if seq := putKV(t, db, fmt.Sprintf("k%02d", i), fmt.Sprintf("v%d", i)); seq != uint64(i+1) {
100 t.Fatalf("seq = %d, want %d", seq, i+1)
101 }
102 }
103 var b Batch
104 b.Delete([]byte("k00"))
105 b.Put([]byte("k01"), []byte("again"))
106 b.Put([]byte("k01"), []byte("last"))
107 if seq, err := db.Apply(bg, &b); err != nil || seq != 51 {
108 t.Fatalf("Apply = %d, %v", seq, err)
109 }
110 if err := db.Close(bg); err != nil {
111 t.Fatalf("Close: %v", err)
112 }
113 if err := db.Close(bg); err != nil {
114 t.Fatalf("second Close: %v", err)
115 }
116
117 db = mustOpen(t, fx, Options{})
118 s := db.Snapshot(bg)
119 if s.Seq() != 51 || s.Len() != 49 {
120 t.Fatalf("reopened: seq %d len %d, want 51 and 49", s.Seq(), s.Len())
121 }
122 mustMissing(t, s, "k00")
123 if it := mustGet(t, s, "k01", "last"); it.Version != 51 {
124 t.Fatalf("k01 version %d, want 51", it.Version)
125 }
126 if it := mustGet(t, s, "k02", "v2"); it.Version != 3 {
127 t.Fatalf("k02 version %d, want 3", it.Version)
128 }
129 if seq := putKV(t, db, "next", "x"); seq != 52 {
130 t.Fatalf("seq after reopen = %d, want 52", seq)
131 }
132}
133
134func TestReopenOS(t *testing.T) {
135 dir := filepath.Join(t.TempDir(), "sub", "db")
136 db, err := Open(testCtx(t), dir, Options{})
137 if err != nil {
138 t.Fatal(err)
139 }
140 for i := range 100 {
141 putKV(t, db, fmt.Sprintf("k%03d", i), strings.Repeat("v", i))
142 }
143 if err := db.Close(bg); err != nil {
144 t.Fatal(err)
145 }
146 if fi, err := os.Stat(dir); err != nil || fi.Mode().Perm() != 0o700 {
147 t.Fatalf("dir stat: %v, %v", fi, err)
148 }
149
150 db, err = Open(testCtx(t), dir, Options{})
151 if err != nil {
152 t.Fatal(err)
153 }
154 defer func() { _ = db.Close(bg) }()
155 s := db.Snapshot(bg)
156 if s.Seq() != 100 || s.Len() != 100 {
157 t.Fatalf("seq %d len %d, want 100", s.Seq(), s.Len())
158 }
159 for i := range 100 {
160 mustGet(t, s, fmt.Sprintf("k%03d", i), strings.Repeat("v", i))
161 }
162 if got := db.Snapshot(bg); got != s {
163 t.Fatal("Snapshot changed without a write")
164 }
165 if it, ok := db.Get(bg, []byte("k005")); !ok || it.Version != 6 {
166 t.Fatalf("DB.Get = %v, %v", it, ok)
167 }
168}
169
170// TestOpenFromSnapshot covers loading a snapshot file (written by the future
171// compactor) and appending to the log created next to it.
172func TestOpenFromSnapshot(t *testing.T) {
173 var buf bytes.Buffer
174 entries := func(yield func(k, v []byte, ver uint64) bool) {
175 _ = yield([]byte("a"), []byte("1"), 3) && yield([]byte("b"), nil, 7) && yield([]byte("c"), []byte("3"), 9)
176 }
177 if err := writeSnapshot(&buf, 9, 3, entries); err != nil {
178 t.Fatal(err)
179 }
180 fx := faultfs.New()
181 fx.WriteFile(filepath.Join(testDir, snapshotName(9)), buf.Bytes())
182
183 db := mustOpen(t, fx, Options{})
184 s := db.Snapshot(bg)
185 if s.Seq() != 9 || s.Len() != 3 {
186 t.Fatalf("seq %d len %d", s.Seq(), s.Len())
187 }
188 if it := mustGet(t, s, "b", ""); it.Version != 7 {
189 t.Fatalf("b = %+v", it)
190 }
191 if seq := putKV(t, db, "d", "4"); seq != 10 {
192 t.Fatalf("seq = %d, want 10", seq)
193 }
194 if err := db.Close(bg); err != nil {
195 t.Fatal(err)
196 }
197 db = mustOpen(t, fx, Options{})
198 mustGet(t, db.Snapshot(bg), "d", "4")
199 mustGet(t, db.Snapshot(bg), "c", "3")
200 if db.Snapshot(bg).Seq() != 10 {
201 t.Fatalf("seq = %d", db.Snapshot(bg).Seq())
202 }
203}
204
205func TestConcurrentWriters(t *testing.T) {
206 fx := faultfs.New()
207 db := mustOpen(t, fx, Options{})
208 const writers = 64
209 per := short(50, 10)
210
211 seqs := make([][]uint64, writers)
212 var wg sync.WaitGroup
213 for g := range writers {
214 wg.Add(1)
215 go func() {
216 defer wg.Done()
217 for n := range per {
218 key := fmt.Sprintf("w%02d/%03d", g, n)
219 var b Batch
220 b.Put([]byte(key), []byte(key))
221 seq, err := db.Apply(bg, &b)
222 if err != nil {
223 t.Errorf("writer %d: %v", g, err)
224 return
225 }
226 if it, ok := db.Get(bg, []byte(key)); !ok || it.Version < seq {
227 t.Errorf("writer %d: no read-your-writes for %s at seq %d", g, key, seq)
228 return
229 }
230 seqs[g] = append(seqs[g], seq)
231 }
232 }()
233 }
234 wg.Wait()
235
236 var all []uint64
237 for g, s := range seqs {
238 if !slices.IsSorted(s) || len(slices.Compact(slices.Clone(s))) != len(s) {
239 t.Fatalf("writer %d seqs not strictly increasing: %v", g, s)
240 }
241 all = append(all, s...)
242 }
243 slices.Sort(all)
244 for i, s := range all {
245 if s != uint64(i+1) {
246 t.Fatalf("seq %d at rank %d: seqs are not 1..%d", s, i, len(all))
247 }
248 }
249 if got := db.Snapshot(bg).Seq(); got != writers*uint64(per) {
250 t.Fatalf("Seq = %d, want %d", got, writers*per)
251 }
252 if got := db.Snapshot(bg).Len(); got != writers*per {
253 t.Fatalf("Len = %d, want %d", got, writers*per)
254 }
255}
256
257func TestRequireAbsentRace(t *testing.T) {
258 for _, window := range []time.Duration{0, 5 * time.Millisecond} {
259 t.Run(window.String(), func(t *testing.T) {
260 fx := faultfs.New()
261 db := mustOpen(t, fx, Options{GroupWindow: window})
262 const n = 64
263 var wins atomic.Int32
264 winner := make([]bool, n)
265 var wg sync.WaitGroup
266 start := make(chan struct{})
267 for g := range n {
268 wg.Add(1)
269 go func() {
270 defer wg.Done()
271 var b Batch
272 b.RequireAbsent([]byte("lock"))
273 b.Put([]byte("lock"), []byte(strconv.Itoa(g)))
274 <-start
275 seq, err := db.Apply(bg, &b)
276 switch {
277 case err == nil && seq > 0:
278 wins.Add(1)
279 winner[g] = true
280 case errors.Is(err, ErrConflict) && seq == 0:
281 default:
282 t.Errorf("writer %d: seq %d err %v", g, seq, err)
283 }
284 }()
285 }
286 close(start)
287 wg.Wait()
288 if wins.Load() != 1 {
289 t.Fatalf("%d winners, want 1", wins.Load())
290 }
291 w := slices.Index(winner, true)
292 mustGet(t, db.Snapshot(bg), "lock", strconv.Itoa(w))
293 if db.Snapshot(bg).Seq() != 1 {
294 t.Fatalf("Seq = %d, want 1 (conflicts consume no seq)", db.Snapshot(bg).Seq())
295 }
296 })
297 }
298}
299
300func TestRequireVersion(t *testing.T) {
301 db := mustOpen(t, faultfs.New(), Options{})
302 v1 := putKV(t, db, "k", "a")
303 cas := func(key string, v uint64, val string) (uint64, error) {
304 var b Batch
305 b.RequireVersion([]byte(key), v)
306 b.Put([]byte(key), []byte(val))
307 return db.Apply(bg, &b)
308 }
309 v2, err := cas("k", v1, "b")
310 if err != nil || v2 != v1+1 {
311 t.Fatalf("cas at current version: %d, %v", v2, err)
312 }
313 if seq, err := cas("k", v1, "c"); !errors.Is(err, ErrConflict) || seq != 0 {
314 t.Fatalf("cas at stale version: %d, %v", seq, err)
315 }
316 if seq, err := cas("missing", 1, "c"); !errors.Is(err, ErrConflict) || seq != 0 {
317 t.Fatalf("cas on missing key: %d, %v", seq, err)
318 }
319 mustGet(t, db.Snapshot(bg), "k", "b")
320 mustMissing(t, db.Snapshot(bg), "missing")
321 if got := db.Snapshot(bg).Seq(); got != v2 {
322 t.Fatalf("Seq = %d, want %d", got, v2)
323 }
324}
325
326// TestPreconditionsBeforeOps checks that all preconditions are evaluated
327// before any op of the same batch and that a conflict contributes nothing.
328func TestPreconditionsBeforeOps(t *testing.T) {
329 db := mustOpen(t, faultfs.New(), Options{})
330 putKV(t, db, "exists", "1")
331
332 var b Batch
333 b.Put([]byte("new"), []byte("x"))
334 b.RequireAbsent([]byte("new")) // holds: checked before the put
335 if _, err := db.Apply(bg, &b); err != nil {
336 t.Fatalf("Apply: %v", err)
337 }
338
339 b.Reset()
340 b.Delete([]byte("exists"))
341 b.Put([]byte("other"), []byte("x"))
342 b.RequireAbsent([]byte("never"))
343 b.RequireAbsent([]byte("exists")) // fails
344 before := db.Snapshot(bg)
345 if seq, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) || seq != 0 {
346 t.Fatalf("Apply = %d, %v; want ErrConflict", seq, err)
347 }
348 if db.Snapshot(bg) != before {
349 t.Fatal("a conflicting batch published a snapshot")
350 }
351 mustGet(t, db.Snapshot(bg), "exists", "1")
352 mustMissing(t, db.Snapshot(bg), "other")
353}
354
355// holdWriter applies a batch whose sync is slow and returns once the writer
356// is inside that sync, so batches applied next queue up and form one group
357// (GroupWindow 0 drains everything already queued). wait returns when the
358// held batch is acknowledged; the sync delay stays set until then. It must be
359// called inside a synctest bubble: the writer sleeps in the sync on the fake
360// clock, so no real time passes.
361func holdWriter(t *testing.T, db *DB, fx *faultfs.FS) (wait func()) {
362 t.Helper()
363 base := fx.Syncs()
364 fx.SetSyncDelay(300 * time.Millisecond)
365 done := make(chan error, 1)
366 go func() {
367 var b Batch
368 b.Put([]byte("hold"), nil)
369 _, err := db.Apply(bg, &b)
370 done <- err
371 }()
372 synctest.Wait() // the writer is parked in the delayed sync
373 if fx.Syncs() == base {
374 t.Fatal("the writer did not reach the sync")
375 }
376 return func() {
377 t.Helper()
378 if err := <-done; err != nil {
379 t.Fatalf("held Apply: %v", err)
380 }
381 fx.SetSyncDelay(0)
382 }
383}
384
385// TestGroupPreconditions checks that preconditions see earlier batches of
386// the same group: the second batch of the group must conflict.
387func TestGroupPreconditions(t *testing.T) {
388 synctest.Test(t, func(t *testing.T) {
389 fx := faultfs.New()
390 db := mustOpen(t, fx, Options{})
391 syncs := fx.Syncs()
392 wait := holdWriter(t, db, fx)
393 errs := make([]error, 2)
394 var wg sync.WaitGroup
395 for i := range errs {
396 wg.Go(func() {
397 var b Batch
398 b.RequireAbsent([]byte("k"))
399 b.Put([]byte("k"), []byte{byte('a' + i)})
400 _, errs[i] = db.Apply(bg, &b)
401 })
402 }
403 wg.Wait()
404 wait()
405 if (errs[0] == nil) == (errs[1] == nil) {
406 t.Fatalf("errors %v, want exactly one success", errs)
407 }
408 if got := fx.Syncs() - syncs; got != 2 {
409 t.Fatalf("%d syncs, want 2 (held batch, then one group)", got)
410 }
411 })
412}
413
414func TestEmptyAndConflictOnlyBatches(t *testing.T) {
415 fx := faultfs.New()
416 db := mustOpen(t, fx, Options{})
417 putKV(t, db, "a", "1")
418 putKV(t, db, "b", "2")
419 writes, syncs := fx.Writes(), fx.Syncs()
420 snap := db.Snapshot(bg)
421
422 var empty Batch
423 if seq, err := db.Apply(bg, &empty); err != nil || seq != 2 {
424 t.Fatalf("empty batch: %d, %v; want current seq 2", seq, err)
425 }
426 var pre Batch
427 pre.RequireVersion([]byte("a"), 1)
428 pre.RequireAbsent([]byte("zzz"))
429 if seq, err := db.Apply(bg, &pre); err != nil || seq != 2 {
430 t.Fatalf("holding precondition-only batch: %d, %v; want seq 2", seq, err)
431 }
432 var bad Batch
433 bad.RequireAbsent([]byte("a"))
434 if seq, err := db.Apply(bg, &bad); !errors.Is(err, ErrConflict) || seq != 0 {
435 t.Fatalf("failing precondition-only batch: %d, %v", seq, err)
436 }
437 var conflict Batch
438 conflict.RequireAbsent([]byte("a"))
439 conflict.Put([]byte("c"), nil)
440 if _, err := db.Apply(bg, &conflict); !errors.Is(err, ErrConflict) {
441 t.Fatalf("conflict: %v", err)
442 }
443 var del Batch // deleting a missing key still writes a record
444 del.Delete([]byte("nope"))
445 if seq, err := db.Apply(bg, &del); err != nil || seq != 3 {
446 t.Fatalf("delete of a missing key: %d, %v; want seq 3", seq, err)
447 }
448
449 // Only the last batch touched the log.
450 if got := fx.Writes() - writes; got != 1 {
451 t.Fatalf("%d writes, want 1", got)
452 }
453 if got := fx.Syncs() - syncs; got != 1 {
454 t.Fatalf("%d syncs, want 1", got)
455 }
456 if db.Snapshot(bg).Len() != snap.Len() || db.Snapshot(bg).Seq() != 3 {
457 t.Fatalf("snapshot after: seq %d len %d", db.Snapshot(bg).Seq(), db.Snapshot(bg).Len())
458 }
459 if err := db.Close(bg); err != nil {
460 t.Fatal(err)
461 }
462 db = mustOpen(t, fx, Options{})
463 if db.Snapshot(bg).Seq() != 3 || db.Snapshot(bg).Len() != 2 {
464 t.Fatalf("reopened: seq %d len %d", db.Snapshot(bg).Seq(), db.Snapshot(bg).Len())
465 }
466}
467
468// TestNoWriteForSkippedGroup checks that a group of only conflicts and empty
469// batches neither writes nor syncs.
470func TestNoWriteForSkippedGroup(t *testing.T) {
471 synctest.Test(t, func(t *testing.T) {
472 fx := faultfs.New()
473 db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
474 putKV(t, db, "a", "1")
475 writes, syncs := fx.Writes(), fx.Syncs()
476 var wg sync.WaitGroup
477 for range 4 {
478 wg.Go(func() {
479 var b Batch
480 b.RequireAbsent([]byte("a"))
481 b.Put([]byte("x"), nil)
482 if _, err := db.Apply(bg, &b); !errors.Is(err, ErrConflict) {
483 t.Errorf("err = %v, want ErrConflict", err)
484 }
485 })
486 }
487 wg.Wait()
488 if fx.Writes() != writes || fx.Syncs() != syncs {
489 t.Fatalf("writes %d→%d syncs %d→%d", writes, fx.Writes(), syncs, fx.Syncs())
490 }
491 })
492}
493
494func TestGroupCommitSyncs(t *testing.T) {
495 fx := faultfs.New()
496 db := mustOpen(t, fx, Options{})
497 fx.SetSyncDelay(time.Millisecond)
498 base := fx.Syncs()
499 const writers = 64
500 per := short(100, 16)
501 var wg sync.WaitGroup
502 for g := range writers {
503 wg.Go(func() {
504 for n := range per {
505 putKV(t, db, fmt.Sprintf("g%02d/%03d", g, n), "v")
506 }
507 })
508 }
509 wg.Wait()
510 total := writers * per
511 syncs := fx.Syncs() - base
512 t.Logf("%d batches, %d syncs", total, syncs)
513 if syncs >= total/4 {
514 t.Fatalf("%d syncs for %d batches: no group commit", syncs, total)
515 }
516 if got := db.Snapshot(bg).Len(); got != total {
517 t.Fatalf("Len = %d, want %d", got, total)
518 }
519}
520
521func TestNoSync(t *testing.T) {
522 fx := faultfs.New()
523 db := mustOpen(t, fx, Options{NoSync: true})
524 base := fx.Syncs()
525 putKV(t, db, "a", "1")
526 if fx.Syncs() != base {
527 t.Fatal("NoSync still syncs")
528 }
529 if err := db.Close(bg); err != nil {
530 t.Fatal(err)
531 }
532 mustGet(t, mustOpen(t, fx, Options{}).Snapshot(bg), "a", "1")
533}
534
535// TestLargeBatches covers a batch above the 4 MiB group cap (a group of one)
536// and a burst of batches that does not fit in one group.
537func TestLargeBatches(t *testing.T) {
538 synctest.Test(t, func(t *testing.T) {
539 fx := faultfs.New()
540 db := mustOpen(t, fx, Options{GroupWindow: 50 * time.Millisecond})
541 big := strings.Repeat("B", 5<<20)
542 putKV(t, db, "big", big)
543
544 const n = 6
545 chunk := strings.Repeat("c", 3<<19) // 1.5 MiB: three do not fit in 4 MiB
546 seqs := make([]uint64, n)
547 var wg sync.WaitGroup
548 for i := range n {
549 wg.Go(func() { seqs[i] = putKV(t, db, fmt.Sprintf("chunk%d", i), chunk) })
550 }
551 wg.Wait()
552 slices.Sort(seqs)
553 for i, s := range seqs {
554 if s != uint64(i+2) {
555 t.Fatalf("seqs = %v", seqs)
556 }
557 }
558 if err := db.Close(bg); err != nil {
559 t.Fatal(err)
560 }
561 s := mustOpen(t, fx, Options{}).Snapshot(bg)
562 mustGet(t, s, "big", big)
563 for i := range n {
564 mustGet(t, s, fmt.Sprintf("chunk%d", i), chunk)
565 }
566 })
567}
568
569func TestInvalidInputs(t *testing.T) {
570 fx := faultfs.New()
571 db := mustOpen(t, fx, Options{MaxBatchBytes: 100})
572
573 var b Batch
574 b.Put(nil, []byte("v"))
575 if _, err := db.Apply(bg, &b); !errors.Is(err, ErrInvalidKey) {
576 t.Fatalf("empty key: %v", err)
577 }
578 b.Reset()
579 b.Put(bytes.Repeat([]byte("k"), maxKeyLen+1), nil)
580 if _, err := db.Apply(bg, &b); !errors.Is(err, ErrInvalidKey) {
581 t.Fatalf("long key: %v", err)
582 }
583 b.Reset()
584 b.RequireVersion([]byte("k"), 0)
585 if _, err := db.Apply(bg, &b); !errors.Is(err, ErrInvalidKey) {
586 t.Fatalf("version 0: %v", err)
587 }
588
589 // Encoded batch = 4 (count) + 1 (kind) + 1 (klen) + 1 (key) + 1 (vlen) + v.
590 b.Reset()
591 b.Put([]byte("k"), make([]byte, 93))
592 if seq, err := db.Apply(bg, &b); !errors.Is(err, ErrBatchTooLarge) || seq != 0 {
593 t.Fatalf("101-byte batch: %d, %v", seq, err)
594 }
595 b.Reset()
596 b.Put([]byte("k"), make([]byte, 92))
597 if _, err := db.Apply(bg, &b); err != nil {
598 t.Fatalf("100-byte batch: %v", err)
599 }
600 if db.Snapshot(bg).Seq() != 1 {
601 t.Fatalf("Seq = %d: rejected batches reached the log", db.Snapshot(bg).Seq())
602 }
603}
604
605func TestOptionValidation(t *testing.T) {
606 bad := []Options{
607 {MaxBatchBytes: maxRecord + 1},
608 {MaxBatchBytes: maxRecord - 11},
609 {MaxBatchBytes: -1},
610 {GroupWindow: -time.Second},
611 }
612 for _, o := range bad {
613 fx := faultfs.New()
614 if db, err := open(testCtx(t), fx, testDir, o); err == nil {
615 _ = db.Close(bg)
616 t.Errorf("open(%+v) succeeded", o)
617 }
618 if _, err := fx.ReadDir(testDir); err == nil {
619 t.Errorf("open(%+v) created the directory", o)
620 }
621 }
622 if _, err := Open(testCtx(t), t.TempDir(), Options{MaxBatchBytes: 65 << 20}); err == nil {
623 t.Error("Open with MaxBatchBytes > 64 MiB succeeded")
624 }
625 for _, o := range []Options{{MaxBatchBytes: maxRecord - 12}, {CompactAfter: -1}, {NoSync: true, CompactAfter: 1}} {
626 db, err := open(testCtx(t), faultfs.New(), testDir, o)
627 if err != nil {
628 t.Errorf("open(%+v): %v", o, err)
629 continue
630 }
631 _ = db.Close(bg)
632 }
633}
634
635func TestLocked(t *testing.T) {
636 t.Run("faultfs", func(t *testing.T) {
637 fx := faultfs.New()
638 db := mustOpen(t, fx, Options{})
639 if _, err := open(testCtx(t), fx, testDir, Options{}); !errors.Is(err, ErrLocked) {
640 t.Fatalf("second open: %v, want ErrLocked", err)
641 }
642 if err := db.Close(bg); err != nil {
643 t.Fatal(err)
644 }
645 mustOpen(t, fx, Options{})
646 })
647 t.Run("os", func(t *testing.T) {
648 dir := t.TempDir()
649 db, err := Open(testCtx(t), dir, Options{})
650 if err != nil {
651 t.Fatal(err)
652 }
653 if _, err := Open(testCtx(t), dir, Options{}); !errors.Is(err, ErrLocked) {
654 t.Fatalf("second Open: %v, want ErrLocked", err)
655 }
656 putKV(t, db, "still", "works")
657 if err := db.Close(bg); err != nil {
658 t.Fatal(err)
659 }
660 db, err = Open(testCtx(t), dir, Options{})
661 if err != nil {
662 t.Fatalf("Open after Close: %v", err)
663 }
664 _ = db.Close(bg)
665 })
666}
667
668func TestApplyAfterClose(t *testing.T) {
669 db := mustOpen(t, faultfs.New(), Options{})
670 putKV(t, db, "a", "1")
671 if err := db.Close(bg); err != nil {
672 t.Fatal(err)
673 }
674 var b Batch
675 b.Put([]byte("b"), nil)
676 if _, err := db.Apply(bg, &b); !errors.Is(err, ErrClosed) {
677 t.Fatalf("Apply after Close: %v", err)
678 }
679 mustGet(t, db.Snapshot(bg), "a", "1") // reads keep working
680 if err := db.Close(bg); err != nil {
681 t.Fatalf("second Close: %v", err)
682 }
683}
684
685// TestContextBeforeDequeue blocks the writer in a slow sync so that a second
686// batch cannot be dequeued, then cancels its context.
687func TestContextBeforeDequeue(t *testing.T) {
688 synctest.Test(t, func(t *testing.T) {
689 fx := faultfs.New()
690 db := mustOpen(t, fx, Options{})
691
692 done := make(chan error, 1)
693 base := fx.Syncs()
694 fx.SetSyncDelay(300 * time.Millisecond)
695 go func() {
696 var b Batch
697 b.Put([]byte("first"), []byte("1"))
698 _, err := db.Apply(bg, &b)
699 done <- err
700 }()
701 synctest.Wait() // the writer is now inside the delayed sync
702 if fx.Syncs() == base {
703 t.Fatal("the writer did not reach the sync")
704 }
705 start := time.Now()
706
707 ctx, cancel := context.WithCancel(bg)
708 second := make(chan error, 1)
709 go func() {
710 var b Batch
711 b.Put([]byte("second"), []byte("2"))
712 _, err := db.Apply(ctx, &b)
713 second <- err
714 }()
715 synctest.Wait() // Apply is blocked in its select: the writer cannot dequeue
716 cancel()
717 if err := <-second; !errors.Is(err, context.Canceled) {
718 t.Fatalf("Apply = %v, want context.Canceled", err)
719 }
720 if d := time.Since(start); d != 0 {
721 t.Fatalf("the cancelled Apply returned after %v, it must not wait for the sync", d)
722 }
723 if err := <-done; err != nil {
724 t.Fatalf("first Apply: %v", err)
725 }
726 fx.SetSyncDelay(0)
727
728 // An already cancelled context never reaches the writer either.
729 var b Batch
730 b.Put([]byte("third"), nil)
731 if _, err := db.Apply(ctx, &b); !errors.Is(err, context.Canceled) {
732 t.Fatalf("Apply with cancelled ctx = %v", err)
733 }
734 putKV(t, db, "fourth", "4") // the writer is still healthy
735 if err := db.Close(bg); err != nil {
736 t.Fatal(err)
737 }
738 s := mustOpen(t, fx, Options{}).Snapshot(bg)
739 mustGet(t, s, "first", "1")
740 mustGet(t, s, "fourth", "4")
741 mustMissing(t, s, "second")
742 mustMissing(t, s, "third")
743 if s.Seq() != 2 {
744 t.Fatalf("Seq = %d, want 2", s.Seq())
745 }
746 })
747}
748
749func TestCloseDuringApplies(t *testing.T) {
750 for i := range short(10, 3) {
751 t.Run(strconv.Itoa(i), func(t *testing.T) {
752 fx := faultfs.New()
753 fx.SetSyncDelay(100 * time.Microsecond)
754 db := mustOpen(t, fx, Options{})
755 const writers = 32
756 acked := make([]uint64, writers) // highest acknowledged n+1
757 var wg sync.WaitGroup
758 for g := range writers {
759 wg.Go(func() {
760 for n := 0; ; n++ {
761 var b Batch
762 b.Put(fmt.Appendf(nil, "g%02d", g), strconv.AppendInt(nil, int64(n), 10))
763 _, err := db.Apply(bg, &b)
764 if errors.Is(err, ErrClosed) {
765 return
766 }
767 if err != nil {
768 t.Errorf("writer %d: %v", g, err)
769 return
770 }
771 acked[g] = uint64(n) + 1
772 }
773 })
774 }
775 time.Sleep(time.Duration(2+i) * time.Millisecond)
776 closed := make(chan error, 1)
777 go func() { closed <- db.Close(bg) }()
778 select {
779 case err := <-closed:
780 if err != nil {
781 t.Errorf("Close: %v", err)
782 }
783 case <-time.After(10 * time.Second):
784 t.Fatal("Close did not return")
785 }
786 wg.Wait()
787
788 fx.SetSyncDelay(0)
789 s := mustOpen(t, fx, Options{}).Snapshot(bg)
790 for g, n := range acked {
791 if n == 0 {
792 continue
793 }
794 it, ok := s.Get(bg, fmt.Appendf(nil, "g%02d", g))
795 got, _ := strconv.Atoi(string(it.Value))
796 // The batch after the last acknowledged one may have been
797 // committed by the final group.
798 if !ok || uint64(got)+1 < n || uint64(got) > n {
799 t.Errorf("writer %d: acked %d batches, recovered n=%d (%v)", g, n, got, ok)
800 }
801 }
802 })
803 }
804}
805
806// failure is one injected I/O fault and what a plain reopen (no power loss)
807// must find of the failed group.
808type failure struct {
809 name string
810 inject func(fx *faultfs.FS)
811 kept bool // the group's bytes reach the file even though the call failed
812 cause error
813}
814
815func TestInjectedFailures(t *testing.T) {
816 failures := []failure{
817 {"write", func(fx *faultfs.FS) { fx.FailWrite(1, nil) }, false, faultfs.ErrInjected},
818 {"short write", func(fx *faultfs.FS) { fx.ShortWrite(1, 10) }, false, nil},
819 {"sync", func(fx *faultfs.FS) { fx.FailSync(1, nil) }, true, faultfs.ErrInjected},
820 }
821 for _, f := range failures {
822 for _, powerLoss := range []bool{false, true} {
823 t.Run(fmt.Sprintf("%s/powerloss=%v", f.name, powerLoss), func(t *testing.T) {
824 synctest.Test(t, func(t *testing.T) { testInjectedFailure(t, f, powerLoss) })
825 })
826 }
827 }
828}
829
830func testInjectedFailure(t *testing.T, f failure, powerLoss bool) {
831 fx := faultfs.New()
832 var logs syncBuffer
833 ctx := zerowrap.WithCtx(bg, zerowrap.New(zerowrap.Config{Level: "debug", Format: "json", Output: &logs}))
834 db := mustOpenCtx(t, ctx, fx, Options{})
835 wait := holdWriter(t, db, fx) // seq 1: "hold"; its write and sync have started
836 f.inject(fx)
837
838 // Three batches queue behind the held one and form the failed group.
839 const group = 3
840 errs := make([]error, group)
841 var wg sync.WaitGroup
842 for i := range group {
843 wg.Go(func() {
844 var b Batch
845 b.Put([]byte(fmt.Sprintf("failed%d", i)), []byte("x"))
846 _, errs[i] = db.Apply(bg, &b)
847 })
848 }
849 wg.Wait()
850 wait()
851 for i, err := range errs {
852 if !errors.Is(err, ErrFailed) {
853 t.Fatalf("batch %d: %v, want ErrFailed", i, err)
854 }
855 if f.cause != nil && !errors.Is(err, f.cause) {
856 t.Fatalf("batch %d: %v does not wrap %v", i, err, f.cause)
857 }
858 }
859 if !strings.Contains(logs.String(), `"level":"error"`) || !strings.Contains(logs.String(), "database failed") {
860 t.Errorf("failure not logged: %q", logs.String())
861 }
862
863 // Every later Apply fails at once; reads serve the last snapshot.
864 for range 3 {
865 var b Batch
866 b.Put([]byte("later"), nil)
867 if _, err := db.Apply(bg, &b); !errors.Is(err, ErrFailed) {
868 t.Fatalf("later Apply: %v, want ErrFailed", err)
869 }
870 }
871 var empty Batch
872 if _, err := db.Apply(bg, &empty); !errors.Is(err, ErrFailed) {
873 t.Fatalf("empty Apply on a failed db: %v", err)
874 }
875 s := db.Snapshot(bg)
876 if s.Seq() != 1 || s.Len() != 1 {
877 t.Fatalf("snapshot after failure: seq %d len %d", s.Seq(), s.Len())
878 }
879 mustGet(t, s, "hold", "")
880 mustMissing(t, s, "failed0")
881 if err := db.Close(bg); !errors.Is(err, ErrFailed) {
882 t.Fatalf("Close = %v, want the failure", err)
883 }
884 if err := db.Close(bg); err != nil {
885 t.Fatalf("second Close = %v", err)
886 }
887
888 if powerLoss {
889 fx.PowerLoss(faultfs.PowerLossOptions{})
890 }
891 s = mustOpen(t, fx, Options{}).Snapshot(bg)
892 mustGet(t, s, "hold", "")
893 present := 0
894 for i := range group {
895 if _, ok := s.Get(bg, fmt.Appendf(nil, "failed%d", i)); ok {
896 present++
897 }
898 }
899 want := 0
900 if f.kept && !powerLoss {
901 want = group
902 }
903 if present != want {
904 t.Fatalf("%d of %d failed batches recovered, want %d", present, group, want)
905 }
906 if s.Seq() != uint64(1+present) {
907 t.Fatalf("Seq = %d, want %d", s.Seq(), 1+present)
908 }
909}
910
911// TestFailedWriterAnswersOverflow checks that no Apply hangs when the group
912// that fails leaves a dequeued request behind.
913func TestFailedWriterAnswersOverflow(t *testing.T) {
914 synctest.Test(t, func(t *testing.T) {
915 fx := faultfs.New()
916 db := mustOpen(t, fx, Options{GroupWindow: 100 * time.Millisecond})
917 fx.FailWrite(1, nil)
918 chunk := strings.Repeat("c", 3<<20) // two do not fit in one group
919 errs := make([]error, 4)
920 var wg sync.WaitGroup
921 for i := range errs {
922 wg.Go(func() {
923 var b Batch
924 b.Put(fmt.Appendf(nil, "k%d", i), []byte(chunk))
925 _, errs[i] = db.Apply(bg, &b)
926 })
927 }
928 wg.Wait()
929 for i, err := range errs {
930 if !errors.Is(err, ErrFailed) {
931 t.Errorf("batch %d: %v", i, err)
932 }
933 }
934 })
935}
936
937type syncBuffer struct {
938 mu sync.Mutex
939 b strings.Builder
940}
941
942func (s *syncBuffer) Write(p []byte) (int, error) {
943 s.mu.Lock()
944 defer s.mu.Unlock()
945 return s.b.Write(p)
946}
947
948func (s *syncBuffer) String() string {
949 s.mu.Lock()
950 defer s.mu.Unlock()
951 return s.b.String()
952}
953
954// powerLossBase returns the first seed. ERMITE_KV_SEED replays the same
955// parameters; goroutine scheduling still varies between runs.
956func powerLossBase(t *testing.T) uint64 {
957 if v := os.Getenv("ERMITE_KV_SEED"); v != "" {
958 seed, err := strconv.ParseUint(v, 10, 64)
959 if err != nil {
960 t.Fatalf("ERMITE_KV_SEED: %v", err)
961 }
962 return seed
963 }
964 return uint64(time.Now().UnixNano())
965}
966
967// TestPowerLoss cuts the power at a random point while writers run, then
968// checks AC2: every acknowledged batch is present and the recovered state is
969// exactly a prefix of the applied batches.
970//
971// Every writer g applies batches n = 0, 1, … that set cur/g = n and
972// all/g/n, and delete all/g/(n-1). The recovered state therefore says how
973// many batches of each writer were applied, and the seq must equal their sum.
974// The modes include KeepPages (a random subset of the unsynced pages
975// survives). They always run with NoSync=false: with NoSync, KeepPages can
976// legitimately make Open fail with ErrCorrupt, which is documented on
977// Options.NoSync and not tested.
978func TestPowerLoss(t *testing.T) {
979 modes := []struct {
980 name string
981 opts faultfs.PowerLossOptions
982 }{
983 {"drop", faultfs.PowerLossOptions{}},
984 {"zerofill", faultfs.PowerLossOptions{ZeroFill: true}},
985 // Pages of the unsynced tail reach the disk in any order. One record
986 // per group makes this safe when commits are synced.
987 {"keeppages", faultfs.PowerLossOptions{KeepPages: true}},
988 {"keeppages+zerofill", faultfs.PowerLossOptions{KeepPages: true, ZeroFill: true}},
989 }
990 base := powerLossBase(t)
991 iterations := short(150, 10)
992 for _, m := range modes {
993 t.Run(m.name, func(t *testing.T) {
994 for i := range iterations {
995 seed := base + uint64(i)
996 if !runPowerLoss(t, seed, m.opts) {
997 t.Fatalf("failed with seed %d (ERMITE_KV_SEED=%d replays its parameters)", seed, seed)
998 }
999 }
1000 })
1001 }
1002}
1003
1004func runPowerLoss(t *testing.T, seed uint64, plo faultfs.PowerLossOptions) (ok bool) {
1005 rng := rand.New(rand.NewPCG(seed, 0x9e3779b97f4a7c15))
1006 fx := faultfs.New()
1007 db, err := open(testCtx(t), fx, testDir, Options{GroupWindow: time.Duration(rng.IntN(3)) * 100 * time.Microsecond})
1008 if err != nil {
1009 t.Errorf("seed %d: open: %v", seed, err)
1010 return false
1011 }
1012 fx.SetSyncDelay(time.Duration(rng.IntN(300)) * time.Microsecond)
1013 writers := 1 + rng.IntN(8)
1014 acked := make([]int, writers) // last acknowledged n, -1 if none
1015 ackSeq := make([]uint64, writers)
1016 var wg sync.WaitGroup
1017 for g := range writers {
1018 acked[g] = -1
1019 wg.Go(func() {
1020 for n := 0; ; n++ {
1021 var b Batch
1022 b.Put(fmt.Appendf(nil, "cur/%d", g), strconv.AppendInt(nil, int64(n), 10))
1023 b.Put(fmt.Appendf(nil, "all/%d/%d", g, n), nil)
1024 if n > 0 {
1025 b.Delete(fmt.Appendf(nil, "all/%d/%d", g, n-1))
1026 }
1027 seq, err := db.Apply(bg, &b)
1028 if err != nil {
1029 return
1030 }
1031 acked[g], ackSeq[g] = n, seq
1032 }
1033 })
1034 }
1035 time.Sleep(time.Duration(rng.IntN(4000)) * time.Microsecond)
1036 plo.Rand = rand.New(rand.NewPCG(seed, 7))
1037 if plo.KeepPages {
1038 plo.PageSize = 64 // small pages: a group record spans several of them
1039 }
1040 fx.PowerLoss(plo)
1041 wg.Wait()
1042 _ = db.Close(bg)
1043 fx.SetSyncDelay(0)
1044
1045 db, err = open(testCtx(t), fx, testDir, Options{})
1046 if err != nil {
1047 t.Errorf("seed %d: reopen after power loss: %v", seed, err)
1048 return false
1049 }
1050 defer func() { _ = db.Close(bg) }()
1051 if !checkPowerLossState(t, seed, db.Snapshot(bg), acked, ackSeq) {
1052 return false
1053 }
1054
1055 // The recovered log accepts new records and survives another reopen.
1056 seq := db.Snapshot(bg).Seq()
1057 if got := putKV(t, db, "after", "x"); got != seq+1 {
1058 t.Errorf("seed %d: seq after recovery = %d, want %d", seed, got, seq+1)
1059 return false
1060 }
1061 if err := db.Close(bg); err != nil {
1062 t.Errorf("seed %d: Close: %v", seed, err)
1063 return false
1064 }
1065 db, err = open(testCtx(t), fx, testDir, Options{})
1066 if err != nil {
1067 t.Errorf("seed %d: second reopen: %v", seed, err)
1068 return false
1069 }
1070 defer func() { _ = db.Close(bg) }()
1071 if db.Snapshot(bg).Seq() != seq+1 {
1072 t.Errorf("seed %d: seq after second reopen = %d, want %d", seed, db.Snapshot(bg).Seq(), seq+1)
1073 return false
1074 }
1075 return true
1076}
1077
1078func checkPowerLossState(t *testing.T, seed uint64, s *Snapshot, acked []int, ackSeq []uint64) bool {
1079 fail := func(format string, args ...any) bool {
1080 t.Errorf("seed %d: "+format, append([]any{seed}, args...)...)
1081 return false
1082 }
1083 var applied uint64
1084 keys := 0
1085 for g := range acked {
1086 it, ok := s.Get(bg, fmt.Appendf(nil, "cur/%d", g))
1087 last := -1
1088 if ok {
1089 n, err := strconv.Atoi(string(it.Value))
1090 if err != nil {
1091 return fail("writer %d: bad cur value %q", g, it.Value)
1092 }
1093 last = n
1094 }
1095 if last < acked[g] {
1096 return fail("writer %d: acknowledged batch %d lost, recovered %d", g, acked[g], last)
1097 }
1098 if last == acked[g] && last >= 0 && it.Version != ackSeq[g] {
1099 return fail("writer %d: batch %d acknowledged at seq %d, recovered at %d", g, last, ackSeq[g], it.Version)
1100 }
1101 if !ok {
1102 continue
1103 }
1104 cur, ok := s.Get(bg, fmt.Appendf(nil, "all/%d/%d", g, last))
1105 if !ok || cur.Version != it.Version {
1106 return fail("writer %d: batch %d is partially applied (%v, %v)", g, last, cur, ok)
1107 }
1108 count := 0
1109 for range s.Prefix(bg, fmt.Appendf(nil, "all/%d/", g)) {
1110 count++
1111 }
1112 if count != 1 {
1113 return fail("writer %d: %d all/ keys, want 1", g, count)
1114 }
1115 applied += uint64(last) + 1
1116 keys += 2
1117 }
1118 if s.Seq() != applied {
1119 return fail("Seq = %d but %d batches are visible: not a prefix", s.Seq(), applied)
1120 }
1121 if s.Len() != keys {
1122 return fail("Len = %d, want %d", s.Len(), keys)
1123 }
1124 return true
1125}
1126
1127// countFS counts the fdatasync calls of the files it opens.
1128type countFS struct {
1129 fsys.FS
1130 syncs atomic.Int64
1131}
1132
1133func (c *countFS) OpenFile(name string, flag int, perm os.FileMode) (fsys.File, error) {
1134 f, err := c.FS.OpenFile(name, flag, perm)
1135 if err != nil {
1136 return nil, err
1137 }
1138 return countFile{f, &c.syncs}, nil
1139}
1140
1141type countFile struct {
1142 fsys.File
1143 n *atomic.Int64
1144}
1145
1146func (f countFile) Sync() error {
1147 f.n.Add(1)
1148 return f.File.Sync()
1149}
1150
1151// BenchmarkApply measures group-commit throughput with parallel writers on a
1152// real directory (NoSync=false). The syncs are counted by a thin wrapper
1153// around the OS file system, so the numbers are real fdatasync calls.
1154func BenchmarkApply(b *testing.B) {
1155 fx := &countFS{FS: fsys.OS()}
1156 db, err := open(bg, fx, filepath.Join(b.TempDir(), "db"), Options{})
1157 if err != nil {
1158 b.Fatal(err)
1159 }
1160 defer func() { _ = db.Close(bg) }()
1161 b.SetParallelism(16)
1162 value := bytes.Repeat([]byte("v"), 100)
1163 var next atomic.Uint64
1164 b.ReportAllocs()
1165 b.ResetTimer()
1166 b.RunParallel(func(pb *testing.PB) {
1167 var batch Batch
1168 for pb.Next() {
1169 batch.Reset()
1170 batch.Put(fmt.Appendf(nil, "key-%d", next.Add(1)), value)
1171 if _, err := db.Apply(bg, &batch); err != nil {
1172 b.Error(err)
1173 return
1174 }
1175 }
1176 })
1177 b.StopTimer()
1178 secs := b.Elapsed().Seconds()
1179 syncs := float64(fx.syncs.Load())
1180 b.ReportMetric(float64(b.N)/secs, "batches/s")
1181 b.ReportMetric(syncs/secs, "syncs/s")
1182 b.ReportMetric(float64(b.N)/syncs, "batches/sync")
1183}
1184
1185func TestErr(t *testing.T) {
1186 tests := []struct {
1187 name string
1188 stop func(t *testing.T, db *DB, fx *faultfs.FS)
1189 want error
1190 }{
1191 {"running", func(*testing.T, *DB, *faultfs.FS) {}, nil},
1192 {"closed", func(t *testing.T, db *DB, _ *faultfs.FS) {
1193 if err := db.Close(bg); err != nil {
1194 t.Fatalf("Close: %v", err)
1195 }
1196 }, ErrClosed},
1197 {"failed", func(t *testing.T, db *DB, fx *faultfs.FS) {
1198 fx.FailWrite(1, nil)
1199 var b Batch
1200 b.Put([]byte("k"), []byte("v"))
1201 if _, err := db.Apply(bg, &b); !errors.Is(err, ErrFailed) {
1202 t.Fatalf("Apply: %v, want ErrFailed", err)
1203 }
1204 }, ErrFailed},
1205 }
1206 for _, tt := range tests {
1207 t.Run(tt.name, func(t *testing.T) {
1208 fx := faultfs.New()
1209 db := mustOpen(t, fx, Options{})
1210 tt.stop(t, db, fx)
1211 err := db.Err()
1212 if tt.want == nil {
1213 if err != nil {
1214 t.Fatalf("Err = %v, want nil", err)
1215 }
1216 return
1217 }
1218 if !errors.Is(err, tt.want) {
1219 t.Fatalf("Err = %v, want %v", err, tt.want)
1220 }
1221 })
1222 }
1223}