1233 lines
59 KiB
Go
1233 lines
59 KiB
Go
package pgstore
|
|
|
|
import (
|
|
"context"
|
|
"crypto/sha256"
|
|
"encoding/base64"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
"unicode/utf8"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
|
|
"textmachine/platform/internal/ingest"
|
|
)
|
|
|
|
// ErrNoBook is a book that does not exist, or does not belong to the caller. ONE error for both, on
|
|
// purpose: telling them apart would let anyone enumerate other people's libraries.
|
|
var ErrNoBook = errors.New("pgstore: no such book")
|
|
|
|
// ErrBadCursor is a pagination cursor this collection cannot use. Rejecting it is the SERVER's duty
|
|
// (contract, NextCursor): the client holds an opaque string and cannot judge it.
|
|
var ErrBadCursor = errors.New("pgstore: cursor does not apply to this collection")
|
|
|
|
// Book is the library row, in contract terms.
|
|
type Book struct {
|
|
ID string
|
|
Title string
|
|
SourceLang string
|
|
TargetLang string
|
|
// Status is the PRODUCT status, already resolved against the book's current or last run
|
|
// (derivedStatus): the book's status is the run's, except the states a run cannot be in. Read
|
|
// off the column alone it disagreed with the run observably — a signing stop is written on the
|
|
// run row, and the card went on saying `translating`.
|
|
Status string
|
|
// RejectReason is the PLATFORM's internal word for why an intake ended; the wire vocabulary is a
|
|
// projection of it (httpapi.contractRejectReason).
|
|
RejectReason string
|
|
StructureVersion int
|
|
// ShapeEpoch is the generation of the book's COUNT: it moves when the pipeline's shape changes
|
|
// under the book (an editor added or removed), which is the one other event besides a re-cut that
|
|
// legitimately recomputes `ChaptersDone`. A coordinate, like StructureVersion — the shape itself
|
|
// is deliberately not published (canon 0.9.0).
|
|
ShapeEpoch int
|
|
ChapterCount int
|
|
// ChaptersDone is the book's own lifetime progress — chapters fully translated — as opposed to
|
|
// the bar of a run, which measures what that run bought.
|
|
ChaptersDone int
|
|
CharacterCount int64
|
|
NoteCount int
|
|
AddedAt time.Time
|
|
// Revision is the book's own scope. Read on the card, where it orders the client's reads against
|
|
// the stream's frames; the LIBRARY has a revision of its own and the two are never compared.
|
|
Revision int64
|
|
}
|
|
|
|
// NewBook is one book being registered in the read model.
|
|
type NewBook struct {
|
|
OwnerID string
|
|
Title string
|
|
SourceLang string
|
|
TargetLang string
|
|
ChapterCount int
|
|
CharacterCount int64
|
|
// Workdir is the engine's project directory. The platform reads the book's journal there and
|
|
// spawns the engine with it as the working directory; it never writes into it (D39.110).
|
|
Workdir string
|
|
Now time.Time
|
|
}
|
|
|
|
// AddBook registers a book. `not_started` and not `uploading`: the file is already on disk, and a
|
|
// status describing an upload this route never performed would be a lie the library then shows.
|
|
func (s *Store) AddBook(ctx context.Context, in NewBook) (string, error) {
|
|
id := NewBookID()
|
|
const q = `
|
|
insert into books (id, owner_id, title, source_lang, target_lang, status,
|
|
chapter_count, character_count, added_at, workdir, engine_book_id, revision)
|
|
values ($1, $2, $3, $4, $5, 'not_started', $6, $7, $8, $9, $10, ` + nextLibraryRevision + `)`
|
|
// engine_book_id is the platform's id until the engine's handshake reports its own. It is not
|
|
// left empty: the column is the join back to whatever the engine calls this book, and an empty
|
|
// one would read as "the engine has no name for it" rather than "we have not been told yet".
|
|
_, err := s.pool.Exec(ctx, q, id, in.OwnerID, in.Title, in.SourceLang, in.TargetLang,
|
|
in.ChapterCount, in.CharacterCount, in.Now, in.Workdir, id)
|
|
if err != nil {
|
|
var pg *pgconn.PgError
|
|
if errors.As(err, &pg) && pg.ConstraintName == "books_owner_id_fkey" {
|
|
return "", ErrNoAccount
|
|
}
|
|
return "", fmt.Errorf("pgstore: add book: %w", err)
|
|
}
|
|
return id, nil
|
|
}
|
|
|
|
// NewBookID mints a book identifier. Exported because the intake needs one BEFORE the row exists:
|
|
// the book's directory is named after it, and the row records that directory.
|
|
func NewBookID() string { return newID("bk") }
|
|
|
|
// nextLibraryRevision is the revision a book JOINS the library with, as a SQL fragment over $2 (the
|
|
// owner).
|
|
//
|
|
// The library's revision is derived as the maximum over the account's books, and the contract asks
|
|
// a client to DROP a read whose revision is not above what it already applied. A book inserted at
|
|
// revision 0 therefore left the library's number unchanged, and the screen that had just uploaded
|
|
// it dropped the very read that carried it (register row PD-122). Starting above the account's
|
|
// current maximum is what makes an addition visible.
|
|
//
|
|
// It is the greatest of BOTH halves the library's own revision is read from — the maximum over the
|
|
// books and the account's floor — because taking only the maximum reproduced the very defect the
|
|
// floor exists to prevent, one step later: after a cancelled upload raised the floor above the
|
|
// maximum, the NEXT book joined below it, and the library then answered the same number for three
|
|
// different states of itself.
|
|
//
|
|
// Two concurrent additions can compute the same number, and that is harmless: the value still rises,
|
|
// so the client still refetches and still sees both. What stays open in PD-122 is the account-scope
|
|
// counter itself — one writer for membership and statuses — which is what would let two SIMULTANEOUS
|
|
// writers get distinct numbers instead of merely rising ones.
|
|
const nextLibraryRevision = `(select greatest(
|
|
coalesce((select max(revision) from books where owner_id = $2), 0),
|
|
coalesce((select library_revision from users where id = $2), 0)) + 1)`
|
|
|
|
// nextRevisionOfThisBooksLibrary is the same number for a statement that has the book but not its
|
|
// owner: the owner is read from the row being updated.
|
|
//
|
|
// ⚠ A status change may NOT simply do `revision + 1`, and that is the same defect as the one above
|
|
// wearing different clothes. A book whose own counter sits below the account's floor — it was
|
|
// uploaded before another upload was cancelled — would walk `uploading → parsing → not_started`
|
|
// three increments deep and still be under the floor, so `greatest(max, floor)` would not move at
|
|
// all. A client obeying the contract drops every read that is not ABOVE what it applied, and the
|
|
// screen that uploaded the book would keep showing it as arriving until some unrelated event pushed
|
|
// the number. Taking the library's own next number instead makes every status visible exactly once.
|
|
const nextRevisionOfThisBooksLibrary = `(select greatest(
|
|
coalesce((select max(b.revision) from books b where b.owner_id = books.owner_id), 0),
|
|
coalesce((select u.library_revision from users u where u.id = books.owner_id), 0)) + 1)`
|
|
|
|
// NewUpload is one book arriving through the contract's intake, before its file has been received.
|
|
type NewUpload struct {
|
|
OwnerID string
|
|
Title string
|
|
SourceLang string
|
|
TargetLang string
|
|
// Workdir is the engine project directory this book will own. It is named after the book id and
|
|
// created before the row, so a row always points at a directory that exists.
|
|
Workdir string
|
|
Now time.Time
|
|
}
|
|
|
|
// CreateUpload registers a book whose file is still arriving.
|
|
//
|
|
// `uploading` is a real, observable state and not a formality: the row is written BEFORE the body is
|
|
// read, so a 60 MB upload is visible in the library — with its languages, without its size — while
|
|
// it is still on the wire. The alternative, inserting once the bytes have landed, gives the status
|
|
// no writer at all and makes a half-finished upload invisible to everything, including the sweep
|
|
// that has to clean it up.
|
|
func (s *Store) CreateUpload(ctx context.Context, id string, in NewUpload) (Book, error) {
|
|
const q = `
|
|
insert into books (id, owner_id, title, source_lang, target_lang, status,
|
|
added_at, workdir, engine_book_id, revision)
|
|
values ($1, $2, $3, $4, $5, 'uploading', $6, $7, $1, ` + nextLibraryRevision + `)`
|
|
var b Book
|
|
err := s.inTx(ctx, func(tx pgx.Tx) error {
|
|
if _, err := tx.Exec(ctx, q, id, in.OwnerID, in.Title, in.SourceLang, in.TargetLang,
|
|
in.Now, in.Workdir); err != nil {
|
|
var pg *pgconn.PgError
|
|
if errors.As(err, &pg) && pg.ConstraintName == "books_owner_id_fkey" {
|
|
return ErrNoAccount
|
|
}
|
|
return fmt.Errorf("pgstore: create upload: %w", err)
|
|
}
|
|
var err error
|
|
if b, err = readBookTx(ctx, tx, id); err != nil {
|
|
return err
|
|
}
|
|
// The book EXISTS from this moment and a watcher of the library is entitled to hear so: the
|
|
// row is written before the bytes, which is the whole reason `uploading` is observable.
|
|
return emitStatus(ctx, tx, id)
|
|
})
|
|
if err != nil {
|
|
return Book{}, err
|
|
}
|
|
return b, nil
|
|
}
|
|
|
|
// StartParsing hands a received file to the parser: `uploading → parsing`, with the size the intake
|
|
// counted while the bytes went past.
|
|
//
|
|
// Conditional on the status, so an upload the sweep has already given up on cannot be resurrected by
|
|
// a request that finally finished.
|
|
//
|
|
// enqueue joins the caller's transaction for the same reason the run admission's does: a status that
|
|
// says "being parsed" with no job to do it waits for the backstop sweep's grace, and a job for a
|
|
// book that never became `parsing` is a worker with nothing to claim.
|
|
func (s *Store) StartParsing(ctx context.Context, id string, characters int64,
|
|
enqueue func(context.Context, Tx, string) error) (Book, error) {
|
|
var b Book
|
|
err := s.inTx(ctx, func(tx pgx.Tx) error {
|
|
const q = `
|
|
update books set status = 'parsing', character_count = $2,
|
|
revision = ` + nextRevisionOfThisBooksLibrary + `
|
|
where id = $1 and status = 'uploading'`
|
|
tag, err := tx.Exec(ctx, q, id, characters)
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: start parsing: %w", err)
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNoBook
|
|
}
|
|
if b, err = readBookTx(ctx, tx, id); err != nil {
|
|
return err
|
|
}
|
|
if err := emitStatus(ctx, tx, id); err != nil {
|
|
return err
|
|
}
|
|
if enqueue != nil {
|
|
return enqueue(ctx, tx, id)
|
|
}
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return Book{}, err
|
|
}
|
|
return b, nil
|
|
}
|
|
|
|
// ParseClaim is one book this process may parse.
|
|
type ParseClaim struct {
|
|
BookID string
|
|
Workdir string
|
|
Attempts int
|
|
// What the USER declared at intake. It travels with the claim because the parse is also where a
|
|
// book's starting engine configuration is rendered (books/render.go), and reading it in a second
|
|
// query would be a second query for a row this one already has open.
|
|
Title string
|
|
SourceLang string
|
|
TargetLang string
|
|
// At is the stamp this claim was taken with. Every write that ENDS the parse carries it back, so
|
|
// a pass whose claim was taken over by another cannot record a verdict about work it is no longer
|
|
// doing — the same re-check-under-the-write the money paths use.
|
|
At time.Time
|
|
}
|
|
|
|
// ErrParseClaimed is a parse someone else is already doing, or a book that is no longer waiting for
|
|
// one. The ordinary case, not a failure: the queue job and the backstop sweep both aim at the same
|
|
// book on purpose.
|
|
var ErrParseClaimed = errors.New("pgstore: the parse of this book is already claimed")
|
|
|
|
// ClaimParse takes the right to parse a book, once.
|
|
//
|
|
// A compare-and-set rather than a plain read, for the same reason the spawn claim is one: the queue
|
|
// worker and the backstop sweep legitimately arrive together, and two `tmctl manifest` processes on
|
|
// one project directory would be two writers of one SQLite file — which is exactly what the engine's
|
|
// exclusive lock exists to refuse, noisily and after the work.
|
|
//
|
|
// staleAfter is what makes the claim recoverable: a process that died holding one leaves a stamp
|
|
// nothing clears, and without a grace the book would wait for parsing forever.
|
|
func (s *Store) ClaimParse(ctx context.Context, id string, now, staleBefore time.Time) (ParseClaim, error) {
|
|
const q = `
|
|
update books set parse_started_at = $2, parse_attempts = parse_attempts + 1
|
|
where id = $1 and status = 'parsing'
|
|
and (parse_started_at is null or parse_started_at < $3)
|
|
returning id, workdir, parse_attempts, title, source_lang, target_lang`
|
|
var c ParseClaim
|
|
err := s.pool.QueryRow(ctx, q, id, now, staleBefore).Scan(&c.BookID, &c.Workdir, &c.Attempts,
|
|
&c.Title, &c.SourceLang, &c.TargetLang)
|
|
c.At = now
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ParseClaim{}, ErrParseClaimed
|
|
}
|
|
if err != nil {
|
|
return ParseClaim{}, fmt.Errorf("pgstore: claim parse: %w", err)
|
|
}
|
|
return c, nil
|
|
}
|
|
|
|
// RefundParseAttempt gives back an attempt that was claimed but never spent on the engine.
|
|
//
|
|
// The budget exists to bound how many times a broken HOST is asked to parse a book, and a book that
|
|
// is only waiting for its configuration asks nobody: counting those claims meant the budget ran out
|
|
// while nothing had been tried, and the first real answer from the engine was terminal the moment it
|
|
// arrived. Conditional on the claim, like every other write that ends a pass.
|
|
func (s *Store) RefundParseAttempt(ctx context.Context, id string, claimedAt time.Time) error {
|
|
_, err := s.pool.Exec(ctx, `
|
|
update books set parse_attempts = greatest(parse_attempts - 1, 0)
|
|
where id = $1 and status = 'parsing' and parse_started_at = $2`, id, claimedAt)
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: refund parse attempt: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ParsedBook is what the engine's manifest told the platform about a book it has now cut.
|
|
type ParsedBook struct {
|
|
Chapters int
|
|
SourceSHA256 []byte
|
|
ChunkerVersion string
|
|
}
|
|
|
|
// FinishParse records a parsed book — `parsing → not_started` — and the materialization it now owes.
|
|
// It returns the moment that debt was stamped, which is what discharges it (ClearReadModelDebt).
|
|
//
|
|
// Conditional on the CLAIM still being this caller's: a pass whose claim was taken over is not the
|
|
// pass whose answer counts, and two answers about one book are one answer too many.
|
|
//
|
|
// The debt is written HERE rather than after the engine call that pays it off, because everything
|
|
// between the two can fail: this is the transaction that makes the book parsed, and a book that is
|
|
// parsed owes a tree.
|
|
func (s *Store) FinishParse(ctx context.Context, id string, claimedAt time.Time, in ParsedBook) (time.Time, error) {
|
|
const q = `
|
|
update books set status = 'not_started', chapter_count = $2, source_sha256 = $3,
|
|
chunker_version = $4, parse_started_at = null,
|
|
` + owesAReadingSurface + `
|
|
revision = ` + nextRevisionOfThisBooksLibrary + `
|
|
where id = $1 and status = 'parsing' and parse_started_at = $5
|
|
returning read_model_owed_at`
|
|
var owed time.Time
|
|
err := s.inTx(ctx, func(tx pgx.Tx) error {
|
|
err := tx.QueryRow(ctx, q, id, in.Chapters, in.SourceSHA256, in.ChunkerVersion, claimedAt).Scan(&owed)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrNoBook
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: finish parse: %w", err)
|
|
}
|
|
// The end of the parse is an ORDINARY status change and reaches the client as one: there is
|
|
// no "parsing finished" frame, deliberately, because the fact is the status (canon
|
|
// §streamBookEvents).
|
|
return emitStatus(ctx, tx, id)
|
|
})
|
|
return owed, err
|
|
}
|
|
|
|
// RejectBook is the terminal end of an intake that failed: `parsing → rejected`.
|
|
//
|
|
// The reason is the platform's own word for the class of failure and is not projected — contract v0
|
|
// has no field for it (migration 00013).
|
|
//
|
|
// Conditional on the CLAIM, like FinishParse and for a sharper reason: rejecting is destructive —
|
|
// the book's source goes with it — and a pass whose claim was stolen may be holding an exit code
|
|
// that says nothing about the book. The engine maps every failure onto exit 1, including a lock held
|
|
// by the parse that took the claim over.
|
|
func (s *Store) RejectBook(ctx context.Context, id string, claimedAt time.Time, reason string) error {
|
|
const q = `
|
|
update books set status = 'rejected', reject_reason = $2, parse_started_at = null,
|
|
revision = ` + nextRevisionOfThisBooksLibrary + `
|
|
where id = $1 and status = 'parsing' and parse_started_at = $3`
|
|
return s.inTx(ctx, func(tx pgx.Tx) error {
|
|
tag, err := tx.Exec(ctx, q, id, reason, claimedAt)
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: reject book: %w", err)
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNoBook
|
|
}
|
|
return emitStatus(ctx, tx, id)
|
|
})
|
|
}
|
|
|
|
// DeleteUpload removes a book whose file never arrived.
|
|
//
|
|
// Not `rejected`: that status means "the file could not be parsed" (contract §BookStatus), and a
|
|
// request the client abandoned produced no file to judge. The user's library must not accumulate a
|
|
// permanent row for every cancelled upload — there is no delete handle in the contract to clear one
|
|
// with.
|
|
//
|
|
// Narrow on purpose: only a book still in `uploading`, and only one no run was ever started for, so
|
|
// this can never become a way to delete a book that has cost money.
|
|
//
|
|
// ⚠ It also carries the library's revision ACROSS the deletion, and that is not bookkeeping: the
|
|
// library's number is derived as the maximum over the account's books, so removing the newest one
|
|
// would lower it — and the contract asks a client to DROP a read whose revision is below what it
|
|
// applied (§Revision). The screen would then keep rendering a book that no longer exists until some
|
|
// other book's counter grew past it. The floor lives in `users.library_revision`, the account-scope
|
|
// counter that has existed unused since 00001 (register row PD-122).
|
|
func (s *Store) DeleteUpload(ctx context.Context, id string) error {
|
|
return s.inTx(ctx, func(tx pgx.Tx) error {
|
|
// Book first, as in every transaction that touches two of these tables (lockBook). `users` is
|
|
// taken after it, and no path in this package takes them the other way round.
|
|
if err := lockBook(ctx, tx, id); err != nil {
|
|
return err
|
|
}
|
|
var owner string
|
|
var revision int64
|
|
err := tx.QueryRow(ctx, `
|
|
delete from books where id = $1 and status = 'uploading'
|
|
and not exists (select 1 from runs where book_id = $1)
|
|
returning owner_id, revision`, id).Scan(&owner, &revision)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return ErrNoBook
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: delete upload: %w", err)
|
|
}
|
|
if _, err := tx.Exec(ctx, `
|
|
update users set library_revision = greatest(library_revision, $2) where id = $1`,
|
|
owner, revision+1); err != nil {
|
|
return fmt.Errorf("pgstore: carry the library revision: %w", err)
|
|
}
|
|
return nil
|
|
})
|
|
}
|
|
|
|
// BookForMigration is one book and whether an engine SCHEMA migration of its project file is safe
|
|
// right now.
|
|
//
|
|
// It exists because an engine upgrade has an order and the order is money (unified backlog row 174).
|
|
// The engine's read path refuses a project file older than the binary reading it, so a deployment
|
|
// that migrates a book somebody is still using breaks that use — and the three ways a book is still
|
|
// in use are different facts that a single "is it running" question misses:
|
|
//
|
|
// - Live: a unit is going right now. Migrating under it is migrating under a writer.
|
|
// - Resumable: a run ENDED in a state its owner can continue from, and a resume runs on the build
|
|
// the run was PINNED to (row 139). Migrating moves the file past that build, so the resume either
|
|
// fails or is forced onto a newer one — and re-pinning a paid run is how already-bought calls get
|
|
// paid for twice.
|
|
// ⚠ `paused` LEFT this set in P7 and it is not an optimisation: since 0.3.0 a run stopped at a
|
|
// limit cannot be continued by `resume` at ALL — it answers 409 `ceiling_reached`, and the
|
|
// remedy is a NEW run, which is spawned with the CURRENT binary and has nothing pinned to
|
|
// migrate around (canon §resumeRun; register row PD-217, which is where a book stuck on the
|
|
// engine's daily ceiling blocked an upgrade for ever). The coupling is real and worth naming:
|
|
// the day `resume` continues a paused run again, this predicate has to come back with it.
|
|
// - Unsettled: an attempt's hold is still open, and the settlement reads the engine's committed
|
|
// figure with that attempt's pinned build. Migrate first and the figure can never be read, so the
|
|
// hold stays reserved with no sweep able to close it.
|
|
type BookForMigration struct {
|
|
ID string
|
|
Title string
|
|
Workdir string
|
|
// Live, Resumable and Unsettled are reported separately rather than folded into one boolean: an
|
|
// operator draining a stand needs to know WHICH of them to wait for, and they are cleared by
|
|
// different actions.
|
|
Live bool
|
|
Resumable bool
|
|
Unsettled bool
|
|
}
|
|
|
|
// Migratable reports whether this book's project file may be migrated now.
|
|
func (b BookForMigration) Migratable() bool { return !b.Live && !b.Resumable && !b.Unsettled }
|
|
|
|
// BooksForMigration lists every book with the three facts an engine upgrade turns on.
|
|
func (s *Store) BooksForMigration(ctx context.Context) ([]BookForMigration, error) {
|
|
const q = `
|
|
select b.id, b.title, b.workdir,
|
|
exists (select 1 from runs r
|
|
where r.book_id = b.id and r.finished_at is null),
|
|
exists (select 1 from runs r
|
|
where r.book_id = b.id and r.finished_at is not null
|
|
and r.status in ('stopped', 'awaiting_bank')),
|
|
exists (select 1 from runs r
|
|
join run_attempts a on a.run_id = r.id
|
|
join reservations res on res.engine_run_id = r.id || '#' || a.attempt_no
|
|
and res.state = 'open'
|
|
where r.book_id = b.id)
|
|
from books b
|
|
order by b.added_at, b.id`
|
|
rows, err := s.pool.Query(ctx, q)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("pgstore: list books for migration: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []BookForMigration
|
|
for rows.Next() {
|
|
var b BookForMigration
|
|
if err := rows.Scan(&b.ID, &b.Title, &b.Workdir, &b.Live, &b.Resumable, &b.Unsettled); err != nil {
|
|
return nil, fmt.Errorf("pgstore: scan book for migration: %w", err)
|
|
}
|
|
out = append(out, b)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// IntakeBook is a book the backstop sweep has to make a decision about.
|
|
type IntakeBook struct {
|
|
ID string
|
|
Status string
|
|
Workdir string
|
|
Attempts int
|
|
}
|
|
|
|
// owesAReadingSurface is what a transaction that ENDS a run adds to its `update books`. It is a
|
|
// fragment rather than three copies of one assignment because there are three such transactions —
|
|
// FinishRun, PauseRun and FinishUnspawnedStop — and the debt is only a mechanism while EVERY one of
|
|
// them writes it: a run whose ending forgets it is a run the user paid for whose text nothing will
|
|
// ever materialize, and nothing looks at that book again.
|
|
//
|
|
// ⚠ It also RESETS the attempt budget, and that is part of the same fragment on purpose: a new
|
|
// boundary is new evidence. A book whose debt was written off as unpayable (read_model_abandoned_at)
|
|
// has just finished a run, so the engine may well answer about it now — and a fresh boundary that
|
|
// inherited a spent budget would be written off without a single try.
|
|
const owesAReadingSurface = `read_model_owed_at = now(), read_model_attempts = 0,
|
|
read_model_abandoned_at = null, read_model_error = null, `
|
|
|
|
// OwedBook is a book whose reading surface is DUE a materialization. OwedAt is the stamp the caller
|
|
// read, and every write below is conditional on it still standing: a debt a later boundary stamped,
|
|
// or another worker claimed, is not the one this caller is holding.
|
|
type OwedBook struct {
|
|
ID string
|
|
Workdir string
|
|
OwedAt time.Time
|
|
// Attempts is how many passes have already failed to pay this debt. It travels with the book
|
|
// because the decision it feeds — try again later, or write it off — belongs to the worker that
|
|
// just failed, and a counter kept in the process would restart with the daemon and never reach
|
|
// any threshold. Same shape and same reason as `books.parse_attempts` at the intake.
|
|
Attempts int
|
|
}
|
|
|
|
// BooksOwedReadModel lists books whose reading surface is DUE a materialization and that are quiet
|
|
// enough to give it — the same "nothing is running" the stream ends on.
|
|
//
|
|
// Quiet, because the contract puts the freshness of a pair's text at the BOUNDARIES of the work: a
|
|
// book with a live run is not at one. (It is not a lock: the engine's read commands open the project
|
|
// WITHOUT its exclusive flock — backend/internal/store/store.go, NewReadOnlyRunner — precisely so an
|
|
// operator is not shut out of `status` for the length of a run.)
|
|
//
|
|
// DUE and not merely owed: a worker claims a book by pushing its stamp out (ClaimReadModelDebt), so
|
|
// the book the intake is materializing right now is not offered to the sweep as well. Without that
|
|
// every upload slower than one sweep interval was materialized twice at once — the cost this pack
|
|
// removed elsewhere, paid back with interest.
|
|
//
|
|
// Bounded, because a host that cannot run the engine would otherwise put every book of the
|
|
// deployment into one pass.
|
|
func (s *Store) BooksOwedReadModel(ctx context.Context, limit int) ([]OwedBook, error) {
|
|
const q = `
|
|
select b.id, b.workdir, b.read_model_owed_at, b.read_model_attempts
|
|
from books b
|
|
where b.read_model_owed_at is not null and b.read_model_owed_at <= now()
|
|
and ` + nothingIsRunning + `
|
|
order by b.read_model_owed_at limit $1`
|
|
rows, err := s.pool.Query(ctx, q, limit)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("pgstore: list books owed a reading surface: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []OwedBook
|
|
for rows.Next() {
|
|
var b OwedBook
|
|
if err := rows.Scan(&b.ID, &b.Workdir, &b.OwedAt, &b.Attempts); err != nil {
|
|
return nil, fmt.Errorf("pgstore: scan book owed a reading surface: %w", err)
|
|
}
|
|
out = append(out, b)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// ClaimReadModelDebt takes a debt for one worker, by pushing its DUE time `window` into the future.
|
|
// It answers the new stamp — which is what that worker must present to discharge or defer it — or
|
|
// the zero time when the debt is no longer the one the caller read: another worker claimed it, or a
|
|
// later boundary stamped a new one.
|
|
//
|
|
// The window is a LEASE and not a promise: a worker that dies holding one simply leaves the book
|
|
// invisible until it lapses, which is the same shape the intake's parse claim has.
|
|
func (s *Store) ClaimReadModelDebt(ctx context.Context, bookID string, owedAt time.Time, window time.Duration) (time.Time, error) {
|
|
var claimed time.Time
|
|
err := s.pool.QueryRow(ctx, `
|
|
update books set read_model_owed_at = now() + make_interval(secs => $3)
|
|
where id = $1 and read_model_owed_at = $2
|
|
returning read_model_owed_at`, bookID, owedAt, window.Seconds()).Scan(&claimed)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return time.Time{}, nil
|
|
}
|
|
if err != nil {
|
|
return time.Time{}, fmt.Errorf("pgstore: claim the reading-surface debt: %w", err)
|
|
}
|
|
return claimed, nil
|
|
}
|
|
|
|
// AttemptCost says whether a deferral counts against the book's attempt budget.
|
|
type AttemptCost bool
|
|
|
|
const (
|
|
// SpendsAnAttempt — the engine answered ABOUT THIS BOOK and the answer was a failure.
|
|
SpendsAnAttempt AttemptCost = true
|
|
// CostsNoAttempt — the failure belongs to the deployment (ingest.DeploymentFault) and applies to
|
|
// every book on the host, so counting it would write them all off in a few passes.
|
|
CostsNoAttempt AttemptCost = false
|
|
)
|
|
|
|
// DeferReadModelDebt moves a debt this pass could not pay to the BACK of the queue, counts the
|
|
// failure when it was the book's own, and answers how many there have now been.
|
|
//
|
|
// The queue is oldest-first, so a book the engine can never answer about — a workdir an operator
|
|
// moved, a project database this build cannot read — would be retried first forever, and enough of
|
|
// them would starve every book behind them out of text that was already paid for. Conditional on the
|
|
// stamp for the same reason the discharge is: a boundary recorded since this pass began is a newer
|
|
// debt and must not be overwritten.
|
|
//
|
|
// ⚠ `now()` was the whole of it, and "the back of the queue" was therefore a queue of one: the
|
|
// pass runs every 15 seconds, `<= now()` is true again immediately, and the same book was re-read
|
|
// by the engine four times a minute for as long as it existed. The caller passes a real deadline
|
|
// now, and the count it gets back is what ends the loop (AbandonReadModelDebt).
|
|
func (s *Store) DeferReadModelDebt(ctx context.Context, bookID string, owedAt, next time.Time, reason string, cost AttemptCost) (int, error) {
|
|
var attempts int
|
|
err := s.pool.QueryRow(ctx, `
|
|
update books set read_model_owed_at = $3, read_model_error = $4,
|
|
read_model_attempts = read_model_attempts + case when $5 then 1 else 0 end
|
|
where id = $1 and read_model_owed_at = $2
|
|
returning read_model_attempts`, bookID, owedAt, next, truncateReason(reason), bool(cost)).Scan(&attempts)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return 0, nil // a newer boundary or another worker owns it now: not this caller's debt to defer
|
|
}
|
|
if err != nil {
|
|
return 0, fmt.Errorf("pgstore: defer the reading-surface debt: %w", err)
|
|
}
|
|
return attempts, nil
|
|
}
|
|
|
|
// AbandonReadModelDebt writes a debt off as unpayable: the book keeps whatever surface it has and
|
|
// stops being asked about.
|
|
//
|
|
// Giving up is SAFE here and that asymmetry is why this exists at all while a stalled RUN is only
|
|
// ever handed to an operator. The money of the boundary that stamped this debt has already settled,
|
|
// so what is lost is the freshness of text — while what was lost by never giving up is three things,
|
|
// all of them ongoing and all of them named by the register: up to five minutes of engine processes
|
|
// per pass forever; a book whose event stream can NEVER end, because `AtRest` requires this column
|
|
// to be null and a browser therefore reconnects to it for good; and, where the manifest reads and
|
|
// the export does not, a committed `SaveStructure` on every pass — a revision bump and a frame every
|
|
// fifteen seconds, to say nothing changed.
|
|
//
|
|
// It is not final in the sense that matters to a user: the next boundary of real work stamps a fresh
|
|
// debt and clears this (owesAReadingSurface), because a new boundary is new evidence.
|
|
func (s *Store) AbandonReadModelDebt(ctx context.Context, bookID string, owedAt, now time.Time, reason string) error {
|
|
// The attempt that failed is COUNTED here as everywhere else. Without it the column said one less
|
|
// than had actually been tried, so `books --abandoned` reported four tries where the log line
|
|
// beside it said five — two numbers about one thing, which is how an operator learns to trust
|
|
// neither.
|
|
if _, err := s.pool.Exec(ctx, `
|
|
update books set read_model_owed_at = null, read_model_attempts = read_model_attempts + 1,
|
|
read_model_abandoned_at = $3, read_model_error = $4
|
|
where id = $1 and read_model_owed_at = $2`,
|
|
bookID, owedAt, now, truncateReason(reason)); err != nil {
|
|
return fmt.Errorf("pgstore: abandon the reading-surface debt: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// truncateReason bounds what an engine's or a driver's error string may put into a column. These
|
|
// fields are an OPERATOR's, never a client's: they carry paths, unit names and driver text, and the
|
|
// wire's vocabulary of failure is a closed enum decided elsewhere.
|
|
//
|
|
// ⚠ IT SANITISES BEFORE IT CUTS, AND BOTH HALVES ARE LOAD-BEARING — this is not defensive habit.
|
|
// What arrives here is the first line of the ENGINE's stderr, verbatim and unbounded
|
|
// (runner.firstLine), for a product whose sources are Chinese and Japanese: a message quoting the
|
|
// book's own text is both long and multi-byte. Postgres refuses a `text` value that is not valid
|
|
// UTF-8 (SQLSTATE 22021), so a byte-slice through the middle of a rune makes the write FAIL — and
|
|
// the write that fails is precisely the one that records the failure. The book or the run then
|
|
// keeps its old deadline, its counter never rises, and it stands at the head of the queue for good:
|
|
// the exact starvation this whole pack exists to end, re-created by the bookkeeping of it. Arbitrary
|
|
// stderr is not guaranteed to be valid UTF-8 to begin with either, which is why the repair comes
|
|
// first and the cut second.
|
|
//
|
|
// ⚠ NUL DEFEATS THE REPAIR ABOVE: U+0000 is valid UTF-8 and Postgres refuses it anyway, same
|
|
// SQLSTATE, same self-defeating failure. No live path carrying one was found — the engine gates NUL
|
|
// on every decode branch — so this closes a hole in the guard rather than a reproduced defect.
|
|
func truncateReason(reason string) string {
|
|
const max = 1000
|
|
reason = strings.ToValidUTF8(reason, "")
|
|
reason = strings.ReplaceAll(reason, "\x00", "")
|
|
if len(reason) <= max {
|
|
return reason
|
|
}
|
|
cut := max
|
|
for cut > 0 && !utf8.RuneStart(reason[cut]) {
|
|
cut--
|
|
}
|
|
return reason[:cut]
|
|
}
|
|
|
|
// RearmReadModelDebt is the operator's handle on a written-off debt: ask for the materialization
|
|
// again, with a clean budget, without waiting for the book's next run.
|
|
//
|
|
// It answers false when there was nothing to re-arm — the book already owes one, or has never owed
|
|
// one — so the CLI can say which of the two happened instead of reporting a success that did nothing.
|
|
func (s *Store) RearmReadModelDebt(ctx context.Context, bookID string, now time.Time) (bool, error) {
|
|
tag, err := s.pool.Exec(ctx, `
|
|
update books set read_model_owed_at = $2, read_model_attempts = 0,
|
|
read_model_abandoned_at = null, read_model_error = null
|
|
where id = $1 and read_model_owed_at is null`, bookID, now)
|
|
if err != nil {
|
|
return false, fmt.Errorf("pgstore: re-arm the reading-surface debt: %w", err)
|
|
}
|
|
return tag.RowsAffected() > 0, nil
|
|
}
|
|
|
|
// AbandonedSurface is a book whose reading surface the platform gave up on, as an operator sees it.
|
|
type AbandonedSurface struct {
|
|
ID string
|
|
Title string
|
|
Workdir string
|
|
Attempts int
|
|
AbandonedAt time.Time
|
|
LastError string
|
|
}
|
|
|
|
// AbandonedSurfaces lists them. The gauge says how many; this says which, and what they said.
|
|
func (s *Store) AbandonedSurfaces(ctx context.Context) ([]AbandonedSurface, error) {
|
|
rows, err := s.pool.Query(ctx, `
|
|
select id, title, workdir, read_model_attempts, read_model_abandoned_at,
|
|
coalesce(read_model_error, '')
|
|
from books where read_model_abandoned_at is not null
|
|
order by read_model_abandoned_at`)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("pgstore: list abandoned reading surfaces: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []AbandonedSurface
|
|
for rows.Next() {
|
|
var b AbandonedSurface
|
|
if err := rows.Scan(&b.ID, &b.Title, &b.Workdir, &b.Attempts, &b.AbandonedAt, &b.LastError); err != nil {
|
|
return nil, fmt.Errorf("pgstore: scan abandoned reading surface: %w", err)
|
|
}
|
|
out = append(out, b)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// ClearReadModelDebt discharges the debt stamped at one boundary, and only that one.
|
|
//
|
|
// The equality is the whole method: materializing a large book is minutes of engine time, and a run
|
|
// can finish inside that window. Clearing unconditionally would answer the NEW boundary with a read
|
|
// taken before it, and the text of that run would never reach its reader.
|
|
func (s *Store) ClearReadModelDebt(ctx context.Context, bookID string, owedAt time.Time) error {
|
|
if _, err := s.pool.Exec(ctx, `
|
|
update books set read_model_owed_at = null, read_model_attempts = 0, read_model_error = null
|
|
where id = $1 and read_model_owed_at = $2`,
|
|
bookID, owedAt); err != nil {
|
|
return fmt.Errorf("pgstore: clear the reading-surface debt: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// StuckIntake lists books that stopped moving through intake.
|
|
//
|
|
// Both halves have the same cause and different cures: an `uploading` row whose request is gone is
|
|
// litter to remove, a `parsing` row whose claim went stale is work to retry. Neither is reachable
|
|
// from the request that created it — the process holding it is the one that died — so the sweep is
|
|
// the only thing that can finish the walk.
|
|
func (s *Store) StuckIntake(ctx context.Context, uploadingBefore, claimedBefore time.Time) ([]IntakeBook, error) {
|
|
// coalesce(claimed, added): an unclaimed parse is stuck when the BOOK has been waiting, a claimed
|
|
// one when the CLAIM has. Written as one expression because they are the same question — how long
|
|
// has this book had nobody working on it — and a NULL claim is simply the first answer to it.
|
|
const q = `
|
|
select id, status, workdir, parse_attempts
|
|
from books
|
|
where (status = 'uploading' and added_at < $1)
|
|
or (status = 'parsing' and coalesce(parse_started_at, added_at) < $2)
|
|
order by added_at`
|
|
rows, err := s.pool.Query(ctx, q, uploadingBefore, claimedBefore)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("pgstore: list stuck intake: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out []IntakeBook
|
|
for rows.Next() {
|
|
var b IntakeBook
|
|
if err := rows.Scan(&b.ID, &b.Status, &b.Workdir, &b.Attempts); err != nil {
|
|
return nil, fmt.Errorf("pgstore: scan stuck intake: %w", err)
|
|
}
|
|
out = append(out, b)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
// Progress is how far a run's WHOLE work has got, as one monotonic fraction (owner's word of 20.08,
|
|
// row 200): `done` counts chapter-passes through both waves and `total` is what the run bought,
|
|
// doubled where the pipeline has an editor. The counting itself is pgstore.runDone/runTotal.
|
|
//
|
|
// ONE counter and not two per wave, and in chapter-passes and not units, and both halves are the
|
|
// same decision: the numerator and the denominator have to be in one unit, the denominator derives
|
|
// from what the run BOUGHT, and what a run buys is declared in chapters (`ceiling_chapters`). The
|
|
// per-phase split stays INSIDE the platform — the wire carried it until 0.3.0, which pinned the
|
|
// engine's architecture to a React component through six files. What crosses the wire beside the
|
|
// numbers is `stage`, the caption's machine value — the client draws its own phrase from it.
|
|
type Progress struct {
|
|
Done int
|
|
Total int
|
|
// Stage is what the run is doing now: `drafting` or `editing` today, an open vocabulary by
|
|
// design (pgstore.runStage).
|
|
Stage string
|
|
// ETASeconds is absent when there is nothing to estimate from.
|
|
ETASeconds *int
|
|
}
|
|
|
|
// Run is the run row, in contract terms.
|
|
type Run struct {
|
|
ID string
|
|
BookID string
|
|
Revision int64
|
|
Status string
|
|
// VerifyBank is the engine flag's name for what the product calls "stop for signing". The column
|
|
// keeps the flag's name because that is what it reaches the engine as; the wire carries the
|
|
// product's.
|
|
VerifyBank bool
|
|
CeilingChapters int
|
|
Progress Progress
|
|
PausedReason string
|
|
// FailureReason is the contract's RunFailureReason, and it is the one state that IS an error
|
|
// carrying why. Written when the run is closed, from what the engine's exit said.
|
|
FailureReason string
|
|
// StopRequested is the user's own click, and it answers what no status can: stopping is not
|
|
// instant, so a stop asked for mid-translation can meet the run reaching a stop of its OWN — the
|
|
// bank signature — and the run then ends in `awaiting_bank`, a status that offers to continue on
|
|
// a click that meant "stop" (canon §Run.stop_requested). Cleared by a resume, with the intent it
|
|
// records.
|
|
StopRequested bool
|
|
StartedAt time.Time
|
|
FinishedAt *time.Time
|
|
}
|
|
|
|
// runRow is the contract's Run, in the ONE shape every path that hands one out reads it — `r` the
|
|
// run and `b` its book. Written once because the three that used it wrote the same fourteen columns
|
|
// three times, in three orders that had to stay in step by hand.
|
|
const runRow = `r.id, b.id, b.revision, r.status, r.verify_bank, r.stop_requested_at is not null,
|
|
r.ceiling_chapters, ` + runDone + `, ` + runTotal + `, ` + runStage + `, r.eta_seconds,
|
|
coalesce(r.paused_reason, ''), coalesce(r.failure_reason, ''), r.started_at, r.finished_at`
|
|
|
|
// scanRun reads runRow into a Run, plus whatever the caller selected after it.
|
|
func scanRun(row pgx.Row, out *Run, extra ...any) error {
|
|
return row.Scan(append([]any{&out.ID, &out.BookID, &out.Revision, &out.Status, &out.VerifyBank,
|
|
&out.StopRequested, &out.CeilingChapters, &out.Progress.Done, &out.Progress.Total,
|
|
&out.Progress.Stage, &out.Progress.ETASeconds, &out.PausedReason, &out.FailureReason,
|
|
&out.StartedAt, &out.FinishedAt}, extra...)...)
|
|
}
|
|
|
|
// Library is one page of a user's books plus the revision of the library scope itself.
|
|
type Library struct {
|
|
Revision int64
|
|
Books []Book
|
|
NextCursor string
|
|
}
|
|
|
|
// ListBooks returns one page of the library, newest first.
|
|
//
|
|
// Keyset, not offset: the library is written to while it is read (a run bumps its book's revision on
|
|
// every materialized transaction), and an offset page silently skips or repeats rows under that.
|
|
//
|
|
// ⚠ The page and the SCOPE's revision are read in ONE transaction, and that is the contract rather
|
|
// than tidiness. The revision travels with the page as "what this page is a picture of", and a
|
|
// client MUST DROP a read whose revision is not above one it has applied (§Revision). Read as a
|
|
// second statement, the number was NEWER than the rows it labelled — anything materialized between
|
|
// the two got a revision the client would then treat as already seen, and every frame in that window
|
|
// was lost for good. Register row PD-163, closed before the stream that would have made it visible.
|
|
func (s *Store) ListBooks(ctx context.Context, userID string, limit int, cursor string) (Library, error) {
|
|
limit = clampPage(limit)
|
|
after, err := decodeCursor(userID, cursor)
|
|
if err != nil {
|
|
return Library{}, err
|
|
}
|
|
var out Library
|
|
err = s.inReadTx(ctx, func(tx pgx.Tx) error {
|
|
var err error
|
|
out, err = listBooksTx(ctx, tx, userID, limit, after)
|
|
return err
|
|
})
|
|
if err != nil {
|
|
return Library{}, err
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func listBooksTx(ctx context.Context, tx pgx.Tx, userID string, limit int, after cursorPos) (Library, error) {
|
|
// The library row carries no progress bar: the bar belongs to the RUN, and the book's own
|
|
// figure is `chapters_done` against `chapter_count` (canon §Book).
|
|
const q = `
|
|
select ` + bookColumns + `
|
|
from books b ` + lastRun + `
|
|
where b.owner_id = $1
|
|
and ($2::timestamptz is null or (b.added_at, b.id) < ($2::timestamptz, $3::text))
|
|
order by b.added_at desc, b.id desc
|
|
limit $4`
|
|
rows, err := tx.Query(ctx, q, userID, after.time, after.id, limit+1)
|
|
if err != nil {
|
|
return Library{}, fmt.Errorf("pgstore: list books: %w", err)
|
|
}
|
|
defer rows.Close()
|
|
var out Library
|
|
for rows.Next() {
|
|
b, err := scanBook(rows)
|
|
if err != nil {
|
|
return Library{}, err
|
|
}
|
|
out.Books = append(out.Books, b)
|
|
}
|
|
if err := rows.Err(); err != nil {
|
|
return Library{}, fmt.Errorf("pgstore: list books: %w", err)
|
|
}
|
|
// One row was asked for beyond the page so that "is there a next page" is answered by data
|
|
// rather than by "the page came back full", which is wrong exactly when the last page is full.
|
|
if len(out.Books) > limit {
|
|
last := out.Books[limit-1]
|
|
out.Books = out.Books[:limit]
|
|
out.NextCursor = encodeCursor(userID, last.AddedAt, last.ID)
|
|
}
|
|
// The revision of the LIBRARY scope — membership and statuses — and never compared with the
|
|
// revision of a book (contract §Revision).
|
|
//
|
|
// The maximum over the account's books, FLOORED by the account's own counter. The floor is what
|
|
// makes the number monotonic across a deletion: the maximum alone falls when the newest book goes
|
|
// (DeleteUpload raises the floor as it deletes), and a library revision that goes backwards is a
|
|
// library the client stops rendering.
|
|
if err := tx.QueryRow(ctx, `
|
|
select greatest(coalesce((select max(revision) from books where owner_id = $1), 0),
|
|
coalesce((select library_revision from users where id = $1), 0))`,
|
|
userID).Scan(&out.Revision); err != nil {
|
|
return Library{}, fmt.Errorf("pgstore: library revision: %w", err)
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// DefaultPage is how many rows a collection returns when `limit` is omitted, and maxPage is the most
|
|
// a client may ASK for. Exported because `GET /capabilities` answers the first one: a client's
|
|
// default and the server's have to be the same number, and the way to make them the same is for
|
|
// there to be only one.
|
|
const (
|
|
DefaultPage = 100
|
|
maxPage = 1000
|
|
)
|
|
|
|
// GetBook returns the book card: the book and its current or last run.
|
|
//
|
|
// ⚠ One TRANSACTION, for the reason ListBooks is one: the card carries a revision, a client drops
|
|
// anything not above what it applied, and the book's number was read before the run's rows. A run
|
|
// that moved in between produced a card labelled with a revision older than the figures on it, and
|
|
// the client obeying the contract discarded exactly that update (register row PD-163).
|
|
func (s *Store) GetBook(ctx context.Context, userID, bookID string) (Book, *Run, error) {
|
|
var b Book
|
|
var run *Run
|
|
err := s.inReadTx(ctx, func(tx pgx.Tx) error {
|
|
var err error
|
|
if b, err = scanBook(tx.QueryRow(ctx,
|
|
`select `+bookColumns+` from books b `+lastRun+` where b.id = $1 and b.owner_id = $2`,
|
|
bookID, userID)); err != nil {
|
|
return err
|
|
}
|
|
run, err = lastRunTx(ctx, tx, bookID)
|
|
return err
|
|
})
|
|
if err != nil {
|
|
return Book{}, nil, err
|
|
}
|
|
if run != nil {
|
|
// ONE counter per book, and this is the single place that says so for every run this store
|
|
// hands out (§Revision: "every book-scoped read … carry the same number"). Read off the run's
|
|
// own column the card lagged — a unit_done bumps the book and the chapter and not the run, so
|
|
// a client that had applied frame id=2 got revision 0 back and, obeying the contract, dropped
|
|
// the read. The stop and resume handles answer from ReadRun/RequestStop, which select the
|
|
// book's revision for the same reason.
|
|
run.Revision = b.Revision
|
|
}
|
|
return b, run, nil
|
|
}
|
|
|
|
// lastRunTx reads a book's current or last run, with the bar of its current segment.
|
|
func lastRunTx(ctx context.Context, tx pgx.Tx, bookID string) (*Run, error) {
|
|
const q = `select ` + runRow + ` from books b ` + lastRun + ` where b.id = $1 and r.id is not null`
|
|
var r Run
|
|
err := scanRun(tx.QueryRow(ctx, q, bookID), &r)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("pgstore: read run: %w", err)
|
|
}
|
|
return &r, nil
|
|
}
|
|
|
|
// BookRunContext is what starting a run needs to know about a book, read under the caller's
|
|
// ownership so a stranger's book is indistinguishable from a missing one.
|
|
type BookRunContext struct {
|
|
Workdir string
|
|
// Status is the book's own state. A run may only be started on a book whose intake finished:
|
|
// the directory of a book still `uploading` holds half a file and the directory of a `rejected`
|
|
// one may hold nothing at all.
|
|
Status string
|
|
// ChaptersLeft is what is still untranslated, and it clamps the ceiling scale. Until the engine
|
|
// persists a chapter manifest (unified backlog row 100) the platform knows the chapter COUNT
|
|
// from intake and nothing per chapter, so a book with no materialized chapters answers with its
|
|
// whole length — which is right for a book that has never run.
|
|
ChaptersLeft int
|
|
// ChapterCount is the book's whole length — what a re-pass hold prices, since a re-pass may in
|
|
// the worst case re-translate everything the correction touched, up to the book.
|
|
ChapterCount int
|
|
HasLiveRun bool
|
|
// LiveRunAwaitingBank narrows HasLiveRun for the correction door: a bank stop moves the run's
|
|
// status to `awaiting_bank` from the journal BEFORE the reconciler reads the exit marker and
|
|
// closes the row (sink TypeBankStop; reconcile.finish) — so for up to a sweep the row is live
|
|
// while the book already shows its signing screen. In that window the door must open, not
|
|
// answer run_in_flight; Start keeps refusing on HasLiveRun alone (the row IS still live, and a
|
|
// second one would break runs_one_live_per_book).
|
|
LiveRunAwaitingBank bool
|
|
// BankMoved is the re-pass door's fact (P10, errata 28.08-к): a correction landed that the
|
|
// already-translated text does not carry, and no --resnapshot run has finished `ready` since.
|
|
// A plain standing flag, deliberately NOT a timestamp comparison: the door stamps it from its
|
|
// own receipt and reconcile.finish clears it explicitly on a ready resnapshot run, so there is
|
|
// no clock pairing to skew and no failed run to retire it early (the adversarial pass's K6 and
|
|
// its clock dispute both die with the predicate). The asymmetry is chosen: a STALE standing
|
|
// fact costs one harmless --resnapshot (the engine reads the flag only when a snapshot actually
|
|
// moved), a falsely-retired one kills the next run on the guard after its hold.
|
|
BankMoved bool
|
|
// HasTree is whether the book's chapter tree has been MATERIALISED — the `chapters` rows the whole
|
|
// read model counts over — as opposed to merely declared by `chapter_count`.
|
|
//
|
|
// The two are separate facts with separate writers, and the window between them is ordinary: the
|
|
// intake commits `not_started` in one transaction and materialises the tree afterwards, outside
|
|
// it. On the healthy path the window is milliseconds; when the materialisation FAILED or was
|
|
// deferred it is unbounded, and a run admitted inside it is a run whose bar can never move — every
|
|
// counter the screen shows is a count over `chapters`, and there are no rows to count (PD-405).
|
|
// Worse, the tree's debt is frozen for the run's whole life, because the sweep that would pay it
|
|
// skips a book with a live run. So the run is refused at admission instead, before its hold.
|
|
HasTree bool
|
|
}
|
|
|
|
// ReadBookForRun gathers the facts a run start is judged on.
|
|
func (s *Store) ReadBookForRun(ctx context.Context, userID, bookID string) (BookRunContext, error) {
|
|
const q = `
|
|
select b.workdir, b.status,
|
|
b.chapter_count - (select count(*) from chapters c
|
|
where c.book_id = b.id and c.units_total > 0
|
|
and ` + finishedUnits + ` >= c.units_total),
|
|
b.chapter_count,
|
|
exists (select 1 from runs lr where lr.book_id = b.id and lr.finished_at is null),
|
|
exists (select 1 from runs lr where lr.book_id = b.id and lr.finished_at is null
|
|
and lr.status = 'awaiting_bank'),
|
|
b.bank_moved_at is not null,
|
|
exists (select 1 from chapters c where c.book_id = b.id)
|
|
from books b ` + lastRun + ` where b.id = $1 and b.owner_id = $2`
|
|
var out BookRunContext
|
|
err := s.pool.QueryRow(ctx, q, bookID, userID).Scan(&out.Workdir, &out.Status, &out.ChaptersLeft,
|
|
&out.ChapterCount, &out.HasLiveRun, &out.LiveRunAwaitingBank, &out.BankMoved, &out.HasTree)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return BookRunContext{}, ErrNoBook
|
|
}
|
|
if err != nil {
|
|
return BookRunContext{}, fmt.Errorf("pgstore: read book for run: %w", err)
|
|
}
|
|
if out.ChaptersLeft < 0 {
|
|
out.ChaptersLeft = 0
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// RecordBankMove is the correction door's write: the bank moved. Stamped from the door's OWN
|
|
// receipt — never from the engine's status projection, which is blind to a correction until the
|
|
// next translate folds the bank into memory (P10 errata 28.08-к, the adversarial pass's K1). A
|
|
// plain flag: what re-passing would COST is unknowable at this moment by construction, so nothing
|
|
// pretends to price it.
|
|
func (s *Store) RecordBankMove(ctx context.Context, bookID string) error {
|
|
tag, err := s.pool.Exec(ctx, `update books set bank_moved_at = now() where id = $1`, bookID)
|
|
if err != nil {
|
|
return fmt.Errorf("pgstore: record bank move: %w", err)
|
|
}
|
|
if tag.RowsAffected() == 0 {
|
|
return ErrNoBook
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ClearBankMove retires the fact — called by the reconciler when a run that CARRIED --resnapshot
|
|
// finished `ready`: the correction has demonstrably reached the translated text. Deliberately not
|
|
// keyed on time and not called for failed/stopped/paused endings: a partial re-pin leaves
|
|
// superseded units behind, and the next admission must still carry the flags.
|
|
func (s *Store) ClearBankMove(ctx context.Context, bookID string) error {
|
|
if _, err := s.pool.Exec(ctx,
|
|
`update books set bank_moved_at = null where id = $1`, bookID); err != nil {
|
|
return fmt.Errorf("pgstore: clear bank move: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Usage is the credit state, in the only form that crosses the boundary: a share, never a sum
|
|
// (D39.84, contract §Usage).
|
|
type Usage struct {
|
|
// RemainingPercent is the balance as a share of everything ever granted to the account. Grants,
|
|
// not "a limit of a period": there are no periods (contract §/usage).
|
|
RemainingPercent int
|
|
// Spendable is whether there is anything left AT ALL. Separate from the percentage because the
|
|
// percentage is floored: $9 of a $1000 grant rounds to 0%, and "0%" is not the same fact as
|
|
// "nothing can be started".
|
|
Spendable bool
|
|
// PausedReason is set when the ACCOUNT is in a halted state.
|
|
PausedReason string
|
|
}
|
|
|
|
// ReadUsage computes the credit state.
|
|
func (s *Store) ReadUsage(ctx context.Context, userID string) (Usage, error) {
|
|
// ⚠ The halt is a state of the ACCOUNT and it is read off the ACCOUNT. It used to be read off the
|
|
// paused reason of each book's latest run, and the canon warns about exactly that at
|
|
// §AccountHaltReason: a run stops for reasons that say nothing about the account. Measured on the
|
|
// wire — $10 on the account, a ten-chapter run that spent the $0.30 IT bought — the answer was
|
|
// `{"state":"ok","remaining_percent":97,"halt_reason":"credit_exhausted"}`: a user with money
|
|
// told they have none, beside a percentage saying the opposite.
|
|
// ⚠ The denominator is everything ever ADDED to the account, not the `grant` rows alone. On the
|
|
// beta the signup grant is zero and an operator tops an account up with `adjust`, which writes a
|
|
// different kind — so counting grants only answered `remaining_percent: 0` to an account holding
|
|
// $20 that could start runs. A NEGATIVE adjustment is a correction and belongs on the other side,
|
|
// like a spend.
|
|
const q = `
|
|
select coalesce((select balance_micro_usd from account_balances where user_id = $1), 0),
|
|
coalesce((select sum(amount_micro_usd) from credit_ledger
|
|
where user_id = $1
|
|
and (kind = 'grant' or (kind = 'adjustment' and amount_micro_usd > 0))), 0)
|
|
from users where id = $1`
|
|
var balance, granted int64
|
|
err := s.pool.QueryRow(ctx, q, userID).Scan(&balance, &granted)
|
|
if errors.Is(err, pgx.ErrNoRows) {
|
|
return Usage{}, ErrNoAccount
|
|
}
|
|
if err != nil {
|
|
return Usage{}, fmt.Errorf("pgstore: read usage: %w", err)
|
|
}
|
|
u := Usage{Spendable: balance > 0}
|
|
switch {
|
|
case granted <= 0:
|
|
// Nothing was ever granted. Zero out of zero is not 100% available; an account that cannot
|
|
// spend anything is exhausted, and saying so is what makes the screen actionable.
|
|
u.RemainingPercent = 0
|
|
case balance <= 0:
|
|
u.RemainingPercent = 0
|
|
case balance >= granted:
|
|
// A correction can push the balance above the sum of grants. The share is still capped: the
|
|
// contract's field is 0..100 and a client that got 137 would have to invent a meaning.
|
|
u.RemainingPercent = 100
|
|
default:
|
|
// Rounded DOWN so "1%" never means "already nothing": the figure is what is LEFT.
|
|
u.RemainingPercent = int(balance * 100 / granted)
|
|
}
|
|
if !u.Spendable {
|
|
// An account with nothing left to spend IS halted, and that is the one account-level reason
|
|
// this contract version has. It travels beside `state` rather than being derived from it by
|
|
// the client: the two vocabularies are separate, so a later reason lands here without touching
|
|
// what `state` means.
|
|
u.PausedReason = PausedCreditExhausted
|
|
}
|
|
return u, nil
|
|
}
|
|
|
|
// The reasons a run can be paused. `credit_exhausted` is the contract's only PausedReason value;
|
|
// `daily_ceiling` is INTERNAL and is deliberately not projected — see CeilingPause and migration
|
|
// 00015.
|
|
const (
|
|
PausedCreditExhausted = ingest.PausedCreditExhausted
|
|
PausedDailyCeiling = ingest.PausedDailyCeiling
|
|
// PausedCeilingUnknown is a ceiling halt with nothing to say WHOSE ceiling it was: an exit code
|
|
// with no event behind it — a journal that was never written, or a projection under quarantine,
|
|
// which never drains at all — or an event whose scope this build cannot place. It is resumable
|
|
// like the platform's own (it may BE the platform's own), and it deliberately is NOT
|
|
// `credit_exhausted`: that value lights the account-level halted flag, and an account whose run
|
|
// hit somebody else's limit has money on it.
|
|
PausedCeilingUnknown = ingest.PausedCeilingUnknown
|
|
)
|
|
|
|
// CeilingPause turns the stream's ceiling SCOPE into the reason a run is paused with.
|
|
//
|
|
// `day` is the engine's own daily limit out of a book.yaml the operator owns (`ceilings.day_usd`):
|
|
// the platform never chose it, cannot read it, and a run it stops has an account with money still
|
|
// on it. That is why it gets a reason of its own rather than `credit_exhausted` — which would tell
|
|
// the user their credit ran out and would light the ACCOUNT-level halted flag, which keys on
|
|
// exactly that value (ReadUsage).
|
|
//
|
|
// `book` is the platform's own, which it sets flush against the hold it took (PD-158). ANYTHING
|
|
// ELSE — an absent scope, a scope a later minor invents — is neither, and says so: this pack gave
|
|
// that state a name for the exit-code path, and the same reasoning applies to an event that names a
|
|
// ceiling this build cannot place. Defaulting it to `credit_exhausted` was the strictly worse of the
|
|
// three available answers: it is resumable either way, so nothing is bought
|
|
// by the guess, and being wrong tells an account with money on it that its credit ran out.
|
|
func CeilingPause(scope string) string {
|
|
switch scope {
|
|
case ingest.ScopeDay:
|
|
return PausedDailyCeiling
|
|
case ingest.ScopeBook:
|
|
return PausedCreditExhausted
|
|
default:
|
|
return PausedCeilingUnknown
|
|
}
|
|
}
|
|
|
|
// validPauseReason is the closed vocabulary the DDL also checks (migration 00015). Both, and not
|
|
// one: this value is produced by our own materializer, so an unknown one is a defect rather than
|
|
// data, and the constraint is what makes that true of a path this function never sees.
|
|
func validPauseReason(reason string) bool {
|
|
switch reason {
|
|
case PausedCreditExhausted, PausedDailyCeiling, PausedCeilingUnknown:
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// cursorPos is the keyset position a cursor encodes.
|
|
type cursorPos struct {
|
|
time *time.Time
|
|
id *string
|
|
}
|
|
|
|
// encodeCursor renders an opaque cursor. Opaque means the client MUST NOT parse it, not that it is
|
|
// secret: it carries nothing the same client cannot already see in the row it came from.
|
|
//
|
|
// It is BOUND to the collection it came from — the owner whose library produced it. Rejecting a
|
|
// cursor that does not apply is the SERVER's duty (contract §NextCursor) and the client cannot
|
|
// perform it, because the token is opaque to it by construction; without the binding, a page token
|
|
// handed to the wrong library was silently accepted and answered with someone else's window.
|
|
func encodeCursor(owner string, t time.Time, id string) string {
|
|
return base64.RawURLEncoding.EncodeToString(
|
|
[]byte(scopeTag(owner) + "\x00" + t.UTC().Format(time.RFC3339Nano) + "\x00" + id))
|
|
}
|
|
|
|
// scopeTag identifies the collection without carrying its identifier: the cursor travels to the
|
|
// client, and an account id in it would be one more place the id exists.
|
|
func scopeTag(owner string) string {
|
|
sum := sha256.Sum256([]byte("library\x00" + owner))
|
|
return base64.RawURLEncoding.EncodeToString(sum[:8])
|
|
}
|
|
|
|
func decodeCursor(owner, s string) (cursorPos, error) {
|
|
if s == "" {
|
|
return cursorPos{}, nil
|
|
}
|
|
raw, err := base64.RawURLEncoding.DecodeString(s)
|
|
if err != nil {
|
|
return cursorPos{}, ErrBadCursor
|
|
}
|
|
scope, rest, ok := strings.Cut(string(raw), "\x00")
|
|
if !ok || scope != scopeTag(owner) {
|
|
return cursorPos{}, ErrBadCursor
|
|
}
|
|
stamp, id, ok := strings.Cut(rest, "\x00")
|
|
if !ok || id == "" {
|
|
return cursorPos{}, ErrBadCursor
|
|
}
|
|
t, err := time.Parse(time.RFC3339Nano, stamp)
|
|
if err != nil {
|
|
return cursorPos{}, ErrBadCursor
|
|
}
|
|
return cursorPos{time: &t, id: &id}, nil
|
|
}
|