// 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"
"net/http"
"os"
"os/exec"
"path/filepath"
"strings"
"sync"
"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
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)
}
// Runner polls, runs and reports until its context ends.
type Runner struct {
cfg Config
slots chan struct{}
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.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
}
return &Runner{cfg: cfg, slots: make(chan struct{}, cfg.MaxTasks)}, 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 {
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)
go r.heartbeat(hbCtx, t.TaskID)
verdict := r.execute(ctx, taskType, in.Checkin, t.Metadata[JobKey], logs, errs)
stopHB()
logs.Close()
errs.Close()
took := time.Since(start).Round(time.Second)
if verdict != nil {
r.cfg.Logf("%s %s @ %.12s: FAIL in %s: %v", taskType, in.Repo, in.Checkin, took, verdict)
r.resolve(ctx, t.TaskID, map[string]any{"action": "fail", "error": verdict.Error()})
return
}
r.cfg.Logf("%s %s @ %.12s: ok in %s", taskType, in.Repo, in.Checkin, took)
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"`
}
// 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, taskType, checkin, job string, 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
}
dir := filepath.Join(r.cfg.WorkDir, "ws", safeName(taskType))
mu, _ := r.locks.LoadOrStore(dir, &sync.Mutex{})
mu.(*sync.Mutex).Lock()
defer mu.(*sync.Mutex).Unlock()
if err := r.checkout(ctx, dir, checkin, logs); err != nil {
return err
}
if job != "" {
j, err := ParseJob(job)
if err != nil {
return err
}
return runJob(ctx, j, dir, 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 := exec.CommandContext(ctx, task.Argv[0], task.Argv[1:]...)
cmd.Dir = filepath.Join(dir, filepath.FromSlash(task.Dir))
if len(task.Env) > 0 {
cmd.Env = os.Environ()
for _, k := range task.EnvNames() {
cmd.Env = append(cmd.Env, k+"="+task.Env[k])
}
}
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.
func (r *Runner) heartbeat(ctx context.Context, taskID string) {
tick := time.NewTicker(r.cfg.Heartbeat)
defer tick.Stop()
for {
select {
case <-ctx.Done():
return
case <-tick.C:
r.resolve(ctx, taskID, map[string]any{"action": "heartbeat"})
}
}
}
func (r *Runner) resolve(ctx context.Context, taskID string, body map[string]any) {
b, _ := json.Marshal(body)
resp, err := r.post(ctx, "/v1/tasks/"+taskID+"/resolve", b)
if err != nil {
if ctx.Err() == nil {
r.cfg.Logf("resolve %s (%v): %v", taskID, body["action"], err)
}
return
}
resp.Body.Close()
}
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()
return nil, fmt.Errorf("%s: %d %s", path, resp.StatusCode, strings.TrimSpace(string(msg)))
}
return resp, nil
}
// 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
}