186 lines
7.5 KiB
Go
186 lines
7.5 KiB
Go
package ingest
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"os"
|
|
"os/exec"
|
|
"time"
|
|
)
|
|
|
|
// Outcome is the engine's ratified exit contract (cmd/tmctl/main.go:30-52). It is read from the
|
|
// exit code rather than from the log, and a bank stop is deliberately NOT a failure.
|
|
type Outcome string
|
|
|
|
const (
|
|
OutcomeClean Outcome = "clean" // 0
|
|
OutcomeFailed Outcome = "failed" // 1 — infra failure, and anything unrecognised
|
|
OutcomeFlagged Outcome = "flagged" // 2 — completed with flagged units
|
|
OutcomeBankStop Outcome = "bank_stop" // 3 — deliberate human-in-the-loop halt
|
|
)
|
|
|
|
func outcomeOf(code int) Outcome {
|
|
switch code {
|
|
case 0:
|
|
return OutcomeClean
|
|
case 2:
|
|
return OutcomeFlagged
|
|
case 3:
|
|
return OutcomeBankStop
|
|
default:
|
|
return OutcomeFailed
|
|
}
|
|
}
|
|
|
|
// stopGrace is how long the engine has to shut down after the interrupt before it is killed. The
|
|
// engine stops gracefully on SIGINT/SIGTERM (signal.NotifyContext in tmctl's main), and it holds an
|
|
// EXCLUSIVE lock on its project file — a SIGKILL first would leave that lock behind.
|
|
const stopGrace = 30 * time.Second
|
|
|
|
// Supervisor runs one tmctl process per attempt.
|
|
//
|
|
// Stream discipline (research/23 §2): stdout belongs to the event stream and to nothing else;
|
|
// the engine's own logs are on stderr and stay there.
|
|
type Supervisor struct {
|
|
// Bin is the tmctl binary. The platform never links the engine — it spawns it (D39.81).
|
|
Bin string
|
|
// Workdir is the book's project directory: book.yaml, the source and the engine's private
|
|
// SQLite live there. That SQLite is never opened by us (D39.85 §4).
|
|
Workdir string
|
|
// Env is the child's environment. Provider keys reach the engine through it and must never be
|
|
// logged; nil means the parent's environment.
|
|
Env []string
|
|
// EngineLog receives the child's stderr verbatim. It is a FILE, not our structured logger: the
|
|
// engine logs per-call cost estimates at INFO, and money must not enter the platform's INFO
|
|
// stream (D39.84). nil discards.
|
|
EngineLog io.Writer
|
|
// Log is the platform's own view of the run. Nil is silent, which is what tests want and what
|
|
// production must not be: a translation runs for hours and its only trace would otherwise be
|
|
// the engine's own file.
|
|
Log *slog.Logger
|
|
}
|
|
|
|
func (s *Supervisor) logger() *slog.Logger {
|
|
if s.Log == nil {
|
|
return slog.New(slog.DiscardHandler)
|
|
}
|
|
return s.Log
|
|
}
|
|
|
|
// Run spawns the engine and ingests its stream. It returns the outcome even when the stream itself
|
|
// failed, because "what did the process do" and "did we materialize all of it" are different
|
|
// questions: the second one is answered by reconciling with Status.
|
|
func (s *Supervisor) Run(ctx context.Context, sink Sink, args ...string) (Outcome, error) {
|
|
// A broken ingest must STOP the run, not watch it (PD-12): with the sink failing, the platform
|
|
// is blind for the hours the engine keeps running and spending, and the ceiling and bank-stop
|
|
// events of that run go to io.Discard with nobody told.
|
|
runCtx, stop := context.WithCancel(ctx)
|
|
defer stop()
|
|
|
|
cmd := exec.CommandContext(runCtx, s.Bin, args...)
|
|
cmd.Dir = s.Workdir
|
|
cmd.Env = s.Env
|
|
cmd.Stderr = s.engineLog()
|
|
setProcessGroup(cmd)
|
|
// CommandContext kills on cancel by default; the engine needs the signal it already handles,
|
|
// and WaitDelay is the backstop if it ignores it.
|
|
cmd.Cancel = func() error { return askToStop(cmd.Process) }
|
|
cmd.WaitDelay = stopGrace
|
|
|
|
stdout, err := cmd.StdoutPipe()
|
|
if err != nil {
|
|
return OutcomeFailed, fmt.Errorf("ingest: stdout pipe: %w", err)
|
|
}
|
|
log := s.logger()
|
|
if err := cmd.Start(); err != nil {
|
|
log.ErrorContext(ctx, "engine did not start", "err", err, "bin", s.Bin)
|
|
return OutcomeFailed, fmt.Errorf("ingest: start %s: %w", s.Bin, err)
|
|
}
|
|
log.InfoContext(ctx, "engine started", "pid", cmd.Process.Pid, "args", args)
|
|
|
|
ingestErr := Ingest(runCtx, stdout, sink)
|
|
// A cancelled run context is OUR stop, not a broken sink. Ingest reports it as its own error, and
|
|
// treating the two alike fired the one ERROR line that is supposed to mean "the platform is blind
|
|
// while money is being spent" on every ordinary shutdown of a live run — and called stop() on a
|
|
// run that was already stopping. Found by review.
|
|
if ingestErr != nil && errors.Is(ingestErr, context.Canceled) && runCtx.Err() != nil {
|
|
ingestErr = nil
|
|
}
|
|
if ingestErr != nil {
|
|
// The run is being ended because we cannot record it. Said once, here, because from the
|
|
// caller's side it is indistinguishable from the engine failing on its own.
|
|
log.ErrorContext(ctx, "stream could not be materialized: stopping the run", "err", ingestErr)
|
|
stop()
|
|
}
|
|
// Drain whatever is left so the child never blocks on a full pipe while we are waiting for it.
|
|
_, _ = io.Copy(io.Discard, stdout)
|
|
|
|
waitErr := cmd.Wait()
|
|
// The exit code is the outcome even when WE stopped the run. exec reports a cancelled command
|
|
// as context.Canceled rather than an *ExitError, so reading the outcome off waitErr alone marks
|
|
// every gracefully stopped run as failed — including all of them on an ordinary SIGTERM.
|
|
if cmd.ProcessState != nil && cmd.ProcessState.Exited() {
|
|
outcome := outcomeOf(cmd.ProcessState.ExitCode())
|
|
log.InfoContext(ctx, "engine finished", "outcome", outcome, "exit_code", cmd.ProcessState.ExitCode())
|
|
// A stopped run is not a finished one, and the engine exits 0 for both. The cancellation
|
|
// travels in the error so the caller can tell "stopped" from "done" — the outcome cannot
|
|
// carry it, because it mirrors the engine's exit contract and nothing else.
|
|
if err := runCtx.Err(); err != nil {
|
|
return outcome, errors.Join(err, ingestErr)
|
|
}
|
|
return outcome, ingestErr
|
|
}
|
|
if waitErr != nil {
|
|
// Did not exit: killed, or never became a process we could wait on.
|
|
log.ErrorContext(ctx, "engine did not exit", "err", waitErr)
|
|
return OutcomeFailed, errors.Join(waitErr, ingestErr)
|
|
}
|
|
return OutcomeClean, ingestErr
|
|
}
|
|
|
|
// Status runs the reconciliation channel: `tmctl status --json` on a stopped or finished run. It
|
|
// is read-only and free, but NOT free of CPU — every call re-ingests and re-chunks the source
|
|
// (1.4-1.5 s on a 23 MB book, engine backlog row 100), so it is a repair path, not a poll.
|
|
func (s *Supervisor) Status(ctx context.Context) (StatusReport, error) {
|
|
cmd := exec.CommandContext(ctx, s.Bin, "status", "--json")
|
|
cmd.Dir = s.Workdir
|
|
cmd.Env = s.Env
|
|
cmd.Stderr = s.engineLog()
|
|
out, err := cmd.Output()
|
|
if err != nil {
|
|
return StatusReport{}, fmt.Errorf("ingest: tmctl status: %w", err)
|
|
}
|
|
return DecodeStatus(out)
|
|
}
|
|
|
|
// retryStop is when the interrupt is repeated. ONE signal is not enough, and this is measured, not
|
|
// defensive: a child interrupted in the first milliseconds of its life misses the signal outright
|
|
// (reproduced on this stand — roughly one run in three), and the only thing left is WaitDelay's
|
|
// SIGKILL, which is exactly what must not happen to a process holding an EXCLUSIVE lock on the
|
|
// book's project file. See PD-20.
|
|
var retryStop = []time.Duration{30 * time.Millisecond, 120 * time.Millisecond, 400 * time.Millisecond}
|
|
|
|
// askToStop asks the engine to shut down, and keeps asking for about half a second.
|
|
func askToStop(p *os.Process) error {
|
|
first := interruptGroup(p)
|
|
for _, d := range retryStop {
|
|
time.Sleep(d)
|
|
// Asks the kernel whether the process is still ours to signal; it goes through os.Process,
|
|
// so a reaped child answers "done" instead of the call reaching a recycled pid.
|
|
if !stillRunning(p) {
|
|
return first
|
|
}
|
|
_ = interruptGroup(p)
|
|
}
|
|
return first
|
|
}
|
|
|
|
func (s *Supervisor) engineLog() io.Writer {
|
|
if s.EngineLog == nil {
|
|
return io.Discard
|
|
}
|
|
return s.EngineLog
|
|
}
|