// Package runner is `quarry runner`: a generic CI worker a repository's
// owner runs on their own machine (f24d78d9).
//
// It speaks the engine's HTTP worker protocol through the forge
// (internal/cibridge), so all it needs is the repository's URL and a
// worker token from Settings → Runners. For each task it checks out the
// check-in under test, reads ci/runner.json AT that check-in, runs the
// argv the manifest names for the task's type, streams its output to
// the run's log, and reports the verdict.
//
// It runs whatever the check-in's manifest says. That is the point on
// the owner's machine and exactly why it must never be pointed at a
// repository whose writers the machine's owner does not trust.
package runner
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
"io"
"log"
"maps"
"net/http"
"os"
"path/filepath"
"strings"
"sync"
"sync/atomic"
"time"
)
// Checkout puts a check-in's tree in dir and returns nothing else: the
// runner reads the manifest from the tree it is about to test.
type Checkout interface {
Checkout(ctx context.Context, dir, checkin string) error
}
// Config is what a runner needs.
type Config struct {
Bridge string // https://forge/r/<repo>/ci/bridge — the four worker routes live under it
Token string // a worker token from Settings → Runners
Tasks []string // task types to poll for
MaxTasks int // tasks run at once
WorkDir string // one workspace per task type under it
// TempRoot holds one temp directory per running task, which a task
// finds as TMPDIR and GOTMPDIR and which goes when the task ends;
// the runner removes its own leftover task dirs as it starts. It defaults to
// "quarry-runner-tmp" under WorkDir; it must be a real directory,
// never a symlink.
TempRoot string
WorkerID string
Version string // printed in the ready line, so a log says which build ran
Checkout Checkout
Client *http.Client
// Heartbeat is how often a running task extends its claim; the
// engine redelivers a task whose claim lapses.
Heartbeat time.Duration
// CheckoutRetry is the wait before each further checkout attempt;
// why the checkout retries here and the step does not is in
// ci/quarry.json's _comment. Every checkout error is retried, so a
// permanent one (a hash the forge does not have) fails after ~65s.
CheckoutRetry []time.Duration
Logf func(format string, args ...any)
// Branch, when set, makes this a runner for one branch's check-ins
// only: a deploy runner, holding a production secret in its
// environment, refuses a branch's code and every premerge pair
// whatever the pipeline that sent the task says. BranchOf answers
// which branch a check-in is on; the forge is the authority.
Branch string
BranchOf func(ctx context.Context, checkin string) (string, error)
// JobUsers, when set, runs every task as one of these users, a pool
// the runner may become through sudo, one per running task (see
// isolate.go). It is what lets a runner receive secrets; without it
// a job asking for one is refused. Self is this program's path, whose
// job-env subcommand starts each command.
JobUsers []string
Self string
}
// Runner polls, runs and reports until its context ends.
type Runner struct {
cfg Config
slots chan struct{}
users *slots
// places numbers the workflow jobs' workspaces, one per task the
// runner may run at once.
places chan int
locks sync.Map // workspace dir -> *sync.Mutex
}
// New checks cfg and fills its defaults.
func New(cfg Config) (*Runner, error) {
if cfg.Bridge == "" || cfg.Token == "" || len(cfg.Tasks) == 0 || cfg.Checkout == nil || cfg.WorkDir == "" {
return nil, errors.New("runner: bridge, token, tasks, workdir and checkout are all required")
}
if cfg.Branch != "" && cfg.BranchOf == nil {
return nil, errors.New("runner: a branch-only runner needs BranchOf")
}
if cfg.MaxTasks < 1 {
cfg.MaxTasks = 1
}
if cfg.TempRoot == "" {
cfg.TempRoot = filepath.Join(cfg.WorkDir, tempRootName)
}
if cfg.Client == nil {
cfg.Client = &http.Client{}
}
if cfg.Heartbeat <= 0 {
cfg.Heartbeat = 20 * time.Second
}
if cfg.CheckoutRetry == nil {
cfg.CheckoutRetry = []time.Duration{5 * time.Second, 15 * time.Second, 45 * time.Second}
}
if cfg.WorkerID == "" {
h, _ := os.Hostname()
cfg.WorkerID = fmt.Sprintf("quarry-runner-%s-%d", h, os.Getpid())
}
if cfg.Logf == nil {
cfg.Logf = log.Printf
}
users, err := newSlots(cfg.JobUsers, cfg.Self)
if err != nil {
return nil, err
}
places := make(chan int, cfg.MaxTasks)
for i := 0; i < cfg.MaxTasks; i++ {
places <- i
}
return &Runner{cfg: cfg, slots: make(chan struct{}, cfg.MaxTasks), users: users, places: places}, nil
}
// Run connects and polls every task type until ctx ends. One poller per
// type, because a poll's answer does not say which of several types a
// task was for, and the type is what picks the manifest entry.
func (r *Runner) Run(ctx context.Context) error {
if err := r.sweepTemp(); err != nil {
return err
}
go r.stayConnected(ctx)
var wg sync.WaitGroup
for _, t := range r.cfg.Tasks {
wg.Add(1)
go func(taskType string) {
defer wg.Done()
r.pollLoop(ctx, taskType)
}(t)
}
r.cfg.Logf("quarry runner %s ready · %s · tasks %s · max-tasks %d", r.cfg.Version, r.cfg.Bridge, strings.Join(r.cfg.Tasks, ", "), r.cfg.MaxTasks)
wg.Wait()
return ctx.Err()
}
// stayConnected holds the registration stream open, reconnecting with
// backoff, so the engine lists this worker and its capacity.
func (r *Runner) stayConnected(ctx context.Context) {
backoff := time.Second
for ctx.Err() == nil {
body, _ := json.Marshal(map[string]any{"worker_id": r.cfg.WorkerID, "task_types": r.cfg.Tasks, "max_tasks": r.cfg.MaxTasks})
resp, err := r.post(ctx, "/v1/workers/connect", body)
if err == nil {
backoff = time.Second
io.Copy(io.Discard, resp.Body) // the heartbeat stream, until it ends
resp.Body.Close()
} else if ctx.Err() == nil {
r.cfg.Logf("connect: %v; retrying in %s", err, backoff)
}
select {
case <-ctx.Done():
return
case <-time.After(backoff):
}
if backoff < time.Minute {
backoff *= 2
}
}
}
type polled struct {
TaskID string `json:"task_id"`
RunID string `json:"run_id"`
StepID string `json:"step_id"`
Input json.RawMessage `json:"input"`
// Metadata is the step's own, copied by the engine: a workflow job
// travels here under JobKey.
Metadata map[string]string `json:"metadata,omitempty"`
}
// pollLoop takes a slot, then asks for one task of its type; a slot is
// held before claiming so a claimed task never waits for a slot here.
func (r *Runner) pollLoop(ctx context.Context, taskType string) {
for ctx.Err() == nil {
select {
case r.slots <- struct{}{}:
case <-ctx.Done():
return
}
body, _ := json.Marshal(map[string]any{"task_types": []string{taskType}, "max_tasks": 1, "timeout_ms": 25000})
resp, err := r.post(ctx, "/v1/tasks/poll", body)
if err != nil {
<-r.slots
if ctx.Err() == nil {
r.cfg.Logf("poll %s: %v", taskType, err)
time.Sleep(5 * time.Second)
}
continue
}
var tasks []polled
err = json.NewDecoder(resp.Body).Decode(&tasks)
resp.Body.Close()
if err != nil || len(tasks) == 0 {
<-r.slots
continue
}
t := tasks[0]
go func() {
defer func() { <-r.slots }()
r.runTask(ctx, taskType, t)
}()
}
}
// runTask checks out, reads the manifest at the check-in, runs the
// task's argv, and resolves the task with the verdict.
func (r *Runner) runTask(ctx context.Context, taskType string, t polled) {
in, carried := subjectOf(t.Input)
start := time.Now()
logs := newLogSink(ctx, r, t.TaskID, "out")
errs := newLogSink(ctx, r, t.TaskID, "err")
hbCtx, stopHB := context.WithCancel(ctx)
taskCtx, stopTask := context.WithCancel(ctx)
defer stopTask()
var cancelled atomic.Bool
go r.heartbeat(hbCtx, t.TaskID, func() { cancelled.Store(true); stopTask() })
var waited time.Duration
verdict := r.execute(taskCtx, t.TaskID, taskType, in.Checkin, in.Tag, t.Metadata[JobKey], &waited, logs, errs)
stopHB()
if cancelled.Load() {
fmt.Fprintln(errs, "stopped: this task's run was cancelled")
}
logs.Close()
errs.Close()
if cancelled.Load() {
// The engine withdrew the task; a resolve would only be refused.
r.cfg.Logf("%s %s @ %.12s: stopped after %s: its run was cancelled", taskType, in.Repo, in.Checkin, time.Since(start).Round(time.Second))
return
}
// The time a task spent waiting for a workspace is not its own:
// said apart, so "ok in 4m" is how long the work took.
took := (time.Since(start) - waited).Round(time.Second)
queue := ""
if waited >= time.Second {
queue = fmt.Sprintf(" (waited %s for a workspace)", waited.Round(time.Second))
}
if verdict != nil {
r.cfg.Logf("%s %s @ %.12s: FAIL in %s%s: %v", taskType, in.Repo, in.Checkin, took, queue, verdict)
r.resolve(ctx, t.TaskID, map[string]any{"action": "fail", "error": verdict.Error()})
return
}
r.cfg.Logf("%s %s @ %.12s: ok in %s%s", taskType, in.Repo, in.Checkin, took, queue)
r.resolve(ctx, t.TaskID, map[string]any{"action": "complete", "output": carried})
}
// subject is what a task is about: the repository and the check-in.
type subject struct {
Repo string `json:"repo"`
Checkin string `json:"checkin"`
// Tag is the tag whose adding started the run, for a run the tag
// event started (RFC 0020); its steps see it as QUARRY_TAG.
Tag string `json:"tag,omitempty"`
}
// subjectOf reads a task's subject from its input, and returns the input
// a step after it should receive. The engine hands a step with no
// depends_on the run's input, but a step with one dependency that
// step's output, and a step with several a map of each one's output
// (dagnats dag.ResolveInput). So a task's output is its subject, which
// is how a deploy that waits for the tests still knows what to deploy;
// with several dependencies the subject is taken from their outputs,
// and they must agree.
func subjectOf(raw json.RawMessage) (subject, json.RawMessage) {
var in subject
if json.Unmarshal(raw, &in) == nil && in.Checkin != "" {
return in, raw
}
var fanIn map[string]json.RawMessage
if json.Unmarshal(raw, &fanIn) != nil {
return subject{}, raw
}
var found subject
var from json.RawMessage
for _, out := range fanIn {
var s subject
if json.Unmarshal(out, &s) != nil || s.Checkin == "" {
continue
}
if found.Checkin != "" && s != found {
return subject{}, raw // the dependencies disagree: no subject, so the task fails naming no check-in
}
found, from = s, out
}
if found.Checkin == "" {
return subject{}, raw
}
return found, from
}
// execute runs one task at a check-in: the workflow job it carries, or
// else the command ci/runner.json names for its type there.
func (r *Runner) execute(ctx context.Context, taskID, taskType, checkin, tag, job string, waited *time.Duration, logs, errs io.Writer) error {
if checkin == "" {
return errors.New("the task names no check-in")
}
if err := r.onBranch(ctx, checkin); err != nil {
return err
}
queued := time.Now()
// A manifest task keeps one workspace per task type, as it always
// has. Every workflow job has the same task type (<group>.job), so
// jobs take a workspace from a pool of --max-tasks, or they would
// run one at a time whatever --max-tasks says.
dir := filepath.Join(r.cfg.WorkDir, "ws", safeName(taskType))
if job != "" {
place := <-r.places
defer func() { r.places <- place }()
dir = fmt.Sprintf("%s-%d", dir, place)
}
mu, _ := r.locks.LoadOrStore(dir, &sync.Mutex{})
mu.(*sync.Mutex).Lock()
defer mu.(*sync.Mutex).Unlock()
if waited != nil {
*waited = time.Since(queued)
}
if err := r.checkout(ctx, dir, checkin, logs); err != nil {
return err
}
// Made before the job user's reclaim defers below, so it is removed
// after that has ended whatever still holds files in it, and removed
// even when that reclaim fails.
tmp, err := r.taskTemp()
if err != nil {
return err
}
defer func() {
if err := removeAll(tmp); err != nil {
r.cfg.Logf("task temp dir %s was not removed: %v", tmp, err)
}
}()
start := withTemp(plainStarter, tmp)
if r.users != nil {
slot := <-r.users.free
defer func() {
if err := r.users.reclaim(slot, dir, tmp); err != nil {
r.cfg.Logf("job user %s is out of the pool until the runner restarts: %v", slot, err)
return
}
r.users.free <- slot
}()
if err := share(dir, slot); err != nil {
return fmt.Errorf("sharing the checkout with job user %s: %w", slot, err)
}
if err := share(tmp, slot); err != nil {
return fmt.Errorf("sharing the temp dir with job user %s: %w", slot, err)
}
start = withTemp(r.users.starter(slot, dir), tmp)
}
if job != "" {
j, err := ParseJob(job)
if err != nil {
return err
}
// The job's run token (RFC 0021 §runs), whatever its file
// declared: the forge decides from its own records, and says
// why when it hands none.
// A runner running jobs as itself cannot keep a token to the
// job, so it does not ask for one: the job runs without.
var tokenEnv map[string]string
if r.users == nil {
fmt.Fprintln(logs, "no run token: this runner runs jobs as itself and cannot keep one to the job; start it with --job-users")
} else {
env, note, err := r.fetchRunToken(ctx, taskID)
if err != nil {
return err
}
if note != "" {
fmt.Fprintln(logs, note)
}
tokenEnv = env
}
var secrets map[string]string
if names := j.secrets(); len(names) > 0 {
if r.users == nil {
// Refused rather than run without them: a deploy that
// quietly got no token would fail later and less clearly.
return fmt.Errorf("the job asks for secrets (%s), and this runner runs jobs as itself, so it cannot receive them; start it with --job-users", strings.Join(names, ", "))
}
if secrets, err = r.fetchSecrets(ctx, taskID, j.Environment, names); err != nil {
return err
}
}
masked := maps.Clone(secrets)
if v := tokenEnv["QUARRY_TOKEN"]; v != "" {
if masked == nil {
masked = map[string]string{}
}
masked["QUARRY_TOKEN"] = v
}
if len(masked) > 0 {
ml, me := newMasker(logs, masked), newMasker(errs, masked)
defer ml.Close()
defer me.Close()
logs, errs = ml, me
}
run := maps.Clone(tokenEnv)
if tag != "" {
if run == nil {
run = map[string]string{}
}
run["QUARRY_TAG"] = tag
}
return runJob(ctx, j, dir, run, secrets, start, logs, errs)
}
raw, err := os.ReadFile(filepath.Join(dir, ManifestPath))
if err != nil {
return fmt.Errorf("%s at %.12s: %v", ManifestPath, checkin, err)
}
m, err := ParseManifest(raw)
if err != nil {
return err
}
task, ok := m.Tasks[taskType]
if !ok {
return fmt.Errorf("%s at %.12s names no task %q (it names %s)", ManifestPath, checkin, taskType, strings.Join(m.Names(), ", "))
}
fmt.Fprintf(logs, "$ %s\n", strings.Join(task.Argv, " "))
cmd := start(ctx, filepath.Join(dir, filepath.FromSlash(task.Dir)), task.Argv, task.Env)
var caught bytes.Buffer
cmd.Stdout = logs
if task.FailOnOutput {
cmd.Stdout = io.MultiWriter(logs, &caught)
}
cmd.Stderr = errs
if err := cmd.Run(); err != nil {
return err
}
if task.FailOnOutput && strings.TrimSpace(caught.String()) != "" {
n := len(strings.Fields(caught.String()))
fmt.Fprintf(errs, "%d line(s) of output from a task that must print nothing\n", n)
return fmt.Errorf("%d line(s) of output from a task that must print nothing", n)
}
return nil
}
// checkout gets the tree, retrying with CheckoutRetry's waits: the
// only part of a task whose failure says nothing about the check-in.
func (r *Runner) checkout(ctx context.Context, dir, checkin string, logs io.Writer) error {
for attempt := 0; ; attempt++ {
fmt.Fprintf(logs, "$ checkout %.12s\n", checkin)
err := r.cfg.Checkout.Checkout(ctx, dir, checkin)
if err == nil {
return nil
}
var conflict *MergeConflict
if errors.As(err, &conflict) {
// A verdict, not a fault: the same pair will not merge on
// the next try either. The paths are the step's log.
fmt.Fprintln(logs, conflict.Error())
return conflict
}
if attempt >= len(r.cfg.CheckoutRetry) || ctx.Err() != nil {
return fmt.Errorf("checkout: %w", err)
}
wait := r.cfg.CheckoutRetry[attempt]
fmt.Fprintf(logs, "checkout failed: %v; retrying in %s\n", err, wait)
select {
case <-time.After(wait):
case <-ctx.Done():
return fmt.Errorf("checkout: %w", err)
}
}
}
// heartbeat extends the task's claim while it runs.
// heartbeat keeps a task's claim alive, and stops the task when the
// engine answers that its run was cancelled: a closed landing's merge
// run, or one cancelled by hand, ends here rather than running to its
// end and holding the runner (dagnats#737).
func (r *Runner) heartbeat(ctx context.Context, taskID string, stop func()) {
tick := time.NewTicker(r.cfg.Heartbeat)
defer tick.Stop()
for {
select {
case <-ctx.Done():
return
case <-tick.C:
if errors.Is(r.resolve(ctx, taskID, map[string]any{"action": "heartbeat"}), errRunCancelled) {
stop()
return
}
}
}
}
func (r *Runner) resolve(ctx context.Context, taskID string, body map[string]any) error {
b, _ := json.Marshal(body)
resp, err := r.post(ctx, "/v1/tasks/"+taskID+"/resolve", b)
if err != nil {
if ctx.Err() == nil && !errors.Is(err, errRunCancelled) {
r.cfg.Logf("resolve %s (%v): %v", taskID, body["action"], err)
}
return err
}
resp.Body.Close()
return nil
}
func (r *Runner) post(ctx context.Context, path string, body []byte) (*http.Response, error) {
req, err := http.NewRequestWithContext(ctx, "POST", strings.TrimSuffix(r.cfg.Bridge, "/")+path, bytes.NewReader(body))
if err != nil {
return nil, err
}
req.Header.Set("Authorization", "Bearer "+r.cfg.Token)
req.Header.Set("Content-Type", "application/json")
resp, err := r.cfg.Client.Do(req)
if err != nil {
return nil, err
}
if resp.StatusCode >= 300 {
msg, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
resp.Body.Close()
err := fmt.Errorf("%s: %d %s", path, resp.StatusCode, strings.TrimSpace(string(msg)))
if resp.StatusCode == http.StatusConflict {
// dagnats#737: every resolve action on a task of a
// cancelled run, heartbeat included, answers 409.
return nil, fmt.Errorf("%w: %v", errRunCancelled, err)
}
if resp.StatusCode == http.StatusNotFound {
return nil, fmt.Errorf("%w: %v", errNoRoute, err)
}
return nil, err
}
return resp, nil
}
// errNoRoute is a 404: a forge older than the route asked for.
var errNoRoute = errors.New("not found")
// errRunCancelled is the engine saying a task's run was cancelled
// (dagnats#737): the task is withdrawn, its work should stop, and
// nothing it would report is recorded.
var errRunCancelled = errors.New("run cancelled")
// logSink batches one stream of a task's output into the run's log: a
// flush every half second or 32 KB, whichever comes first, so a quiet
// step still shows progress and a loud one does not make a request per
// line. The engine names its two streams "out" and "err"; anything else
// is refused (measured: "stdout" was, so the first live run showed no
// output at all).
type logSink struct {
r *Runner
stream string
ctx context.Context
taskID string
mu sync.Mutex
buf bytes.Buffer
warned bool
done chan struct{}
wg sync.WaitGroup
}
func newLogSink(ctx context.Context, r *Runner, taskID, stream string) *logSink {
s := &logSink{r: r, ctx: ctx, taskID: taskID, stream: stream, done: make(chan struct{})}
s.wg.Add(1)
go func() {
defer s.wg.Done()
tick := time.NewTicker(500 * time.Millisecond)
defer tick.Stop()
for {
select {
case <-s.done:
s.flush()
return
case <-tick.C:
s.flush()
}
}
}()
return s
}
func (s *logSink) Write(p []byte) (int, error) {
s.mu.Lock()
s.buf.Write(p)
big := s.buf.Len() >= 32<<10
s.mu.Unlock()
if big {
s.flush()
}
return len(p), nil
}
func (s *logSink) flush() {
s.mu.Lock()
if s.buf.Len() == 0 {
s.mu.Unlock()
return
}
data := append([]byte(nil), s.buf.Bytes()...)
s.buf.Reset()
s.mu.Unlock()
body, _ := json.Marshal(map[string]any{"chunks": []map[string]string{{"stream": s.stream, "data": base64.StdEncoding.EncodeToString(data)}}})
resp, err := s.r.post(s.ctx, "/v1/tasks/"+s.taskID+"/logs", body)
if err != nil {
// A lost batch does not fail the task, but it must not be
// silent: a run page that shows no output while the runner
// printed plenty is the confident-nothing shape. Said once per
// task, so a dead log lane does not flood the runner's output.
s.mu.Lock()
first := !s.warned
s.warned = true
s.mu.Unlock()
if first {
s.r.cfg.Logf("logs for %s could not be sent (the run page will show none): %v", s.taskID, err)
}
return
}
resp.Body.Close()
}
func (s *logSink) Close() {
close(s.done)
s.wg.Wait()
}
func safeName(s string) string {
if s == "" {
return "_"
}
return strings.Map(func(c rune) rune {
switch {
case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9', c == '-', c == '_', c == '.':
return c
}
return '-'
}, s)
}
// onBranch refuses a check-in that is not on the runner's branch, before
// anything of it is fetched or run. A runner with no branch runs all.
func (r *Runner) onBranch(ctx context.Context, checkin string) error {
if r.cfg.Branch == "" {
return nil
}
if IsPair(checkin) {
return fmt.Errorf("this runner runs %s check-ins only, and a landing's merge is not one", r.cfg.Branch)
}
got, err := r.cfg.BranchOf(ctx, checkin)
if err != nil {
return fmt.Errorf("which branch %.12s is on: %w", checkin, err)
}
if got != r.cfg.Branch {
return fmt.Errorf("this runner runs %s check-ins only, and %.12s is on %q", r.cfg.Branch, checkin, got)
}
return nil
}