textmachine/platform/internal/runs/control_test.go

1356 lines
60 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package runs
import (
"errors"
"os"
"path/filepath"
"slices"
"testing"
"time"
"textmachine/platform/internal/ingest"
"textmachine/platform/internal/money"
"textmachine/platform/internal/pgstore"
"textmachine/platform/internal/pricing"
"textmachine/platform/internal/runner"
)
// The control handles — stop and resume — are the two places where a USER's decision reaches a run
// that is not this process's child. What they have to get right is not the systemd call, which is
// one line, but the record around it: a stop that is not written down before the signal cannot be
// told from a crash afterwards, and a resume that forgets the previous attempt's hold reserves the
// run's ceiling twice.
// stopped drives a fixture to a run the user stopped: admitted, spawned, asked to stop, and then
// reconciled from the marker the unit left. It is the state resume starts from.
//
// spent is what the engine reports AFTER the spawn, which is the only order that models the real
// one: the spawn reads the book's meter to take this attempt's baseline, so a figure set before it
// would be read as money some earlier run had spent and this attempt would settle at nothing.
func (f *fixture) stopped(t *testing.T, chapters int, spent money.MicroUSD, marker runner.Marker) string {
t.Helper()
return f.stoppedRun(t, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(chapters)}, spent, marker)
}
// stoppedRun is the same walk for a run whose START matters — a signing run, say.
func (f *fixture) stoppedRun(t *testing.T, in StartRequest, spent money.MicroUSD, marker runner.Marker) string {
t.Helper()
run, err := f.svc.Start(f.ctx, in)
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
live := f.live(t)
marker.Unit = live.UnitName
if err := runner.WriteMarker(f.svc.markerPath(live.RunID, live.AttemptNo), marker); err != nil {
t.Fatal(err)
}
f.engine.set(statusSpending(spent), nil)
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
return run.ID
}
// A run is REFUSED over a book whose chapter tree was never materialised, and refused BEFORE its
// hold — which is the whole point of where the guard sits (PD-405).
//
// The window is the intake's own and it is ordinary: `not_started` is committed in one transaction
// and the tree is materialised after it, outside. On the healthy path that is milliseconds (measured
// at 0.018 s, act D39.162); when the materialisation FAILED or was deferred it is unbounded, and that
// is the shape that costs money. A run admitted there can never move: every counter the screen shows
// is a count over `chapters` and there are no rows to count, so the bar reads 0/total for the run's
// whole life — and the tree's own debt is frozen meanwhile, because the sweep that would pay it skips
// a book with a live run. The user pays for a run that shows nothing and cannot be repaired.
//
// Mutation caught: moving the guard below the hold; dropping the `ChapterCount > 0` term (which would
// refuse a book that legitimately declares no chapters).
func TestARunIsRefusedOverABookWhoseChaptersWereNeverMaterialised(t *testing.T) {
f := newFixture(t, "10", 500)
book := f.bookID(t)
// The tree goes away: the state the intake is in between its commit and its materialisation, and
// the state it STAYS in when that materialisation broke.
if _, err := f.store.Pool().Exec(f.ctx, `delete from chapters where book_id = $1`, book); err != nil {
t.Fatal(err)
}
before := f.account(t)
_, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, Chapters: order(10)})
if !errors.Is(err, ErrBookNotReady) {
t.Fatalf("a run over a book with no chapter tree was answered %v, want book_not_ready: its bar"+
" can never move and its tree's debt is frozen for as long as it runs", err)
}
// BEFORE the hold, and this is the half that is money rather than honesty.
after := f.account(t)
if after.Balance != before.Balance || after.LedgerSum != before.LedgerSum {
t.Errorf("the refusal moved money: balance %s → %s, ledger %s → %s",
before.Balance.USD(), after.Balance.USD(), before.LedgerSum.USD(), after.LedgerSum.USD())
}
// And the same book with its tree back is startable: the guard refuses the STATE, not the book.
//
// ⚠ THE TREE COMES BACK WITH ITS UNITS, and it did not have to before this pack. An order is now
// priced from the UNDELIVERED UNITS themselves rather than from a chapter's total divided by a
// count of them, so a chapter row with no unit rows under it has nothing to sell and is excluded —
// which is right, and which no writer produces: `SaveStructure` writes a chapter and its units in
// one pass. Re-inserting chapters alone described a tree the materializer cannot make, and the
// book then answered «there is nothing left of this book to buy» instead of starting.
if _, err := f.store.Pool().Exec(f.ctx, `
insert into chapters (id, book_id, number, units_total)
select 'c' || g, $1, g, 1 from generate_series(1, 500) g`, book); err != nil {
t.Fatal(err)
}
if _, err := f.store.Pool().Exec(f.ctx, `
insert into units (id, chapter_id, ordinal, source, target, state, source_chars, expected_micro_usd)
select 'u' || g, 'c' || g, 0, 'src', '', 'pending', 1000, $1 from generate_series(1, 500) g`,
int64(fixtureChapterUSD)); err != nil {
t.Fatal(err)
}
if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, Chapters: order(10)}); err != nil {
t.Fatalf("the book was refused with its tree materialised: %v", err)
}
}
// The OTHER half of the same guard, and it exists because this pack's adversarial pass showed the
// header above claiming a mutation nothing caught: dropping the `ChapterCount > 0` term left the
// whole eighteen-package battery green, because no test built the book it would newly refuse.
//
// A book that declares NO chapters has no tree to be missing, so the guard must not touch it. It is
// refused, but by the thing that should refuse it — there is nothing to buy — and the two refusals
// carry different remedies: "wait for the intake" is a lie told about a book whose intake is over.
//
// Mutation caught: dropping the `ChapterCount > 0` term from the admission guard.
func TestABookThatDeclaresNoChaptersIsNotRefusedAsUnmaterialised(t *testing.T) {
f := newFixture(t, "10", 0)
book := f.bookID(t)
var count int
if err := f.store.Pool().QueryRow(f.ctx,
`select chapter_count from books where id = $1`, book).Scan(&count); err != nil {
t.Fatal(err)
}
if count != 0 {
t.Fatalf("the fixture book declares %d chapters: this test is about the one that declares none", count)
}
_, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: book, Chapters: order(1)})
if errors.Is(err, ErrBookNotReady) {
t.Errorf("a book that declares no chapters was told its chapters are still being materialised:"+
" %v — its intake is OVER, and the honest refusal is that there is nothing to buy", err)
}
if err == nil {
t.Error("a run was admitted over a book with nothing in it")
}
}
// secondBook registers another book of the same account, with a directory of its own.
func (f *fixture) secondBook(t *testing.T) string {
t.Helper()
id, err := f.store.AddBook(f.ctx, pgstore.NewBook{OwnerID: "u1", Title: "second", SourceLang: "zh",
TargetLang: "ru", ChapterCount: 100, Workdir: t.TempDir(), Now: f.now})
if err != nil {
t.Fatal(err)
}
// …with its chapter TREE, for the same reason the main fixture has one: a book that declares
// chapters and materialised none cannot be started over (PD-405).
if _, err := f.store.Pool().Exec(f.ctx, `
insert into chapters (id, book_id, number, units_total)
select $1 || ':c' || g, $1, g, 1 from generate_series(1, 100) g`, id); err != nil {
t.Fatal(err)
}
// …and with its PRICE, for the same reason again: an unpriced book is refused rather than sold at
// a constant (runs.ErrNotPriced), so a second book without one would fail the test that uses it
// for a reason that has nothing to do with what the test is about.
if _, err := f.store.Pool().Exec(f.ctx, `
update books set expected_micro_usd = $2, book_once_micro_usd = 0, step_max_micro_usd = $3,
source_chars = 100000, structure = 'detected' where id = $1`,
id, int64(fixtureChapterUSD)*100, int64(fixtureStepMaxUSD)); err != nil {
t.Fatal(err)
}
if _, err := f.store.Pool().Exec(f.ctx, `
insert into units (id, chapter_id, ordinal, source, target, state, source_chars, expected_micro_usd)
select $1 || ':u' || g, $1 || ':c' || g, 0, 'src', '', 'pending', 1000, $2
from generate_series(1, 100) g`, id, int64(fixtureChapterUSD)); err != nil {
t.Fatal(err)
}
return id
}
// lastAttempt is the snapshot a slow sweep pass would still be holding: the run as it was, with the
// attempt that was live at the time.
func (f *fixture) lastAttempt(t *testing.T, runID string) pgstore.LiveRun {
t.Helper()
l, err := f.store.ReadRunForResume(f.ctx, "u1", runID)
if err != nil {
t.Fatal(err)
}
return l
}
func (f *fixture) run(t *testing.T, runID string) pgstore.Run {
t.Helper()
r, err := f.store.ReadRun(f.ctx, "u1", runID)
if err != nil {
t.Fatal(err)
}
return r
}
// The heart of PD-152. The engine catches SIGTERM and exits 1, so the marker of a stop and the
// marker of a crash are the same bytes; what tells them apart is that the platform wrote down its
// own intent first.
func TestARunTheUserStoppedIsNotReportedAsFailed(t *testing.T) {
f := newFixture(t, "10", 500)
runID := f.stopped(t, 100, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
At: f.now.Add(time.Second)})
if got := f.run(t, runID); got.Status != "stopped" {
t.Fatalf("a stop the user asked for came back as %q, want stopped", got.Status)
}
// And the same marker WITHOUT the intent is still a failure: the discriminator has to be the
// record, not a new reading of the exit code.
g := newFixture(t, "10", 500)
run, err := g.svc.Start(g.ctx, StartRequest{UserID: "u1", BookID: g.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
if err := g.svc.Spawn(g.ctx, run.ID); err != nil {
t.Fatal(err)
}
live := g.live(t)
if err := runner.WriteMarker(g.svc.markerPath(live.RunID, live.AttemptNo),
runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "1"}); err != nil {
t.Fatal(err)
}
if err := g.svc.Sweep(g.ctx); err != nil {
t.Fatal(err)
}
if got := g.run(t, run.ID); got.Status != "failed" {
t.Fatalf("a run nobody stopped came back as %q, want failed", got.Status)
}
}
// The race the design owes an answer to: a run that finished by itself a moment before someone
// pressed stop was not stopped by them, and calling it `stopped` would hide a completed translation
// behind a cancelled one.
func TestARunThatEndedBeforeTheStopKeepsItsOwnOutcome(t *testing.T) {
f := newFixture(t, "10", 500)
// The marker is written a minute BEFORE the stop request the fixture makes.
runID := f.stopped(t, 100, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
At: f.now.Add(-time.Minute)})
if got := f.run(t, runID); got.Status != "failed" {
t.Fatalf("a run that had already ended came back as %q, want failed", got.Status)
}
}
// The engine's own clean answers outrank a stop that arrived while it was already finishing: a
// translation that COMPLETED must not be shown as cancelled.
func TestACleanExitOutranksAStopThatArrivedTooLate(t *testing.T) {
f := newFixture(t, "10", 500)
runID := f.stopped(t, 100, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "0",
At: f.now.Add(time.Second)})
if got := f.run(t, runID); got.Status != "ready" {
t.Fatalf("a run that finished cleanly came back as %q, want ready", got.Status)
}
}
// The intent is committed BEFORE systemd is asked, and that order is the whole mechanism: a platform
// that died between the two must still know, on its next sweep, that the run was stopped on purpose.
func TestTheStopIsRecordedBeforeSystemdIsAsked(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
var recordedFirst bool
f.runner.onStop = func(string) {
var at *time.Time
if err := f.store.Pool().QueryRow(f.ctx,
`select stop_requested_at from runs where id = $1`, run.ID).Scan(&at); err != nil {
t.Error(err)
return
}
recordedFirst = at != nil
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
if !recordedFirst {
t.Fatal("systemd was asked to stop the unit before the intent was committed")
}
}
// A stop that systemd did not take must not be lost. The intent is durable, so the next sweep asks
// again — which is the only thing that closes "the platform died between the write and the call".
func TestAStopTheUnitDidNotTakeIsAskedAgainByTheSweep(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
f.runner.alive = true
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
if n := len(f.runner.stops()); n != 2 {
t.Fatalf("the unit was asked to stop %d times, want 2 (the handle and the sweep)", n)
}
}
// The dangerous half: a stopped unit that left no marker looks exactly like a reboot, and the reboot
// path RESTARTS the run — which would spend the account's money on work its owner had just
// cancelled.
func TestAStoppedRunIsNotRestartedWhenItsMarkerIsMissing(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
// The unit is gone, no marker was written, and the spawn grace is long past.
f.runner.alive = false
f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
if got := f.run(t, run.ID); got.Status != "stopped" {
t.Fatalf("run is %q, want stopped", got.Status)
}
if n := len(f.runner.starts()); n != 1 {
t.Fatalf("%d units were started; a stopped run must not be restarted", n)
}
if acct := f.account(t); acct.Reserved != 0 {
t.Fatalf("the hold of a stopped run is still open: %s", acct.Reserved.USD())
}
}
// A run stopped before its unit ever existed: there is nothing to signal, and the money must come
// back whole rather than wait for a unit that will never be created.
func TestStoppingARunThatNeverSpawnedGivesTheHoldBackWhole(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
if n := len(f.runner.stops()); n != 0 {
t.Fatalf("systemd was asked about a unit that does not exist (%d times)", n)
}
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
if got := f.run(t, run.ID); got.Status != "stopped" {
t.Fatalf("run is %q, want stopped", got.Status)
}
acct := f.account(t)
if acct.Reserved != 0 || acct.Balance != money.MicroUSD(10_000_000) {
t.Fatalf("the hold of a run that never started did not come back whole: balance %s reserved %s",
acct.Balance.USD(), acct.Reserved.USD())
}
if n := len(f.runner.starts()); n != 0 {
t.Fatalf("a stopped run was spawned anyway (%d units)", n)
}
}
// Ownership, on the handle that ENDS a paid run: another account's run is not "forbidden", it is not
// there at all (API1/BOLA, and the same rule the library reads by).
func TestARunOfAnotherAccountCannotBeStoppedOrResumed(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
if _, err := f.store.Pool().Exec(f.ctx,
`insert into users (id, email) values ('u2','u2@example.org')`); err != nil {
t.Fatal(err)
}
if _, err := f.svc.Stop(f.ctx, "u2", run.ID); !errors.Is(err, pgstore.ErrNoRun) {
t.Fatalf("stop by a stranger: %v, want ErrNoRun", err)
}
if _, err := f.svc.Resume(f.ctx, "u2", run.ID); !errors.Is(err, pgstore.ErrNoRun) {
t.Fatalf("resume by a stranger: %v, want ErrNoRun", err)
}
}
// Stopping a run that is already over is a conflict and NOT a not-found: the owner can see the run,
// and answering 404 would tell them their own run does not exist.
func TestStoppingARunThatIsOverIsAConflict(t *testing.T) {
f := newFixture(t, "10", 500)
runID := f.stopped(t, 10, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
At: f.now.Add(time.Second)})
if _, err := f.svc.Stop(f.ctx, "u1", runID); !errors.Is(err, ErrNotStoppable) {
t.Fatalf("stopping a finished run: %v, want ErrNotStoppable", err)
}
}
// Resume is the reconciler's restart with a user behind it: a NEW attempt, holding what is LEFT of
// the run's budget. The money is the assertion — one run may not reserve its ceiling twice.
func TestResumeContinuesTheRunWithWhatIsLeftOfItsBudget(t *testing.T) {
f := newFixture(t, "10", 500)
spent := money.MicroUSD(500_000)
runID := f.stopped(t, 100, spent, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
At: f.now.Add(time.Second)})
budget := fixtureHold(100)
if acct := f.account(t); acct.Reserved != 0 || acct.Balance != money.MicroUSD(10_000_000)-spent {
t.Fatalf("before the resume: balance %s reserved %s", acct.Balance.USD(), acct.Reserved.USD())
}
got, err := f.svc.Resume(f.ctx, "u1", runID)
if err != nil {
t.Fatal(err)
}
if got.Status != "translating" || got.FinishedAt != nil {
t.Fatalf("the resumed run is %+v, want a live translating run", got)
}
// The new attempt holds the REMAINDER, not the whole ceiling: the first attempt already spent
// part of it and settled at what it spent.
if acct := f.account(t); acct.Reserved != budget-spent {
t.Fatalf("the resumed run holds %s, want %s (the ceiling less what was already spent)",
acct.Reserved.USD(), (budget - spent).USD())
}
// The engine is QUEUED, not started inside the request: reading the book's meter costs seconds of
// the engine's CPU and creating a unit is a round trip to systemd, and a call that answers 202
// must not hold either. (Measured, not reasoned about: the live probe's resume held its request
// until the engine exited.) With no queue wired here, the reconciler is the backstop — which is
// the same backstop the admission path relies on.
live := f.live(t)
if live.AttemptNo != 2 || live.UnitName != "" {
t.Fatalf("the resumed run's attempt is %+v, want a second attempt waiting to be spawned", live)
}
if n := len(f.runner.starts()); n != 1 {
t.Fatalf("%d units started; the resume spawned the engine inside the request", n)
}
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
if n := len(f.runner.starts()); n != 2 {
t.Fatalf("%d units started across the run, want 2 once the backstop ran", n)
}
// The book follows the run back into translation, and the library's revision moves with it.
book, _, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
if err != nil {
t.Fatal(err)
}
if book.Status != "translating" {
t.Fatalf("the book of a resumed run is %q, want translating", book.Status)
}
}
// The bank stop is the one resume the contract describes in detail, and its condition — a COMPLETE
// set of decisions — cannot be met by any deployment today: nothing materializes the bank at all
// (companion §3). Answering 202 would start a run that walks straight into the same stop.
// Signing the bank is ONE act over the whole of it, and `resume` lifts that stop with the decisions
// AS THEY STAND (owner 16.08, D39.144). This test replaces the one that pinned the opposite: the
// gate it protected — "the stop clears only on a complete set of decisions" — was an invention of
// the contract line, and the engine has always continued with an unsigned bank by default.
//
// Mutation caught: restoring the completeness gate in Resume; treating `awaiting_bank` as a status
// that falls through to the default 409.
func TestResumeLiftsABankStopWithTheDecisionsAsTheyStand(t *testing.T) {
f := newFixture(t, "10", 500)
runID := f.stopped(t, 100, 0, runner.Marker{Result: "exit-code", Code: "exited", Status: "3",
At: f.now.Add(time.Second)})
if got := f.run(t, runID); got.Status != "awaiting_bank" {
t.Fatalf("run after the bank stop is %q", got.Status)
}
// No decision has been recorded at all, which is the state every book is in until somebody opens
// the screen — and the one the old gate refused.
got, err := f.svc.Resume(f.ctx, "u1", runID)
if err != nil {
t.Fatalf("resume of a bank stop: %v, want it continued", err)
}
if got.Status != "translating" {
t.Fatalf("the resumed run is %q, want it going again", got.Status)
}
}
// A run that spent the whole ceiling it bought is REFUSED: there is no work this call could pay for,
// so the remedy is a new run. Nothing moves — no second hold, no second unit — and the run's own
// state is not rewritten on the way out: a run the user stopped stays stopped.
//
// ⚠ It used to answer 202 with the run unchanged, which is a success a client cannot tell from a
// continuation (PD-282). The canon settled it the other way and now says the opposite in as many
// words: `ceiling_reached` in EVERY status, and a 202 means work was actually re-opened.
func TestResumeOfARunWithNothingLeftIsRefusedWithTheCeilingReached(t *testing.T) {
f := newFixture(t, "10", 500)
// The attempt spends its entire ceiling, so the run has no budget left at all.
runID := f.stopped(t, 10, fixtureHold(10), runner.Marker{Result: "exit-code",
Code: "exited", Status: "1", At: f.now.Add(time.Second)})
before := f.account(t)
_, err := f.svc.Resume(f.ctx, "u1", runID)
if !errors.Is(err, ErrCeilingReached) {
t.Fatalf("resume of a run with nothing left: %v, want ErrCeilingReached", err)
}
if got := f.run(t, runID); got.Status != "stopped" || got.PausedReason != "" {
t.Fatalf("the refusal rewrote the run to %q/%q, want the stopped run it was", got.Status, got.PausedReason)
}
if acct := f.account(t); acct.Reserved != 0 || acct.Balance != before.Balance {
t.Fatalf("a resume that changed nothing moved money: %+v -> %+v", before, acct)
}
if n := len(f.runner.starts()); n != 1 {
t.Fatalf("%d units started, want 1: nothing was resumed", n)
}
}
// The OTHER refusal, and the whole reason the two are told apart: this run has chapters left in the
// ceiling it bought and the ACCOUNT cannot cover them. The remedy is money — after which this same
// call continues this same run — so answering `ceiling_reached` would send the user off to buy a run
// they do not need.
//
// Mutation caught: folding the two verdicts back into one.
func TestResumeOverAnEmptyAccountSaysTheCreditIsUnavailableAndNotTheCeiling(t *testing.T) {
f := newFixture(t, "10", 500)
// Half the ceiling spent, so the RUN has room left; then the account is emptied under it.
runID := f.stopped(t, 10, fixtureHold(5), runner.Marker{Result: "exit-code",
Code: "exited", Status: "1", At: f.now.Add(time.Second)})
if _, err := f.store.Adjust(f.ctx, "u1", -f.account(t).Balance, "test", "drain", "spent", f.now); err != nil {
t.Fatal(err)
}
_, err := f.svc.Resume(f.ctx, "u1", runID)
if errors.Is(err, ErrCeilingReached) {
t.Fatalf("a run with room left was refused as if its ceiling were spent: %v", err)
}
if !errors.Is(err, ErrCreditUnavailable) {
t.Fatalf("resume over an empty account: %v, want ErrCreditUnavailable", err)
}
// …and topping the account up is all it takes: the same call then continues the same run.
if _, err := f.store.Grant(f.ctx, "u1", money.MicroUSD(5_000_000), "test", "topup", "", f.now); err != nil {
t.Fatal(err)
}
if _, err := f.svc.Resume(f.ctx, "u1", runID); err != nil {
t.Fatalf("a topped-up account still could not continue the run: %v", err)
}
if n := len(f.runner.starts()); n != 1 {
t.Fatalf("%d units started: the resume is queued, not spawned inline", n)
}
if got := f.run(t, runID); got.Status != "translating" {
t.Errorf("the resumed run is %q, want it going again", got.Status)
}
}
// The reconciler's half of the same decision, which the shared body must not have taken away: a run
// that was LIVE and has spent its ORDER is `paused` with the reason the contract has a word for, and
// a resume of that comes back paused too.
//
// ⚠ THE WORD CHANGED ON 05.09 AND THE FACT DID NOT (PD-446, unified backlog row 279). What is
// exhausted here is the PURCHASE — the account still holds the rest of its $10 — and the old word,
// `credit_exhausted`, said the opposite of that to the one person who reads it. `run_limit_reached`
// is the same state under its own name; `credit_exhausted` now means only what it says, and is
// asserted where it is true (TestAnInterruptedRunThatTheBalanceCannotCarryIsPaused).
func TestAnInterruptedRunWithNothingLeftIsPausedAndStaysPausedThroughAResume(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
// The engine spent the whole ceiling and then its unit vanished without a marker, which is what a
// reboot leaves behind.
f.engine.set(statusSpending(fixtureHold(10)), nil)
f.runner.alive = false
f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
got := f.run(t, run.ID)
if got.Status != "paused" || got.PausedReason != pgstore.PausedRunLimitReached {
t.Fatalf("an interrupted run that spent its order is %q/%q, want paused/run_limit_reached",
got.Status, got.PausedReason)
}
// ⚠ 409 and NOT a silent 202 with the run unchanged. The limit travels with the START of a run
// and nothing changes it afterwards, so this call can never move such a run — and an answer a
// client cannot tell from a real continuation is the defect 0.3.0 named (canon §resumeRun). The
// remedy is a NEW run with a larger ceiling, which is legal from any paused book.
if _, err := f.svc.Resume(f.ctx, "u1", run.ID); !errors.Is(err, ErrCeilingReached) {
t.Fatalf("resume of a run stopped at a limit: %v, want ErrCeilingReached", err)
}
if got := f.run(t, run.ID); got.Status != "paused" || got.PausedReason != pgstore.PausedRunLimitReached {
t.Fatalf("the refused resume rewrote the run to %q/%q", got.Status, got.PausedReason)
}
if n := len(f.runner.starts()); n != 1 {
t.Fatalf("%d units started, want 1", n)
}
}
// A settlement is allowed to defer, and a resume that ignored that would open a SECOND hold on a run
// whose first is still reserved.
func TestResumeRefusesWhileThePreviousAttemptIsStillHoldingMoney(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
live := f.live(t)
if err := runner.WriteMarker(f.svc.markerPath(live.RunID, live.AttemptNo),
runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "1",
At: f.now.Add(time.Second)}); err != nil {
t.Fatal(err)
}
// The engine cannot be asked what it spent, so the run finishes and its hold stays open.
f.engine.set(statusSpending(0), errors.New("status unavailable"))
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
if acct := f.account(t); acct.Reserved == 0 {
t.Fatal("the hold was released even though the engine could not be asked")
}
if _, err := f.svc.Resume(f.ctx, "u1", run.ID); !errors.Is(err, ErrNotResumable) {
t.Fatalf("resume while unsettled: %v, want ErrNotResumable", err)
}
if n := len(f.runner.starts()); n != 1 {
t.Fatalf("%d units started: an unsettled run must not be resumed", n)
}
}
// A book that is still arriving, or one the engine refused, has no chapter tree to translate — and
// while it is `uploading` it has half a file on disk. The refusal happens BEFORE the money moves.
func TestARunIsRefusedOnABookThatIsNotThroughIntake(t *testing.T) {
f := newFixture(t, "10", 500)
for _, status := range []string{"uploading", "parsing", "rejected"} {
if _, err := f.store.Pool().Exec(f.ctx,
`update books set status = $1 where id = $2`, status, f.bookID(t)); err != nil {
t.Fatal(err)
}
_, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
if !errors.Is(err, ErrBookNotReady) {
t.Fatalf("starting a run on a %s book: %v, want ErrBookNotReady", status, err)
}
if acct := f.account(t); acct.Reserved != 0 {
t.Fatalf("a refused run held money: %s", acct.Reserved.USD())
}
}
}
// statusSpending is a report of an engine that has committed exactly this much on the book.
func statusSpending(v money.MicroUSD) ingest.StatusReport {
return ingest.StatusReport{Spend: usd(v), Reserved: usd(0)}
}
// PD-169. A pass of the sweep has one budget for every run in it, and the list is ordered the same
// way every time: a run whose engine hangs used to eat the pass, and the tail of the list — other
// people's books — was never reached at all. The per-run budget is what bounds that, and this is the
// assertion that it does.
func TestOneSlowRunDoesNotEatThePassOfTheWholeSweep(t *testing.T) {
f := newFixture(t, "20", 500)
f.svc.Cfg.RunBudget = 100 * time.Millisecond
// Two books, two live runs, and the sweep visits them in the order they started.
first, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
second := f.secondBook(t)
if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: second, Chapters: order(10)}); err != nil {
t.Fatal(err)
}
// The engine hangs for the FIRST run only — until the context that carries it is cancelled.
f.engine.block(first.ID, f.workdir)
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
// The second run was spawned in the same pass, which is the whole property: the slow one cost its
// own budget and nothing else's.
started := f.runner.starts()
if len(started) != 1 {
t.Fatalf("%d units started; the second run was starved by the first", len(started))
}
}
// A run standing at the bank-signing stop is NOT an interruption, so the sweep does not restart it —
// it closes it AT the stop, and only the user's resume takes it further.
//
// ⚠ RE-SIGNED by pack P12 (unified-backlog row 240), and the re-signing is the point rather than an
// edit to keep a test green. This test used to pin a guard that keyed on WHICH BIT a restart wrote
// (`LiftBankStop`), and that guard was written against an engine that re-halted on everything still
// undecided: keeping `--verify-bank` on the restarted attempt was what re-created the pause. The
// engine's presented memory (storage v16, D39.158) ended that — a map whose clusters were already
// shown does not stop again, whatever the flag says — so from that day the guard protected NOTHING:
// the sweep's restart marched straight through the paid pause with one WARN in the engine's own
// journal. What holds instead is the decision one level up, and it is the same rule the ceiling
// branch beside it already follows: an engine that ended ON PURPOSE is not restarted.
//
// Mutation caught: restarting a run whose status is `awaiting_bank` when its unit vanishes.
func TestTheReconcilerDoesNotRestartPastABankStopNobodySigned(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), VerifyBank: true,
Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
// What the sink writes when the engine reports the bank stop: the status moves and the run stays
// LIVE, because the attempt has not ended yet (pgstore/sink.go, TypeBankStop).
if _, err := f.store.Pool().Exec(f.ctx,
`update runs set status = 'awaiting_bank' where id = $1`, run.ID); err != nil {
t.Fatal(err)
}
// …and then the unit dies leaving no marker — a reboot, or a SIGTERM from outside. The SERVICE's
// clock is what the grace is measured on, and it is not f.now.
f.runner.alive = false
late := f.now.Add(2 * time.Hour)
f.svc.Now = func() time.Time { return late }
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
var status string
var attempts int
var finished *time.Time
if err := f.store.Pool().QueryRow(f.ctx,
`select r.status, r.finished_at, (select count(*) from run_attempts where run_id = r.id)
from runs r where r.id = $1`, run.ID).Scan(&status, &finished, &attempts); err != nil {
t.Fatal(err)
}
// The sweep must have LOOKED at this run, or the test observes nothing: the same guard against a
// fixture that never reaches the decision which caught the first version of this test.
if !f.runner.askedAlive() {
t.Fatal("the sweep never reached this run: the fixture cannot observe the property")
}
if attempts != 1 {
t.Errorf("the sweep restarted a run standing at the signing stop (%d attempts): the pause the user paid for was spent on a march through it", attempts)
}
if status != "awaiting_bank" {
t.Errorf("the run left the signing stop as %q, want it still standing there for the user to answer", status)
}
if finished == nil {
t.Error("the run was left LIVE at a stop whose engine is gone: nothing ends it and nothing resumes it")
}
// Whatever the sweep did spawn, it carried the flag: after D39.158 the platform never masks it,
// and where the engine stops is the engine's decision.
for i, spec := range f.runner.starts() {
if !slices.Contains(spec.Args, "--verify-bank") {
t.Errorf("attempt %d was spawned without the signing stop: %v", i+1, spec.Args)
}
}
}
// THE PAID STOP SURVIVES THE TWO PATHS THAT REACH IT THROUGH THIS SWEEP'S OWN DRAIN, and both were
// open until this pack's adversarial pass measured them.
//
// The trap they share: the sweep's `LiveRun` is a by-value snapshot taken BEFORE the journal is
// drained, and the bank-stop event lands in that very drain — it writes `awaiting_bank` on the row
// and nothing the snapshot carries. A guard that asked the snapshot therefore asked a value that
// could not yet know, and the run was restarted past the pause the user paid for. After D39.158 that
// restart does not even re-stop: it carries `--verify-bank` and the engine's presented memory
// auto-continues over a map it has already shown, one WARN in the engine's own journal.
//
// Sub-test A — the unit VANISHED, and the stop arrived in this pass's drain.
// Sub-test B — a marker exists reading exit 5, an outside SIGTERM. That branch (`interruptedBySomeoneElse`)
// answers BEFORE `outcome` is ever consulted, so the rule written into `outcome` could not see it.
//
// Mutation caught: reading l.Status from the pre-drain snapshot; dropping the awaiting_bank exclusion
// from the interrupted-by-someone-else branch.
func TestABankStopArrivingInThisSweepsOwnDrainIsNotRestartedPast(t *testing.T) {
bankStop := func(t *testing.T, f *fixture) {
t.Helper()
body := hello(t, f) + `{"seq":2,"type":"bank_stop","data":{}}` + "\n"
if err := os.WriteFile(filepath.Join(f.workdir, ingest.JournalFile), []byte(body), 0o600); err != nil {
t.Fatal(err)
}
}
start := func(t *testing.T) (*fixture, string) {
t.Helper()
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), VerifyBank: true,
Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
bankStop(t, f)
return f, run.ID
}
assertStandsAtTheStop := func(t *testing.T, f *fixture, runID string) {
t.Helper()
var status string
var attempts int
if err := f.store.Pool().QueryRow(f.ctx,
`select r.status, (select count(*) from run_attempts where run_id = r.id)
from runs r where r.id = $1`, runID).Scan(&status, &attempts); err != nil {
t.Fatal(err)
}
if attempts != 1 {
t.Errorf("the sweep opened attempt %d over a run standing at the signing stop: the pause the"+
" user paid for was spent on a march through it", attempts)
}
if status != "awaiting_bank" {
t.Errorf("the run left the signing stop as %q, want it still standing there", status)
}
}
t.Run("the unit vanished", func(t *testing.T) {
f, runID := start(t)
f.runner.alive = false
late := f.now.Add(2 * time.Hour)
f.svc.Now = func() time.Time { return late }
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
assertStandsAtTheStop(t, f, runID)
})
t.Run("an outside signal left a marker", func(t *testing.T) {
f, runID := start(t)
f.runner.alive = false
live := f.live(t)
// Exit 5 with NO stop recorded by this platform: the shape interruptedBySomeoneElse exists for.
if err := runner.WriteMarker(f.svc.markerPath(runID, live.AttemptNo),
runner.Marker{Unit: live.UnitName, Result: "exit-code", Code: "exited", Status: "5"}); err != nil {
t.Fatal(err)
}
late := f.now.Add(2 * time.Hour)
f.svc.Now = func() time.Time { return late }
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
assertStandsAtTheStop(t, f, runID)
})
}
// `blocked` answers ONE question — why is this scale shorter than the account could afford — so it
// names another book only when giving that book's hold back would lengthen the scale. A hold that
// costs this book nothing must not send the user off to stop a run for no gain.
//
// Mutation caught: filling BlockedBy from CreditHeldBy alone, which is what shipped.
func TestBlockedNamesAnotherBookOnlyWhenItsHoldShortensTheScale(t *testing.T) {
// $20 buys far more than either book has chapters, so nothing money does can shorten this scale.
f := newFixture(t, "20", 4)
own := f.bookID(t)
second := f.secondBook(t)
if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: second, Chapters: order(10)}); err != nil {
t.Fatal(err)
}
opts, err := f.svc.Order(f.ctx, "u1", own)
if err != nil {
t.Fatal(err)
}
if opts.AffordableChapters != 4 || opts.Verdict != pricing.VerdictCoversAll {
t.Fatalf("the order affords %d chapters with verdict %q, want the book's own 4 and covers_all — "+
"the fixture does not test what it claims", opts.AffordableChapters, opts.Verdict)
}
if opts.BlockedBy != "" {
t.Errorf("blocked names %q while the order is bounded by the book's length, not by money", opts.BlockedBy)
}
// Now money IS the bound: a balance that buys less than the book has left, with a hold open
// elsewhere that would have bought more.
poor := newFixture(t, "0.3", 500)
poorOwn := poor.bookID(t)
other := poor.secondBook(t)
if _, err := poor.svc.Start(poor.ctx, StartRequest{UserID: "u1", BookID: other, Chapters: order(5)}); err != nil {
t.Fatal(err)
}
opts, err = poor.svc.Order(poor.ctx, "u1", poorOwn)
if err != nil {
t.Fatal(err)
}
if opts.BlockedBy != other {
t.Errorf("blocked is %q while another book's hold is what shortens the order (affordable %d)",
opts.BlockedBy, opts.AffordableChapters)
}
}
// Two clicks on one button are pinned next door, in resumeorder_test.go: this fixture used to live
// here and reached only ONE of the two orders the call has to answer identically, which is what the
// rebuilt pair now makes inevitable rather than likely.
// A stopped run can be resumed — unless the account started ANOTHER run of the same book in the
// meantime. Re-opening then puts two live runs under an index that forbids exactly that, and the
// answer has to be the conflict a second admission gets, not an internal error. The money is the
// other half: the hold of the refused resume is taken in the same transaction and dies with it.
func TestResumeIsRefusedWhenTheBookHasAnotherLiveRun(t *testing.T) {
f := newFixture(t, "10", 500)
stoppedRun := f.stopped(t, 10, money.MicroUSD(10_000), runner.Marker{Result: "exit-code",
Code: "exited", Status: "1", At: f.now.Add(time.Second)})
if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)}); err != nil {
t.Fatal(err)
}
before := f.account(t)
if _, err := f.svc.Resume(f.ctx, "u1", stoppedRun); !errors.Is(err, pgstore.ErrRunInFlight) {
t.Fatalf("resume while another run is live: %v, want ErrRunInFlight", err)
}
acct := f.account(t)
if acct.Reserved != before.Reserved || acct.Balance != before.Balance {
t.Fatalf("a refused resume moved money: %+v -> %+v", before, acct)
}
if acct.Balance != acct.LedgerSum {
t.Fatalf("the cached balance and the ledger disagree: %s vs %s", acct.Balance.USD(), acct.LedgerSum.USD())
}
}
// One counter per book, and every run-carrying answer uses it (contract §Revision). The stop handle
// used to answer from the run's own column, which sits BELOW the card's number the moment anything
// materializes — and a client that has applied the card's revision is required by the contract to
// drop the lower one, i.e. to drop the answer to the button it just pressed.
func TestEveryRunCarryingAnswerUsesTheBooksRevision(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
// Something materializes: the book's counter moves and the run's own column does not.
if _, err := f.store.Pool().Exec(f.ctx,
`update books set revision = revision + 5 where id = $1`, f.bookID(t)); err != nil {
t.Fatal(err)
}
book, card, err := f.store.GetBook(f.ctx, "u1", f.bookID(t))
if err != nil || card == nil {
t.Fatalf("card: %v", err)
}
if card.Revision != book.Revision {
t.Fatalf("the card's run carries revision %d and its book %d: one counter per book", card.Revision, book.Revision)
}
stopped, err := f.svc.Stop(f.ctx, "u1", run.ID)
if err != nil {
t.Fatal(err)
}
if stopped.Revision != book.Revision {
t.Fatalf("stop answered revision %d, the card answers %d — the client drops the lower one",
stopped.Revision, book.Revision)
}
read, err := f.store.ReadRun(f.ctx, "u1", run.ID)
if err != nil {
t.Fatal(err)
}
if read.Revision != book.Revision {
t.Fatalf("ReadRun answered revision %d, the card answers %d", read.Revision, book.Revision)
}
}
// A run that is not the caller's, or one that does not exist, is a 404 — even on a deployment that
// could not resume anything anyway. The other order answers "this service cannot start runs" to a
// question about someone else's run, which is both a worse answer and a small oracle.
func TestOwnershipIsJudgedBeforeTheDeploymentsHealth(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
f.svc.Cfg.MarkerArgv = nil // a deployment that cannot record how a run ends
if _, err := f.svc.Resume(f.ctx, "u1", "run_does_not_exist"); !errors.Is(err, pgstore.ErrNoRun) {
t.Fatalf("resume of a missing run on an unwired deployment: %v, want ErrNoRun", err)
}
if _, err := f.svc.Resume(f.ctx, "u1", run.ID); !errors.Is(err, ErrRunnerIncomplete) {
t.Fatalf("resume of an OWN run on an unwired deployment: %v, want the deployment's refusal", err)
}
}
// The reconciler decides from a SNAPSHOT and writes seconds later; in between, the user presses
// stop. Three windows, one cure — the row is re-read under the lock that does the write. Each of
// these was reproduced by an independent reviewer before it was closed.
// Window 1 (money): a stop pressed while the reconciler is settling an interrupted attempt. The
// restart used to clear the fresh intent, open a second attempt and take a new hold — a 202 for a
// stop that never happened, and the user paying for the work they had just cancelled.
func TestAStopPressedWhileTheReconcilerRestartsIsNotLost(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
// The unit is gone with no marker — the shape of a reboot — and the grace is past, so the
// reconciler will restart. The stop lands while it is settling.
f.runner.alive = false
f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
f.engine.onStatus = func() {
f.engine.onStatus = nil
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Error(err)
}
}
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
got := f.run(t, run.ID)
if got.Status != "stopped" {
t.Fatalf("the run is %q, want stopped: the stop outranks the restart", got.Status)
}
if n := len(f.runner.starts()); n != 1 {
t.Fatalf("%d units started; the restart went ahead over a stop", n)
}
if acct := f.account(t); acct.Reserved != 0 {
t.Fatalf("a run stopped mid-restart still holds %s", acct.Reserved.USD())
}
if acct := f.account(t); acct.Balance != acct.LedgerSum {
t.Fatalf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
}
}
// Window 2 (money): the sweep is about to close a run that was stopped before it spawned, and the
// queue worker creates the unit in between. Closing it anyway strands a live engine that no list
// looks at — not the live runs (finished), not the unsettled ones (its hold was released).
func TestARunSpawnedWhileTheSweepWasClosingItIsNotAbandoned(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
// The snapshot a sweep would decide from: taken while the attempt still had no unit.
live := f.live(t)
// The worker claims and creates the unit — BEFORE the stop, because the claim itself refuses once
// an intent is standing. This is the window that remains: the sweep is holding a snapshot older
// than the unit.
claimed, err := f.store.RecordSpawn(f.ctx, pgstore.SpawnRecord{AttemptID: live.AttemptID,
Unit: "tm-run-raced", Binary: "/opt/engine/tmctl", Ceiling: live.Ceiling,
CeilingArg: live.Ceiling, Baseline: 1})
if err != nil || !claimed {
t.Fatalf("the worker could not claim the attempt: %v %v", claimed, err)
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
if err := f.svc.finishStopped(f.ctx, live); err != nil {
t.Fatal(err)
}
got := f.run(t, run.ID)
if got.Status == "stopped" || got.FinishedAt != nil {
t.Fatalf("a run whose engine had just started was closed anyway: %+v", got)
}
if acct := f.account(t); acct.Reserved == 0 {
t.Fatal("the hold of a run whose engine is running was released")
}
// And the next pass does the right thing with it: a live unit plus a standing intent is a stop.
f.runner.alive = true
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
if n := len(f.runner.stops()); n == 0 {
t.Fatal("the raced run was never asked to stop")
}
}
// Window 3: a stop pressed on a run whose queue job has not been picked up yet. The worker must not
// start an engine for it — the stop arrived before the run did anything at all.
func TestTheQueueWorkerDoesNotStartARunThatWasAlreadyStopped(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
if n := len(f.runner.starts()); n != 0 {
t.Fatalf("%d units started for a run that was stopped before its worker ran", n)
}
// The claim itself refuses too, which is the half that covers an intent written after the read.
live := f.live(t)
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
claimed, err := f.store.RecordSpawn(f.ctx, pgstore.SpawnRecord{AttemptID: live.AttemptID,
Unit: "tm-run-late", Binary: "/opt/engine/tmctl", Ceiling: live.Ceiling, CeilingArg: live.Ceiling})
if err != nil {
t.Fatal(err)
}
if claimed {
t.Fatal("the spawn claim was granted for an attempt of a run that is over and settled")
}
if acct := f.account(t); acct.Reserved != 0 || acct.Balance != money.MicroUSD(10_000_000) {
t.Fatalf("the hold did not come back whole: balance %s reserved %s",
acct.Balance.USD(), acct.Reserved.USD())
}
}
// The window a RESUME opened and nothing else could: a run comes back to life, and the exit marker of
// the attempt that ended is still on disk — nothing deletes markers of attempts that are over. A pass
// holding the old snapshot would close the resumed run from it, and attempt 2's hold would then be
// in NO worklist at all: live runs are selected by `finished_at is null`, unsettled ones by an
// attempt with `ended_at`, and a re-finished resumed run matches neither.
func TestAStaleSweepDoesNotReFinishAResumedRunFromTheOldAttemptsMarker(t *testing.T) {
f := newFixture(t, "10", 500)
spent := money.MicroUSD(100_000)
runID := f.stopped(t, 100, spent, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
At: f.now.Add(time.Second)})
stale := f.lastAttempt(t, runID) // the snapshot a slow pass is still holding: attempt 1
if _, err := f.svc.Resume(f.ctx, "u1", runID); err != nil {
t.Fatal(err)
}
held := f.account(t).Reserved
if held == 0 {
t.Fatal("the resumed run holds nothing")
}
// The slow pass arrives with attempt 1 and its marker, which is still on disk.
if _, err := f.svc.reconcile(f.ctx, stale); err != nil {
t.Fatal(err)
}
got := f.run(t, runID)
if got.Status != "translating" || got.FinishedAt != nil {
t.Fatalf("a stale pass closed the resumed run: %+v", got)
}
if acct := f.account(t); acct.Reserved != held {
t.Fatalf("the resumed run's hold moved: %s → %s", held.USD(), acct.Reserved.USD())
}
// And the run is still findable, which is the half that makes the money recoverable at all.
live, err := f.store.ListLiveRuns(f.ctx)
if err != nil {
t.Fatal(err)
}
if len(live) != 1 || live[0].AttemptNo != 2 {
t.Fatalf("live runs after the stale pass: %+v", live)
}
}
// A claim that was GIVEN BACK is not proof that no engine exists: `systemd-run` killed after it had
// already asked leaves one running, which is why the spend baseline is kept on such an attempt. The
// stop path must ask systemd about the unit's deterministic name rather than close over it.
func TestAStopDoesNotCloseARunWhoseGivenBackClaimLeftAnEngineRunning(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
// The unit could "not be created" — but it was: the claim is given back and the baseline stays.
f.runner.startErr = errors.New("systemd-run was killed after it had asked")
if err := f.svc.Spawn(f.ctx, run.ID); err == nil {
t.Fatal("a failed spawn reported success")
}
live := f.live(t)
if live.UnitName != "" || live.SpendBaseline == nil {
t.Fatalf("after a given-back claim: unit %q, baseline %v", live.UnitName, live.SpendBaseline)
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
// systemd says the orphan is alive.
f.runner.alive = true
f.runner.startErr = nil
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
got := f.run(t, run.ID)
if got.FinishedAt != nil {
t.Fatalf("the run was closed while its engine was still running: %+v", got)
}
if n := len(f.runner.stops()); n == 0 {
t.Fatal("the orphaned unit was never asked to stop")
}
if acct := f.account(t); acct.Reserved == 0 {
t.Fatal("the hold of a run whose engine is alive was released")
}
}
// The window the queue worker opens: it reads the run, then spends seconds in `bookMeter` asking the
// engine what the book has cost. A stop committed inside that window has no unit to signal — so the
// only thing that can refuse the engine is the claim itself.
func TestAStopCommittedWhileTheWorkerReadsTheMeterStopsTheSpawn(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
// The stop lands while the worker is inside the meter call, i.e. after it read its snapshot.
f.engine.onStatus = func() {
f.engine.onStatus = nil
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Error(err)
}
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
if n := len(f.runner.starts()); n != 0 {
t.Fatalf("%d units started for a run stopped while the worker was reading the meter", n)
}
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
if got := f.run(t, run.ID); got.Status != "stopped" {
t.Fatalf("the run is %q, want stopped", got.Status)
}
if acct := f.account(t); acct.Reserved != 0 || acct.Balance != money.MicroUSD(10_000_000) {
t.Fatalf("the hold did not come back whole: balance %s reserved %s",
acct.Balance.USD(), acct.Reserved.USD())
}
}
// FP5-2 (acceptance, money). The guard FinishRun received (PD-181) was missing from the path that
// closes a stop which never spawned: a stale pass could close a RESUMED run through it, and attempt
// 2's hold would then be in no worklist at all — with the added twist that every later resume answers
// 409 forever, because the run it would continue is finished.
func TestAStaleUnspawnedStopDoesNotCloseAResumedRun(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(100)})
if err != nil {
t.Fatal(err)
}
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
stale := f.live(t) // attempt 1, no unit — the snapshot a slow pass holds
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
if _, err := f.svc.Resume(f.ctx, "u1", run.ID); err != nil {
t.Fatal(err)
}
held := f.account(t).Reserved
if held == 0 {
t.Fatal("the resumed run holds nothing")
}
// The slow pass arrives with attempt 1 and the intent that was on it.
if err := f.svc.finishStopped(f.ctx, stale); err != nil {
t.Fatal(err)
}
got := f.run(t, run.ID)
if got.Status != "translating" || got.FinishedAt != nil {
t.Fatalf("a stale unspawned-stop closed the resumed run: %+v", got)
}
if acct := f.account(t); acct.Reserved != held {
t.Fatalf("the resumed run's hold moved: %s → %s", held.USD(), acct.Reserved.USD())
}
live, err := f.store.ListLiveRuns(f.ctx)
if err != nil {
t.Fatal(err)
}
if len(live) != 1 || live[0].AttemptNo != 2 {
t.Fatalf("live runs after the stale pass: %+v", live)
}
}
// FP5-8(а) (acceptance). A stop that lands while the reconciler is settling an exhausted run must not
// come back as `paused/credit_exhausted`: that says the money ran out, when what happened is that the
// owner stopped it.
func TestAStopDuringSettlementOutranksThePause(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), Chapters: order(10)})
if err != nil {
t.Fatal(err)
}
if err := f.svc.Spawn(f.ctx, run.ID); err != nil {
t.Fatal(err)
}
// The attempt spends the whole ceiling, so the restart path will find nothing left; the stop lands
// while that settlement is in flight.
f.engine.set(statusSpending(fixtureHold(10)), nil)
f.engine.onStatus = func() {
f.engine.onStatus = nil
if _, err := f.svc.Stop(f.ctx, "u1", run.ID); err != nil {
t.Error(err)
}
}
f.runner.alive = false
f.svc.Now = func() time.Time { return f.now.Add(2 * time.Hour) }
if err := f.svc.Sweep(f.ctx); err != nil {
t.Fatal(err)
}
got := f.run(t, run.ID)
if got.Status != "stopped" || got.PausedReason != "" {
t.Fatalf("the run came back as %q/%q, want stopped", got.Status, got.PausedReason)
}
if acct := f.account(t); acct.Balance != acct.LedgerSum {
t.Fatalf("balance %s and ledger %s disagree", acct.Balance.USD(), acct.LedgerSum.USD())
}
}
// waveShape reads what a progress line actually folds into the read model: the ETA it carried, and
// the SHAPE the engine announced with it — whether this book's pipeline has an editor, and which
// generation of that answer the book is on.
//
// ⚠ It replaces `waveCounts`, which read `runs.draft_done/edit_done`. Those columns were dropped with
// register row PD-411 — two writers, no reader — so an assertion over them proved that a fold had
// happened, not that anything a client can see was right. The re-point is to the numbers this event
// STILL materialises, which is the honest remainder rather than a weakened claim: read the wrong
// field of the line and the shape comes out wrong, which is what the assertion catches.
func (f *fixture) waveShape(t *testing.T, bookID string) (editor *bool, epoch int) {
t.Helper()
if err := f.store.Pool().QueryRow(f.ctx,
`select epoch_editor, shape_epoch from books where id = $1`, bookID).Scan(&editor, &epoch); err != nil {
t.Fatal(err)
}
return editor, epoch
}
// runETA is the other half of what a progress line leaves behind.
func (f *fixture) runETA(t *testing.T, runID string) *int {
t.Helper()
var eta *int
if err := f.store.Pool().QueryRow(f.ctx,
`select eta_seconds from runs where id = $1`, runID).Scan(&eta); err != nil {
t.Fatal(err)
}
return eta
}
// The budget of a RESUME is what the run was sold for — PD-168's second caller, the one a user
// presses. The shape is PD-168's own measurement: 100 chapters bought at $0.03, $0.50 spent, the rate
// doubled between the stop and the resume — a rate-derived budget holds $5.50 for a $2.50 remainder.
//
// Mutation caught: `budget, err := s.Store.RunBudget(...)` in reopen replaced by a fresh quote of the order.
func TestAResumeHoldsWhatTheRunWasSoldForWhenTheRateHasMovedSince(t *testing.T) {
f := newFixture(t, "10", 500)
spent := money.MicroUSD(500_000)
runID := f.stopped(t, 100, spent, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
At: f.now.Add(time.Second)})
sold := fixtureHold(100)
doubled, err := pricing.New(2 * pricing.DefaultHoldFactorPercent)
if err != nil {
t.Fatal(err)
}
f.svc.Pricing = doubled
if _, err := f.svc.Resume(f.ctx, "u1", runID); err != nil {
t.Fatal(err)
}
if acct := f.account(t); acct.Reserved != sold-spent {
t.Fatalf("the resumed run holds %s, want %s (sold for %s, spent %s); priced again at today's rate it would hold %s",
acct.Reserved.USD(), (sold - spent).USD(), sold.USD(), spent.USD(), (fixtureHoldAt(2*pricing.DefaultHoldFactorPercent, 100) - spent).USD())
}
if acct := f.account(t); acct.Balance != acct.LedgerSum {
t.Errorf("balance %s and ledger %s disagree after a resume", acct.Balance.USD(), acct.LedgerSum.USD())
}
}
// A run whose first hold cannot be found is NOT priced again from the rate: that would be the defect
// in another coat. Every admission opens the row in the same transaction as the run and closing it
// keeps the row, so its absence is a broken invariant — answered by name, with nothing moved.
//
// Mutation caught: quoting the order again instead of returning ErrNoFirstHold.
func TestAContinuationWithoutTheFirstHoldIsRefusedRatherThanRepriced(t *testing.T) {
f := newFixture(t, "10", 500)
runID := f.stopped(t, 100, 500_000, runner.Marker{Result: "exit-code", Code: "exited", Status: "1",
At: f.now.Add(time.Second)})
if _, err := f.store.Pool().Exec(f.ctx, `delete from reservations where engine_run_id = $1`, runID+"#1"); err != nil {
t.Fatal(err)
}
before := f.account(t)
_, err := f.svc.Resume(f.ctx, "u1", runID)
if !errors.Is(err, pgstore.ErrNoFirstHold) {
t.Fatalf("a resume with no first hold answered %v, want ErrNoFirstHold", err)
}
if live, err := f.store.ListLiveRuns(f.ctx); err != nil || len(live) != 0 {
t.Fatalf("a run whose budget cannot be read was re-opened: %+v (%v)", live, err)
}
if after := f.account(t); after.Reserved != before.Reserved || after.Balance != before.Balance {
t.Fatalf("money moved on a refused resume: before %+v, after %+v", before, after)
}
}