// Package readmodel materializes what the client READS: a book's chapter tree with the text of // every pair, and its memory bank. // // It is the consumer half of three engine channels and it is deliberately ONE place, because the // three have to agree with each other: the manifest carries the structure and the identities, the // export carries the text and the state derived from it, and the bank sidecar carries the terms. A // tree written from one of them and text written from another would drift apart at exactly the // moment they disagree — a book cut again between two calls. // // WHEN it runs is the contract's own answer, not a schedule: "`target` is updated at the boundaries // of the work and at stops, not continuously" (canon §Unit). Those boundaries are the end of an // intake and the end of a run, and they are also the only moments the engine's project is not held // exclusively by a translation. package readmodel import ( "context" "errors" "fmt" "log/slog" "time" "textmachine/platform/internal/ingest" "textmachine/platform/internal/pgstore" "textmachine/platform/internal/runner" ) // Engine is what this needs from the engine's $0 read commands. type Engine interface { Manifest(ctx context.Context, binary, workdir string) (ingest.Manifest, error) Export(ctx context.Context, binary, workdir string) (ingest.Export, error) } // Store is the write side of the read model, plus the ledger of what it still owes. type Store interface { SaveStructure(ctx context.Context, bookID string, in pgstore.Structure) error SaveBank(ctx context.Context, bookID string, terms []ingest.BankTerm) error BooksOwedReadModel(ctx context.Context, limit int) ([]pgstore.OwedBook, error) ClaimReadModelDebt(ctx context.Context, bookID string, owedAt time.Time, window time.Duration) (time.Time, error) ClearReadModelDebt(ctx context.Context, bookID string, owedAt time.Time) error DeferReadModelDebt(ctx context.Context, bookID string, owedAt time.Time) error } // MaterializeBudget is what ONE book's materialization may take. Generous because it is two full // re-chunks of the source, and bounded because a host that cannot run the engine must not spend a // whole pass finding that out on one book. const MaterializeBudget = 5 * time.Minute // booksPerDrain bounds one drain pass, for the same reason the budget above bounds one book. const booksPerDrain = 4 // dischargeBudget is what writing DOWN a finished materialization gets. Short: it is one statement // against a database this process is already connected to. const dischargeBudget = 30 * time.Second // Service refreshes the reading surface of one book. type Service struct { Store Store Engine Engine // Binary is the versioned tmctl path. Empty means this deployment cannot read the engine at all, // and then a refresh is a no-op rather than a failure: an instance with no engine is a read // replica, and the surface it serves is whatever was materialized before. Binary string Log *slog.Logger } func (s *Service) log() *slog.Logger { if s.Log == nil { return slog.New(slog.DiscardHandler) } return s.Log } // Claim takes a book's debt for this worker before it starts reading the engine, and answers the // book to work on — or `ok` false when the debt is no longer the caller's to pay. // // It is what stops two materializations of one book running at once: the intake pays its own debt // inline, the sweep drains the rest every few seconds, and without a claim every upload slower than // one sweep interval was read twice over. func (s *Service) Claim(ctx context.Context, b pgstore.OwedBook) (pgstore.OwedBook, bool, error) { claimed, err := s.Store.ClaimReadModelDebt(ctx, b.ID, b.OwedAt, MaterializeBudget) if err != nil || claimed.IsZero() { return pgstore.OwedBook{}, false, err } b.OwedAt = claimed return b, true, nil } // Refresh re-reads a book from the engine and writes what a client reads. // // The three channels are read in one order and for one reason: the STRUCTURE first, because the text // is addressed by it and a pair whose chapter this build has not seen has nowhere to go; the bank // last, because it is independent of both and its failure must not cost the tree. // // A failure of one channel does not abandon the others. Each is a different question about the same // book, and answering two of the three is strictly better than answering none — the alternative is a // reader screen that stays empty because a bank read-out that has never existed could not be found. // But a partial answer does not DISCHARGE the debt: everything that failed is returned, and the book // stays in the queue until one pass answers all three. func (s *Service) Refresh(ctx context.Context, b pgstore.OwedBook) error { return s.refresh(ctx, b, nil) } // RefreshCut is Refresh for a caller that has ALREADY read the manifest — the intake, which decoded // the same document a moment earlier to learn the book's chapter count. Re-reading it there is a // third full re-chunk of the source per upload (PD-248), paid for nothing. func (s *Service) RefreshCut(ctx context.Context, b pgstore.OwedBook, cut ingest.Manifest) error { return s.refresh(ctx, b, &cut) } func (s *Service) refresh(ctx context.Context, b pgstore.OwedBook, cut *ingest.Manifest) error { if s.Binary == "" || s.Engine == nil { // A replica with no engine serves what was materialized before, and — the debt being untouched // — leaves the work to an instance that has one. return nil } // Bounded HERE and not by each caller: the budget is a property of this work — two full re-chunks // of the source — rather than of whoever asks for it, and a caller with less time than that still // wins, since the shorter deadline is the one that fires. ctx, cancel := context.WithTimeout(ctx, MaterializeBudget) defer cancel() var failed error if err := s.refreshStructure(ctx, b.ID, b.Workdir, cut); err != nil { failed = err s.log().ErrorContext(ctx, "the chapter tree could not be refreshed", "err", err) } if err := s.refreshBank(ctx, b.ID, b.Workdir); err != nil { failed = errors.Join(failed, err) s.log().ErrorContext(ctx, "the bank could not be refreshed", "err", err) } if failed != nil { return failed } // The discharge is a RECORD, and it gets a context of its own — detached and short, the same rule // the intake's terminal writes follow. The reads above can legitimately consume the whole budget, // and a write that then runs on an expired context is a materialization that HAPPENED and was not // written down: the next pass would do two full re-chunks of the same source again. c, cancel := context.WithTimeout(context.WithoutCancel(ctx), dischargeBudget) defer cancel() return s.Store.ClearReadModelDebt(c, b.ID, b.OwedAt) } // Drain materializes the books that owe a reading surface, oldest debt first. // // This is the ONLY retry there is, and every way a materialization can be lost ends here: an engine // call that failed, a process that died between the boundary and the read, an instance that has no // engine to read with. The debt is a column, so none of those forget it. func (s *Service) Drain(ctx context.Context) error { if s.Binary == "" || s.Engine == nil { return nil } owed, err := s.Store.BooksOwedReadModel(ctx, booksPerDrain) if err != nil { return err } for _, b := range owed { if deadline, ok := ctx.Deadline(); ok && time.Until(deadline) < MaterializeBudget { // What is left of the pass is shorter than one book. Starting it anyway spends the engine's // CPU on a read that will be killed; the debt keeps the book for the next pass. return nil } b, ok, err := s.Claim(ctx, b) if err != nil { return err } if !ok { continue // somebody else is paying it, or a later boundary has replaced it } if err := s.Refresh(ctx, b); err != nil { // Loud: the boundary is past and its text is what the reader screen shows, so a book that // keeps failing here is a book that finished and looks empty. s.log().ErrorContext(ctx, "a book still owes its reading surface", "err", err) // …and it 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 starve the books behind it out // of text they were paid for. c, cancel := context.WithTimeout(context.WithoutCancel(ctx), dischargeBudget) if err := s.Store.DeferReadModelDebt(c, b.ID, b.OwedAt); err != nil { s.log().ErrorContext(ctx, "the unpaid debt could not be moved to the back of the queue", "err", err) } cancel() } } return nil } func (s *Service) refreshStructure(ctx context.Context, bookID, workdir string, cut *ingest.Manifest) error { manifest := ingest.Manifest{} if cut != nil { manifest = *cut } else { read, err := s.Engine.Manifest(ctx, s.Binary, workdir) if err != nil { return fmt.Errorf("readmodel: read the manifest: %w", err) } manifest = read } // The text is a SEPARATE call and a failing one does not cost the tree, which lands either way. // // ⚠ But this read must not SPEAK about text it does not have. A pair carries `TextKnown` and the // store then leaves the stored text alone — which covers the whole failure (nothing is known) and // the partial one alike: the two calls are two re-cuts of the same source, and a pair the export // did not carry is a skew between them, not an empty pair. // // ⚠ And it is RETURNED, not merely logged. A tree with no text is not a materialized book: the // caller counted this as done, the debt was discharged, and a first materialization that lost its // text left the reader unable to see even their own source. text, textErr := s.Engine.Export(ctx, s.Binary, workdir) if textErr != nil { textErr = fmt.Errorf("readmodel: read the pairs: %w", textErr) } byKey := make(map[[2]int]ingest.UnitText, len(text.Units)) for _, u := range text.Units { byKey[[2]int{u.Chapter, u.Unit}] = u } in := pgstore.Structure{ManifestKey: manifest.Key, Chapters: make([]pgstore.StructureChapter, 0, len(manifest.Chapters))} for _, c := range manifest.Chapters { chapter := pgstore.StructureChapter{ EngineID: c.ID, Number: c.Number, Units: make([]pgstore.StructureUnit, 0, len(c.Units)), } for _, u := range c.Units { pair := pgstore.StructureUnit{ EngineID: u.ID, Ordinal: u.FirstChunkIdx, State: ingest.StatePending, } if t, ok := byKey[[2]int{c.Number, u.FirstChunkIdx}]; ok { pair.Source, pair.Target, pair.State, pair.TextKnown = t.Source, t.Target, t.State, true } chapter.Units = append(chapter.Units, pair) } in.Chapters = append(in.Chapters, chapter) } if err := s.Store.SaveStructure(ctx, bookID, in); err != nil { return errors.Join(err, textErr) } // Counts, never a book id: the same rule as everywhere else on this side of the log. s.log().InfoContext(ctx, "the reading surface was refreshed", "chapters", len(in.Chapters), "pairs", manifest.UnitsTotal, "text", len(text.Units)) return textErr } func (s *Service) refreshBank(ctx context.Context, bookID, workdir string) error { bank, err := runner.ReadBank(workdir) if errors.Is(err, runner.ErrNoBank) { // A book that has never produced terms. Not an empty bank written over a full one: nothing is // saved at all, so a read-out that has simply not been made yet cannot erase one that was. return nil } if err != nil { return err } return s.Store.SaveBank(ctx, bookID, bank.Terms) }