1// SPDX-License-Identifier: MIT
2
3package kv
4
5import (
6 "bytes"
7 "errors"
8 "math/rand/v2"
9 "os"
10 "os/exec"
11 "strconv"
12 "strings"
13 "sync"
14 "syscall"
15 "testing"
16 "time"
17)
18
19// crashChildEnv names the directory the child process writes to; the child
20// test is skipped when it is unset.
21const crashChildEnv = "ERMITE_KV_CRASH_CHILD"
22
23// crashIterations returns ERMITE_KV_CRASH_ITERATIONS (default 20).
24func crashIterations(t *testing.T) int {
25 v := os.Getenv("ERMITE_KV_CRASH_ITERATIONS")
26 if v == "" {
27 return 20
28 }
29 n, err := strconv.Atoi(v)
30 if err != nil || n < 1 {
31 t.Fatalf("ERMITE_KV_CRASH_ITERATIONS=%q: want a positive integer", v)
32 }
33 return n
34}
35
36// crashOptions makes the child compact every few dozen batches.
37func crashOptions() Options { return Options{CompactAfter: 20 * recSize} }
38
39// TestCrashChild is the writer run by TestCrash in a child process. It
40// continues the sequence k<i> → v<i> where the previous run stopped and prints
41// the seq of each acknowledged batch on its own line, until it is killed.
42func TestCrashChild(t *testing.T) {
43 dir := os.Getenv(crashChildEnv)
44 if dir == "" {
45 t.Skip("only runs as the child of TestCrash")
46 }
47 db, err := Open(bg, dir, crashOptions())
48 if err != nil {
49 t.Fatalf("open: %v", err)
50 }
51 deadline := time.Now().Add(time.Minute) // do not outlive a parent that died
52 for i := int(db.Snapshot(bg).Seq()) + 1; time.Now().Before(deadline); i++ {
53 if err := applySeq(db, i); err != nil {
54 t.Fatalf("apply %d: %v", i, err)
55 }
56 // One write per line: the parent never sees half of a number.
57 if _, err := os.Stdout.WriteString(strconv.Itoa(i) + "\n"); err != nil {
58 t.Fatalf("report %d: %v", i, err)
59 }
60 }
61}
62
63// ackReader records the highest seq the child reported and keeps every
64// other line (test failures are written to stdout).
65type ackReader struct {
66 mu sync.Mutex
67 rest []byte
68 last int
69 other strings.Builder
70}
71
72func (a *ackReader) Write(p []byte) (int, error) {
73 a.mu.Lock()
74 defer a.mu.Unlock()
75 a.rest = append(a.rest, p...)
76 for {
77 line, rest, ok := bytes.Cut(a.rest, []byte{'\n'})
78 if !ok {
79 return len(p), nil
80 }
81 if n, err := strconv.Atoi(string(line)); err == nil {
82 a.last = max(a.last, n)
83 } else {
84 a.other.Write(line)
85 a.other.WriteByte('\n')
86 }
87 a.rest = rest
88 }
89}
90
91func (a *ackReader) output() string {
92 a.mu.Lock()
93 defer a.mu.Unlock()
94 return a.other.String()
95}
96
97func (a *ackReader) lastAcked() int {
98 a.mu.Lock()
99 defer a.mu.Unlock()
100 return a.last
101}
102
103// TestCrash checks AC2 against SIGKILL: a child process applies sequential
104// batches with NoSync=false and frequent compactions on a real directory and
105// is killed at a random time. After each kill the database must reopen, hold
106// every batch the child reported as acknowledged, and be exactly the batches
107// 1..N for some N at least as large. The next child continues from there, so
108// the directory goes through every kind of interrupted compaction.
109func TestCrash(t *testing.T) {
110 if testing.Short() {
111 t.Skip("spawns child processes")
112 }
113 seed := powerLossBase(t)
114 t.Logf("seed %d (ERMITE_KV_SEED=%d reproduces the kill delays)", seed, seed)
115 rng := rand.New(rand.NewPCG(seed, 0x9e3779b97f4a7c15))
116 dir := t.TempDir()
117 exe, err := os.Executable()
118 if err != nil {
119 t.Fatal(err)
120 }
121
122 total := 0
123 for i := range crashIterations(t) {
124 acks := new(ackReader)
125 cmd := exec.Command(exe, "-test.run=^TestCrashChild$", "-test.count=1")
126 cmd.Env = append(os.Environ(), crashChildEnv+"="+dir)
127 cmd.Stdout = acks
128 var stderr strings.Builder
129 cmd.Stderr = &stderr
130 if err := cmd.Start(); err != nil {
131 t.Fatal(err)
132 }
133 delay := time.Duration(20+rng.IntN(180)) * time.Millisecond
134 time.Sleep(delay)
135 _ = cmd.Process.Kill()
136 // Wait also waits for the stdout copy. The child must have died from
137 // the kill: an earlier exit means the writer itself failed.
138 werr := cmd.Wait()
139 var exitErr *exec.ExitError
140 if !errors.As(werr, &exitErr) || exitErr.Sys().(syscall.WaitStatus).Signal() != syscall.SIGKILL {
141 t.Fatalf("iteration %d: child did not die from SIGKILL (%v)\nchild output: %s\nchild stderr: %s",
142 i, werr, acks.output(), stderr.String())
143 }
144 acked := acks.lastAcked()
145
146 db, err := Open(bg, dir, Options{})
147 if err != nil {
148 t.Fatalf("iteration %d (killed after %v, %d acknowledged, seed %d): reopen: %v\nchild stderr: %s",
149 i, delay, acked, seed, err, stderr.String())
150 }
151 n, err := checkSequential(db.Snapshot(bg), acked)
152 if cerr := db.Close(bg); err == nil {
153 err = cerr
154 }
155 if err != nil {
156 t.Fatalf("iteration %d (killed after %v, %d acknowledged, seed %d): %v", i, delay, acked, seed, err)
157 }
158 total = n
159 }
160 if total == 0 {
161 t.Fatal("the child never acknowledged a batch")
162 }
163 t.Logf("%d batches recovered after %d kills", total, crashIterations(t))
164}