textmachine/platform/cmd/tmplatformd/runner.go

package main

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

	"textmachine/platform/internal/config"
	"textmachine/platform/internal/httpapi"
	"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 returns the function that stops it.
//
// 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, deps *httpapi.Deps) (func(), error) {
	deps.Library = db
	if !cfg.RunsEnabled() {
		log.Warn("no TM_PLATFORM_ENGINE_BIN: the library is served read-only and no run can be started")
		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,
	}
	queue, err := runs.NewQueue(db.Pool(), svc, log, cfg.Runner.Workers)
	if err != nil {
		return nil, err
	}
	svc.Queue = queue
	deps.Runs = svc
	if err := queue.Start(ctx); err != nil {
		return nil, fmt.Errorf("start queue: %w", err)
	}
	go sweepRuns(ctx, svc, 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
}

// sweepRuns runs the reconciler 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).
func sweepRuns(ctx context.Context, svc *runs.Service, every time.Duration, log *slog.Logger) {
	sweep := func() {
		c, cancel := context.WithTimeout(ctx, 2*time.Minute)
		defer cancel()
		if err := svc.Sweep(c); err != nil && !errors.Is(err, context.Canceled) {
			log.Error("run sweep failed", "err", err)
		}
	}
	sweep()
	t := time.NewTicker(every)
	defer t.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-t.C:
			sweep()
		}
	}
}

// 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.
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 _, err := os.Stat(bin); err != nil {
		return nil, fmt.Errorf("exit-marker command %s: %w", bin, err)
	}
	return []string{bin, "exit-marker"}, nil
}