test(crash): add SIGKILL subprocess crash test

commit: aa8d6de920d97448421fd1aa86b1bbb3ba80a815
author: brice · 13d ago · 2026-10-10 02:53 UTC
parent: f83ddf7
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

1 file · +144 -0

A crash_test.go +144 -0

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

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