textmachine/platform/internal/runs/control_test.go

1125 lines
47 KiB
Go
Raw 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"
"slices"
"sync"
"testing"
"time"
"textmachine/platform/internal/ingest"
"textmachine/platform/internal/money"
"textmachine/platform/internal/pgstore"
"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), CeilingChapters: 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
}
// 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)
}
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
}
// bankReleased reads the bit that tells the first segment of a run from the second. Read with SQL
// because it is deliberately NOT on the wire: the client sees one counter, and which pass it is
// counting is the platform's own business.
func (f *fixture) bankReleased(t *testing.T, runID string) bool {
t.Helper()
var released bool
if err := f.store.Pool().QueryRow(f.ctx,
`select bank_released from runs where id = $1`, runID).Scan(&released); err != nil {
t.Fatal(err)
}
return released
}
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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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 := f.svc.Pricing.Ceiling(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)
}
// The SEGMENT moved with it: the bar the client reads starts again from zero for the work that
// follows the stop, which is what `bank_released` is for.
if !f.bankReleased(t, runID) {
t.Error("the signing stop was lifted without recording that the second segment had begun")
}
}
// 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, f.svc.Pricing.Ceiling(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, f.svc.Pricing.Ceiling(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 nothing left is `paused` with the reason the contract has a word for, and a
// resume of that comes back paused too.
func TestAnInterruptedRunWithNothingLeftIsPausedAndStaysPausedThroughAResume(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), CeilingChapters: 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(f.svc.Pricing.Ceiling(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.PausedCreditExhausted {
t.Fatalf("an interrupted run with no budget is %q/%q, want paused/credit_exhausted",
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.PausedCreditExhausted {
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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 10})
if err != nil {
t.Fatal(err)
}
second := f.secondBook(t)
if _, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: second, CeilingChapters: 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))
}
}
// ONLY the user's resume may clear a bank-signing stop. The reconciler restarts INTERRUPTIONS, and a
// run standing at the signing stop is not one: the stop writes `awaiting_bank` without `finished_at`,
// so the run stays in the sweep's list, and a restart that lifted the stop would spawn the rest of the
// ceiling without `--verify-bank` — running past the pause the user paid for, with no click of theirs.
//
// Mutation caught: keying LiftBankStop on l.Status alone, which is the form that shipped this morning.
func TestTheReconcilerDoesNotLiftABankStopNobodySigned(t *testing.T) {
f := newFixture(t, "10", 500)
run, err := f.svc.Start(f.ctx, StartRequest{UserID: "u1", BookID: f.bookID(t), VerifyBank: true,
CeilingChapters: 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 released bool
var status string
var attempts int
if err := f.store.Pool().QueryRow(f.ctx,
`select r.bank_released, r.status, (select count(*) from run_attempts where run_id = r.id)
from runs r where r.id = $1`, run.ID).Scan(&released, &status, &attempts); err != nil {
t.Fatal(err)
}
// The fixture must actually reach the restart, or it observes nothing: this assertion is what
// caught the first version of this test, which passed under its own mutation.
if attempts < 2 {
t.Fatalf("the sweep did not restart the run at all (%d attempts): this fixture cannot observe the property", attempts)
}
if released {
t.Errorf("the sweep lifted a signing stop the user never signed (status now %q)", status)
}
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)
}
}
}
// `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, CeilingChapters: 10}); err != nil {
t.Fatal(err)
}
opts, err := f.svc.Bounds(f.ctx, "u1", own)
if err != nil {
t.Fatal(err)
}
if opts.Ceiling.Max != 4 {
t.Fatalf("the scale is %d chapters, want the book's own 4 — the fixture does not test what it claims", opts.Ceiling.Max)
}
if opts.BlockedBy != "" {
t.Errorf("blocked names %q while the scale 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, CeilingChapters: 5}); err != nil {
t.Fatal(err)
}
opts, err = poor.svc.Bounds(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 scale (max %d)",
opts.BlockedBy, opts.Ceiling.Max)
}
}
// Two clicks on one button. Both calls pass the state check together and race for attempt N+1 on the
// unique index; the loser must answer with the run the winner re-opened, not with a not-found — and
// exactly ONE hold may be taken.
func TestTwoResumesOfOneRunTakeOneHoldAndBothAnswer(t *testing.T) {
f := newFixture(t, "10", 500)
runID := f.stopped(t, 100, money.MicroUSD(500_000), runner.Marker{Result: "exit-code",
Code: "exited", Status: "1", At: f.now.Add(time.Second)})
var wg sync.WaitGroup
runs := make([]pgstore.Run, 2)
errs := make([]error, 2)
for i := range runs {
wg.Add(1)
go func() {
defer wg.Done()
runs[i], errs[i] = f.svc.Resume(f.ctx, "u1", runID)
}()
}
wg.Wait()
for i, err := range errs {
if err != nil {
t.Fatalf("resume %d: %v, want the run back", i, err)
}
if runs[i].Status != "translating" {
t.Errorf("resume %d answered %q, want translating", i, runs[i].Status)
}
}
if acct := f.account(t); acct.Reserved != f.svc.Pricing.Ceiling(100)-money.MicroUSD(500_000) {
t.Fatalf("two resumes reserved %s, want exactly one remainder", acct.Reserved.USD())
}
if acct := f.account(t); acct.Balance != acct.LedgerSum {
t.Fatalf("the cached balance and the ledger disagree: %s vs %s", acct.Balance.USD(), acct.LedgerSum.USD())
}
var attempts int
if err := f.store.Pool().QueryRow(f.ctx,
`select count(*) from run_attempts where run_id = $1`, runID).Scan(&attempts); err != nil {
t.Fatal(err)
}
if attempts != 2 {
t.Fatalf("%d attempts after two resumes, want 2", attempts)
}
}
// 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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), CeilingChapters: 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(f.svc.Pricing.Ceiling(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())
}
}
// waveCounts reads what the projection actually folded out of the journal, per wave.
//
// ⚠ It exists because `Book.Progress` left the wire in 0.3.0 and the sweep's own test was then
// weakened to counting FRAMES — which says a frame was emitted, not that the numbers behind it are
// right. The counters are still in the database; only the way to read them changed.
func (f *fixture) waveCounts(t *testing.T, runID string) (draft, edit int) {
t.Helper()
if err := f.store.Pool().QueryRow(f.ctx,
`select draft_done, edit_done from runs where id = $1`, runID).Scan(&draft, &edit); err != nil {
t.Fatal(err)
}
return draft, edit
}