1236 lines
49 KiB
Go
1236 lines
49 KiB
Go
package pgstore
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"errors"
|
||
"sync"
|
||
"testing"
|
||
"time"
|
||
|
||
"textmachine/platform/internal/ingest"
|
||
"textmachine/platform/internal/money"
|
||
)
|
||
|
||
// The claim on an attempt can be taken, given back and taken again — that is what happens when a
|
||
// unit could not be created — and the SECOND claim must not overwrite what the first decided. The
|
||
// caller passes the stored values back on a retry, so this pins the guard the DATABASE keeps
|
||
// underneath it: two answers to "what limit did that process have" is one too many.
|
||
func TestASecondClaimOnOneAttemptKeepsWhatTheFirstRecorded(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
first := SpawnRecord{AttemptID: run.AttemptID, Unit: "tm-run-x-1", Binary: "/opt/engine/1/tmctl",
|
||
Ceiling: 300_000, CeilingArg: 3_000_000, Baseline: 1_000_000}
|
||
if claimed, err := s.RecordSpawn(ctx, first); err != nil || !claimed {
|
||
t.Fatalf("first claim: %v %v", claimed, err)
|
||
}
|
||
if err := s.ReleaseSpawnClaim(ctx, run.AttemptID); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// The meter has moved since — by the engine the first claim may well have started.
|
||
second := first
|
||
second.CeilingArg, second.Baseline = 3_400_000, 1_400_000
|
||
if claimed, err := s.RecordSpawn(ctx, second); err != nil || !claimed {
|
||
t.Fatalf("second claim: %v %v", claimed, err)
|
||
}
|
||
var baseline, arg int64
|
||
if err := s.pool.QueryRow(ctx, `
|
||
select spend_baseline_micro_usd, ceiling_arg_micro_usd from run_attempts where id = $1`,
|
||
run.AttemptID).Scan(&baseline, &arg); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if money.MicroUSD(baseline) != first.Baseline {
|
||
t.Errorf("baseline %s after a re-claim, want the one the first claim recorded (%s)",
|
||
money.MicroUSD(baseline).USD(), first.Baseline.USD())
|
||
}
|
||
if money.MicroUSD(arg) != first.CeilingArg {
|
||
t.Errorf("recorded limit %s after a re-claim, want %s",
|
||
money.MicroUSD(arg).USD(), first.CeilingArg.USD())
|
||
}
|
||
}
|
||
|
||
func seedBook(t testing.TB, s *Store, ctx context.Context, id, owner string, chapters int) {
|
||
t.Helper()
|
||
exec(t, s, ctx, `insert into books (id, owner_id, title, source_lang, target_lang, status,
|
||
chapter_count, workdir, engine_book_id)
|
||
values ($1,$2,'蛊真人','zh','ru','not_started',$3,'/srv/books/'||$1,$1)`, id, owner, chapters)
|
||
}
|
||
|
||
func fundedAccount(t testing.TB, s *Store, ctx context.Context, user string, usd string) time.Time {
|
||
t.Helper()
|
||
now := time.Now().UTC().Truncate(time.Millisecond)
|
||
seedUser(t, s, ctx, user)
|
||
amount, err := money.ParseUSD(usd)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := s.Grant(ctx, user, amount, "test", "seed-"+user, "", now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
return now
|
||
}
|
||
|
||
// Admission is ONE transaction, and this is the assertion that says so: the run row, its attempt,
|
||
// the hold and the queue entry either all exist or none of them do. A hold that outlived a failed
|
||
// admission is credit reserved for a run that does not exist, and nothing would ever release it.
|
||
func TestAdmittingARunWritesTheRunTheAttemptAndTheHoldTogether(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
|
||
var enqueued string
|
||
run, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", VerifyBank: true, OrderedChapters: 100,
|
||
Ceiling: money.MicroUSD(3_000_000), Now: now,
|
||
}, 0, func(_ context.Context, _ Tx, id string) error { enqueued = id; return nil })
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if enqueued != run.ID {
|
||
t.Errorf("the queue was handed %q, the run is %q", enqueued, run.ID)
|
||
}
|
||
acct, err := s.ReadAccount(ctx, "u1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// The hold is a DEBIT at the moment it is taken: the balance already excludes it, and Reserved is
|
||
// a separate, informational sum. Subtracting Reserved again is the arithmetic D39.115 §2a
|
||
// corrected — it would halve the run-ceiling scale.
|
||
if acct.Balance != money.MicroUSD(7_000_000) || acct.Reserved != money.MicroUSD(3_000_000) {
|
||
t.Fatalf("balance %s, reserved %s; want 7.000000 and 3.000000", acct.Balance.USD(), acct.Reserved.USD())
|
||
}
|
||
if acct.LedgerSum != acct.Balance {
|
||
t.Errorf("the cached balance and the ledger disagree: %s vs %s", acct.Balance.USD(), acct.LedgerSum.USD())
|
||
}
|
||
live, err := s.ListLiveRuns(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(live) != 1 || live[0].RunID != run.ID || live[0].AttemptNo != 1 {
|
||
t.Fatalf("live runs: %+v", live)
|
||
}
|
||
if live[0].Ceiling != money.MicroUSD(3_000_000) {
|
||
t.Errorf("the attempt's ceiling is %s, want the amount that was held", live[0].Ceiling.USD())
|
||
}
|
||
}
|
||
|
||
// Nothing partial survives a refusal. The interesting half is the ROLLBACK: the run row is inserted
|
||
// before the hold is attempted, so a broke account must not leave one behind.
|
||
func TestARunThatCannotBePaidForLeavesNothingBehind(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "1")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
|
||
_, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 100, Ceiling: money.MicroUSD(5_000_000), Now: now,
|
||
}, 0, nil)
|
||
if !errors.Is(err, ErrInsufficientCredit) {
|
||
t.Fatalf("started a run on an account that cannot pay: %v", err)
|
||
}
|
||
live, err := s.ListLiveRuns(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(live) != 0 {
|
||
t.Fatalf("a refused admission left %d live runs", len(live))
|
||
}
|
||
var attempts int
|
||
if err := s.pool.QueryRow(ctx, `select count(*) from run_attempts`).Scan(&attempts); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if attempts != 0 {
|
||
t.Errorf("a refused admission left %d attempts", attempts)
|
||
}
|
||
acct, err := s.ReadAccount(ctx, "u1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if acct.Balance != money.MicroUSD(1_000_000) || acct.Reserved != 0 {
|
||
t.Errorf("money moved on a refused admission: balance %s reserved %s", acct.Balance.USD(), acct.Reserved.USD())
|
||
}
|
||
}
|
||
|
||
// The engine holds an EXCLUSIVE lock on the project file, so two live runs on one book is not a
|
||
// state to explain — it is one the database refuses. The guard is a partial unique index, and the
|
||
// point of this test is that the refusal surfaces as a DOMAIN error a handler can turn into 409,
|
||
// not as a raw SQLSTATE.
|
||
func TestABookCannotHaveTwoLiveRuns(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
in := StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10, Ceiling: money.MicroUSD(300_000), Now: now}
|
||
if _, err := s.StartRun(ctx, in, 0, nil); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := s.StartRun(ctx, in, 0, nil); !errors.Is(err, ErrRunInFlight) {
|
||
t.Fatalf("a second live run gave %v, want ErrRunInFlight", err)
|
||
}
|
||
acct, err := s.ReadAccount(ctx, "u1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if acct.Reserved != money.MicroUSD(300_000) {
|
||
t.Errorf("the refused second run reserved money: %s", acct.Reserved.USD())
|
||
}
|
||
}
|
||
|
||
// A book that is not the caller's is indistinguishable from one that does not exist.
|
||
func TestARunCannotBeStartedOnSomeoneElsesBook(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
fundedAccount(t, s, ctx, "u2", "10")
|
||
seedBook(t, s, ctx, "bk1", "u2", 500)
|
||
_, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 10, Ceiling: money.MicroUSD(300_000), Now: now,
|
||
}, 0, nil)
|
||
if !errors.Is(err, ErrNoBook) {
|
||
t.Fatalf("starting a run on another account's book gave %v, want ErrNoBook", err)
|
||
}
|
||
}
|
||
|
||
// The cursor and the effect move in ONE transaction. The assertion that matters is the second Apply
|
||
// of the same event: at-least-once delivery is the ratified norm (PD-105), so a duplicate must be a
|
||
// no-op rather than a second increment.
|
||
func TestAnEventAndItsCursorMoveTogetherAndADuplicateChangesNothing(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 10, Ceiling: money.MicroUSD(300_000), Now: now,
|
||
}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sink := s.NewRunSink(run.AttemptID, run.ID, "bk1")
|
||
if err := sink.Begin(ctx, ingest.Hello{EngineRunID: EngineStreamID(run.ID, 1), ChunkerVersion: "chunk-7"}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ev := ingest.Envelope{Seq: 2, Type: ingest.TypeProgress,
|
||
Data: json.RawMessage(`{"draft":{"done":3,"total":10},"edit":{"done":1,"total":10},"eta_seconds":42}`)}
|
||
cur := ingest.Cursor{Offset: 128, SHA256: []byte("0123456789abcdef0123456789abcdef")}
|
||
if err := sink.Apply(ctx, ev, cur); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var lastSeq, offset int64
|
||
var epoch int
|
||
var editor *bool
|
||
var eta *int
|
||
// ⚠ The observable is the ETA and the announced SHAPE, not the per-wave counters this line also
|
||
// used to write: those columns had two writers and no reader and were dropped with PD-411, so an
|
||
// assertion over them said only that a fold ran. These two are what a progress line still leaves
|
||
// in the read model.
|
||
//
|
||
// ⚠ What the DUPLICATE half of this test proves is the CURSOR, and only the cursor — said plainly
|
||
// because the first edition of this comment claimed the epoch made it bite, and this pack's
|
||
// adversarial pass showed it cannot: a re-delivery carries the same shape, so the epoch's
|
||
// `epoch_editor <> $2` guard is false by construction and the counter would not move even with the
|
||
// dedup guard deleted. The idempotence of the FOLD is pinned where it can be observed, on the
|
||
// counters a duplicate would actually double: TestAUnitAnnouncedTwiceIsCountedOnce.
|
||
read := func() {
|
||
t.Helper()
|
||
if err := s.pool.QueryRow(ctx, `select last_seq, last_offset from run_attempts where id = $1`,
|
||
run.AttemptID).Scan(&lastSeq, &offset); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.pool.QueryRow(ctx, `select eta_seconds from runs where id = $1`,
|
||
run.ID).Scan(&eta); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.pool.QueryRow(ctx, `select epoch_editor, shape_epoch from books where id = 'bk1'`).
|
||
Scan(&editor, &epoch); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
read()
|
||
if lastSeq != 2 || offset != 128 || eta == nil || *eta != 42 {
|
||
t.Fatalf("after one event: seq=%d offset=%d eta=%v", lastSeq, offset, eta)
|
||
}
|
||
if editor == nil || !*editor || epoch != 0 {
|
||
t.Fatalf("the line announced an edit total of 10: shape folded to editor=%v epoch=%d", editor, epoch)
|
||
}
|
||
// The same line again — the ordinary consequence of a reader restart.
|
||
if err := sink.Apply(ctx, ev, cur); err != nil {
|
||
t.Fatalf("a redelivered event was refused: %v", err)
|
||
}
|
||
read()
|
||
if epoch != 0 || lastSeq != 2 {
|
||
t.Errorf("a duplicate moved something: epoch %d, seq %d", epoch, lastSeq)
|
||
}
|
||
// The handshake's chunker version is what tells the platform its chapter numbering was produced
|
||
// by a different chunker (row 100).
|
||
var chunker string
|
||
if err := s.pool.QueryRow(ctx, `select chunker_version from books where id='bk1'`).Scan(&chunker); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if chunker != "chunk-7" {
|
||
t.Errorf("chunker version %q", chunker)
|
||
}
|
||
}
|
||
|
||
// The high-water mark inside the transaction, pinned on the one effect that COUNTS.
|
||
//
|
||
// ⚠ Written after a mutation survived: the first version of this pin used a progress event, whose
|
||
// effect is an absolute assignment, so applying it twice produces the same row whether the guard is
|
||
// there or not. A `unit_done` increments, and that is where a redelivered line — the ordinary
|
||
// consequence of at-least-once delivery — turns into a chapter that reports more finished units than
|
||
// it has.
|
||
func TestARedeliveredCountingEventDoesNotCountTwice(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
exec(t, s, ctx, `insert into chapters (id, book_id, number, units_total) values ('c1','bk1',1,5)`)
|
||
run, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 10, Ceiling: money.MicroUSD(300_000), Now: now,
|
||
}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sink := s.NewRunSink(run.AttemptID, run.ID, "bk1")
|
||
if err := sink.Begin(ctx, ingest.Hello{EngineRunID: EngineStreamID(run.ID, 1)}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
body, err := json.Marshal(ingest.UnitDone{Chapter: 1, Unit: 1, Wave: "edit", Shipped: true})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ev := ingest.Envelope{Seq: 2, Type: ingest.TypeUnitDone, Data: body}
|
||
cur := ingest.Cursor{Offset: 64}
|
||
for range 3 {
|
||
if err := sink.Apply(ctx, ev, cur); err != nil {
|
||
t.Fatalf("a redelivered event was refused: %v", err)
|
||
}
|
||
}
|
||
var editDone int
|
||
if err := s.pool.QueryRow(ctx,
|
||
`select units_edit_done from chapters where id='c1'`).Scan(&editDone); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if editDone != 1 {
|
||
t.Errorf("one unit delivered three times counted as %d", editDone)
|
||
}
|
||
}
|
||
|
||
// The spend counter is CUMULATIVE, so the materializer keeps the maximum instead of adding: a
|
||
// redelivered line that was summed would inflate what the account is charged.
|
||
func TestSpendKeepsTheHighestFigureRatherThanASum(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 10, Ceiling: money.MicroUSD(300_000), Now: now,
|
||
}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sink := s.NewRunSink(run.AttemptID, run.ID, "bk1")
|
||
if err := sink.Begin(ctx, ingest.Hello{EngineRunID: EngineStreamID(run.ID, 1)}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
spend := func(seq int64, micro int64) {
|
||
t.Helper()
|
||
body, _ := json.Marshal(ingest.Spend{CommittedMicroUSD: micro})
|
||
if err := sink.Apply(ctx, ingest.Envelope{Seq: seq, Type: ingest.TypeSpend, Data: body}, ingest.Cursor{Offset: seq * 10}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
}
|
||
spend(2, 120_000)
|
||
spend(3, 90_000) // an out-of-order or stale figure must not lower it either
|
||
var got int64
|
||
if err := s.pool.QueryRow(ctx, `select spend_micro_usd from run_attempts where id=$1`, run.AttemptID).Scan(&got); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got != 120_000 {
|
||
t.Errorf("spend %d, want the highest seen (120000)", got)
|
||
}
|
||
}
|
||
|
||
// A ceiling halt is `paused` and never `failed` (contract §BookStatus): the stop is resumable, and
|
||
// calling it a failure lies about that.
|
||
func TestACeilingHaltPausesTheRunWithItsReason(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 10, Ceiling: money.MicroUSD(300_000), Now: now,
|
||
}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sink := s.NewRunSink(run.AttemptID, run.ID, "bk1")
|
||
if err := sink.Begin(ctx, ingest.Hello{EngineRunID: EngineStreamID(run.ID, 1)}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// The platform's OWN ceiling, which is the one it sets flush against the hold. WHICH ceiling the
|
||
// scope names is decided by CeilingPause and pinned by its own table (sink_test.go); what this
|
||
// test is about is that any of them is `paused` and not `failed`.
|
||
body, _ := json.Marshal(ingest.Ceiling{Halted: true, Scope: ingest.ScopeBook})
|
||
if err := sink.Apply(ctx, ingest.Envelope{Seq: 2, Type: ingest.TypeCeiling, Data: body}, ingest.Cursor{Offset: 10}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var status, reason string
|
||
if err := s.pool.QueryRow(ctx, `select status, coalesce(paused_reason,'') from runs where id=$1`, run.ID).
|
||
Scan(&status, &reason); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if status != "paused" || reason != PausedRunLimitReached {
|
||
t.Errorf("after a ceiling halt: status=%q reason=%q", status, reason)
|
||
}
|
||
}
|
||
|
||
// An interrupted run gets a NEW attempt with what is LEFT of its budget. Reserving the full ceiling
|
||
// again would let one run spend it twice.
|
||
func TestARestartedRunOnlyGetsTheBudgetItHasNotSpent(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 100, Ceiling: money.MicroUSD(3_000_000), Now: now,
|
||
}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// The first attempt is settled at what the engine actually spent.
|
||
if err := s.Settle(ctx, ReservationKey(run.ID, 1), money.MicroUSD(1_000_000), BasisComplete, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
spent, err := s.RunSpent(ctx, run.ID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if spent != money.MicroUSD(1_000_000) {
|
||
t.Fatalf("run spend %s, want 1.000000", spent.USD())
|
||
}
|
||
next, err := s.RestartRun(ctx, RestartInput{
|
||
RunID: run.ID, AttemptID: run.AttemptID, UserID: "u1", BookID: "bk1",
|
||
Ceiling: money.MicroUSD(3_000_000) - spent, Offset: 512, Now: now,
|
||
})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if next.AttemptNo != 2 || next.Position.Offset != 512 {
|
||
t.Fatalf("next attempt: %+v", next)
|
||
}
|
||
acct, err := s.ReadAccount(ctx, "u1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// $10 granted − $1 spent − $2 reserved for the new attempt.
|
||
if acct.Balance != money.MicroUSD(7_000_000) || acct.Reserved != money.MicroUSD(2_000_000) {
|
||
t.Fatalf("balance %s reserved %s", acct.Balance.USD(), acct.Reserved.USD())
|
||
}
|
||
live, err := s.ListLiveRuns(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(live) != 1 || live[0].AttemptNo != 2 {
|
||
t.Fatalf("live runs after a restart: %+v", live)
|
||
}
|
||
}
|
||
|
||
// Finishing is idempotent by refusal: the reconciler retries, and a second pass must not move a
|
||
// finished run or reopen its attempt.
|
||
func TestFinishingARunTwiceChangesNothingTheSecondTime(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 10, Ceiling: money.MicroUSD(300_000), Now: now,
|
||
}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
code := 0
|
||
if _, err := s.FinishRun(ctx, RunEnding{RunID: run.ID, AttemptID: run.AttemptID,
|
||
Status: "ready", ExitResult: "success", ExitCode: &code, Now: now}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
later := now.Add(time.Hour)
|
||
if _, err := s.FinishRun(ctx, RunEnding{RunID: run.ID, AttemptID: run.AttemptID,
|
||
Status: "failed", ExitResult: "exit-code", Now: later}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var status string
|
||
var finished time.Time
|
||
if err := s.pool.QueryRow(ctx, `select status, finished_at from runs where id=$1`, run.ID).Scan(&status, &finished); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if status != "ready" || !finished.Equal(now) {
|
||
t.Errorf("a second finish rewrote the run: status=%q finished=%s", status, finished)
|
||
}
|
||
if _, err := s.ListLiveRuns(ctx); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var bookStatus string
|
||
if err := s.pool.QueryRow(ctx, `select status from books where id='bk1'`).Scan(&bookStatus); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if bookStatus != "ready" {
|
||
t.Errorf("book status %q", bookStatus)
|
||
}
|
||
}
|
||
|
||
// The unsettled list is what keeps a hold from being reserved forever when the settlement figure
|
||
// could not be read.
|
||
func TestAFinishedRunWithAnOpenHoldIsListedAsUnsettled(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 10, Ceiling: money.MicroUSD(300_000), Now: now,
|
||
}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
code := 0
|
||
if _, err := s.FinishRun(ctx, RunEnding{RunID: run.ID, AttemptID: run.AttemptID,
|
||
Status: "ready", ExitResult: "success", ExitCode: &code, Now: now}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
open, err := s.UnsettledRuns(ctx, time.Now())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(open) != 1 || open[0].RunID != run.ID {
|
||
t.Fatalf("unsettled: %+v", open)
|
||
}
|
||
if err := s.Settle(ctx, ReservationKey(run.ID, 1), money.MicroUSD(120_000), BasisComplete, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.MarkSettled(ctx, run.ID, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
open, err = s.UnsettledRuns(ctx, time.Now())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(open) != 0 {
|
||
t.Fatalf("a settled run is still listed: %+v", open)
|
||
}
|
||
}
|
||
|
||
// Quarantine stops the PROJECTION and leaves the run alone: the engine is spending money the account
|
||
// already reserved, and our inability to read its journal is not a reason to throw that away.
|
||
func TestQuarantineStopsTheProjectionAndNotTheRun(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{
|
||
UserID: "u1", BookID: "bk1", OrderedChapters: 10, Ceiling: money.MicroUSD(300_000), Now: now,
|
||
}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Quarantine(ctx, run.AttemptID, "a seq was re-read with a different payload"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
live, err := s.ListLiveRuns(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(live) != 1 || !live[0].Quarantined {
|
||
t.Fatalf("after quarantine: %+v", live)
|
||
}
|
||
// The first reason is kept: a later sweep must not overwrite why materialization stopped.
|
||
if err := s.Quarantine(ctx, run.AttemptID, "something else"); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var reason string
|
||
if err := s.pool.QueryRow(ctx, `select quarantine_reason from run_attempts where id=$1`, run.AttemptID).Scan(&reason); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if reason != "a seq was re-read with a different payload" {
|
||
t.Errorf("quarantine reason was overwritten: %q", reason)
|
||
}
|
||
}
|
||
|
||
// PD-82: a mistyped account id used to come back as "insufficient credit", which reads as "this
|
||
// account is broke" — the one answer an operator acts on differently.
|
||
func TestMoneyOperationsTellAMissingAccountFromAnEmptyOne(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
seedUser(t, s, ctx, "u1")
|
||
seedBook(t, s, ctx, "bk1", "u1", 10)
|
||
now := time.Now().UTC()
|
||
if err := s.Hold(ctx, "nobody", "bk1", "r#1", money.MicroUSD(1), now); !errors.Is(err, ErrNoAccount) {
|
||
t.Errorf("hold on a missing account gave %v, want ErrNoAccount", err)
|
||
}
|
||
if err := s.Hold(ctx, "u1", "bk1", "r#1", money.MicroUSD(1), now); !errors.Is(err, ErrInsufficientCredit) {
|
||
t.Errorf("hold on an account with no credit gave %v, want ErrInsufficientCredit", err)
|
||
}
|
||
}
|
||
|
||
// PD-81: with a LIVE reservation the second hold collides on the primary key, and the declared error
|
||
// was reachable only through a shape nobody produces. A worker cannot act on a raw SQLSTATE.
|
||
func TestASecondHoldOnALiveReservationIsADuplicateNotASqlstate(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 10)
|
||
if err := s.Hold(ctx, "u1", "bk1", "r#1", money.MicroUSD(1_000_000), now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
err := s.Hold(ctx, "u1", "bk1", "r#1", money.MicroUSD(1_000_000), now)
|
||
if !errors.Is(err, ErrDuplicateHold) {
|
||
t.Fatalf("a second hold on a live reservation gave %v, want ErrDuplicateHold", err)
|
||
}
|
||
acct, err := s.ReadAccount(ctx, "u1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if acct.Reserved != money.MicroUSD(1_000_000) || acct.Balance != acct.LedgerSum {
|
||
t.Errorf("the refused duplicate moved money: %+v", acct)
|
||
}
|
||
}
|
||
|
||
// PD-97: the release of a hold used to discard whether it applied, so a spent release key closed the
|
||
// reservation and returned nothing — a silent no-op on the money path.
|
||
func TestAReleaseWhoseKeyWasSpentIsRefusedRatherThanSilent(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 10)
|
||
if err := s.Hold(ctx, "u1", "bk1", "r#1", money.MicroUSD(1_000_000), now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Release(ctx, "r#1", now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// The out-of-band sequence the register named: the reservation row is gone but the release key is
|
||
// spent, and the same attempt id is opened again.
|
||
if err := s.Hold(ctx, "u1", "bk1", "r#1", money.MicroUSD(1_000_000), now); !errors.Is(err, ErrDuplicateHold) {
|
||
t.Fatalf("re-opening a spent attempt id gave %v", err)
|
||
}
|
||
// The surgery removes the reservation and the HOLD's ledger key while leaving the RELEASE key
|
||
// spent — the state the register describes and no code path produces. The cached balance is
|
||
// recomputed with it, or the fixture itself would break `balance == SUM(ledger)` and the
|
||
// assertion below would be measuring the test rather than the code.
|
||
exec(t, s, ctx, `delete from reservations where engine_run_id='r#1'`)
|
||
exec(t, s, ctx, `delete from credit_ledger where source='run' and source_id='r#1'`)
|
||
exec(t, s, ctx, `update account_balances b set balance_micro_usd =
|
||
(select coalesce(sum(amount_micro_usd),0) from credit_ledger where user_id = b.user_id)`)
|
||
if err := s.Hold(ctx, "u1", "bk1", "r#1", money.MicroUSD(1_000_000), now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Release(ctx, "r#1", now); !errors.Is(err, ErrReleaseKeySpent) {
|
||
t.Fatalf("a release against a spent key gave %v, want ErrReleaseKeySpent", err)
|
||
}
|
||
// The refusal rolls the whole transaction back: the reservation stays OPEN and visible to an
|
||
// operator, instead of being closed while the money never came back.
|
||
acct, err := s.ReadAccount(ctx, "u1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if acct.Balance != acct.LedgerSum {
|
||
t.Errorf("balance %s and ledger %s disagree after the refusal", acct.Balance.USD(), acct.LedgerSum.USD())
|
||
}
|
||
if acct.Reserved != money.MicroUSD(1_000_000) {
|
||
t.Errorf("the reservation was closed by a release that returned nothing: reserved %s", acct.Reserved.USD())
|
||
}
|
||
}
|
||
|
||
// Readiness has to cover the QUEUE's schema too: River keeps its own migration ledger, so goose's
|
||
// version says nothing about it, and an instance that answers "ready" and then cannot accept a run
|
||
// has told the load balancer a falsehood (the class of PD-68).
|
||
func TestReadinessCoversTheQueueSchema(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
if err := s.Ready(ctx); err != nil {
|
||
t.Fatalf("a fully migrated database is not ready: %v", err)
|
||
}
|
||
if _, err := s.pool.Exec(ctx, `drop table river_job cascade`); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Ready(ctx); !errors.Is(err, ErrSchemaBehind) {
|
||
t.Fatalf("readiness without the queue schema gave %v, want ErrSchemaBehind", err)
|
||
}
|
||
}
|
||
|
||
// The high-water mark has to hold under CONCURRENCY, not only under repetition. Three sequential
|
||
// deliveries are absorbed by the seq check alone; two SIMULTANEOUS ones are absorbed only by the row
|
||
// lock, and this is the direct analogue of TestConcurrentHoldsCannotOvercommitAnAccount on the
|
||
// credits side, which had no counterpart here.
|
||
func TestConcurrentDeliveriesOfOneEventCountItOnce(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
exec(t, s, ctx, `insert into chapters (id, book_id, number, units_total) values ('c1','bk1',1,50)`)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sink := s.NewRunSink(run.AttemptID, run.ID, "bk1")
|
||
if err := sink.Begin(ctx, ingest.Hello{EngineRunID: EngineStreamID(run.ID, 1)}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
body, err := json.Marshal(ingest.UnitDone{Chapter: 1, Unit: 1, Wave: "edit", Shipped: true})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
ev := ingest.Envelope{Seq: 2, Type: ingest.TypeUnitDone, Data: body}
|
||
var wg sync.WaitGroup
|
||
for range 8 {
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
_ = sink.Apply(ctx, ev, ingest.Cursor{Offset: 64})
|
||
}()
|
||
}
|
||
wg.Wait()
|
||
var done int
|
||
if err := s.pool.QueryRow(ctx, `select units_edit_done from chapters where id='c1'`).Scan(&done); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if done != 1 {
|
||
t.Errorf("one event delivered by eight goroutines counted as %d", done)
|
||
}
|
||
}
|
||
|
||
// An attempt already bound to one engine run must refuse a different one: two engine processes
|
||
// writing into one journal under one attempt would mix two runs' counters into one projection.
|
||
func TestAnAttemptRefusesASecondEnginesHandshake(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sink := s.NewRunSink(run.AttemptID, run.ID, "bk1")
|
||
if err := sink.Begin(ctx, ingest.Hello{EngineRunID: EngineStreamID(run.ID, 1)}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := sink.Begin(ctx, ingest.Hello{EngineRunID: EngineStreamID(run.ID, 1)}); err != nil {
|
||
t.Fatalf("re-reading our own handshake was refused: %v", err)
|
||
}
|
||
if err := sink.Begin(ctx, ingest.Hello{EngineRunID: "eng-2"}); err == nil {
|
||
t.Error("a second engine rebound the attempt")
|
||
}
|
||
var bound string
|
||
if err := s.pool.QueryRow(ctx, `select engine_run_id from run_attempts where id=$1`, run.AttemptID).Scan(&bound); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if bound != EngineStreamID(run.ID, 1) {
|
||
t.Errorf("the attempt is bound to %q, want the name the PLATFORM gave it", bound)
|
||
}
|
||
}
|
||
|
||
// RunSpent decides how much budget a restarted run has left, so it must count THIS run's settlements
|
||
// and nothing else: not another run's, and not the holds and releases that cancel out only while an
|
||
// attempt is balanced.
|
||
func TestRunSpentCountsOnlyThisRunsSettlements(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "20")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
seedBook(t, s, ctx, "bk2", "u1", 500)
|
||
mine, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 100,
|
||
Ceiling: money.MicroUSD(3_000_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
other, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk2", OrderedChapters: 100,
|
||
Ceiling: money.MicroUSD(3_000_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Settle(ctx, ReservationKey(mine.ID, 1), money.MicroUSD(1_000_000), BasisComplete, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Settle(ctx, ReservationKey(other.ID, 1), money.MicroUSD(2_000_000), BasisComplete, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
got, err := s.RunSpent(ctx, mine.ID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got != money.MicroUSD(1_000_000) {
|
||
t.Errorf("RunSpent = %s, want 1.000000 — another run's settlement leaked in", got.USD())
|
||
}
|
||
// An OPEN hold is not spend. This is the reachable case: settlement deferred, the hold still open,
|
||
// and counting it would overstate the spend by the whole ceiling and pause a run with budget left.
|
||
seedBook(t, s, ctx, "bk3", "u1", 500)
|
||
third, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk3", OrderedChapters: 50,
|
||
Ceiling: money.MicroUSD(1_500_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got, err = s.RunSpent(ctx, third.ID); err != nil || got != 0 {
|
||
t.Errorf("a run with an open hold and no settlement reports %s spent (%v)", got.USD(), err)
|
||
}
|
||
}
|
||
|
||
// The settlement worklist must contain only money that is still reserved, and the settled stamp must
|
||
// be written once: a second stamp with a later clock would rewrite when the money resolved.
|
||
func TestTheSettlementWorklistAndItsStampAreBothIdempotent(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
code := 0
|
||
if _, err := s.FinishRun(ctx, RunEnding{RunID: run.ID, AttemptID: run.AttemptID,
|
||
Status: "ready", ExitResult: "success", ExitCode: &code, Now: now}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Settle(ctx, ReservationKey(run.ID, 1), money.MicroUSD(100_000), BasisComplete, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
open, err := s.UnsettledRuns(ctx, time.Now())
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(open) != 0 {
|
||
t.Errorf("a run whose reservation is closed is still on the worklist: %+v", open)
|
||
}
|
||
if err := s.MarkSettled(ctx, run.ID, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
later := now.Add(time.Hour)
|
||
if err := s.MarkSettled(ctx, run.ID, later); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var at time.Time
|
||
if err := s.pool.QueryRow(ctx, `select settled_at from runs where id=$1`, run.ID).Scan(&at); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if !at.Equal(now) {
|
||
t.Errorf("the settled stamp was rewritten: %s, want %s", at, now)
|
||
}
|
||
}
|
||
|
||
// A hold that debited NOTHING must be refused. The ordinary duplicate collides on the reservation's
|
||
// primary key one level up; this is the other shape — the reservation row gone and the ledger key
|
||
// spent — which is what a book deletion leaves behind for CLOSED reservations.
|
||
func TestAHoldThatDebitedNothingIsRefused(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
if err := s.Hold(ctx, "u1", "bk1", "r#1", money.MicroUSD(1_000_000), now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
before, err := s.ReadAccount(ctx, "u1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
exec(t, s, ctx, `delete from reservations where engine_run_id='r#1'`)
|
||
err = s.Hold(ctx, "u1", "bk1", "r#1", money.MicroUSD(1_000_000), now)
|
||
if !errors.Is(err, ErrDuplicateHold) {
|
||
t.Fatalf("a hold against a spent ledger key gave %v, want ErrDuplicateHold", err)
|
||
}
|
||
after, err := s.ReadAccount(ctx, "u1")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if after.Balance != before.Balance {
|
||
t.Errorf("the refused hold moved the balance: %s → %s", before.Balance.USD(), after.Balance.USD())
|
||
}
|
||
}
|
||
|
||
// The byte hint is monotone: a stale sweep must not walk it back and re-read applied lines forever.
|
||
func TestTheByteHintNeverGoesBackwards(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.SaveCursor(ctx, run.AttemptID, Position{Offset: 1000}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.SaveCursor(ctx, run.AttemptID, Position{Offset: 100}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
var got int64
|
||
if err := s.pool.QueryRow(ctx, `select last_offset from run_attempts where id=$1`, run.AttemptID).Scan(&got); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got != 1000 {
|
||
t.Errorf("the byte hint went back to %d", got)
|
||
}
|
||
}
|
||
|
||
// The store's own guard on the chapter ceiling, and the pause vocabulary: both are values the API
|
||
// layer also checks, and both must be refused here too — the store is what the reconciler calls.
|
||
func TestTheStoreRefusesAnImpossibleCeilingAndAnUnknownPauseReason(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
for _, n := range []int{0, -5} {
|
||
if _, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: n,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil); err == nil {
|
||
t.Errorf("a ceiling of %d chapters was accepted", n)
|
||
}
|
||
}
|
||
var runs int
|
||
if err := s.pool.QueryRow(ctx, `select count(*) from runs`).Scan(&runs); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if runs != 0 {
|
||
t.Errorf("a refused start left %d run rows", runs)
|
||
}
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := s.PauseRun(ctx, run.ID, run.AttemptID, "because", now); err == nil {
|
||
t.Error("an unknown pause reason was accepted; the vocabulary is closed")
|
||
}
|
||
// The daily ceiling is the second value, and it has to be accepted by BOTH gates — this one and
|
||
// the DDL check of migration 00015 — or a run stopped by the engine's own limit cannot be
|
||
// recorded at all.
|
||
if paused, err := s.PauseRun(ctx, run.ID, run.AttemptID, PausedDailyCeiling, now); err != nil || !paused {
|
||
t.Fatalf("PauseRun(daily_ceiling) = %v, %v", paused, err)
|
||
}
|
||
}
|
||
|
||
// Only ONE sweep may restart an interrupted attempt: two that both succeeded would each open a next
|
||
// attempt and each take a hold.
|
||
func TestOnlyOneSweepCanRestartAnInterruptedAttempt(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 100,
|
||
Ceiling: money.MicroUSD(3_000_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Settle(ctx, ReservationKey(run.ID, 1), money.MicroUSD(500_000), BasisComplete, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
in := RestartInput{RunID: run.ID, AttemptID: run.AttemptID, UserID: "u1", BookID: "bk1",
|
||
Ceiling: money.MicroUSD(1_000_000), Now: now}
|
||
var wg sync.WaitGroup
|
||
results := make([]error, 4)
|
||
for i := range results {
|
||
wg.Add(1)
|
||
go func() {
|
||
defer wg.Done()
|
||
_, results[i] = s.RestartRun(ctx, in)
|
||
}()
|
||
}
|
||
wg.Wait()
|
||
won := 0
|
||
for _, err := range results {
|
||
if err == nil {
|
||
won++
|
||
}
|
||
}
|
||
if won != 1 {
|
||
t.Fatalf("%d of four concurrent restarts succeeded, want exactly 1", won)
|
||
}
|
||
var attempts int
|
||
if err := s.pool.QueryRow(ctx, `select count(*) from run_attempts where run_id=$1`, run.ID).Scan(&attempts); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if attempts != 2 {
|
||
t.Errorf("%d attempts after four concurrent restarts", attempts)
|
||
}
|
||
}
|
||
|
||
// Cross-family review of the acceptance dofix (M1): a pause is a CLOSING path, and every closing path
|
||
// must check that the attempt it closes is still a live attempt of this run. `FinishRun` and
|
||
// `FinishUnspawnedStop` do; `PauseRun` did not, so a pass holding a snapshot of the attempt that was
|
||
// already restarted answered "the credit ran out" for a run whose second attempt is spending right
|
||
// now — and that second hold then belongs to no list at all.
|
||
func TestAPauseFromAnOldSnapshotDoesNotCloseARunOverALiveAttempt(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 100,
|
||
Ceiling: money.MicroUSD(3_000_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Settle(ctx, ReservationKey(run.ID, 1), money.MicroUSD(500_000), BasisComplete, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
next, err := s.RestartRun(ctx, RestartInput{RunID: run.ID, AttemptID: run.AttemptID, UserID: "u1",
|
||
BookID: "bk1", Ceiling: money.MicroUSD(1_000_000), Now: now})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
paused, err := s.PauseRun(ctx, run.ID, run.AttemptID, PausedCreditExhausted, now)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
// It refused, and it SAYS it refused. A silent nil here let the caller log "the run was paused"
|
||
// about a run somebody else had already moved on (register row PD-141).
|
||
if paused {
|
||
t.Error("a stale pass reported that it paused a run whose attempt had been restarted")
|
||
}
|
||
var status string
|
||
var finished *time.Time
|
||
if err := s.pool.QueryRow(ctx,
|
||
`select status, finished_at from runs where id = $1`, run.ID).Scan(&status, &finished); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if finished != nil {
|
||
t.Fatalf("a stale pause closed the run as %q while attempt %d is live", status, next.AttemptID)
|
||
}
|
||
var ended *time.Time
|
||
if err := s.pool.QueryRow(ctx,
|
||
`select ended_at from run_attempts where id = $1`, next.AttemptID).Scan(&ended); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if ended != nil {
|
||
t.Fatal("a stale pause ended the LIVE attempt")
|
||
}
|
||
// The money half of the same fact: the live attempt's hold must be findable. A closed run with an
|
||
// open hold is in neither list — not in the live runs (the run is finished) and not in the
|
||
// unsettled ones (the attempt is not) — which is how a hold is lost for good.
|
||
live, err := s.ListLiveRuns(ctx)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
found := false
|
||
for _, l := range live {
|
||
if l.AttemptID == next.AttemptID {
|
||
found = true
|
||
}
|
||
}
|
||
if !found {
|
||
t.Fatalf("the live attempt %d is not in ListLiveRuns after a stale pause", next.AttemptID)
|
||
}
|
||
}
|
||
|
||
// Re-check of the dofix (FP5-11): the rule the intake writers got — take the LIBRARY's next number,
|
||
// not your own plus one — has to hold for the run's writers too. A book that is not the account's
|
||
// newest carries a counter below the library's, so `translating`, `paused` and `stopped` on it left
|
||
// `greatest(max, floor)` exactly where it was and the library screen never saw the run start.
|
||
func TestARunTransitionOnAnOlderBookStillMovesTheLibrary(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
// A second, NEWER book carries the library's maximum, so bk1's own counter is not it.
|
||
seedBook(t, s, ctx, "bk2", "u1", 500)
|
||
exec(t, s, ctx, `update books set revision = 41 where id = 'bk2'`)
|
||
before, err := s.ListBooks(ctx, "u1", 50, "")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
after, err := s.ListBooks(ctx, "u1", 50, "")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if after.Revision <= before.Revision {
|
||
t.Fatalf("a run started on an older book and the library still answers %d (was %d): a client "+
|
||
"obeying the contract drops that read and never sees the book start translating",
|
||
after.Revision, before.Revision)
|
||
}
|
||
// And the same for the closing half of the lifecycle.
|
||
mid := after.Revision
|
||
if paused, err := s.PauseRun(ctx, run.ID, run.AttemptID, PausedCreditExhausted, now); err != nil || !paused {
|
||
t.Fatalf("PauseRun = %v, %v", paused, err)
|
||
}
|
||
end, err := s.ListBooks(ctx, "u1", 50, "")
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if end.Revision <= mid {
|
||
t.Fatalf("the run was paused and the library still answers %d (was %d)", end.Revision, mid)
|
||
}
|
||
}
|
||
|
||
// `stop_requested` is the user's own click, and it survives the reload a client's own pending state
|
||
// does not. It travels on every shape a Run is handed out in — the stop receipt, the run handle and
|
||
// the book card — and a resume CLEARS it with the intent it records.
|
||
//
|
||
// Mutation caught: reading the flag from the status; leaving stop_requested_at set on a restart.
|
||
func TestTheStopIntentTravelsOnEveryRunAndAResumeClearsIt(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
fresh, err := s.ReadRun(ctx, "u1", run.ID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if fresh.StopRequested {
|
||
t.Fatal("a run nobody stopped already reports the intent")
|
||
}
|
||
stopped, _, err := s.RequestStop(ctx, "u1", run.ID, now.Add(time.Minute))
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if !stopped.StopRequested {
|
||
t.Error("the stop receipt does not carry the intent it just recorded")
|
||
}
|
||
for _, tc := range []struct {
|
||
name string
|
||
read func() (Run, error)
|
||
}{
|
||
{"the run handle", func() (Run, error) { return s.ReadRun(ctx, "u1", run.ID) }},
|
||
{"the book card", func() (Run, error) {
|
||
_, card, err := s.GetBook(ctx, "u1", "bk1")
|
||
if card == nil {
|
||
return Run{}, err
|
||
}
|
||
return *card, err
|
||
}},
|
||
} {
|
||
got, err := tc.read()
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if !got.StopRequested {
|
||
t.Errorf("%s does not carry the stop the user asked for", tc.name)
|
||
}
|
||
}
|
||
// A resume clears the intent, because it is the answer to it: the run continues, and a flag left
|
||
// standing would have every later read say the user is still asking it to stop. It works on a run
|
||
// that has ENDED, which is the state a stop leaves behind.
|
||
if closed, err := s.FinishRun(ctx, RunEnding{RunID: run.ID, AttemptID: run.AttemptID,
|
||
Status: "stopped", ExitResult: "exit-code", Now: now.Add(2 * time.Minute)}); err != nil || !closed {
|
||
t.Fatalf("closing the stopped run: %v %v", closed, err)
|
||
}
|
||
if _, err := s.RestartRun(ctx, RestartInput{RunID: run.ID, AttemptID: run.AttemptID,
|
||
UserID: "u1", BookID: "bk1", Ceiling: money.MicroUSD(300_000), Now: now.Add(3 * time.Minute)}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
after, err := s.ReadRun(ctx, "u1", run.ID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if after.StopRequested {
|
||
t.Error("a resumed run still reports the stop it was resumed out of")
|
||
}
|
||
}
|
||
|
||
// EVERY transaction that ends a run leaves the book owing a materialization, and there are three of
|
||
// them. The debt is only a mechanism while all three write it: a run whose ending forgets it is a run
|
||
// the user PAID for whose text nothing will ever materialize — the drain lists by that column alone,
|
||
// and nothing else looks at a finished book again.
|
||
//
|
||
// ⚠ The two beside FinishRun are the easy ones to miss, and both had. A run paused at a ceiling has
|
||
// translated everything up to it; a stop that lands on an attempt whose spawn claim was released may
|
||
// still have an engine behind it — the reconciler keeps that attempt's spend baseline for exactly
|
||
// that reason, and settles it.
|
||
//
|
||
// Mutation caught: dropping the stamp from any one of the three endings.
|
||
func TestEveryEndingOfARunLeavesTheBookOwingASurface(t *testing.T) {
|
||
for _, tc := range []struct {
|
||
name string
|
||
end func(t *testing.T, s *Store, ctx context.Context, run StartedRun, now time.Time)
|
||
}{
|
||
{"the ordinary ending", func(t *testing.T, s *Store, ctx context.Context, run StartedRun, now time.Time) {
|
||
if closed, err := s.FinishRun(ctx, RunEnding{RunID: run.ID, AttemptID: run.AttemptID,
|
||
Status: "ready", ExitResult: "exit-code", Now: now}); err != nil || !closed {
|
||
t.Fatalf("FinishRun = %v, %v", closed, err)
|
||
}
|
||
}},
|
||
{"a pause at a ceiling", func(t *testing.T, s *Store, ctx context.Context, run StartedRun, now time.Time) {
|
||
if paused, err := s.PauseRun(ctx, run.ID, run.AttemptID, PausedCreditExhausted, now); err != nil || !paused {
|
||
t.Fatalf("PauseRun = %v, %v", paused, err)
|
||
}
|
||
}},
|
||
{"a stop that landed before the unit existed", func(t *testing.T, s *Store, ctx context.Context, run StartedRun, now time.Time) {
|
||
if _, _, err := s.RequestStop(ctx, "u1", run.ID, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if finished, err := s.FinishUnspawnedStop(ctx, run.ID, run.AttemptID, now); err != nil || !finished {
|
||
t.Fatalf("FinishUnspawnedStop = %v, %v", finished, err)
|
||
}
|
||
}},
|
||
} {
|
||
t.Run(tc.name, func(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "10")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 10,
|
||
Ceiling: money.MicroUSD(300_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
owed, err := s.BooksOwedReadModel(ctx, 10)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(owed) != 0 {
|
||
t.Fatalf("the book owed a surface before the run ended: %+v", owed)
|
||
}
|
||
tc.end(t, s, ctx, run, now.Add(time.Minute))
|
||
owed, err = s.BooksOwedReadModel(ctx, 10)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if len(owed) != 1 || owed[0].ID != "bk1" {
|
||
t.Errorf("after this ending the materializer's queue holds %+v, want the book", owed)
|
||
}
|
||
})
|
||
}
|
||
}
|
||
|
||
// The budget of a run is what its FIRST attempt was sold for, pinned in the package that reads it —
|
||
// beside RunSpent, the other money reader of this file, so a swap to «the latest hold» is caught
|
||
// where it is written rather than two packages away.
|
||
//
|
||
// The shape is the one that tells the two readings apart: a run interrupted twice, whose second hold
|
||
// is smaller than its first. The budget must not follow the attempts down.
|
||
//
|
||
// Mutation caught: keying RunBudget on the run's latest reservation, or on its highest attempt.
|
||
func TestARunsBudgetIsTheFirstAttemptsHoldWhateverTheLaterOnesHold(t *testing.T) {
|
||
s, ctx := testDB(t)
|
||
now := fundedAccount(t, s, ctx, "u1", "20")
|
||
seedBook(t, s, ctx, "bk1", "u1", 500)
|
||
run, err := s.StartRun(ctx, StartRunInput{UserID: "u1", BookID: "bk1", OrderedChapters: 100,
|
||
Ceiling: money.MicroUSD(3_000_000), Now: now}, 0, nil)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
sold, err := s.RunBudget(ctx, run.ID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if sold != money.MicroUSD(3_000_000) {
|
||
t.Fatalf("the run was sold for %s, and the budget reads %s", money.MicroUSD(3_000_000).USD(), sold.USD())
|
||
}
|
||
// Two continuations, each holding the remainder: the second attempt takes 2.5, the third 2.2.
|
||
if err := s.Settle(ctx, ReservationKey(run.ID, 1), money.MicroUSD(500_000), BasisComplete, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
second, err := s.RestartRun(ctx, RestartInput{RunID: run.ID, AttemptID: run.AttemptID, UserID: "u1",
|
||
BookID: "bk1", Ceiling: money.MicroUSD(2_500_000), Now: now})
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if err := s.Settle(ctx, ReservationKey(run.ID, 2), money.MicroUSD(300_000), BasisComplete, now); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if _, err := s.RestartRun(ctx, RestartInput{RunID: run.ID, AttemptID: second.AttemptID, UserID: "u1",
|
||
BookID: "bk1", Ceiling: money.MicroUSD(2_200_000), Now: now}); err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
got, err := s.RunBudget(ctx, run.ID)
|
||
if err != nil {
|
||
t.Fatal(err)
|
||
}
|
||
if got != sold {
|
||
t.Errorf("after two continuations the budget reads %s, want the %s the run was sold for: "+
|
||
"a budget that follows the attempts down under-holds every restart by everything spent before it",
|
||
got.USD(), sold.USD())
|
||
}
|
||
// A run whose first hold is gone is a broken invariant, and it is answered by name rather than
|
||
// with a figure — the caller must not fall back to the rate.
|
||
exec(t, s, ctx, `delete from reservations where engine_run_id = $1`, ReservationKey(run.ID, 1))
|
||
if _, err := s.RunBudget(ctx, run.ID); !errors.Is(err, ErrNoFirstHold) {
|
||
t.Errorf("a run with no first hold answered %v, want ErrNoFirstHold", err)
|
||
}
|
||
}
|