271 lines
11 KiB
Go
271 lines
11 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"path/filepath"
|
|
"time"
|
|
|
|
"textmachine/platform/internal/books"
|
|
"textmachine/platform/internal/config"
|
|
"textmachine/platform/internal/httpapi"
|
|
"textmachine/platform/internal/jobs"
|
|
"textmachine/platform/internal/metrics"
|
|
"textmachine/platform/internal/money"
|
|
"textmachine/platform/internal/pgstore"
|
|
"textmachine/platform/internal/pricing"
|
|
"textmachine/platform/internal/runner"
|
|
"textmachine/platform/internal/runs"
|
|
)
|
|
|
|
// startRunner wires the run lifecycle and the book intake, and returns the function that stops them.
|
|
//
|
|
// The reads are mounted whether or not runs can be spawned: a library and a book card are not run
|
|
// machinery, and an instance without an engine binary is a perfectly useful read replica. What an
|
|
// unconfigured runner refuses is starting a run, and it says so at boot rather than at the first
|
|
// click.
|
|
func startRunner(ctx context.Context, cfg config.Config, db *pgstore.Store, log *slog.Logger,
|
|
m *metrics.Metrics, deps *httpapi.Deps) (func(), error) {
|
|
deps.Library = db
|
|
if !cfg.RunsEnabled() {
|
|
log.Warn("no TM_PLATFORM_ENGINE_BIN: the library is served read-only, no run can be started and no book can be uploaded")
|
|
return func() {}, nil
|
|
}
|
|
model, err := pricing.New(money.MicroUSD(cfg.Runner.PerChapterMicroUSD))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ceiling, err := runner.ParseCeilingTemplate(cfg.Runner.CeilingArg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(ceiling) == 0 {
|
|
// Loud at boot, refused at the handle. The alternative — starting runs anyway — spends the
|
|
// account's money under the BOOK's ceiling instead of the one the user chose (row 145).
|
|
log.Warn("TM_PLATFORM_ENGINE_CEILING_ARG is empty: starting a run will be refused rather than run under the book's own ceiling (row 145)")
|
|
}
|
|
marker, err := markerArgv(cfg.Runner.MarkerBinary)
|
|
if err != nil {
|
|
// Reads keep working. Losing the exit-marker command means a finished run could not be
|
|
// recorded and its hold would stay reserved, so runs are refused — but the library is not
|
|
// run machinery and taking it down with them turns a binary upgrade into a full outage.
|
|
log.Warn("the exit-marker command is not usable; starting a run will be refused", "err", err)
|
|
}
|
|
rn := runner.New(log)
|
|
if err := rn.Available(ctx); err != nil {
|
|
// Not fatal: reads still work, and an operator who has not enabled lingering yet gets a line
|
|
// that names the cause instead of a run that never starts.
|
|
log.Warn("transient units are not available; runs cannot be spawned on this host", "err", err)
|
|
}
|
|
svc := &runs.Service{
|
|
Store: db,
|
|
Runner: rn,
|
|
Engine: rn,
|
|
Pricing: model,
|
|
Cfg: runs.Config{
|
|
StateDir: cfg.Runner.StateDir,
|
|
EngineBinary: cfg.Runner.EngineBinary,
|
|
MarkerArgv: marker,
|
|
Ceiling: ceiling,
|
|
MemoryMax: cfg.Runner.MemoryMax,
|
|
TasksMax: cfg.Runner.TasksMax,
|
|
AllowEngineVersionChange: cfg.Runner.AllowEngineVersionChange,
|
|
ResyncEvery: cfg.Runner.ResyncEvery,
|
|
},
|
|
Log: log,
|
|
}
|
|
intake, err := startIntake(cfg, db, rn, log, deps)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// A typed nil is not nil once it is inside an interface, and the queue decides which workers to
|
|
// register by exactly that check. Spelled out so an instance without intake registers no parse
|
|
// worker rather than one that calls a nil service.
|
|
var parser jobs.Parser
|
|
if intake != nil {
|
|
parser = intake
|
|
}
|
|
queue, err := jobs.New(db.Pool(), svc, parser, log, cfg.Runner.Workers)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
svc.Queue = queue
|
|
deps.Runs = svc
|
|
if intake != nil {
|
|
intake.Queue = queue
|
|
}
|
|
if err := queue.Start(ctx); err != nil {
|
|
return nil, fmt.Errorf("start queue: %w", err)
|
|
}
|
|
go sweep(ctx, sweeps{runs: svc, books: intake, db: db, metrics: m},
|
|
cfg.Runner.SweepEvery, log)
|
|
return func() {
|
|
// The queue is drained; the RUNS are not touched. They are transient units, not children,
|
|
// and outliving this process is the whole point of the seam (D39.106).
|
|
stop, cancel := context.WithTimeout(context.WithoutCancel(ctx), 10*time.Second)
|
|
defer cancel()
|
|
if err := queue.Stop(stop); err != nil {
|
|
log.Warn("queue did not stop cleanly", "err", err)
|
|
}
|
|
}, nil
|
|
}
|
|
|
|
// startIntake wires the book upload, or explains at boot why this deployment takes none.
|
|
//
|
|
// A nil service leaves POST /books unmounted, which is the same shape every unbuilt contract route
|
|
// has: an instance that accepted a file it had nowhere to put would take the whole upload before it
|
|
// could say so.
|
|
func startIntake(cfg config.Config, db *pgstore.Store, engine books.Manifester, log *slog.Logger,
|
|
deps *httpapi.Deps) (*books.Service, error) {
|
|
if !cfg.IntakeEnabled() {
|
|
log.Warn("no TM_PLATFORM_BOOKS_DIR: this instance accepts no book uploads")
|
|
return nil, nil
|
|
}
|
|
// Created at boot, where a permission problem is an operator's to see, rather than on the first
|
|
// upload, where it would be one user's mysterious failure.
|
|
if err := os.MkdirAll(cfg.Intake.BooksDir, 0o750); err != nil {
|
|
return nil, fmt.Errorf("books directory %s: %w", cfg.Intake.BooksDir, err)
|
|
}
|
|
svc := &books.Service{
|
|
Store: db,
|
|
Engine: engine,
|
|
Cfg: books.Config{
|
|
BooksDir: cfg.Intake.BooksDir,
|
|
EngineBinary: cfg.Runner.EngineBinary,
|
|
BookTemplate: cfg.Intake.BookTemplate,
|
|
},
|
|
Log: log,
|
|
}
|
|
deps.Intake = svc
|
|
deps.Upload = httpapi.UploadLimits{
|
|
MaxBytes: cfg.Intake.MaxUploadBytes,
|
|
Deadline: cfg.Intake.UploadDeadline,
|
|
}
|
|
return svc, nil
|
|
}
|
|
|
|
// sweepBudget is what ONE pass of the RUN sweep may take. Per pass rather than per tick: see the
|
|
// note inside sweep().
|
|
const sweepBudget = 2 * time.Minute
|
|
|
|
// intakeSweepBudget is the same for the intake, and it is larger because the work inside it is:
|
|
// re-driving one book's parse is an engine call the queue itself bounds at jobs.JobTimeout, and a
|
|
// pass shorter than that turns "this book is large" into "this host cannot run the engine".
|
|
const intakeSweepBudget = jobs.JobTimeout + time.Minute
|
|
|
|
// sweeps is everything one tick of the reconciler covers.
|
|
type sweeps struct {
|
|
runs *runs.Service
|
|
books *books.Service
|
|
db *pgstore.Store
|
|
metrics *metrics.Metrics
|
|
}
|
|
|
|
// sweep runs the reconcilers at boot and then on a ticker.
|
|
//
|
|
// The boot pass is not a special case and is not skipped: after a reboot every transient unit is
|
|
// gone, and this is what notices and restarts the runs they were carrying (unified backlog row 138).
|
|
// The intake's own pass rides the same ticker — it answers the same question about a different
|
|
// object, "whose walk stopped and nobody is coming back for it".
|
|
func sweep(ctx context.Context, s sweeps, every time.Duration, log *slog.Logger) {
|
|
// EACH pass gets its OWN budget, and that is not tidiness: sharing one deadline let the runs pass
|
|
// spend all of it — two runs whose engine hangs are two minutes — and hand an already-expired
|
|
// context to the intake sweep and to the telemetry, every tick, for as long as those runs sat
|
|
// there. The starvation the per-run budget bounds inside one pass would simply have moved up a
|
|
// level, and the metric that shows it would have stopped updating with it.
|
|
pass := func(name string, budget time.Duration, fn func(context.Context) error) {
|
|
c, cancel := context.WithTimeout(ctx, budget)
|
|
defer cancel()
|
|
start := time.Now()
|
|
err := fn(c)
|
|
// A pass that ran out of its budget left work untouched, and the list is ordered the same way
|
|
// every time, so the tail starves. Counted rather than only logged: this is the number that
|
|
// says whether it is happening (register row PD-169).
|
|
s.metrics.ObserveSweep(name, time.Since(start), errors.Is(err, context.DeadlineExceeded))
|
|
if err != nil && !errors.Is(err, context.Canceled) {
|
|
log.Error(name+" sweep failed", "err", err)
|
|
}
|
|
}
|
|
one := func() {
|
|
pass("runs", sweepBudget, s.runs.Sweep)
|
|
if s.books != nil {
|
|
// The intake's pass is allowed MORE than the runs' pass, and the number is not a taste: one
|
|
// book's parse is the same engine call the queue gives `jobs.JobTimeout`, and a pass that
|
|
// cannot contain one kills it — which the intake then counts as a host that cannot run the
|
|
// engine and spends an attempt on. The pass has to be able to hold at least one book.
|
|
pass("intake", intakeSweepBudget, s.books.Sweep)
|
|
}
|
|
c, cancel := context.WithTimeout(ctx, sweepBudget)
|
|
defer cancel()
|
|
observe(c, s, log)
|
|
}
|
|
one()
|
|
t := time.NewTicker(every)
|
|
defer t.Stop()
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-t.C:
|
|
one()
|
|
}
|
|
}
|
|
}
|
|
|
|
// observe publishes the state of the control plane. A failure to measure never fails the sweep: it
|
|
// is one WARN and the next tick tries again.
|
|
func observe(ctx context.Context, s sweeps, log *slog.Logger) {
|
|
o, err := s.db.Observe(ctx)
|
|
if err != nil {
|
|
log.Warn("the control plane's own state could not be read", "err", err)
|
|
return
|
|
}
|
|
s.metrics.ObserveRunner(metrics.Runner{
|
|
QueueDepth: o.QueueDepth,
|
|
OldestHoldSeconds: o.OldestHoldSeconds,
|
|
QuarantinedAttempts: o.QuarantinedAttempts,
|
|
LiveRuns: o.LiveRuns,
|
|
BooksUploading: o.BooksUploading,
|
|
BooksParsing: o.BooksParsing,
|
|
})
|
|
lag, err := s.runs.Lag(ctx)
|
|
if err != nil {
|
|
log.Warn("the tailer's lag could not be read", "err", err)
|
|
return
|
|
}
|
|
s.metrics.ObserveTailerLag(lag)
|
|
}
|
|
|
|
// markerArgv is the command systemd runs when a unit ends. It defaults to the admin CLI next to the
|
|
// running daemon, because that is the one binary guaranteed to be the same build as this process.
|
|
//
|
|
// ⚠ ABSOLUTE or refused, and the refusal is here rather than at the first run because of what
|
|
// systemd does with the alternative — measured on this stand (systemd 259):
|
|
//
|
|
// Failed to start transient service unit: "./tmplatformctl" is neither a valid executable
|
|
// name nor an absolute path
|
|
//
|
|
// That is the whole unit refusing to START, so every run would take a claim, fail to spawn, give the
|
|
// claim back and try again on the next sweep — a loop whose only symptom is that nothing ever runs.
|
|
// `os.Stat` is happy with a relative path, so it cannot be what catches this (register row PD-165,
|
|
// the same class as StateDir's own boot refusal, PD-149).
|
|
func markerArgv(configured string) ([]string, error) {
|
|
bin := configured
|
|
if bin == "" {
|
|
self, err := os.Executable()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("locate tmplatformctl: %w", err)
|
|
}
|
|
bin = filepath.Join(filepath.Dir(self), "tmplatformctl")
|
|
}
|
|
if !filepath.IsAbs(bin) {
|
|
return nil, fmt.Errorf("exit-marker command %s must be an absolute path: systemd refuses any other kind in ExecStopPost", bin)
|
|
}
|
|
if _, err := os.Stat(bin); err != nil {
|
|
return nil, fmt.Errorf("exit-marker command %s: %w", bin, err)
|
|
}
|
|
return []string{bin, "exit-marker"}, nil
|
|
}
|