textmachine/platform/internal/readmodel/readmodel_test.go

721 lines
32 KiB
Go

package readmodel
import (
"context"
"errors"
"fmt"
"os"
"os/exec"
"path/filepath"
"strconv"
"strings"
"testing"
"time"
"textmachine/platform/internal/ingest"
"textmachine/platform/internal/money"
"textmachine/platform/internal/pgstore"
)
type fakeEngine struct {
manifest ingest.Manifest
export ingest.Export
manifestErr error
exportErr error
// noEnvelope models an engine build from before the artifacts envelope (D39.158): its manifest
// names no bank read-out place at all.
noEnvelope bool
}
func (f *fakeEngine) Manifest(_ context.Context, _, workdir string) (ingest.Manifest, error) {
m := f.manifest
if !f.noEnvelope {
// Derived from the workdir the way the real engine derives it — beside the project database.
m.Artifacts = ingest.StatusArtifacts{BankExport: filepath.Join(workdir, "bk_1.db.bank.json")}
}
return m, f.manifestErr
}
func (f *fakeEngine) Export(context.Context, string, string) (ingest.Export, error) {
return f.export, f.exportErr
}
type fakeStore struct {
structure pgstore.Structure
saved bool
bank []ingest.BankTerm
bankSaved bool
owed []pgstore.OwedBook
cleared []pgstore.OwedBook
deferred []pgstore.OwedBook
// deferredUntil is WHEN each deferral pushed the debt to. It is the whole of the fix the register
// row is about: `now()` put the book back at the head of the queue fifteen seconds later.
deferredUntil []time.Time
deferCost []pgstore.AttemptCost
attempts int
abandoned []pgstore.OwedBook
abandonReason string
claimed []pgstore.OwedBook
// unclaimable models a debt somebody else already holds.
unclaimable bool
// beforeClear runs just before the discharge, so a test can spend the caller's context exactly
// where the reads would have spent it; clearCtxErr is what the discharge saw.
beforeClear func()
clearCtxErr error
}
func (f *fakeStore) SaveStructure(_ context.Context, _ string, in pgstore.Structure) error {
f.structure, f.saved = in, true
return nil
}
func (f *fakeStore) SaveBank(_ context.Context, _ string, terms []ingest.BankTerm) error {
f.bank, f.bankSaved = terms, true
return nil
}
func (f *fakeStore) BooksOwedReadModel(_ context.Context, limit int) ([]pgstore.OwedBook, error) {
if len(f.owed) > limit {
return f.owed[:limit], nil
}
return f.owed, nil
}
func (f *fakeStore) ClaimReadModelDebt(_ context.Context, bookID string, owedAt time.Time, window time.Duration) (time.Time, error) {
if f.unclaimable {
return time.Time{}, nil
}
f.claimed = append(f.claimed, pgstore.OwedBook{ID: bookID, OwedAt: owedAt})
return owedAt.Add(window), nil
}
func (f *fakeStore) DeferReadModelDebt(_ context.Context, bookID string, owedAt, next time.Time, reason string, cost pgstore.AttemptCost) (int, error) {
f.deferred = append(f.deferred, pgstore.OwedBook{ID: bookID, OwedAt: owedAt})
f.deferredUntil = append(f.deferredUntil, next)
f.deferCost = append(f.deferCost, cost)
if cost == pgstore.SpendsAnAttempt {
f.attempts++
}
return f.attempts, nil
}
func (f *fakeStore) AbandonReadModelDebt(_ context.Context, bookID string, owedAt, _ time.Time, reason string) error {
f.abandoned = append(f.abandoned, pgstore.OwedBook{ID: bookID, OwedAt: owedAt})
f.abandonReason = reason
return nil
}
func (f *fakeStore) ClearReadModelDebt(ctx context.Context, bookID string, owedAt time.Time) error {
if f.beforeClear != nil {
f.beforeClear()
}
f.clearCtxErr = ctx.Err()
f.cleared = append(f.cleared, pgstore.OwedBook{ID: bookID, OwedAt: owedAt})
return nil
}
// owedBook is the book every test below refreshes: an id, a workdir and the boundary that owes it.
func owedBook(workdir string) pgstore.OwedBook {
return pgstore.OwedBook{ID: "bk_1", Workdir: workdir, OwedAt: boundary}
}
var boundary = time.Date(2026, 8, 20, 12, 0, 0, 0, time.UTC)
func tree() ingest.Manifest {
return ingest.Manifest{
Version: "tm-manifest-v2", Key: "k1", ChaptersTotal: 2, UnitsTotal: 3,
Chapters: []ingest.ManifestChapter{
{ID: "c1", Number: 1, UnitsTotal: 2, Units: []ingest.ManifestUnit{
{ID: "c1:cut:0", FirstChunkIdx: 0}, {ID: "c1:cut:4", FirstChunkIdx: 4},
}},
{ID: "c2", Number: 2, UnitsTotal: 1, Units: []ingest.ManifestUnit{{ID: "c2:cut:0", FirstChunkIdx: 0}}},
},
}
}
// The two documents are joined by the key the engine publishes for exactly this — the chapter's
// ordinal and the unit's LEADER chunk index — and a pair the export says nothing about stays
// `pending` with no text rather than inheriting a neighbour's.
//
// Mutation caught: joining by position in the array instead of by (chapter, first_chunk_idx).
func TestTheTextIsJoinedOntoTheTreeByTheEnginesOwnKey(t *testing.T) {
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
manifest: tree(),
noEnvelope: true,
export: ingest.Export{TotalUnits: 3, Units: []ingest.UnitText{
{Chapter: 1, Unit: 4, Source: "第二节", Target: "Второй", State: ingest.StateTranslated},
{Chapter: 2, Unit: 0, Source: "第三节", State: ingest.StateWithheld},
}},
}}
// ⚠ The BANK half fails here — this engine publishes no artifacts envelope — and the tree lands
// anyway: each channel is a different question about the same book, and answering two of three
// is strictly better than answering none.
if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err == nil {
t.Error("a workdir with no bank read-out reported no failure at all")
}
if !store.saved || len(store.structure.Chapters) != 2 {
t.Fatalf("structure: %+v", store.structure)
}
first := store.structure.Chapters[0]
if first.Units[0].Ordinal != 0 || first.Units[0].State != ingest.StatePending || first.Units[0].Target != "" {
t.Errorf("a pair the export did not mention: %+v", first.Units[0])
}
if first.Units[1].Target != "Второй" || first.Units[1].State != ingest.StateTranslated {
t.Errorf("the pair at leader index 4: %+v", first.Units[1])
}
second := store.structure.Chapters[1]
if second.Units[0].Source != "第三节" || second.Units[0].State != ingest.StateWithheld {
t.Errorf("the pair of the second chapter: %+v", second.Units[0])
}
if store.structure.ManifestKey != "k1" {
t.Errorf("the validity key did not travel: %q", store.structure.ManifestKey)
}
}
// A book that has never been translated has a tree and no translations, and so does a book whose
// export this build could not read. The TREE lands either way: a reader screen that stayed empty
// because the text channel failed would hide a book the user can already navigate.
//
// Mutation caught: returning early when Export fails; claiming to KNOW the text that failed to read.
func TestAFailedTextReadStillLandsTheTree(t *testing.T) {
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
manifest: tree(), exportErr: errors.New("the engine refused"),
}}
_ = svc.Refresh(t.Context(), owedBook(t.TempDir()))
if !store.saved {
t.Fatal("a failed text read cost the whole tree")
}
for _, c := range store.structure.Chapters {
for _, u := range c.Units {
if u.State != ingest.StatePending || u.Target != "" {
t.Errorf("a pair claims a translation nothing produced: %+v", u)
}
// …and it says so, which is what stops the store overwriting the stored text.
if u.TextKnown {
t.Errorf("a pair claims its text is known after the export failed: %+v", u)
}
}
}
}
// The two engine calls are two re-cuts of the same source, and they can disagree: a pair in the
// manifest that the export did not carry is a SKEW, not an empty pair, and must not be written as
// one over text already materialized.
//
// Mutation caught: setting TextKnown for every pair of a successful export.
func TestAPairTheExportDidNotCarryDoesNotSpeakAboutItsText(t *testing.T) {
// The manifest has three pairs; the export carries ONE of them, as a cut that moved between the
// two calls would leave it.
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
manifest: tree(),
export: ingest.Export{TotalUnits: 3, Units: []ingest.UnitText{
{Chapter: 1, Unit: 0, Source: "第一节", Target: "Первый", State: ingest.StateTranslated},
}},
}}
_ = svc.Refresh(t.Context(), owedBook(t.TempDir()))
var known, unknown int
for _, c := range store.structure.Chapters {
for _, u := range c.Units {
if u.TextKnown {
known++
} else {
unknown++
}
}
}
if known != 1 || unknown == 0 {
t.Errorf("%d pairs claim known text and %d do not, want only the exported one", known, unknown)
}
}
// A book with no bank read-out has never produced terms, and that is not an empty bank: saving one
// would erase a bank the engine simply has not written yet.
//
// Mutation caught: treating ErrNoBank as an empty read-out.
func TestAMissingBankReadOutSavesNothing(t *testing.T) {
// ⚠ The engine PUBLISHED the read-out's place (the artifacts envelope, row 213) and no file is
// there yet — which is the branch this pins. An engine that published no place at all is the
// loud failure next door, and telling the two apart is what the envelope is for.
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err != nil {
t.Fatalf("a book that has never produced terms is not a failure: %v", err)
}
if store.bankSaved {
t.Error("an absent bank read-out was saved as an empty bank")
}
if !store.saved {
t.Error("the tree was not materialized, though only the bank was missing")
}
}
// An engine that publishes no artifacts envelope — a build from before D39.158 — is a different
// fact, and it IS a failure: the bank's place cannot be located, and reading that as "no bank"
// would silently stop every bank refresh on such a deployment.
func TestAnEngineWithoutTheEnvelopeIsAFailureRatherThanAnEmptyBank(t *testing.T) {
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree(), noEnvelope: true}}
if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err == nil {
t.Fatal("a manifest naming no bank read-out place was read as a book with no bank")
}
if store.bankSaved {
t.Error("a missing envelope saved a bank anyway")
}
}
// garbageBank is a workdir whose published bank read-out exists and is not a bank — the one way a
// book's bank channel fails while its neighbours' answer fine.
func garbageBank(t *testing.T) string {
t.Helper()
dir := t.TempDir()
if err := os.WriteFile(filepath.Join(dir, "bk_1.db.bank.json"), []byte("not a bank"), 0o644); err != nil {
t.Fatal(err)
}
return dir
}
// The debt is what brings the materializer back to a book, so only a materialization that answered
// EVERY channel may discharge it. A tree with no text is not a materialized book: before this the
// text channel's failure was logged and the caller was told the book was done.
//
// Mutation caught: returning nil from refreshStructure when Export fails.
func TestAPartialMaterializationDoesNotDischargeTheDebt(t *testing.T) {
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
manifest: tree(), exportErr: errors.New("the engine refused"),
}}
if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err == nil {
t.Fatal("a materialization with no text reported success")
}
if !store.saved {
t.Fatal("this fixture never reached the write, so it cannot observe the property")
}
if len(store.cleared) != 0 {
t.Errorf("a partial materialization discharged the debt: %+v", store.cleared)
}
}
// And a complete one discharges THAT boundary, carried through unchanged. Whether the STORE then
// refuses a debt stamped since is pgstore's own property and is pinned there, on a live database —
// here what is checked is that the boundary reaches it at all.
//
// Mutation caught: discharging with the current time instead of the boundary the caller was given.
func TestACompleteMaterializationDischargesTheBoundaryItRead(t *testing.T) {
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
manifest: tree(),
export: ingest.Export{TotalUnits: 3, Units: []ingest.UnitText{
{Chapter: 1, Unit: 0, Source: "第一节", Target: "Первый", State: ingest.StateTranslated},
}},
}}
if err := svc.Refresh(t.Context(), owedBook(t.TempDir())); err != nil {
t.Fatal(err)
}
if len(store.cleared) != 1 || store.cleared[0].ID != "bk_1" || !store.cleared[0].OwedAt.Equal(boundary) {
t.Errorf("the debt discharged: %+v, want the boundary this read answered", store.cleared)
}
}
// A materialization that HAPPENED is written down even when the context that paid for the reads is
// already spent. The reads legitimately consume the whole budget on a large book, and a discharge
// that then fails leaves the debt standing — so the next pass re-does two full re-chunks of the same
// source, and every pass after it.
//
// Mutation caught: discharging on the caller's own context.
func TestTheDischargeSurvivesAContextTheReadsUsedUp(t *testing.T) {
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{
manifest: tree(),
export: ingest.Export{TotalUnits: 3, Units: []ingest.UnitText{
{Chapter: 1, Unit: 0, Source: "第一节", Target: "Первый", State: ingest.StateTranslated},
}},
}}
ctx, cancel := context.WithCancel(t.Context())
store.beforeClear = cancel // the budget runs out exactly between the last read and the record
if err := svc.Refresh(ctx, owedBook(t.TempDir())); err != nil {
t.Fatal(err)
}
if len(store.cleared) != 1 {
t.Fatalf("the debt was not discharged at all: %+v", store.cleared)
}
if store.clearCtxErr != nil {
t.Errorf("the discharge ran on a context that was already over: %v", store.clearCtxErr)
}
}
// The drain is the only retry there is, so it must reach every owed book and must not let one book's
// failure end the pass: these are independent books of independent accounts.
//
// Mutation caught: returning on the first error.
func TestTheDrainMaterializesEveryOwedBookAndKeepsWhatFailed(t *testing.T) {
store := &fakeStore{owed: []pgstore.OwedBook{
{ID: "bk_1", Workdir: garbageBank(t), OwedAt: boundary}, // its read-out is not a bank: the read fails
{ID: "bk_2", Workdir: t.TempDir(), OwedAt: boundary},
}}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
if err := svc.Drain(t.Context()); err != nil {
t.Fatal(err)
}
if len(store.cleared) != 1 || store.cleared[0].ID != "bk_2" {
t.Errorf("discharged %+v, want only the book that was materialized whole", store.cleared)
}
// …and the one that failed goes to the BACK of the queue. The list is oldest-first, so a book the
// engine can never answer about would hold the front of it forever and the books behind it would
// never be materialized at all.
if len(store.deferred) != 1 || store.deferred[0].ID != "bk_1" {
t.Errorf("deferred %+v, want the book whose materialization failed", store.deferred)
}
}
// A deployment with no engine binary is a read replica, not a broken one: a refresh is a no-op and
// the surface it serves is whatever was materialized before.
func TestWithoutAnEngineARefreshIsANoOp(t *testing.T) {
store := &fakeStore{}
svc := &Service{Store: store, Engine: &fakeEngine{manifest: tree()}}
if err := svc.Refresh(t.Context(), owedBook("/srv/books/bk_1")); err != nil {
t.Fatal(err)
}
if store.saved || store.bankSaved {
t.Error("a deployment with no engine binary materialized something")
}
// …and it does not discharge the debt either: the work belongs to an instance that has an engine.
if len(store.cleared) != 0 {
t.Errorf("a replica cleared a debt it did not pay: %+v", store.cleared)
}
}
// The engine's manifest carries a `heading` and the contract forbids a deployment to project it —
// it is a rendered ordinal («Глава N»), not a label out of the book's data. The allowlist is
// enforced by the DECODER: there is nowhere for the field to land.
//
// Mutation caught: adding a Heading field to ingest.ManifestChapter.
func TestTheManifestsRenderedHeadingHasNowhereToLand(t *testing.T) {
doc := `{"manifest_version":"tm-manifest-v2","key":"k","chapters":[
{"id":"c1","number":1,"heading":"Глава 1","units_total":1,
"units":[{"id":"c1:cut:0","first_chunk_idx":0}]}]}`
m, err := ingest.DecodeManifest([]byte(doc))
if err != nil {
t.Fatal(err)
}
if len(m.Chapters) != 1 || m.Chapters[0].ID != "c1" {
t.Fatalf("manifest: %+v", m)
}
// Whatever the engine renders, this side has no field holding it: the check is over the decoded
// STRUCT, so a field added later fails here rather than on a reader's screen.
if rendered := fmt.Sprintf("%#v", m.Chapters[0]); strings.Contains(rendered, "Глава") {
t.Errorf("the engine's rendered ordinal survived the decode: %s", rendered)
}
}
// Two materializations of one book must not run at once. The intake pays its own debt inline while
// the sweep drains the rest every few seconds, so without a claim every upload slower than one sweep
// interval was read by both at the same time — two full re-chunks of the source, which is the cost
// this package removed elsewhere.
//
// Mutation caught: draining without claiming; ignoring a claim somebody else holds.
func TestTheDrainSkipsABookSomebodyElseIsAlreadyMaterializing(t *testing.T) {
store := &fakeStore{owed: []pgstore.OwedBook{{ID: "bk_1", Workdir: t.TempDir(), OwedAt: boundary}}}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
store.unclaimable = true
if err := svc.Drain(t.Context()); err != nil {
t.Fatal(err)
}
if store.saved {
t.Error("the drain read the engine for a book it does not hold the debt of")
}
// …and it DOES take a debt nobody holds — the claim is a lease, not a refusal.
store.unclaimable = false
if err := svc.Drain(t.Context()); err != nil {
t.Fatal(err)
}
if len(store.claimed) != 1 || !store.saved {
t.Errorf("claimed %+v, saved=%v — want the free debt taken and paid", store.claimed, store.saved)
}
// The discharge presents the stamp the CLAIM produced, not the one the queue was listed with: the
// claim moved it, and a discharge holding the old value would clear nothing.
if len(store.cleared) != 1 || !store.cleared[0].OwedAt.Equal(boundary.Add(MaterializeBudget)) {
t.Errorf("discharged %+v, want the stamp the claim produced", store.cleared)
}
}
// A manifest whose lists are shorter than its own counts NEVER reaches the store.
//
// The tree is written by replacement — `SaveStructure` deletes every chapter outside the list it is
// handed and the cascade takes the text with it — so an empty list is the deletion of the whole
// book. No engine produces one (`buildManifest` fills the chapters and the file is written
// atomically), which is exactly why the danger is silent: the case this refuses is not a broken
// engine but a document THIS build read wrongly, and the counts printed beside the lists are the
// only witness it has of that.
//
// The asymmetry that made it a defect: the intake refuses precisely this document and has since P6,
// because there "no chapters" is the verdict that deletes a user's upload. The materializer's write
// is as destructive and checked nothing.
//
// Mutation caught: dropping either floor from refreshStructure — the self-description check or the
// "no chapters at all" one.
func TestAManifestThatDoesNotDescribeItselfNeverReachesTheTree(t *testing.T) {
for name, broken := range map[string]func(*ingest.Manifest){
"a chapter list this build could not read": func(m *ingest.Manifest) { m.Chapters = nil },
"one chapter short": func(m *ingest.Manifest) { m.Chapters = m.Chapters[:1] },
"a unit list this build could not read": func(m *ingest.Manifest) { m.Chapters[0].Units = nil },
"counts no chapters at all": func(m *ingest.Manifest) {
m.Chapters, m.ChaptersTotal, m.UnitsTotal = nil, 0, 0
},
// The BOOK-level pair count disagreeing with the sum over chapters. Its own case because the
// per-chapter check cannot see it: every chapter can describe itself correctly while the
// document as a whole says a different number of pairs — which is what a renamed field does.
"counts pairs the chapters do not carry": func(m *ingest.Manifest) { m.UnitsTotal = 99 },
// …and a chapter numbered from zero, which chapter numbering cannot produce.
"a chapter numbered zero": func(m *ingest.Manifest) { m.Chapters[0].Number = 0 },
// A per-chapter count that is wrong while the BOOK's total still adds up. Its own case because
// nothing else reaches it: every other broken shape here also breaks the book-level sum, so the
// per-chapter equality could be deleted and the battery would not notice.
"chapter counts that swap while the book's total holds": func(m *ingest.Manifest) {
m.Chapters[0].UnitsTotal, m.Chapters[1].UnitsTotal = 1, 2
},
// The identities, which no counter stands beside: a renamed key decodes to "" for every row and
// the replacement write then keeps one of them.
"a chapter with no id": func(m *ingest.Manifest) { m.Chapters[0].ID = "" },
"a pair with no id": func(m *ingest.Manifest) { m.Chapters[0].Units[1].ID = "" },
"no identities at all, as a renamed key decodes": func(m *ingest.Manifest) {
for i := range m.Chapters {
m.Chapters[i].ID = ""
for j := range m.Chapters[i].Units {
m.Chapters[i].Units[j].ID = ""
}
}
},
} {
t.Run(name, func(t *testing.T) {
m := tree()
broken(&m)
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: m}}
err := svc.Refresh(t.Context(), owedBook(t.TempDir()))
if err == nil {
t.Fatal("the refresh reported success")
}
if store.saved {
t.Errorf("the tree was rewritten from a document that does not describe itself: %+v", store.structure)
}
// …and the debt is NOT discharged: the reader keeps the tree it has and the book comes back.
if len(store.cleared) != 0 {
t.Errorf("the debt was discharged by a materialization that wrote nothing: %+v", store.cleared)
}
})
}
}
// …and the same document reaching through the INTAKE's door is refused too. `RefreshCut` exists to
// save a third re-chunk per upload by passing the manifest the intake already decoded, and a floor
// on only one of the two entrances is a floor on neither.
func TestTheIntakesOwnCutIsHeldToTheSameFloor(t *testing.T) {
m := tree()
m.Chapters = nil
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
if err := svc.RefreshCut(t.Context(), owedBook(t.TempDir()), m); err == nil {
t.Fatal("the cut handed in by the intake was written without being checked")
}
if store.saved {
t.Errorf("the tree was rewritten from the intake's own broken cut: %+v", store.structure)
}
}
// A host that cannot run the engine does not write off every book on it.
//
// The attempt budget answers "this book cannot be read". An engine binary that is missing, locked
// out by another process or refusing an unmigrated schema says nothing about any book and applies to
// all of them at once — so counting it would abandon the whole library within a few passes, each
// book keeping whatever stale surface it had. The intake holds the same rule (books.defer_).
//
// Mutation caught: spending an attempt on a deployment fault; not deferring one at all.
func TestABrokenDeploymentDoesNotSpendABooksAttempts(t *testing.T) {
for _, c := range []struct {
name string
err error
}{
{"the binary cannot be run", fmt.Errorf("readmodel: read the manifest: %w", exec.ErrNotFound)},
{"the project is locked", exitError(t, ingest.ExitProjectLocked)},
{"the schema is not migrated", exitError(t, ingest.ExitSchemaMismatch)},
} {
t.Run(c.name, func(t *testing.T) {
store := &fakeStore{owed: []pgstore.OwedBook{
// One failure short of the write-off: if this pass counts, the book is given up on.
{ID: "bk_1", Workdir: t.TempDir(), OwedAt: boundary, Attempts: maxAttempts - 1},
}}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifestErr: c.err}}
if err := svc.Drain(t.Context()); err != nil {
t.Fatal(err)
}
if len(store.abandoned) != 0 {
t.Errorf("a book was given up on over a fault of the host: every book on it would be, within a few passes")
}
if len(store.deferred) != 1 {
t.Fatalf("the debt was not put back: %+v", store.deferred)
}
if store.deferCost[0] != pgstore.CostsNoAttempt {
t.Errorf("the deployment's fault was charged to the book's budget")
}
})
}
}
// exitError is what an engine that ANSWERED with a refusal looks like from the runner: the platform
// tells that from "could not be run" by the type, so a fixture has to produce a real one.
func exitError(t *testing.T, code int) error {
t.Helper()
err := exec.CommandContext(t.Context(), "sh", "-c", "exit "+strconv.Itoa(code)).Run()
var exit *exec.ExitError
if !errors.As(err, &exit) || exit.ExitCode() != code {
t.Fatalf("the fixture could not produce exit %d: %v", code, err)
}
return fmt.Errorf("readmodel: read the manifest: %w", err)
}
// A materialization that failed waits a REAL interval, and after enough of them the debt is given up
// on rather than retried for the life of the deployment.
//
// What was there before is the whole defect: claim, fail, `DeferReadModelDebt` writing `now()`, and
// the sweep running again fifteen seconds later — so "the back of the queue" was a queue of one and
// the book was re-read by the engine four times a minute, forever. Three things followed and every
// one of them was ongoing: up to five minutes of engine processes per pass; a book whose event
// stream can NEVER end, because `AtRest` requires the debt to be null and a browser reconnects to it
// for good; and, where the manifest reads and the export does not, a committed write on every pass —
// a revision bump and a frame every fifteen seconds to say nothing changed.
//
// Giving up is safe HERE in a way it never is for a run: the money of the boundary that stamped this
// debt has already settled, so what is lost is freshness of text. It is also not final — the next
// boundary of real work stamps a fresh debt (owesAReadingSurface resets the budget).
//
// Mutation caught: deferring to `now()`; not counting the attempt; never abandoning; abandoning on
// the first failure.
func TestAnUnpayableDebtBacksOffAndIsEventuallyGivenUpOn(t *testing.T) {
// A book whose published bank read-out exists and is not a bank: the read fails on every pass, so
// the materialization never completes — the shape of a project directory an operator broke.
broken := garbageBank(t)
for attempts := range maxAttempts {
store := &fakeStore{owed: []pgstore.OwedBook{
{ID: "bk_1", Workdir: broken, OwedAt: boundary, Attempts: attempts},
}}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: tree()}}
before := time.Now()
if err := svc.Drain(t.Context()); err != nil {
t.Fatal(err)
}
if len(store.cleared) != 0 {
t.Fatalf("attempt %d: a failed materialization discharged the debt: %+v", attempts, store.cleared)
}
last := attempts+1 >= maxAttempts
switch {
case last && len(store.abandoned) != 1:
t.Errorf("after %d failures the debt is still being retried; nothing ever ends this loop", attempts+1)
case last && store.abandonReason == "":
t.Error("the debt was written off with no reason recorded: an operator cannot tell why")
case !last && len(store.abandoned) != 0:
t.Errorf("the debt was given up on after %d failures, before its budget was spent", attempts+1)
case !last && len(store.deferredUntil) != 1:
t.Fatalf("attempt %d: deferred %d times, want once", attempts, len(store.deferredUntil))
case !last && !store.deferredUntil[0].After(before.Add(30*time.Second)):
// The number that matters: a deferral shorter than the sweep interval is not a deferral.
t.Errorf("attempt %d: the debt was pushed to %v, barely past %v — the next pass picks it straight back up",
attempts, store.deferredUntil[0], before)
case !last && !store.deferredUntil[0].After(before.Add(retryIn(attempts+1)-time.Second)):
// …and it GROWS with the count. Asserting the function alone leaves the CALL free to hand it
// a constant, which is a flat one-minute retry wearing the shape of a backoff.
t.Errorf("attempt %d: deferred to %v, which is not the %v this many failures have earned",
attempts, store.deferredUntil[0], retryIn(attempts+1))
}
}
}
// …and the delay grows and stops growing. Without a ceiling a book that healed would wait days;
// without growth the first retry is the only one that costs little.
func TestTheMaterializationBackoffGrowsAndIsCapped(t *testing.T) {
if retryIn(1) != time.Minute {
t.Errorf("the first retry is %v, want a minute", retryIn(1))
}
if retryIn(3) <= retryIn(1) {
t.Errorf("the retry does not grow: %v then %v", retryIn(1), retryIn(3))
}
if got := retryIn(1000); got != 30*time.Minute {
t.Errorf("the retry of a long-dead book is %v, want the cap", got)
}
}
// ⛔ THE SEAM BETWEEN THE ENGINE'S DOCUMENT AND THE STORE'S ROW, and it went untested until it broke.
//
// Both halves had tests: `ingest` decides whether a manifest is priced, `pgstore` writes what it is
// handed. What nobody asked was whether this function hands over the RIGHT THING — and the answer was
// no. `pgstore.Structure` carried the book's character count in two places at once (its own field and
// the wire struct's), this mapping filled one of them, and the row went in null. The type no longer
// admits that, so what is pinned here is the mapping's SHAPE: every money figure arrives, and the
// character count arrives on the CUT, where it survives a price this build cannot read.
//
// Mutation caught: dropping a member of the Projection literal; reading SourceChars off the price
// object again; gating the character count on `Priced`.
func TestTheProjectionReachesTheStoreWholeAndTheCharacterCountTravelsBesideIt(t *testing.T) {
m := tree()
m.Structure = ingest.StructureDetected
m.Price = &ingest.BookPrice{ExpectedUSD: 2_090_000, BookOnceUSD: 2_000_000, StepMaxUSD: 69_828, SourceChars: 3000}
pricedTree(&m)
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: m, noEnvelope: true}}
_ = svc.Refresh(t.Context(), owedBook(t.TempDir())) // the bank half fails; the tree lands anyway
if !store.saved {
t.Fatal("the tree did not land at all")
}
got := store.structure
if got.Price == nil {
t.Fatal("a manifest this build read whole reached the store with no projection")
}
if got.Price.Expected != 2_090_000 || got.Price.BookOnce != 2_000_000 || got.Price.StepMax != 69_828 {
t.Errorf("a money figure was lost in the mapping: %+v", *got.Price)
}
// ⛔ ON THE CUT, and this is the assertion the defect would have failed: the count is a property of
// the TEXT and the store keeps it apart from the money.
if got.SourceChars != 3000 {
t.Errorf("the engine's rune count reached the store as %d", got.SourceChars)
}
if got.Structure != ingest.StructureDetected {
t.Errorf("the cut's provenance reached the store as %q", got.Structure)
}
}
// …and the other direction of the same independence: a manifest whose MONEY this build cannot read
// still delivers the count of its text. The refusal is `ingest`'s (a chapter with no price), so what
// is pinned here is that this mapping does not widen it into the character count.
func TestAPriceThisBuildCannotReadDoesNotCostTheStoreItsCharacterCount(t *testing.T) {
m := tree()
m.Structure = ingest.StructureDetected
m.Price = &ingest.BookPrice{ExpectedUSD: 2_090_000, BookOnceUSD: 2_000_000, StepMaxUSD: 69_828, SourceChars: 3000}
pricedTree(&m)
// One unit loses its bill, and `ingest` then refuses the projection as a whole — the half-read
// shape is the dangerous one. The book's own `source_chars` is untouched and must still travel.
m.Chapters[0].Units[0].Price = nil
store := &fakeStore{}
svc := &Service{Store: store, Binary: "tmctl", Engine: &fakeEngine{manifest: m, noEnvelope: true}}
_ = svc.Refresh(t.Context(), owedBook(t.TempDir()))
if store.structure.Price != nil {
t.Errorf("a projection this build could not read whole was written anyway: %+v", *store.structure.Price)
}
if store.structure.SourceChars != 3000 {
t.Errorf("the character count went down with the price: %d", store.structure.SourceChars)
}
}
// pricedTree gives `tree()` a projection its own witness accepts: 30_000 micro-USD and 1000 runes a
// unit, the chapters' roll-up summing to the book's figure less the flat book-level bond.
func pricedTree(m *ingest.Manifest) {
for i := range m.Chapters {
total := money.MicroUSD(0)
for j := range m.Chapters[i].Units {
m.Chapters[i].Units[j].Price = &ingest.UnitPrice{ExpectedUSD: 30_000, SourceChars: 1000}
total += 30_000
}
m.Chapters[i].Price = &ingest.UnitPrice{ExpectedUSD: total, SourceChars: int64(len(m.Chapters[i].Units)) * 1000}
}
}