1// Package oprunner runs one pipeline operation at a time and streams its voice2// back to whoever asked for it.3//4// The workbench does not reimplement the pipeline: an operation is the `atlas`5// binary invoked exactly as a person at a terminal would invoke it, and what6// comes back is that run's own event stream, one row at a time, flushed as it7// arrives (issue #5 §3.1, §5.6). This package is the whole of that mechanism8// and is deliberately not workbench-internal code: the safety properties below9// are the interesting part, they are testable without a page around them, and a10// second consumer of the pipeline gets them by importing rather than by copying.11//12// # The safety properties13//14// A page that runs pipeline operations is a page that runs commands, so these15// four are the contract rather than an implementation detail, and they are16// named here so that a change to any of them is a change somebody argued for:17//18// - **Origin-checked.** A browser sends Origin on any cross-site POST.19// [CheckOrigin] refuses a foreign one, so a stray page cannot operate the20// workbench through the visitor's browser.21// - **One operation at a time.** [Runner.Acquire] is the single slot; a22// second submission is answered 409 rather than queued. The tools are safe23// to interleave in principle, but two crawls sharing one pane help nobody.24// - **Validated targets.** [ValidTarget] admits only what is safe as a25// command argument and is actually a slug; a leading dash is refused so a26// target can never be read as a flag.27// - **It dies with the request.** The subprocess runs under the request's28// context: a page abandoned mid-operation stops its operation, so nothing29// crawls on with nobody watching.30//31// # The stream32//33// Rows are data, not markup. The runner parses the subprocess's event stream34// -- `atlas` writes slog records to stderr, JSON when asked (docs/logging.md)35// -- into [Row] values carrying level, message and the documented attribute36// vocabulary; anything that is not a record is carried as its own line.37// Rendering rows as HTML is the consumer's business, which is what keeps the38// framework vocabulary in templates (issue #5 §4.3).39package oprunner4041import (42 "bufio"43 "context"44 "encoding/json"45 "errors"46 "fmt"47 "os/exec"48 "sort"49 "strings"50 "sync"51)5253// Operation is one subprocess to run: what it is called, where it runs, and the54// argv a person would have typed. Nothing here is derived from a request55// without passing through [ValidTarget] first.56type Operation struct {57 // Name is the operation as the page named it -- "crawl", "tiles",58 // "compose", "enrich", "measure". It is what a row is labelled with and59 // what a refusal talks about.60 Name string61 // Dir is the working directory. Empty means the runner's own, which is62 // what an absolute-path invocation wants.63 Dir string64 // Argv is the command, program first. It is passed to the operating65 // system as-is: no shell, no expansion, no interpolation.66 Argv []string67}6869// Validate refuses an operation that is not a command. It is the last gate70// before a subprocess exists, and it exists so that a caller assembling argv71// from form values cannot accidentally assemble nothing.72func (o Operation) Validate() error {73 switch {74 case o.Name == "":75 return errors.New("an operation must be named")76 case len(o.Argv) == 0:77 return fmt.Errorf("operation %s has no command", o.Name)78 case strings.TrimSpace(o.Argv[0]) == "":79 return fmt.Errorf("operation %s names no program", o.Name)80 }81 for at, arg := range o.Argv {82 if strings.ContainsAny(arg, "\x00\n") {83 return fmt.Errorf("operation %s: argument %d carries a line break or a NUL", o.Name, at)84 }85 }86 return nil87}8889// Command is the operation as one line, for the row that announces it.90func (o Operation) Command() string { return strings.Join(o.Argv, " ") }9192// Kind says what a row is. The four kinds are the four things a run says: what93// was asked, what the event stream reported, what the program printed outside94// its event stream, and how it ended.95type Kind string9697const (98 // KindCommand is the first row of every run: the command itself.99 KindCommand Kind = "command"100 // KindEvent is one parsed slog record from the subprocess's event stream.101 KindEvent Kind = "event"102 // KindOutput is a line the subprocess wrote that is not a record: product103 // output on stdout, or a message from something that does not speak slog.104 KindOutput Kind = "output"105 // KindResult is the last row: how the run ended.106 KindResult Kind = "result"107)108109// Stream names which of a subprocess's two mouths a row came from.110const (111 StreamOut = "stdout"112 StreamErr = "stderr"113)114115// Row is one line of an operation's voice.116//117// It is deliberately flat and stringly: a row exists to be rendered, and a118// consumer that has to reach into a nested value to write a table cell is119// deciding things a template should be reading.120type Row struct {121 // Seq numbers rows from one, in arrival order across both streams.122 Seq int123 // Kind is what this row is.124 Kind Kind125 // Stream is StreamOut or StreamErr for the rows that came from the126 // subprocess, empty for the rows the runner speaks itself.127 Stream string128 // Level is the event's level ("INFO", "WARN", "ERROR", "DEBUG") when the129 // row is a parsed record, empty otherwise.130 Level string131 // Time is the record's own timestamp, verbatim, when it carried one.132 Time string133 // Message is the record's message, the command line, the raw output line,134 // or the result -- whichever this row is.135 Message string136 // Attrs are the record's remaining attributes in a stable order: the137 // documented vocabulary of docs/logging.md first, in the order that138 // document lists it, then anything else alphabetically.139 Attrs []Attr140 // Failed marks a result row whose operation did not succeed, and an event141 // row at ERROR. It is a styling fact as much as a semantic one.142 Failed bool143}144145// Attr is one attribute of an event.146type Attr struct{ Key, Value string }147148// vocabulary is the documented attribute order of docs/logging.md. Attributes149// outside it sort after it, alphabetically: a row reads the same way every150// time, and the facts a person groups by come first.151var vocabulary = []string{"op", "volume", "world", "lens", "stamp", "source", "enricher", "dur", "path"}152153// Runner owns the one operation slot. The zero value is ready to use.154type Runner struct {155 slot sync.Mutex156 // held is the operation running right now, for a refusal that can say157 // what it is waiting on.158 mu sync.Mutex159 held string160}161162// ErrBusy is what a second operation is told. It is the 409 of §5.6.163var ErrBusy = errors.New("an operation is already running")164165// Busy reports the operation running right now, or the empty string.166func (r *Runner) Busy() string {167 r.mu.Lock()168 defer r.mu.Unlock()169 return r.held170}171172// Acquire takes the single operation slot, or reports that it is taken. The173// returned release is safe to call exactly once and must be called.174//175// It is exported because the refusal has to happen before a response is176// committed: a 409 cannot be sent after the first row has gone out.177func (r *Runner) Acquire(name string) (release func(), ok bool) {178 if !r.slot.TryLock() {179 return nil, false180 }181 r.mu.Lock()182 r.held = name183 r.mu.Unlock()184 return func() {185 r.mu.Lock()186 r.held = ""187 r.mu.Unlock()188 r.slot.Unlock()189 }, true190}191192// Run streams one operation to emit, holding the slot for its whole life.193//194// The rows arrive in the order the subprocess produced them, interleaved across195// its two streams. An emit that fails ends the run: the consumer is the page,196// and a page that has gone away is the operation's reason to stop.197func (r *Runner) Run(ctx context.Context, op Operation, emit func(Row) error) error {198 if err := op.Validate(); err != nil {199 return err200 }201 release, ok := r.Acquire(op.Name)202 if !ok {203 return ErrBusy204 }205 defer release()206 return Stream(ctx, op, emit)207}208209// Stream runs one operation and emits its rows, taking no slot.210//211// It is the mechanism without the mutex, exported for the consumer that has212// already acquired the slot itself -- [Runner.Serve] does exactly that, because213// it must answer 409 before it starts writing a body.214func Stream(ctx context.Context, op Operation, emit func(Row) error) error {215 if err := op.Validate(); err != nil {216 return err217 }218 // A run that ends early -- an emit that fails, a caller that gives up --219 // takes its subprocess with it.220 ctx, stop := context.WithCancel(ctx)221 defer stop()222223 seq := 0224 say := func(row Row) error {225 seq++226 row.Seq = seq227 return emit(row)228 }229 if err := say(Row{Kind: KindCommand, Message: op.Command(), Attrs: commandAttrs(op)}); err != nil {230 return err231 }232233 cmd := exec.CommandContext(ctx, op.Argv[0], op.Argv[1:]...)234 cmd.Dir = op.Dir235 stdout, err := cmd.StdoutPipe()236 if err != nil {237 return err238 }239 stderr, err := cmd.StderrPipe()240 if err != nil {241 return err242 }243 if err := cmd.Start(); err != nil {244 // A program that will not start is the operation's whole story, and245 // it is told through the stream rather than beside it.246 return say(Row{Kind: KindResult, Message: err.Error(), Failed: true})247 }248249 lines := make(chan Row, 64)250 var readers sync.WaitGroup251 readers.Add(2)252 go func() { defer readers.Done(); scan(stdout, StreamOut, lines) }()253 go func() { defer readers.Done(); scan(stderr, StreamErr, lines) }()254 go func() { readers.Wait(); close(lines) }()255256 var emitErr error257 for row := range lines {258 if emitErr != nil {259 continue // drain, so the readers finish and Wait can reap260 }261 if err := say(row); err != nil {262 emitErr = err263 stop()264 }265 }266 waitErr := cmd.Wait()267 if emitErr != nil {268 return emitErr269 }270 if waitErr != nil {271 return say(Row{Kind: KindResult, Message: waitErr.Error(), Failed: true})272 }273 return say(Row{Kind: KindResult, Message: "done"})274}275276func commandAttrs(op Operation) []Attr {277 attrs := []Attr{{Key: "op", Value: op.Name}}278 if op.Dir != "" {279 attrs = append(attrs, Attr{Key: "path", Value: op.Dir})280 }281 return attrs282}283284// scan reads one of the subprocess's streams line by line and turns each line285// into a row. A line that parses as a slog record becomes an event; anything286// else is carried verbatim, because a tool's own words are worth more than a287// runner's opinion of them.288func scan(from interface{ Read([]byte) (int, error) }, stream string, out chan<- Row) {289 reader := bufio.NewScanner(from)290 // A ledger line or a path can be long; a megabyte is past anything the291 // pipeline writes and short of anything that matters.292 reader.Buffer(make([]byte, 0, 64*1024), 1<<20)293 for reader.Scan() {294 out <- rowOf(reader.Text(), stream)295 }296}297298// rowOf reads one line as a row.299func rowOf(line, stream string) Row {300 if row, ok := eventOf(line); ok {301 row.Stream = stream302 return row303 }304 return Row{Kind: KindOutput, Stream: stream, Message: line}305}306307// eventOf reads a JSON slog record. Anything missing its message is not one:308// the pipeline's product output is JSON too (`atlas measure -json`), and a309// score is not an event.310func eventOf(line string) (Row, bool) {311 trimmed := strings.TrimSpace(line)312 if !strings.HasPrefix(trimmed, "{") {313 return Row{}, false314 }315 var record map[string]any316 if err := json.Unmarshal([]byte(trimmed), &record); err != nil {317 return Row{}, false318 }319 message, ok := record["msg"].(string)320 if !ok {321 return Row{}, false322 }323 level, _ := record["level"].(string)324 when, _ := record["time"].(string)325 row := Row{326 Kind: KindEvent,327 Level: level,328 Time: when,329 Message: message,330 Failed: strings.EqualFold(level, "ERROR"),331 }332 row.Attrs = attrsOf(record)333 return row, true334}335336// attrsOf is every attribute but the record's own three, in the documented337// order first and alphabetically after it.338func attrsOf(record map[string]any) []Attr {339 rest := make([]string, 0, len(record))340 for key := range record {341 switch key {342 case "msg", "level", "time", "source":343 continue344 }345 rest = append(rest, key)346 }347 sort.Slice(rest, func(a, b int) bool {348 rankA, rankB := rank(rest[a]), rank(rest[b])349 if rankA != rankB {350 return rankA < rankB351 }352 return rest[a] < rest[b]353 })354 attrs := make([]Attr, 0, len(rest))355 for _, key := range rest {356 attrs = append(attrs, Attr{Key: key, Value: valueOf(record[key])})357 }358 return attrs359}360361func rank(key string) int {362 for at, known := range vocabulary {363 if key == known {364 return at365 }366 }367 return len(vocabulary)368}369370// valueOf spells one attribute value. Numbers lose their JSON float shape:371// a count of 37 reads as 37, never as 3.7e+01.372func valueOf(value any) string {373 switch held := value.(type) {374 case string:375 return held376 case bool:377 if held {378 return "true"379 }380 return "false"381 case float64:382 if held == float64(int64(held)) {383 return fmt.Sprintf("%d", int64(held))384 }385 return fmt.Sprintf("%g", held)386 case nil:387 return ""388 default:389 encoded, err := json.Marshal(held)390 if err != nil {391 return fmt.Sprint(held)392 }393 return string(encoded)394 }395}