// quarry's own CI worker (RFC 0009, dogfooded): each task checks out
// the check-in under test in a fresh clone and runs one Go toolchain
// step, streaming its output to the engine's log lane. Deliberately
// boring — three task types, one helper, no state. Anything worth
// knowing is either in the run's logs (what the tools said) or on this
// process's stdout (what the worker did, one line per task).
//
// Its own module: the log-lane SDK (dagnats#624) is newer than the
// dagnats release quarry's main module pins, and a worker is a
// deployment artifact, not part of the forge.
package main
import (
"bytes"
"flag"
"fmt"
"io"
"log"
"os"
"os/exec"
"os/signal"
"path/filepath"
"strings"
"sync"
"time"
"github.com/danmestas/dagnats/worker"
"github.com/nats-io/nats.go"
)
var (
natsURL = flag.String("nats", nats.DefaultURL, "engine NATS address")
zitBin = flag.String("zit", "zit", "zit binary the worker shells (pin one so a zit upgrade cannot blind CI mid-flight)")
maxTasks = flag.Int("max-tasks", 3, "tasks this worker runs at once; the rest queue (three suites racing on one box turned 2-minute steps into 14)")
workdir = flag.String("work", filepath.Join(os.TempDir(), "quarry-ci-worker"), "scratch directory for clones")
keep = flag.Bool("keep", false, "keep a failed step's tree for debugging")
fresh = flag.Bool("fresh", false, "clone from scratch every step instead of reusing a workspace (slower; use to rule the workspace out)")
)
// wsLocks serialises access to one workspace path. Reuse means two
// tasks must never hold the same directory at once — the slot
// semaphore bounds how many run, not which.
var (
wsMu sync.Mutex
wsLocks = map[string]*sync.Mutex{}
)
func workspaceLock(path string) *sync.Mutex {
wsMu.Lock()
defer wsMu.Unlock()
m, ok := wsLocks[path]
if !ok {
m = &sync.Mutex{}
wsLocks[path] = m
}
return m
}
// stepPath is the PATH every step runs with: the pinned zit first,
// under the name "zit". A suite under test shells "zit" by name
// (internal/zitcli does), and the worker's own PATH is whatever shell
// launched it — one box moved zit to ~/.local/bin and every run went
// red with "executable file not found" while the checkout steps, which
// use the pin by path, carried on. Steps and checkouts must see the
// same zit, so the pin is the one on PATH.
var stepPath string
func exposeZit() error {
abs, err := exec.LookPath(*zitBin)
if err != nil {
return fmt.Errorf("zit %q: %w", *zitBin, err)
}
if abs, err = filepath.Abs(abs); err != nil {
return err
}
bin := filepath.Join(*workdir, "bin")
if err := os.MkdirAll(bin, 0o755); err != nil {
return err
}
link := filepath.Join(bin, "zit")
os.Remove(link)
if err := os.Symlink(abs, link); err != nil {
return err
}
stepPath = bin + string(os.PathListSeparator) + os.Getenv("PATH")
return nil
}
// slots gates handler entry: acquired before any clone, released when
// the step ends either way. The engine's dispatch stays greedy; the
// box's capacity is the worker's own truth to enforce.
var slots chan struct{}
func main() {
flag.Parse()
if *maxTasks < 1 {
*maxTasks = 1
}
slots = make(chan struct{}, *maxTasks)
if err := os.MkdirAll(*workdir, 0o755); err != nil {
log.Fatal(err)
}
if err := exposeZit(); err != nil {
log.Fatal(err)
}
nc, err := nats.Connect(*natsURL)
if err != nil {
log.Fatalf("connect %s: %v", *natsURL, err)
}
defer nc.Close()
w := worker.NewWorker(nc)
step(w, "quarry-test", []string{"go", "test", "./..."}, false)
step(w, "quarry-build", []string{"go", "build", "./..."}, false)
// gofmt -l exits 0 even when files need formatting; its OUTPUT is
// the verdict, so this step fails on any.
step(w, "quarry-fmt", []string{"gofmt", "-l", "cmd", "internal"}, true)
log.Printf("quarry-ci-worker ready · nats %s · work %s · zit %s · max-tasks %d", *natsURL, *workdir, *zitBin, *maxTasks)
w.Start()
sig := make(chan os.Signal, 1)
signal.Notify(sig, os.Interrupt)
<-sig
w.Stop()
}
// step registers one task type: clone, open the check-in, run argv.
func step(w *worker.Worker, task string, argv []string, failOnOutput bool) {
worker.HandleTyped(w, task, func(ctx worker.TaskContext, in map[string]any) (string, error) {
repo, _ := in["repo"].(string)
checkin, _ := in["checkin"].(string)
clone, _ := in["cloneUrl"].(string)
if clone == "" || checkin == "" {
return "", fmt.Errorf("input missing cloneUrl or checkin")
}
// Take a slot before spending anything; say so when the wait
// is real, so a slow run reads as "queued behind N" in the
// worker's own log rather than as mystery latency.
select {
case slots <- struct{}{}:
default:
log.Printf("%s %s @ %.12s: waiting for a slot (%d busy)", task, repo, checkin, len(slots))
slots <- struct{}{}
}
defer func() { <-slots }()
start := time.Now()
out, errw := ctx.LogOut(), ctx.LogErr()
// A PERSISTENT workspace per (repo, task), not a fresh clone.
//
// Cloning quarry from the forge measured 94 SECONDS, and every
// step of every run paid it — three steps meant nearly five
// minutes of identical transfer per run, which is most of why a
// 2-minute suite reported 6-9 minutes. Updating a workspace
// that already holds the history costs about six.
//
// Correctness comes from revert+clean, not from newness: managed
// files go back to their checked-in contents and untracked
// leftovers are deleted, so a reused tree enters the step
// looking like a fresh one. Anything unexpected — a failed
// pull, a tree we cannot clean — falls back to a full clone
// rather than running a step against a tree we cannot vouch
// for.
//
// This is language-agnostic: it is the clone that was slow, not
// any toolchain, so every project on the forge gets it.
dir := filepath.Join(*workdir, "ws", safeName(repo), safeName(task))
if *fresh {
var err error
if dir, err = os.MkdirTemp(*workdir, task+"-"); err != nil {
return "", err
}
}
lock := workspaceLock(dir)
lock.Lock()
defer lock.Unlock()
failed := false
defer func() {
if *fresh {
if failed && *keep {
log.Printf("%s: kept %s", task, dir)
return
}
os.RemoveAll(dir)
return
}
// A reused workspace is kept by definition; that is the
// point. It is left at the check-in that was tested, which
// is also what a reader wants when a step fails.
if failed {
log.Printf("%s: workspace kept at %s", task, dir)
}
}()
fail := func(err error) (string, error) {
failed = true
log.Printf("%s %s @ %.12s: FAIL in %s: %v", task, repo, checkin, time.Since(start).Round(time.Second), err)
return "", err
}
run := func(extra io.Writer, name string, args ...string) error {
fmt.Fprintf(out, "$ %s %s\n", name, strings.Join(args, " "))
cmd := exec.Command(name, args...)
cmd.Dir = dir
cmd.Env = append(os.Environ(), "PATH="+stepPath)
cmd.Stdout, cmd.Stderr = out, errw
if extra != nil {
cmd.Stdout = io.MultiWriter(out, extra)
}
return cmd.Run()
}
// Reuse the workspace when it already holds this repository;
// clone only when there is nothing to reuse or reuse failed.
reused := false
if !*fresh {
if _, serr := os.Stat(filepath.Join(dir, ".zit")); serr == nil {
reused = refresh(run, out, clone, checkin) == nil
if !reused {
// Do not run a step against a tree we could not
// bring to a known state.
fmt.Fprintf(out, "$ workspace could not be refreshed; recloning\n")
os.RemoveAll(dir)
}
}
}
if !reused {
if err := os.MkdirAll(dir, 0o755); err != nil {
return fail(err)
}
if err := run(nil, *zitBin, "clone", clone, "."); err != nil {
return fail(fmt.Errorf("clone: %w", err))
}
// zit 0.2.0 opens the tree during clone; moving to the
// check-in under test is then an update. Older zits still
// want the open.
if err := run(nil, *zitBin, "update", checkin); err != nil {
if oerr := run(nil, *zitBin, "open", checkin); oerr != nil {
return fail(fmt.Errorf("open %s: %w", checkin, err))
}
}
}
var caught bytes.Buffer
var sink io.Writer
if failOnOutput {
sink = &caught
}
if err := run(sink, argv[0], argv[1:]...); err != nil {
return fail(err)
}
if failOnOutput && strings.TrimSpace(caught.String()) != "" {
n := len(strings.Fields(caught.String()))
fmt.Fprintf(errw, "%d file(s) need gofmt\n", n)
return fail(fmt.Errorf("%d file(s) need gofmt", n))
}
log.Printf("%s %s @ %.12s: ok in %s", task, repo, checkin, time.Since(start).Round(time.Second))
return "ok", nil
})
}
// refresh brings an existing workspace to the check-in under test.
// Every step matters: pull fetches artifacts the tree has never seen,
// revert undoes whatever the last step wrote to managed files, clean
// removes untracked leftovers so a stale file cannot satisfy a build
// that should have failed, and update materialises the check-in.
//
// Any failure returns an error and the caller reclones rather than
// running against a tree in an unknown state.
func refresh(run func(io.Writer, string, ...string) error, out io.Writer, clone, checkin string) error {
if err := run(nil, *zitBin, "pull", clone); err != nil {
return fmt.Errorf("pull: %w", err)
}
if err := run(nil, *zitBin, "revert"); err != nil {
return fmt.Errorf("revert: %w", err)
}
// clean is best-effort: a workspace with nothing untracked is the
// normal case, and some zits exit non-zero when there is nothing to
// delete. revert has already restored everything tracked.
_ = run(nil, *zitBin, "clean")
if err := run(nil, *zitBin, "update", checkin); err != nil {
return fmt.Errorf("update %s: %w", checkin, err)
}
return nil
}
// safeName keeps a repository or task name usable as one path element.
func safeName(s string) string {
if s == "" {
return "_"
}
return strings.Map(func(r rune) rune {
switch {
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9':
return r
case r == '-', r == '_', r == '.':
return r
}
return '-'
}, s)
}