textmachine/platform/cmd/tmplatformd/runner.go

package main

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"time"

	"textmachine/platform/internal/books"
	"textmachine/platform/internal/config"
	"textmachine/platform/internal/exports"
	"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/readmodel"
	"textmachine/platform/internal/runner"
	"textmachine/platform/internal/runs"
)

// runsConfig maps the operator's environment onto the reconciler's settings.
//
// A function and not a literal inside startRunner so that the mapping has a witness: startRunner
// needs a database, a systemd bus and a queue, so nothing could observe a field that stopped being
// assigned — which is how RunBudget came to be declared, documented and left at zero.
func runsConfig(cfg config.Config, marker []string, ceiling []string) runs.Config {
	return runs.Config{
		StateDir:                 cfg.Runner.StateDir,
		EngineBinary:             cfg.Runner.EngineBinary,
		MarkerArgv:               marker,
		Ceiling:                  ceiling,
		KeysFile:                 cfg.Runner.KeysFile,
		MemoryMax:                cfg.Runner.MemoryMax,
		TasksMax:                 cfg.Runner.TasksMax,
		AllowEngineVersionChange: cfg.Runner.AllowEngineVersionChange,
		ResyncEvery:              cfg.Runner.ResyncEvery,
		RunBudget:                cfg.Runner.RunBudget,
	}
}

// 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
	// The keys survive without a runner: `Idempotency-Key` is a property of a WRITE, and the writes
	// that take one are mounted independently of whether a run can be spawned.
	deps.Keys = db
	deps.Capabilities = capabilities(cfg)
	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)")
	}
	if cfg.Runner.KeysFile == "" {
		// Runs still start: a stand may keep a `.env` beside each book. What cannot be allowed is the
		// silent version — a SaaS deployment whose engine gets no keys from anywhere fails at the first
		// provider call, hours of queueing later (row 211).
		log.Warn("no TM_PLATFORM_ENGINE_KEYS_PATH: the engine is passed no provider keys; a run finds them only in a .env beside its own book.yaml")
	}
	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)
	}
	reader := &readmodel.Service{Store: db, Engine: rn, Binary: cfg.Runner.EngineBinary, Log: log}
	svc := &runs.Service{
		Store:   db,
		Runner:  rn,
		Engine:  rn,
		Bank:    rn,
		Pricing: model,
		Cfg:     runsConfig(cfg, marker, ceiling),
		Log:     log,
	}
	// Before the door serves: a decision document still on disk here outlived its process, and
	// nothing else ever deletes by that mask (PD-409's cure).
	svc.SweepCorrectionScratch()
	intake, err := startIntake(cfg, db, rn, reader, log, deps)
	if err != nil {
		return nil, err
	}
	exporter, err := startExports(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
	}
	var builder jobs.Exporter
	if exporter != nil {
		builder = exporter
	}
	queue, err := jobs.New(db.Pool(), svc, parser, builder, log, cfg.Runner.Workers)
	if err != nil {
		return nil, err
	}
	svc.Queue = queue
	deps.Runs = svc
	// The correction door mounts with the run machinery: it needs the same engine binary, the same
	// state directory and the same per-book serialization. The capability flag says the same fact.
	deps.Bank = svc
	if intake != nil {
		intake.Queue = queue
	}
	if exporter != nil {
		exporter.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, reader: reader, exports: exporter, db: db, metrics: m},
		cfg.Runner.SweepEvery, cfg.Runner.SweepBudget, 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, reader books.Reader,
	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,
		Reader: reader,
		Cfg: books.Config{
			BooksDir:     cfg.Intake.BooksDir,
			EngineBinary: cfg.Runner.EngineBinary,
			BookTemplate: cfg.Intake.BookTemplate,
			Pairs:        intakePairs(cfg),
		},
		Log: log,
	}
	deps.Intake = svc
	deps.Upload = httpapi.UploadLimits{
		MaxBytes: cfg.Intake.MaxUploadBytes,
		Deadline: cfg.Intake.UploadDeadline,
	}
	return svc, nil
}

// startExports wires the export door, or explains at boot why this deployment builds none.
//
// A nil service leaves all three export routes unmounted, which is the same shape every unbuilt
// contract route has — and `export_formats` answers `[]` from the same fact, so a client never
// learns of a format by having a user's download fail.
func startExports(cfg config.Config, db *pgstore.Store, engine exports.Builder,
	log *slog.Logger, deps *httpapi.Deps) (*exports.Service, error) {
	if !cfg.ExportsEnabled() {
		log.Warn("no TM_PLATFORM_EXPORT_FORMATS: this instance builds no exports and `export_formats` is empty")
		return nil, nil
	}
	// Created at boot, where a permission problem is an operator's to see, rather than on the first
	// download, where it would be one user's mysterious failure. The same reasoning as the intake's
	// books directory, and the same mode: artifacts are a user's book text.
	if err := os.MkdirAll(cfg.Export.Dir, 0o750); err != nil {
		return nil, fmt.Errorf("exports directory %s: %w", cfg.Export.Dir, err)
	}
	svc := &exports.Service{Store: db, Engine: engine, Cfg: exportsConfig(cfg), Log: log}
	deps.Exports = svc
	return svc, nil
}

// exportsConfig maps the operator's environment onto the export door's settings.
//
// A function and not a literal inside startExports for the same reason runsConfig is one: the
// literal needs a database and a queue to be reached at all, so an assignment that stopped happening
// would be observable only by running the daemon — which is exactly how a documented knob came to do
// nothing for a whole pack (PD-331).
func exportsConfig(cfg config.Config) exports.Config {
	return exports.Config{
		EngineBinary: cfg.Runner.EngineBinary,
		Formats:      cfg.Export.Formats,
		Dir:          cfg.Export.Dir,
		TTL:          cfg.Export.TTL,
		StaleAfter:   exportStaleAfter,
		QueuedGrace:  exportQueuedGrace,
	}
}

// exportStaleAfter is how old a `pending` export has to be before the sweep calls its build lost.
//
// It must OUTLIVE one whole job or the sweep would give up on builds that are merely slow, and the
// margin is the sweep's own tick plus room for a queue that took its time picking the job up. Same
// shape as the intake's claim grace, and written as the job timeout plus a margin rather than as a
// number, so the two cannot drift apart.
const exportStaleAfter = jobs.JobTimeout + 5*time.Minute

// exportQueuedGrace is the same question for a build that has NOT been picked up yet, and it is four
// whole jobs rather than one.
//
// The number is a share of `jobs.JobTimeout` and not a constant, for the reason the figure exists at
// all: ONE queue serves spawns, parses and builds, a parse may take a whole job timeout, and the
// default worker count is four — so an export can legitimately sit behind four full-length jobs
// without anything being wrong with it. Judged on the started-build clock it would be buried at
// twenty minutes and its user told it was interrupted, which is false. The poll still ends, which is
// the canon's actual requirement; a backlog longer than this is an operator's problem, and the line
// the sweep writes is where they see it.
const exportQueuedGrace = 4 * jobs.JobTimeout

// capabilities is what `GET /capabilities` answers, assembled from what this deployment IS rather
// than from a document somebody keeps in step with it: the pairs an operator declared, whether an
// intake is mounted at all, the cap that intake enforces, and the page size every read defaults to.
func capabilities(cfg config.Config) httpapi.Capabilities {
	caps := httpapi.Capabilities{
		IntakeEnabled:  cfg.IntakeEnabled() && cfg.RunsEnabled(),
		IntakeMaxBytes: cfg.Intake.MaxUploadBytes,
		// What this deployment DECLARES its engine can write, and the same condition that mounts
		// the three export routes (startExports runs exactly when ExportsEnabled). The capability
		// and the mount are one fact, or a client learns the truth by failing a user's download.
		ExportFormats: exportFormats(cfg),
		// The same condition that mounts the door (startRunner sets deps.Bank exactly when runs are
		// enabled): the flag and the 404 must be one fact, or a client learns the truth by failing
		// a user's save.
		BankCorrectionsEnabled: cfg.RunsEnabled(),
		PageSizeDefault:        pgstore.DefaultPage,
	}
	for _, p := range cfg.LanguagePairs {
		caps.Pairs = append(caps.Pairs, httpapi.LanguagePair{
			Source: p.Source, Target: p.Target, Available: p.Available,
		})
	}
	return caps
}

// exportFormats is what `export_formats` answers: the declared list on a deployment that can build,
// and an empty one everywhere else. Never nil — an empty collection is an empty array on the wire.
func exportFormats(cfg config.Config) []string {
	if !cfg.ExportsEnabled() {
		return []string{}
	}
	return append([]string{}, cfg.Export.Formats...)
}

// intakePairs is the AVAILABLE half of the same list: what the intake judges an upload against.
func intakePairs(cfg config.Config) []books.Pair {
	var out []books.Pair
	for _, p := range cfg.AvailablePairs() {
		out = append(out, books.Pair{Source: p.Source, Target: p.Target})
	}
	return out
}

// refreshSweepBudget is what materializing the owed reading surfaces may take in one pass. Larger
// than the sweep's own because the work is a re-chunk of the source per book, and it is spent on
// nobody else's behalf: the money of every run that owes one has already settled.
const refreshSweepBudget = 10 * 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".
//
// ⚠ It carries the intake's materialization too: the parse is followed by two more engine reads of
// the same source, and the sweep's own guard — "do not start a book with less than one book's worth
// of pass left" — is only true if the pass can hold both.
const intakeSweepBudget = jobs.JobTimeout + readmodel.MaterializeBudget + time.Minute

// idempotencyWindow is how long a key is remembered. The contract promises "at least 24 hours, then
// the key is forgotten" — both halves: without the sweep the table only grows, and a month-old key
// still replays a response about a book that may no longer exist.
const idempotencyWindow = 25 * time.Hour

// sweeps is everything one tick of the reconciler covers.
type sweeps struct {
	runs    *runs.Service
	books   *books.Service
	reader  *readmodel.Service
	exports *exports.Service
	db      *pgstore.Store
	metrics *metrics.Metrics
}

// sweepPass is one reconciler's share of a tick: its name in the telemetry, what it may take, and
// what it does.
type sweepPass struct {
	name   string
	budget time.Duration
	run    func(context.Context) error
}

// passes is EVERY pass of one tick, in order, as a list rather than a dozen conditionals inside the
// loop.
//
// A list for the same reason the contract surface is one (httpapi.contractSurface): "every
// reconciler this daemon runs" becomes something CODE can enumerate, so a pass wired into the daemon
// and into nothing else is not expressible. It was: the whole of an assignment like this used to be
// observable only by running the daemon against a database, a systemd bus and a queue — which is how
// `RunBudget` came to be declared, documented and left at zero (PD-331), and how a new pass could be
// added and silently never armed.
//
// ⚠ The BUDGETS are not equal and the differences are argued where they are chosen: the
// materialization and the intake get their own because the work inside them is engine calls over a
// whole book, the export GC and the idempotency sweep take the ordinary one because neither calls
// anything. What this function does NOT do is decide how the tick's time is shared between them —
// that is `pass`, and register row PD-386 is open on it.
func (s sweeps) passes(sweepBudget time.Duration, log *slog.Logger) []sweepPass {
	out := []sweepPass{{"runs", sweepBudget, s.runs.Sweep}}
	// Materializing what a boundary left is TWO full engine reads of the source — minutes on a large
	// book — so it gets a pass of its own after the money has settled. Inside the runs pass it held up
	// every account's settlement behind it.
	if s.reader != nil {
		out = append(out, sweepPass{"readmodel", refreshSweepBudget, s.reader.Drain})
	}
	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.
		out = append(out, sweepPass{"intake", intakeSweepBudget, s.books.Sweep})
	}
	if s.exports != nil {
		// The export GC: take away what has lapsed, and end the polls of builds nobody is coming back
		// for. It rides this tick because both questions are one indexed statement plus a few unlinks,
		// and it takes the ORDINARY budget: nothing inside it calls the engine.
		out = append(out, sweepPass{"exports", sweepBudget, s.exports.Sweep})
	}
	if s.db != nil {
		// Forgetting idempotency keys past their window is the second half of the promise the header
		// makes; it rides this tick because it is one indexed DELETE and needs no schedule of its own.
		out = append(out, sweepPass{"idempotency", sweepBudget, func(c context.Context) error {
			forgotten, err := s.db.SweepIdempotency(c, time.Now().Add(-idempotencyWindow))
			if err == nil && forgotten > 0 {
				log.Info("idempotency keys forgotten", "keys", forgotten)
			}
			return err
		}})
	}
	return out
}

// 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, sweepBudget 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() {
		for _, p := range s.passes(sweepBudget, log) {
			pass(p.name, p.budget, p.run)
		}
		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, runs.StalledAfter)
	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,
		StalledRuns:         o.StalledRuns,
		AbandonedSurfaces:   o.AbandonedSurfaces,
	})
	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
}