732 lines
35 KiB
Go
732 lines
35 KiB
Go
// Package books owns a book's INTAKE: receiving the file the user uploads, putting it where the
|
|
// engine will look for it, and turning it into a chapter tree.
|
|
//
|
|
// The shape mirrors the run lifecycle deliberately (package runs): the request only gets the book as
|
|
// far as "the bytes are here", and everything after that is a step some later sweep can finish. A
|
|
// parse is one $0 call of the engine that takes seconds on a large book, and holding a request open
|
|
// for it — or losing the book when the process doing it is restarted — are the two failures this
|
|
// split exists to avoid.
|
|
package books
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"io/fs"
|
|
"log/slog"
|
|
"os"
|
|
"path/filepath"
|
|
"regexp"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"unicode"
|
|
"unicode/utf8"
|
|
|
|
"textmachine/platform/internal/ingest"
|
|
"textmachine/platform/internal/pgstore"
|
|
)
|
|
|
|
// Manifester is the engine's $0 producer of a chapter tree: `tmctl manifest`. An interface so the
|
|
// intake's decisions can be pinned without an engine binary.
|
|
type Manifester interface {
|
|
Manifest(ctx context.Context, binary, workdir string) (ingest.Manifest, error)
|
|
}
|
|
|
|
// Reader materializes what a client reads — the chapter tree and the pairs — once the engine has cut
|
|
// the book. Nil where this deployment serves no reading surface.
|
|
//
|
|
// RefreshCut takes the manifest the caller has already read, so an intake does not pay for a second
|
|
// re-chunk of the same source to learn the same cut. A parse that never reaches it, or reaches it and
|
|
// fails, has still recorded the DEBT — the materializer's own pass is what answers that.
|
|
type Reader interface {
|
|
// Claim takes the debt this intake has just recorded, so the materializer's own sweep does not
|
|
// read the same book at the same time. `ok` false means somebody else already holds it.
|
|
Claim(ctx context.Context, b pgstore.OwedBook) (pgstore.OwedBook, bool, error)
|
|
RefreshCut(ctx context.Context, b pgstore.OwedBook, cut ingest.Manifest) error
|
|
}
|
|
|
|
// Enqueuer hands a book to the queue inside the caller's transaction.
|
|
type Enqueuer interface {
|
|
EnqueueParse(ctx context.Context, tx pgstore.Tx, bookID string) error
|
|
}
|
|
|
|
// Config is what an operator chooses about intake.
|
|
type Config struct {
|
|
// BooksDir is the root the platform creates book directories under. Absolute, and NOT the place
|
|
// an operator keeps hand-made books: everything below it is written and, on a rejected intake,
|
|
// removed by this service.
|
|
BooksDir string
|
|
// EngineBinary is the versioned tmctl path used to parse. Intake is not mounted without it: a
|
|
// deployment that cannot parse would take uploads it can only reject.
|
|
EngineBinary string
|
|
// BookTemplate is the operator's starting `book.yaml`, from which each new book's own is rendered
|
|
// once (form Б, D39.130 — see render.go). Empty means this deployment provisions books by hand,
|
|
// which is what every deployment did before the form was ratified: an unprovisioned book then
|
|
// WAITS rather than being rejected.
|
|
BookTemplate string
|
|
// MaxCuts caps how many books this deployment lets the engine cut at once, across the upload that
|
|
// cuts its own book, the queue's workers and the backstop sweep together. Zero takes
|
|
// DefaultMaxCuts. See limit.go for why the number is the only memory bound this path has.
|
|
MaxCuts int
|
|
// Pairs is what this deployment declares it can translate, from its configuration — the AVAILABLE
|
|
// half of it. EMPTY refuses every upload: "declares nothing" is not "declares this pair", and the
|
|
// boot refuses to mount an intake with an empty list at all, so this is the second half of one
|
|
// rule. Which pairs exist is DATA (a prompt pack the operator deploys), never a list in this
|
|
// package.
|
|
Pairs []Pair
|
|
}
|
|
|
|
// Service is the intake.
|
|
type Service struct {
|
|
Store *pgstore.Store
|
|
Engine Manifester
|
|
Reader Reader
|
|
Queue Enqueuer
|
|
Cfg Config
|
|
Log *slog.Logger
|
|
// Now is injectable so the graces below are testable without sleeping.
|
|
Now func() time.Time
|
|
// writeBudget shortens what a terminal write gets. Unexported and zero by default: only this
|
|
// package's own tests set it, to reach the case where the cut outlives that budget.
|
|
writeBudget time.Duration
|
|
// uploadSettle shortens the whole tail of an upload (walk). Unexported and zero by default, for
|
|
// the same reason as writeBudget: reaching the end of a three-and-a-half-minute budget is a test
|
|
// nobody would run otherwise, and the boot compares the CONSTANT, never this.
|
|
uploadSettle time.Duration
|
|
// The host's cap on concurrent engine cuts, built on first use — see limit.go.
|
|
cutsOnce sync.Once
|
|
cutSlots *cutSlots
|
|
}
|
|
|
|
func (s *Service) now() time.Time {
|
|
if s.Now != nil {
|
|
return s.Now()
|
|
}
|
|
return time.Now()
|
|
}
|
|
|
|
func (s *Service) log() *slog.Logger {
|
|
if s.Log == nil {
|
|
return slog.New(slog.DiscardHandler)
|
|
}
|
|
return s.Log
|
|
}
|
|
|
|
// ErrBadIntake is a request this route cannot make a book out of. It is the contract's 400.
|
|
var ErrBadIntake = errors.New("books: the intake form is not usable")
|
|
|
|
// ErrUnsupportedPair is a book in a direction this deployment cannot translate.
|
|
//
|
|
// It is refused AT INTAKE and that is the whole point of the value: a well-formed language code is
|
|
// not the same as a supported one, and until 0.3.0 such a book was accepted, written to disk, cut
|
|
// into chapters, walked through the money screen — and then died at the start of a run on the
|
|
// engine's own configuration check, with no reason that reached the user (companion §2.1). No money
|
|
// burned; a user's time did.
|
|
var ErrUnsupportedPair = fmt.Errorf("%w: this deployment cannot translate that pair", ErrBadIntake)
|
|
|
|
// ErrMalformedLanguage is a language code that is not one. Told apart from the rest of ErrBadIntake
|
|
// so the refusal can NAME the field: the canon asks the intake's 400 to say which part was wrong,
|
|
// and "the request could not be read" is the answer 0.2.3 gave to six different conditions.
|
|
var ErrMalformedLanguage = fmt.Errorf("%w: the languages must be codes", ErrBadIntake)
|
|
|
|
// Pair is one direction this deployment declares it can run.
|
|
type Pair struct{ Source, Target string }
|
|
|
|
// SourceName is the file name the intake writes a book's source under, extension aside. The name is
|
|
// FIXED and predictable because the engine finds the source through `source_file:` in the book's own
|
|
// config, which somebody else writes (see the provisioning seam in parse.go).
|
|
const SourceName = "source"
|
|
|
|
// MaxTitle bounds a book's display name. The library lists it, and the client-supplied name is
|
|
// otherwise the one unbounded string on that screen.
|
|
//
|
|
// Exported because the rename door enforces the same bound (httpapi.patchedTitle) and it is one
|
|
// column: two copies of the number is how the intake and the patch come to disagree about what fits.
|
|
const MaxTitle = 200
|
|
|
|
// langCode is the contract's LangCode: a code, never a name (§LangCode). Validated here as well as
|
|
// in the read model because this is where a value from a browser enters.
|
|
var langCode = regexp.MustCompile(`^[a-z]{2,3}(-[A-Za-z0-9]{2,8})*$`)
|
|
|
|
// Intake is one accepted call of POST /books.
|
|
type Intake struct {
|
|
UserID string
|
|
// Title is what the person named the book, and the EMPTY STRING is a value rather than an
|
|
// absence: it means "name it from the file" (canon §BookIntake.title).
|
|
Title string
|
|
SourceLang string
|
|
TargetLang string
|
|
// Filename is the name the client gave the part. It is used for two things and trusted for
|
|
// neither: a title for the library and the extension that tells the engine which reader to use.
|
|
Filename string
|
|
// File is the body of the upload, already bounded by the route's own limit. It is streamed to
|
|
// disk and never held in memory.
|
|
File io.Reader
|
|
}
|
|
|
|
// Accept receives one book.
|
|
//
|
|
// The order — row, then bytes — is what makes `uploading` a state anything can observe: a second tab
|
|
// listing the library while a 60 MB file is still on the wire sees the book with its languages and
|
|
// without its size. It also makes an abandoned upload FINDABLE, which is the half that matters
|
|
// operationally: the row is the only record that a directory under BooksDir belongs to anyone.
|
|
func (s *Service) Accept(ctx context.Context, in Intake) (pgstore.Book, error) {
|
|
if !langCode.MatchString(in.SourceLang) {
|
|
return pgstore.Book{}, fmt.Errorf("%w: source", ErrMalformedLanguage)
|
|
}
|
|
if !langCode.MatchString(in.TargetLang) {
|
|
return pgstore.Book{}, fmt.Errorf("%w: target", ErrMalformedLanguage)
|
|
}
|
|
if !s.canTranslate(in.SourceLang, in.TargetLang) {
|
|
return pgstore.Book{}, ErrUnsupportedPair
|
|
}
|
|
if s.Cfg.BooksDir == "" {
|
|
return pgstore.Book{}, errors.New("books: no books directory is configured")
|
|
}
|
|
id := pgstore.NewBookID()
|
|
dir := filepath.Join(s.Cfg.BooksDir, id)
|
|
if err := os.MkdirAll(dir, 0o750); err != nil {
|
|
return pgstore.Book{}, fmt.Errorf("books: create book directory: %w", err)
|
|
}
|
|
// The first upload is what marks this storage as ours — see storageIsThere. Written here and
|
|
// nowhere else, and deliberately NOT at boot: a marker the boot recreates says "the storage is
|
|
// here" about a directory the boot itself just made.
|
|
if err := s.markStorage(); err != nil {
|
|
s.removeDir(dir)
|
|
return pgstore.Book{}, err
|
|
}
|
|
// The created row is not kept: what the caller gets back is the row as it stands AFTER the file
|
|
// landed, and between the two the status has moved from `uploading` to `parsing`.
|
|
_, err := s.Store.CreateUpload(ctx, id, pgstore.NewUpload{
|
|
OwnerID: in.UserID,
|
|
Title: in.title(),
|
|
SourceLang: in.SourceLang,
|
|
TargetLang: in.TargetLang,
|
|
Workdir: dir,
|
|
Now: s.now(),
|
|
})
|
|
if err != nil {
|
|
s.removeDir(dir)
|
|
return pgstore.Book{}, err
|
|
}
|
|
// `streamRunes` and not `characters`: for an EPUB or a UTF-16 source the number is not one — the
|
|
// full account is on `counter`, and the wire says which of the two it is carrying through
|
|
// `character_count_exact`.
|
|
streamRunes, err := s.receive(filepath.Join(dir, SourceName+extensionOf(in.Filename)), in.File)
|
|
if err != nil {
|
|
// The upload did not finish, so there is nothing to parse and nothing to keep. The row is
|
|
// DELETED rather than rejected: `rejected` means the file could not be parsed (contract
|
|
// §BookStatus), and there is no delete handle in the contract for the user to clear a row an
|
|
// abandoned upload would otherwise leave in their library forever.
|
|
//
|
|
// On its own context: the ordinary cause of getting here is the client going away, and the
|
|
// request's context is already cancelled by then.
|
|
c, cancel := s.writeCtx(ctx)
|
|
defer cancel()
|
|
s.abandon(c, id, dir)
|
|
return pgstore.Book{}, err
|
|
}
|
|
// Every byte is in, so the upload's TAIL starts here and everything past this line is spent from
|
|
// one budget. It outlives the request — the ROW is what makes the book findable, and a client
|
|
// that hung up while waiting for the 201 must not cost the upload it already finished — and it is
|
|
// bounded as a whole rather than step by step, which is what keeps its end where the boot was
|
|
// promised no matter how many steps it grows (walk, step, UploadSettle).
|
|
walk, cancelWalk := s.walk(ctx)
|
|
defer cancelWalk()
|
|
start, cancelStart := s.writeCtx(walk)
|
|
defer cancelStart()
|
|
// The job is enqueued here ONLY where no cut runs in this request. Where one does, a job that
|
|
// exists while the cut is going races it for the parse claim — and both outcomes cost the book:
|
|
// the worker winning takes the fail-fast away, and the worker losing spends the job on nothing,
|
|
// leaving a released claim no job comes back for. cutNow enqueues instead, on the one path that
|
|
// needs it (ReleaseParseClaim).
|
|
//
|
|
// The price, named: a process that dies between this row and the cut leaves the book `parsing`
|
|
// with no job, and the backstop sweep takes it after ClaimGrace rather than at once.
|
|
var enqueue func(context.Context, pgstore.Tx, string) error
|
|
if !s.cutsItsOwnUploads() {
|
|
enqueue = s.enqueue
|
|
}
|
|
book, err := s.Store.StartParsing(start, id, streamRunes, enqueue)
|
|
if err != nil {
|
|
// The row is gone or unreachable, and the directory holds a file nothing points at — which is
|
|
// the one thing the row-first order exists to prevent, so it is undone here too. The ordinary
|
|
// cause is the sweep having abandoned this upload while it was still arriving.
|
|
c, cancel := s.writeCtx(walk)
|
|
defer cancel()
|
|
s.abandon(c, id, dir)
|
|
return pgstore.Book{}, err
|
|
}
|
|
// The cut runs synchronously: `tmctl manifest` needs no provider call and no key, so "there is no
|
|
// book in this file" and the book's size are both knowable before the upload is over (backlog row
|
|
// 285). Only the engine's own verdict about the SOURCE refuses; a deployment fault falls through
|
|
// to the asynchronous path this route always had — see cutNow.
|
|
if cut := s.cutNow(walk, book); cut.err != nil {
|
|
c, cancel := s.writeCtx(walk)
|
|
defer cancel()
|
|
// Nothing was accepted, so nothing is left behind: no row, no file (row 285 closes row 254).
|
|
s.discard(c, id, dir, cut.claimedAt)
|
|
return pgstore.Book{}, cut.err
|
|
}
|
|
// The cut moved the book past `parsing`, so the response carries the row as it stands now.
|
|
//
|
|
// On its OWN budget: `start` was opened before the bytes were received and the cut runs inside
|
|
// this call, so by now that context can be spent — and a re-read on a dead context answers with
|
|
// the pre-cut row, which is the one thing this line exists to avoid.
|
|
read, cancelRead := s.writeCtx(walk)
|
|
defer cancelRead()
|
|
fresh, err := s.Store.ReadBook(read, id)
|
|
if err == nil {
|
|
book = fresh
|
|
} else {
|
|
// The book is parsed and only this read failed: a stale 201 beats failing an upload that
|
|
// succeeded.
|
|
s.log().WarnContext(ctx, "the parsed book could not be re-read; the response carries the row as it stood before the cut", "err", err)
|
|
}
|
|
// No book id: an INFO line must not identify a user's library (ENGINEERING_STANDARDS
|
|
// §Наблюдаемость, the same rule that keeps raw paths out of the access log — PD-3). The request
|
|
// id the handler carries is what ties this line to the response the user got.
|
|
s.log().InfoContext(ctx, "book accepted", "source_stream_runes", streamRunes, "status", book.Status)
|
|
return book, nil
|
|
}
|
|
|
|
// cutResult is the intake cut's verdict together with the claim it was taken under, which is what
|
|
// lets the caller delete a row that is still `parsing`.
|
|
type cutResult struct {
|
|
err error
|
|
claimedAt time.Time
|
|
}
|
|
|
|
// cutNow runs the intake's own parse and returns the refusal to answer the uploader with, if any.
|
|
//
|
|
// An empty result covers both "the book is cut" and "nothing could be established": the second is not
|
|
// a refusal, and the queue finishes the book as before. A deployment fault must never be answered to
|
|
// a user as a verdict about their file.
|
|
//
|
|
// Bounded by its own budget, which keeps a large book from holding the request open: past it the
|
|
// upload is accepted `parsing`. That budget is a STEP of the upload's walk, so a cut cannot spend
|
|
// what the rest of the tail still needs (step). The claim is taken here so this and the queue cannot
|
|
// run the engine over one project directory at once (see Parse).
|
|
func (s *Service) cutNow(ctx context.Context, book pgstore.Book) cutResult {
|
|
if !s.cutsItsOwnUploads() {
|
|
return cutResult{}
|
|
}
|
|
// The claim is a WRITE and gets a write's budget, not the cut's: it is what lets this pass end the
|
|
// book at all, and tying it to the budget of the work it guards was what made «no room for a cut»
|
|
// arrive as «the store could not be asked».
|
|
cl, cancelClaim := s.writeCtx(ctx)
|
|
defer cancelClaim()
|
|
now := s.now()
|
|
claim, err := s.Store.ClaimParse(cl, book.ID, now, now.Add(-ClaimGrace))
|
|
// The two ways this does not produce a claim are told apart, because one of them needs a human
|
|
// and the other is how the walk is supposed to go. Neither is an answer about the FILE, so both
|
|
// leave the upload to be accepted `parsing`.
|
|
switch {
|
|
case errors.Is(err, pgstore.ErrParseClaimed):
|
|
// Somebody else holds this book: on this route that is the backstop sweep, since no job was
|
|
// enqueued for a book the intake cuts itself. Whoever holds it finishes the walk.
|
|
s.log().InfoContext(ctx, "the intake did not get the parse claim; the pass that holds it finishes the book")
|
|
return cutResult{}
|
|
case err != nil:
|
|
// NOT the race above: the claim could not be ASKED for. Said out loud, because the book is now
|
|
// `parsing` with no claim and no job — this route enqueues none — and nothing comes back for it
|
|
// until the backstop sweep does, a claim grace later. Silent, this is indistinguishable from
|
|
// the ordinary line above, which needs no operator at all.
|
|
s.log().ErrorContext(ctx, "the parse claim could not be taken, so this upload is not cut here; the backstop sweep finishes the book after its grace", "err", err)
|
|
return cutResult{}
|
|
}
|
|
// The cut itself, on what the walk can spare after the writes that must follow it (stepLeaving).
|
|
// Below zero there is no cut to run, and it is not ATTEMPTED: a pass started on a spent context
|
|
// fails somewhere inside itself and is diagnosed as whatever failed first, which is never the
|
|
// truth. Answered as «no verdict», which is what the queue finishing the book already means.
|
|
c, cancel := s.stepLeaving(ctx, CutBudget, s.cutTailReserve())
|
|
defer cancel()
|
|
cut := c.Err()
|
|
if cut == nil {
|
|
cut = s.parseClaimed(c, claim, true)
|
|
} else {
|
|
s.log().InfoContext(ctx, "what is left of this upload is shorter than a cut plus the writes that follow it; the queue takes the book",
|
|
"reason", ReasonNoTimeToCut)
|
|
cut = notConclusive(ReasonNoTimeToCut)
|
|
}
|
|
switch err := cut; {
|
|
case errors.Is(err, ErrBadIntake):
|
|
return cutResult{err: err, claimedAt: claim.At}
|
|
case err != nil: // errNotConclusive, or a context this pass could not finish inside
|
|
// Nothing was established, so the claim goes back at once — and the job that finishes the book
|
|
// goes in WITH it, in the one transaction: this route enqueued none at StartParsing, precisely
|
|
// so that no worker could race this cut for the claim (Accept above, and ReleaseParseClaim).
|
|
s.log().WarnContext(ctx, "the intake's own cut was not conclusive; the queue takes the book", "err", err)
|
|
w, wcancel := s.writeCtx(ctx)
|
|
defer wcancel()
|
|
if err := s.Store.ReleaseParseClaim(w, book.ID, claim.At, s.enqueue); err != nil {
|
|
s.log().ErrorContext(ctx, "the intake's parse claim could not be given back; the backstop sweep takes the book", "err", err)
|
|
}
|
|
}
|
|
return cutResult{}
|
|
}
|
|
|
|
// canTranslate judges an upload against what this deployment declared.
|
|
//
|
|
// An empty list refuses everything rather than accepting it: "declared nothing" and "declares this
|
|
// pair" are not the same answer, and the permissive reading made a deployment whose available half is
|
|
// empty accept books it can only fail later. The boot refuses that configuration outright, so this is
|
|
// the second half of one rule rather than a policy of its own.
|
|
//
|
|
// Compared case-INSENSITIVELY: a language tag is case-insensitive by BCP 47, and `zh-Hans` declared
|
|
// against `zh-hans` uploaded is the same pair.
|
|
func (s *Service) canTranslate(source, target string) bool {
|
|
for _, p := range s.Cfg.Pairs {
|
|
if strings.EqualFold(p.Source, source) && strings.EqualFold(p.Target, target) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// receive streams the upload to disk and counts the RUNES OF THE STREAM that went past — see
|
|
// `counter` for what that number is and is not.
|
|
//
|
|
// Streamed, never buffered: the route's limit is tens of megabytes and reading that into memory
|
|
// would make one upload per concurrent request the platform's memory profile. The file is created
|
|
// with O_EXCL so a book directory can never be written twice by two requests that somehow minted the
|
|
// same id.
|
|
func (s *Service) receive(path string, body io.Reader) (int64, error) {
|
|
f, err := os.OpenFile(path, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o640)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("books: create source file: %w", err)
|
|
}
|
|
c := &counter{w: f}
|
|
if _, err := io.Copy(c, body); err != nil {
|
|
f.Close()
|
|
// The error is returned as it came: the handler tells a body that outgrew the route's limit
|
|
// from a client that went away, and both are the caller's to name (http.MaxBytesError).
|
|
return 0, err
|
|
}
|
|
if err := f.Close(); err != nil {
|
|
return 0, fmt.Errorf("books: close source file: %w", err)
|
|
}
|
|
return c.runes, nil
|
|
}
|
|
|
|
// counter writes through and counts the RUNES OF THE BYTE STREAM IT IS WRITING — every byte that is
|
|
// not a UTF-8 continuation byte, which needs no state across the chunk boundaries io.Copy hands it.
|
|
//
|
|
// ⚠ NAME WHAT THIS IS, BECAUSE THE FIELD IT FEEDS IS CALLED `character_count` AND THE SCREEN CALLS IT
|
|
// «Знаков» (unified backlog row 282). What the number means depends entirely on what was uploaded,
|
|
// and only the first of these three is the count the name promises:
|
|
//
|
|
// - **A UTF-8 text source** — it IS the character count, exactly.
|
|
// - **An EPUB** — the source is a ZIP archive, and this counts the non-continuation bytes of
|
|
// COMPRESSED DATA. The figure is not a character count, not an approximation of one, and not
|
|
// even the same order of magnitude reliably: it is a property of the container. The intake
|
|
// accepts `.epub` (extensionOf below, and the engine dispatches an EPUB reader by it), so this
|
|
// is a live case and not a hypothetical.
|
|
// - **A text source in GB18030 or UTF-16** — the engine accepts both and decodes them itself; here
|
|
// the figure is an approximation, and for UTF-16 a poor one.
|
|
//
|
|
// This is the FALLBACK, not the answer: the engine's manifest carries `source_chars` (runes of the
|
|
// ingested text, container and encoding dealt with) and the read model prefers it once a cut has been
|
|
// read. `character_count_exact` on the wire says which of the two a response carries.
|
|
//
|
|
// The counting stays byte-shaped deliberately: decoding here would make this a second reader of the
|
|
// source, disagreeing with the engine's on exactly the shapes above.
|
|
//
|
|
// Pinned by TestTheIntakeCounterCountsTheWriteStreamAndNotCharacters.
|
|
type counter struct {
|
|
w io.Writer
|
|
// runes is deliberately not called `characters`: for two of the three source shapes above it is
|
|
// not one.
|
|
runes int64
|
|
}
|
|
|
|
func (c *counter) Write(p []byte) (int, error) {
|
|
n, err := c.w.Write(p)
|
|
for _, b := range p[:n] {
|
|
if b&0xC0 != 0x80 {
|
|
c.runes++
|
|
}
|
|
}
|
|
return n, err
|
|
}
|
|
|
|
// cutsItsOwnUploads reports whether this deployment answers an upload with a verdict about the file,
|
|
// or hands the book to the queue and answers `parsing`.
|
|
//
|
|
// ONE function because it decides TWO things that must never disagree: whether the row's transaction
|
|
// carries a job (Accept), and whether a cut runs at all (cutNow). Written twice, the two readings of
|
|
// one predicate drift, and each way of drifting costs the book — a job enqueued for a cut that then
|
|
// runs races it for the claim, and a cut skipped where no job was enqueued leaves the book waiting
|
|
// out the backstop sweep's whole grace.
|
|
//
|
|
// In a deployment it is always true: the boot refuses to mount an intake without an engine binary
|
|
// (config.IntakeEnabled), so the false branch belongs to the development path and to this package's
|
|
// own tests, where it stands for a deployment that has a queue and no engine.
|
|
func (s *Service) cutsItsOwnUploads() bool { return s.Engine != nil }
|
|
|
|
func (s *Service) enqueue(ctx context.Context, tx pgstore.Tx, bookID string) error {
|
|
if s.Queue == nil {
|
|
return nil // no queue configured: the sweep picks the book up on its next pass
|
|
}
|
|
return s.Queue.EnqueueParse(ctx, tx, bookID)
|
|
}
|
|
|
|
// abandon undoes an upload that did not finish. Both halves are best effort and both are logged:
|
|
// what must not happen is a silent leak of either the row or the directory.
|
|
func (s *Service) abandon(ctx context.Context, id, dir string) {
|
|
s.removeIntake(ctx, id, dir, func(c context.Context) error { return s.Store.DeleteUpload(c, id) })
|
|
}
|
|
|
|
// discard removes a book the intake's own cut refused: same halves as abandon, different stage.
|
|
func (s *Service) discard(ctx context.Context, id, dir string, claimedAt time.Time) {
|
|
s.removeIntake(ctx, id, dir, func(c context.Context) error {
|
|
return s.Store.DeleteRefusedIntake(c, id, claimedAt)
|
|
})
|
|
}
|
|
|
|
func (s *Service) removeIntake(ctx context.Context, id, dir string, deleteRow func(context.Context) error) {
|
|
// The ROW is deleted first here, and unlike the reject path that order is forced: the directory
|
|
// may only go once it is certain nobody owns it. The sweep decides from a snapshot, and a request
|
|
// that finished in the meantime has moved the book to `parsing` — DeleteUpload's status guard then
|
|
// refuses, and removing the directory anyway would delete the source of a live book under it.
|
|
//
|
|
// The crash window that leaves is a directory with no row, and it is named rather than closed:
|
|
// nothing walks BooksDir looking for orphans (register row PD-175, where the retention sweep that
|
|
// would is filed).
|
|
if err := deleteRow(ctx); err != nil {
|
|
if errors.Is(err, pgstore.ErrNoBook) {
|
|
// The ordinary race: another instance's sweep got there first, or the upload finished. Not
|
|
// an error — an ERROR line on a routine race is noise that teaches operators to skim.
|
|
s.log().InfoContext(ctx, "the abandoned upload was already gone; its directory is left alone")
|
|
return
|
|
}
|
|
s.log().ErrorContext(ctx, "an abandoned upload was not removed; its directory is left alone", "err", err)
|
|
return
|
|
}
|
|
s.removeDir(dir)
|
|
}
|
|
|
|
// removeDir deletes a directory this service created, and NOTHING else.
|
|
//
|
|
// The guard is not ceremony: a book registered by the dev CLI carries a workdir the operator chose —
|
|
// their own project directory, with their own source and their own project database — and no path
|
|
// here may ever remove one. Everything under BooksDir was created by this service and holds nothing
|
|
// the platform did not put there.
|
|
func (s *Service) removeDir(dir string) {
|
|
if !s.owns(dir) {
|
|
s.log().Error("refusing to remove a directory this service did not create", "dir", dir)
|
|
return
|
|
}
|
|
if err := os.RemoveAll(dir); err != nil {
|
|
s.log().Error("book directory could not be removed", "err", err)
|
|
}
|
|
}
|
|
|
|
// StorageMarker is the file that says "this is the storage this platform has been writing books
|
|
// into". Exported so an operator provisioning a volume by hand can put one there.
|
|
const StorageMarker = ".tmplatform-books"
|
|
|
|
const storageMarkerText = `This directory holds book sources written by tmplatformd.
|
|
Its presence is what tells the intake that the storage is mounted: without it a book whose directory
|
|
is missing is treated as "the storage is gone" and WAITS, instead of being rejected with its source
|
|
deleted. Do not remove it while books live here.
|
|
`
|
|
|
|
// markStorage writes the marker if it is not there. Idempotent and race-free by O_EXCL: two uploads
|
|
// arriving together are both correct, and the loser's EEXIST is the state it wanted.
|
|
func (s *Service) markStorage() error {
|
|
f, err := os.OpenFile(filepath.Join(s.Cfg.BooksDir, StorageMarker),
|
|
os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o640)
|
|
if errors.Is(err, fs.ErrExist) {
|
|
return nil
|
|
}
|
|
if err != nil {
|
|
return fmt.Errorf("books: mark the storage: %w", err)
|
|
}
|
|
defer f.Close()
|
|
if _, err := f.WriteString(storageMarkerText); err != nil {
|
|
return fmt.Errorf("books: mark the storage: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// WithoutItsPath strikes a book's own directory out of text that is about to be logged, and it is the
|
|
// zone's rule about book identity in logs turned from a checklist into a mechanism.
|
|
//
|
|
// ⛔ WHY A REMOVER AND NOT A CLASSIFIER. The directory of every book this platform takes in is
|
|
// `<books dir>/<book id>` (Receive), so the path IS the identifier the standard keeps out of logs
|
|
// (PD-139, PD-99). The first two attempts at that rule tried to RECOGNISE the cases where a path
|
|
// would appear — the directory is gone, the volume is not mounted — and both were short: an engine
|
|
// that fails on a FILE INSIDE a directory that is perfectly present names the path too, and so does
|
|
// this platform's own sweep when it cannot remove a copy. Each round closed one case and reported the
|
|
// class (acceptance of 11.09, F3/F4). A remover has no cases: whatever the text says, the book's own
|
|
// name is not in it afterwards.
|
|
//
|
|
// What an operator keeps: everything else the text said — the operation, the errno, the file BELOW
|
|
// the directory — and the run id, which every one of these lines already carries and which
|
|
// `tmplatformctl` resolves to the book for whoever is allowed to ask.
|
|
//
|
|
// It is deliberately dumb: a literal replacement of the directory and of its last segment, in that
|
|
// order. It cannot be defeated by a message this zone has not seen, and it does not pretend to find
|
|
// identifiers it was not given — a text naming a book some other way is out of its reach, and that
|
|
// limit is real rather than papered over.
|
|
func WithoutItsPath(workdir, text string) string {
|
|
if workdir == "" || text == "" {
|
|
return text
|
|
}
|
|
const marker = "<the book's own directory>"
|
|
out := strings.ReplaceAll(text, workdir, marker)
|
|
if id := filepath.Base(workdir); id != "" && id != "." && id != string(filepath.Separator) {
|
|
out = strings.ReplaceAll(out, id, "<the book>")
|
|
}
|
|
return out
|
|
}
|
|
|
|
// StorageIsThere reports whether the books storage under booksDir is the one this platform wrote
|
|
// into.
|
|
//
|
|
// ⚠ It asks about the MARKER and not about the directory, and the difference is the whole point. A
|
|
// volume mounted AT BooksDir leaves an empty mountpoint behind when it is unmounted, so the directory
|
|
// still exists and `Stat` still succeeds — which is how the first version of this guard (PD-192) let
|
|
// an unmount reject every book in intake, with their sources deleted, exactly as if each book's own
|
|
// directory had been removed. The boot's MkdirAll made it worse by recreating the root after a
|
|
// restart. The marker is written by the FIRST upload and by nothing else, so neither an unmount nor a
|
|
// fresh MkdirAll can forge it (re-check of the dofix, FP5-10).
|
|
//
|
|
// A host that has never taken an upload has no marker either, and answers "not there" — which is the
|
|
// safe direction: it has no books to reject.
|
|
//
|
|
// ⚠ Exported and taking the root as an argument because the RUN DOOR asks the same question of the
|
|
// same storage (runs.sourceThere): a book admitted over a vanished directory holds the account's
|
|
// credit for good, and telling that apart from an unmounted volume is this predicate, not a second
|
|
// edition of it. A copy would be the same defect twice — the first edition of THIS one cost the
|
|
// intake every book on the host.
|
|
func StorageIsThere(booksDir string) bool {
|
|
if booksDir == "" {
|
|
return false
|
|
}
|
|
_, err := os.Stat(filepath.Join(booksDir, StorageMarker))
|
|
return err == nil
|
|
}
|
|
|
|
func (s *Service) storageIsThere() bool { return StorageIsThere(s.Cfg.BooksDir) }
|
|
|
|
func (s *Service) owns(dir string) bool { return Owns(s.Cfg.BooksDir, dir) }
|
|
|
|
// Owns reports whether dir is a path this platform's intake created, i.e. one UNDER booksDir.
|
|
//
|
|
// ⚠ Exported for the run door (runs.sourceThere), and the question it answers there is which fault a
|
|
// missing directory is. The storage marker only says something about booksDir, so it may only be
|
|
// consulted about books that live under it: a book placed by hand somewhere else — `tmplatformctl
|
|
// book add --workdir` — sits on a volume this platform never wrote and has no sentinel on, and
|
|
// reading the intake's marker as evidence about THAT volume is the PD-192 mistake with a longer
|
|
// path.
|
|
func Owns(booksDir, dir string) bool {
|
|
if booksDir == "" || dir == "" {
|
|
return false
|
|
}
|
|
rel, err := filepath.Rel(booksDir, dir)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
return rel != "." && rel != ".." && !strings.HasPrefix(rel, ".."+string(filepath.Separator))
|
|
}
|
|
|
|
// title is the name the book joins the library under: the one the person typed, or — when they
|
|
// typed none — the name of the file they uploaded.
|
|
//
|
|
// The empty string is the DECLARED way to ask for the second (canon §BookIntake.title), which is why
|
|
// it is not an absence: a value present means the person named the book themselves, and no later
|
|
// parse overwrites it.
|
|
func (in Intake) title() string {
|
|
if t := strings.TrimSpace(in.Title); t != "" {
|
|
// ⚠ THE SAME CLEANING THE DERIVED BRANCH GETS, and it was missing here — the defect the
|
|
// exported MaxTitle above exists to prevent, one field over. A title the user TYPED went
|
|
// through length bounding alone, so a control character reached Postgres, which cannot hold
|
|
// U+0000 in a `text` column: measured on a live database as
|
|
// `ERROR: invalid byte sequence for encoding "UTF8": 0x00`, while U+2028 landed and the book
|
|
// lived with a line separator in its name. The rename door refuses both (httpapi, the PATCH
|
|
// handler); a deployment where one writer of a column is stricter than the other is the class
|
|
// this file's own comment calls "how the intake and the patch come to disagree".
|
|
//
|
|
// Cleaned rather than refused HERE, unlike at the rename door, and the difference is the
|
|
// contract's: `BookIntake.title` is one field of a multipart upload whose body has already
|
|
// been received, and failing the whole upload over a stray character would throw away tens of
|
|
// megabytes the user has just sent. The rename door has nothing to throw away, so it refuses
|
|
// and says which member was wrong.
|
|
if t = withoutControls(t); t == "" {
|
|
// Nothing legible was left, so this is the same as having named nothing at all.
|
|
return titleFrom(in.Filename)
|
|
}
|
|
return boundedTitle(t)
|
|
}
|
|
return titleFrom(in.Filename)
|
|
}
|
|
|
|
// withoutControls drops what may not appear in a display name: Unicode Cc (C0 and C1 — NUL, the
|
|
// newline, the escape) and the two line separators U+2028/U+2029.
|
|
//
|
|
// ⚠ AND DELIBERATELY NOT THE FORMAT CATEGORY (Cf). U+200E/U+200F and ZWJ/ZWNJ are ordinary content in
|
|
// Hebrew, Arabic, Devanagari and Persian, and dropping "everything invisible" would quietly mangle a
|
|
// legitimate title in a language pair this repository does not contain yet — the generality invariant,
|
|
// not a hypothetical. The engine's own inbound fence draws the line in the same place.
|
|
func withoutControls(s string) string {
|
|
return strings.TrimSpace(strings.Map(func(r rune) rune {
|
|
if unicode.IsControl(r) || r == '\u2028' || r == '\u2029' {
|
|
return -1
|
|
}
|
|
return r
|
|
}, s))
|
|
}
|
|
|
|
// titleFrom is the book's name derived from the file's own.
|
|
func titleFrom(filename string) string {
|
|
name := filepath.Base(filepath.FromSlash(filename))
|
|
if i := strings.LastIndexByte(name, '.'); i > 0 {
|
|
name = name[:i]
|
|
}
|
|
name = strings.Map(func(r rune) rune {
|
|
if unicode.IsControl(r) {
|
|
return -1
|
|
}
|
|
return r
|
|
}, name)
|
|
name = strings.TrimSpace(name)
|
|
if name == "." || name == ".." || name == string(filepath.Separator) {
|
|
return ""
|
|
}
|
|
return boundedTitle(name)
|
|
}
|
|
|
|
// boundedTitle is the one bound on a title, wherever it came from: the library lists it, and it is
|
|
// otherwise the one unbounded string on that screen.
|
|
func boundedTitle(name string) string {
|
|
if utf8.RuneCountInString(name) > MaxTitle {
|
|
return string([]rune(name)[:MaxTitle])
|
|
}
|
|
return name
|
|
}
|
|
|
|
// extensionOf is the one thing about the FORMAT the platform is allowed to know: the engine
|
|
// dispatches its reader by extension (`.epub` → the epub reader, anything else → plain text,
|
|
// backend/internal/chunk/ingest.go), so the extension has to survive intake or an EPUB is read as
|
|
// text.
|
|
//
|
|
// It is not a format allowlist, and that is deliberate: which formats exist is the engine's
|
|
// question, and a platform that refused an extension the engine had just learned would be a second
|
|
// place to teach. What it does refuse is a name that is not an extension — the value goes into a
|
|
// path, so anything but lowercase alphanumerics is dropped and the source becomes plain text.
|
|
func extensionOf(filename string) string {
|
|
ext := strings.ToLower(filepath.Ext(filepath.Base(filepath.FromSlash(filename))))
|
|
if len(ext) < 2 || len(ext) > 9 {
|
|
return ".txt"
|
|
}
|
|
for _, r := range ext[1:] {
|
|
if (r < 'a' || r > 'z') && (r < '0' || r > '9') {
|
|
return ".txt"
|
|
}
|
|
}
|
|
return ext
|
|
}
|