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/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,
		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)")
	}
	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,
		Pricing: model,
		Cfg:     runsConfig(cfg, marker, ceiling),
		Log:     log,
	}
	intake, err := startIntake(cfg, db, rn, reader, 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, reader: reader, 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
}

// 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,
		// None are built here yet: the export path is deferred, and answering a format this
		// deployment cannot produce would be a promise the first click discovers is empty.
		ExportFormats:   []string{},
		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
}

// 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
	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, 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() {
		pass("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 {
			pass("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.
			pass("intake", intakeSweepBudget, s.books.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.
			pass("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
			})
		}
		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
}