260 lines
12 KiB
Go
260 lines
12 KiB
Go
package store
|
||
|
||
import (
|
||
"bufio"
|
||
"errors"
|
||
"fmt"
|
||
"io"
|
||
"math"
|
||
"math/rand"
|
||
"os"
|
||
"os/exec"
|
||
"path/filepath"
|
||
"syscall"
|
||
"testing"
|
||
"time"
|
||
)
|
||
|
||
// kill9_test.go — Р6/Phase 0 acceptance test: kill -9 mid-work loses at most one in-flight call and
|
||
// NEVER desyncs the money from the checkpoints. A real SIGKILL against a subprocess (the standard
|
||
// library's helper-process pattern), not emulation.
|
||
//
|
||
// Invariant after any kill: committed_usd == SUM(checkpoints.cost_usd) — settle and the checkpoint are
|
||
// written in ONE transaction. Orphaned reserved_usd are cleared by the recovery pass on open.
|
||
//
|
||
// The helper announces its first committed checkpoint on stdout and the parent kills it only once that
|
||
// line has arrived. That handshake is what gives the assertions their subject: the kill is known to land
|
||
// on a process that has already committed, so a checkpoint the helper CONFIRMED before dying has to be
|
||
// on disk when the database is reopened. That is durability, not merely internal consistency, and it is
|
||
// the stronger of the two claims this test can make. Liveness is a fact the helper reports rather than
|
||
// an inference from a clock, so a helper that never commits is a loud failure instead of a round that
|
||
// quietly asserts nothing.
|
||
//
|
||
// The database is shared across rounds, so every stored figure is cumulative and only the INCREMENT
|
||
// speaks about the round that just ran: a total is non-zero for the rest of the run once any round has
|
||
// written, whatever the helper does afterwards. Each round therefore compares against the previous
|
||
// round's own reading and requires strict growth, which is also why no round needs an exemption.
|
||
//
|
||
// Two shapes this test deliberately does not take:
|
||
// - a line per commit would fill the 64 KB pipe and stall the helper, leaving the test measuring its
|
||
// own plumbing. The helper prints exactly once, and the parent drains to EOF regardless so nothing
|
||
// the testing framework writes later can block either.
|
||
// - polling the database while the helper runs would block on the exclusive flock store.Open takes.
|
||
// Nothing polls: the previous round's reading is this round's baseline, and no second connection is
|
||
// ever opened against a live helper.
|
||
|
||
const (
|
||
killLoopEnv = "TM_STORE_KILL_LOOP_DB"
|
||
// killLoopRoundEnv scopes the helper's request hashes to its round. The rest of a hash is the
|
||
// helper's pid and a counter that restarts at zero every round, so two rounds can only collide when
|
||
// the OS hands the second helper the pid the first one had. Checkpoints are idempotent by
|
||
// request_hash (ON CONFLICT DO NOTHING), so that collision would settle once between the two rounds,
|
||
// leave the count flat, and make the increment check below report a durability failure that never
|
||
// happened. The round makes each round's writes unambiguously its own.
|
||
killLoopRoundEnv = "TM_STORE_KILL_LOOP_ROUND"
|
||
// killLoopReady is printed by the helper exactly once, after its FIRST settle+checkpoint has
|
||
// committed. It is the whole liveness signal: everything the parent asserts afterwards is about work
|
||
// that is known to be on disk.
|
||
killLoopReady = "KILL9-HELPER-COMMITTED-FIRST-CHECKPOINT"
|
||
// killLoopReadyWait bounds the wait for that signal. It is deliberately far larger than any plausible
|
||
// first-commit latency (the race detector's ~15× on this loop included): it is not a performance
|
||
// budget, it is the difference between "the helper is slow" and "the helper is not committing", and
|
||
// only the second is a failure.
|
||
killLoopReadyWait = 60 * time.Second
|
||
)
|
||
|
||
func TestHelperKillLoop(t *testing.T) {
|
||
path := os.Getenv(killLoopEnv)
|
||
if path == "" {
|
||
t.Skip("helper process only")
|
||
}
|
||
s, err := Open(path)
|
||
if err != nil {
|
||
fmt.Fprintln(os.Stderr, "helper open:", err)
|
||
os.Exit(1)
|
||
}
|
||
if err := s.UpsertSnapshot("snap", "brief", `{}`); err != nil {
|
||
os.Exit(1)
|
||
}
|
||
job, err := s.EnsureJob("book", 1, "draft", "snap")
|
||
if err != nil {
|
||
os.Exit(1)
|
||
}
|
||
round := os.Getenv(killLoopRoundEnv)
|
||
// Infinite reserve → settle+checkpoint loop; the parent will SIGKILL us
|
||
// at a random moment. No Release — recovery will clear the "crashed" reservations.
|
||
for i := 0; ; i++ {
|
||
res, verdict, err := s.Reserve("book", 0.002, Ceilings{BookUSD: 10_000})
|
||
if err != nil || verdict != ReserveOK {
|
||
os.Exit(1)
|
||
}
|
||
cp := Checkpoint{
|
||
RequestHash: fmt.Sprintf("hash-%s-%d-%d", round, os.Getpid(), i),
|
||
JobID: job.ID, ChunkIdx: i, Stage: "draft", Role: "translator",
|
||
ModelRequested: "m", ModelActual: "m",
|
||
ResponseText: "ответ", UsageJSON: "{}", CostUSD: 0.001,
|
||
}
|
||
if err := s.SettleWithCheckpoint(res, 0.001, cp, nil); err != nil {
|
||
os.Exit(1)
|
||
}
|
||
if i == 0 {
|
||
// ONCE, and only after the write is durable. Announcing before the settle would put the
|
||
// parent's whole assertion back on a guess.
|
||
fmt.Println(killLoopReady)
|
||
}
|
||
}
|
||
}
|
||
|
||
// killHelper is a running helper plus the one piece of bookkeeping its lifecycle needs: os/exec is
|
||
// explicit that "it is incorrect to call Wait before all reads from the pipe have completed", because
|
||
// Wait closes the pipe out from under the reader. drained is closed when the reader has reached EOF,
|
||
// and killAndReap is the only place Wait is called.
|
||
type killHelper struct {
|
||
cmd *exec.Cmd
|
||
drained <-chan struct{}
|
||
}
|
||
|
||
// killAndReap SIGKILLs a helper that is expected to be running and collects it in the order os/exec
|
||
// requires. A helper that has already exited means the round lost its subject before the kill landed,
|
||
// so that is named rather than absorbed.
|
||
func (h *killHelper) killAndReap(t *testing.T, round int) {
|
||
t.Helper()
|
||
if err := h.cmd.Process.Signal(syscall.SIGKILL); err != nil {
|
||
t.Fatalf("round %d: SIGKILL: %v", round, err)
|
||
}
|
||
h.reap()
|
||
}
|
||
|
||
// reap leaves nothing of the helper behind. The signal is best-effort because reap also serves the
|
||
// paths where the helper may already have exited on its own; what it guarantees is that a live helper
|
||
// is stopped BEFORE the reader is waited on. The reader drains until EOF, and a process still running
|
||
// never produces one — so waiting for it first would hang the test where it means to fail it.
|
||
func (h *killHelper) reap() {
|
||
_ = h.cmd.Process.Signal(syscall.SIGKILL)
|
||
<-h.drained // every read from the pipe has completed…
|
||
_ = h.cmd.Wait() // …and only now may Wait close it
|
||
}
|
||
|
||
// startKillHelper launches the helper and returns once it has COMMITTED its first checkpoint, so the
|
||
// caller knows the kill it is about to send will land on a process that has already done real work.
|
||
// Everything it can go wrong with is a loud failure, never a skip and never a silent pass.
|
||
func startKillHelper(t *testing.T, round int, dbPath string) *killHelper {
|
||
t.Helper()
|
||
exe, err := os.Executable()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
cmd := exec.Command(exe, "-test.run", "^TestHelperKillLoop$", "-test.v")
|
||
cmd.Env = append(os.Environ(), killLoopEnv+"="+dbPath, fmt.Sprintf("%s=%d", killLoopRoundEnv, round))
|
||
stdout, err := cmd.StdoutPipe()
|
||
if err != nil {
|
||
t.Fatalf("round %d: helper stdout pipe: %v", round, err)
|
||
}
|
||
if err := cmd.Start(); err != nil {
|
||
t.Fatalf("round %d: start helper: %v", round, err)
|
||
}
|
||
// One reader for the whole life of the process: it reports the marker and then keeps draining, so the
|
||
// pipe can never fill and stall the helper mid-loop. It reports reaching EOF without the marker as an
|
||
// error rather than as silence, because "the helper never committed" and "the helper died" are
|
||
// different failures and a test that confuses them sends the next reader down the wrong path.
|
||
ready := make(chan error, 1)
|
||
drained := make(chan struct{})
|
||
go func() {
|
||
defer close(drained)
|
||
sc := bufio.NewScanner(stdout)
|
||
seen := false
|
||
for sc.Scan() {
|
||
if !seen && sc.Text() == killLoopReady {
|
||
seen = true
|
||
ready <- nil
|
||
}
|
||
}
|
||
if !seen {
|
||
// A scanner failure and a clean EOF are different facts about the helper — one is a broken
|
||
// read, the other a process that ran and said nothing — and the reader keeps them apart so
|
||
// the parent's diagnosis names which one happened.
|
||
if err := sc.Err(); err != nil {
|
||
ready <- fmt.Errorf("the helper's stdout could not be read to the end: %w", err)
|
||
} else {
|
||
ready <- errors.New("the helper's stdout reached a clean EOF without the commit marker")
|
||
}
|
||
}
|
||
// An over-long line would have stopped the scanner above; keep draining regardless so a stalled
|
||
// pipe can never be mistaken for a stalled store.
|
||
_, _ = io.Copy(io.Discard, stdout)
|
||
}()
|
||
h := &killHelper{cmd: cmd, drained: drained}
|
||
select {
|
||
case err := <-ready:
|
||
if err != nil {
|
||
h.reap()
|
||
t.Fatalf("round %d: %v — there was nothing for SIGKILL to land on, so everything asserted afterwards would be vacuous", round, err)
|
||
}
|
||
case <-time.After(killLoopReadyWait):
|
||
h.reap()
|
||
t.Fatalf("round %d: the helper did not commit a first checkpoint within %s — the kill would have landed on a process that had done nothing, and everything asserted afterwards would be vacuous",
|
||
round, killLoopReadyWait)
|
||
}
|
||
return h
|
||
}
|
||
|
||
func TestKillMinus9LosesAtMostOneCall(t *testing.T) {
|
||
if testing.Short() {
|
||
t.Skip("subprocess test")
|
||
}
|
||
path := filepath.Join(t.TempDir(), "kill9.db")
|
||
|
||
// The database is shared across rounds, so every stored figure is cumulative. The previous round's own
|
||
// reading is this round's baseline — which is also why no second connection is ever opened while the
|
||
// helper holds the project lock.
|
||
var prevCheckpoints int
|
||
var prevCommitted float64
|
||
|
||
for round := 0; round < 5; round++ { // D23.4: 3→5 rounds — more SIGKILL windows per run
|
||
helper := startKillHelper(t, round, path)
|
||
// The kill lands at a random point of a reserve→settle cycle. This delay varies WHERE it lands;
|
||
// whether there is anything for it to land on was settled by the readiness signal.
|
||
time.Sleep(time.Duration(rand.Intn(200)) * time.Millisecond)
|
||
helper.killAndReap(t, round)
|
||
|
||
s, err := Open(path) // the recovery pass clears the stuck reservations
|
||
if err != nil {
|
||
t.Fatalf("round %d: reopen after kill: %v", round, err)
|
||
}
|
||
|
||
var checkpointSum float64
|
||
var checkpointCount int
|
||
if err := s.r.QueryRow(`SELECT COALESCE(SUM(cost_usd),0), COUNT(*) FROM checkpoints`).Scan(&checkpointSum, &checkpointCount); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
committed, reserved, err := s.SpentUSD("book")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
s.Close()
|
||
|
||
if reserved != 0 {
|
||
t.Fatalf("round %d: recovery must clear reservations, reserved=%v", round, reserved)
|
||
}
|
||
// Acceptance core: money and checkpoints are atomic — a mismatch would mean
|
||
// a paid-but-lost (or free-but-recorded) call.
|
||
if math.Abs(committed-checkpointSum) > 1e-9 {
|
||
t.Fatalf("round %d: committed=%v != sum(checkpoints)=%v over %d checkpoints — settle+checkpoint not atomic",
|
||
round, committed, checkpointSum, checkpointCount)
|
||
}
|
||
// DURABILITY: the helper reported a committed checkpoint before it was killed, so this round must
|
||
// have added at least one, in every round including the first.
|
||
if checkpointCount <= prevCheckpoints {
|
||
t.Fatalf("round %d: %d checkpoints before the kill, %d after reopening — a checkpoint the helper confirmed as committed did not survive SIGKILL, or the round did no work the previous rounds had not already done",
|
||
round, prevCheckpoints, checkpointCount)
|
||
}
|
||
// The money must have moved with them, by the same increment logic.
|
||
if committed <= prevCommitted {
|
||
t.Fatalf("round %d: committed went from $%.6f to $%.6f across a round that added %d checkpoint(s)",
|
||
round, prevCommitted, committed, checkpointCount-prevCheckpoints)
|
||
}
|
||
t.Logf("round %d: +%d checkpoints (%d total), committed=$%.6f — a confirmed commit survived SIGKILL and the ledger matches",
|
||
round, checkpointCount-prevCheckpoints, checkpointCount, committed)
|
||
prevCheckpoints, prevCommitted = checkpointCount, committed
|
||
}
|
||
}
|