package ci
import (
"encoding/json"
"errors"
"fmt"
"log"
"net/url"
"path"
"regexp"
"slices"
"strings"
"time"
)
// errTransient marks a launch failure that says nothing about the
// check-in's own pipelines — the store answered strangely, or the
// engine could not be reached. The watermark must NOT advance past
// one: the next round retries, and nothing is silently lost. Its
// counterpart is the content fault (a pipeline that does not compile,
// an engine that refused the document), which DOES advance, because
// retrying a bad definition forever would wedge the watermark — the
// original rationale, kept.
var errTransient = errors.New("transient")
// transient wraps err so the trigger holds the watermark.
func transient(err error) error {
return fmt.Errorf("%w: %w", errTransient, err)
}
// engineUnreachable reports a network-shaped engine error — the engine
// never answered, as opposed to answering with a refusal.
func engineUnreachable(err error) bool {
var u *url.Error
return errors.As(err, &u)
}
// Store is the record-keeping the trigger needs; the OLTP database
// satisfies it.
type Store interface {
CIWatermark(repo string) (string, bool)
SetCIWatermark(repo, checkin string) error
AddRun(runID, repo, checkin, workflow, pipeline, status string) error
// The deferral (029c4730): check-ins waiting for a worker slot, and
// how many runs are in flight forge-wide to compare with capacity.
Defer(repo string, c Checkin) error
Deferred(repo string) ([]Checkin, error)
Undefer(repo, checkin string) error
// RunPipelines lists the pipelines with a run at a check-in that
// has, or will have, a verdict: not cancelled, not still deferred.
RunPipelines(repo, checkin string) ([]string, error)
}
// Repo is the read side the trigger needs: which check-ins are new, and
// what the repository's pipelines look like AT that check-in.
type Repo interface {
// NewCheckins returns check-ins after the watermark, oldest first.
// An empty watermark means "the tip only" — a first run must not
// launch a pipeline for every check-in in a repository's history,
// which on an imported corpus would be tens of thousands.
NewCheckins(after string, limit int) ([]Checkin, error)
// Files lists a check-in's paths.
Files(version string) ([]string, error)
// Read returns one file's bytes at a check-in.
Read(path, version string) ([]byte, error)
// HasOpenLanding says a branch is proposed somewhere, which is what
// exempts its check-ins from waiting when the worker is full.
HasOpenLanding(branch string) bool
// DefaultBranch names the line a deploy builds from (trunk, main).
DefaultBranch() string
}
// Checkin is the context a pipeline receives.
type Checkin struct {
Hash string
Branch string
Date string
User string
Comment string
}
// Trigger launches pipelines for check-ins a repository has not seen.
// It is deliberately quiet about the ordinary case — most repositories
// have no ci/ directory, and that is not a condition worth logging on
// every sync round.
func Trigger(e *Engine, st Store, repo string, r Repo, cloneURL string) error {
if !e.Enabled() {
return nil
}
mark, ok := st.CIWatermark(repo)
if !ok {
// First sight of this repository: take the tip as the watermark
// without running anything. Launching the whole history would
// be a denial of service against the engine and would say
// nothing useful about code nobody just changed.
tip, err := r.NewCheckins("", 1)
if err != nil || len(tip) == 0 {
return err
}
return st.SetCIWatermark(repo, tip[len(tip)-1].Hash)
}
fresh, err := r.NewCheckins(mark, 20)
if err != nil {
return err
}
// The workers are asked about room only once something is actually
// about to start: a repository with no pipelines, or a round with
// nothing new, never touches the engine.
var rm *room
need := func() *room {
if rm == nil {
x := slots(e, st)
rm = &x
}
return rm
}
drain(e, st, repo, r, cloneURL, need)
for _, c := range fresh {
pipelines, err := pipelinesAt(r, c)
if err != nil {
log.Printf("ci: %s %.12s: %v — watermark held, retrying next round", repo, c.Hash, err)
return nil
}
if len(pipelines) > 0 && defers(*need(), c, r) {
if err := st.Defer(repo, c); err != nil {
return err
}
log.Printf("ci: %s %.12s on %s: the worker is full; waiting for a slot", repo, c.Hash, c.Branch)
if err := st.SetCIWatermark(repo, c.Hash); err != nil {
return err
}
continue
}
if err := launch(e, st, repo, r, c, cloneURL, pipelines, scopePush); err != nil {
if errors.Is(err, errTransient) {
// The failure says nothing about this check-in's own
// pipelines. Hold the watermark and stop — the next
// round retries from here, in order, losing nothing.
// (2026-08-29: a drifted zit parser answered "no
// files" for three trunk check-ins and the old code
// walked the watermark past all of them, silently.)
log.Printf("ci: %s %.12s: %v — watermark held, retrying next round", repo, c.Hash, err)
return nil
}
// A content fault — the pipeline itself, or the engine's
// judgment of it. Advancing is right: retrying a bad
// definition forever would wedge the watermark.
log.Printf("ci: %s %.12s: %v", repo, c.Hash, err)
}
if err := st.SetCIWatermark(repo, c.Hash); err != nil {
return err
}
if len(pipelines) > 0 {
need().take()
}
}
return nil
}
// scope says which pipelines a launch starts. A JSON pipeline whose
// top-level "quarry_when" is "landing" (quarry's key, namespaced so the
// engine can grow its own "when"; the engine ignores unknown keys) runs
// on the default branch and on proposed branches only.
type scope int
const (
scopePush scope = iota // a new check-in: landing pipelines only on trunk or a proposed branch
scopeMissing // an opened landing's tip: every pipeline with no verdict there yet
scopeAll // an asked-for re-run: everything
)
// landingOnly reads a JSON pipeline's "quarry_when".
func landingOnly(def []byte) bool {
var p struct {
When string `json:"quarry_when"`
}
return json.Unmarshal(def, &p) == nil && p.When == "landing"
}
// yamlQuarryWhen finds quarry_when as a top-level key of a ci.yml,
// where it is not honoured: refusing beats running everywhere a
// pipeline its author meant for landings only.
var yamlQuarryWhen = regexp.MustCompile(`(?m)^quarry_when\s*:`)
// room is what the deferral knows about the workers: how many more
// STEPS they will take right now, or that nobody can say. Unknown never
// defers — a forge whose engine does not list its workers (dagnats#628)
// behaves exactly as before.
type room struct {
capacity int
free int
known bool
e *Engine
}
func (r *room) full() bool { return r.known && r.free <= 0 }
// take re-measures after a launch. A run holds as many steps as its
// pipeline runs at once — three for quarry's own — so reserving one
// slot per run let a drain start four runs into a worker with room for
// one (measured on 7ead79c34e8f: four waiting branches all launched in
// the same second). The engine already lists the new run's steps as
// queued, so asking again is exact where a guess per run is not.
func (r *room) take() {
if !r.known {
return
}
busy, err := r.e.Occupied(time.Now())
if err != nil {
r.free-- // the engine stopped answering: at least one is gone
return
}
r.free = r.capacity - busy
}
// slots asks the engine for capacity (the sum of every worker's
// max_tasks) and for what occupies it (Engine.Occupied: steps, not
// runs). Either answer missing means unknown. The forge's own ci_run
// cache is not asked: its "running" rows include runs the engine will
// never schedule, which made the first version defer every branch.
func slots(e *Engine, st Store) room {
workers, err := e.Workers()
if err != nil || len(workers) == 0 {
return room{}
}
capacity := 0
now := time.Now()
for _, w := range workers {
// A registration the engine has not heard from lately is a
// worker that restarted: the deploy restarts it, and for a
// while the engine lists both, doubling the capacity.
if t, err := time.Parse(time.RFC3339Nano, w.LastSeen); err == nil && now.Sub(t) > workerStale {
continue
}
capacity += w.MaxTasks
}
if capacity <= 0 {
return room{}
}
busy, err := e.Occupied(now)
if err != nil {
return room{}
}
return room{capacity: capacity, free: capacity - busy, known: true, e: e}
}
// defers decides whether a fresh check-in waits (029c4730): only when
// the worker is full, only on a branch, never on trunk — the line a
// deploy builds from — and never on a branch somebody has proposed,
// because a landing is exactly what the wait exists to protect.
func defers(rm room, c Checkin, r Repo) bool {
if !rm.full() || c.Branch == "" || c.Branch == r.DefaultBranch() {
return false
}
return !r.HasOpenLanding(c.Branch)
}
// drain launches waiting check-ins, longest wait first, while there is
// room. A check-in whose branch has since been proposed goes regardless
// of room, since a landing is waiting on it.
func drain(e *Engine, st Store, repo string, r Repo, cloneURL string, need func() *room) {
waiting, err := st.Deferred(repo)
if err != nil || len(waiting) == 0 {
return
}
for _, c := range waiting {
pipelines, err := pipelinesAt(r, c)
if err != nil {
log.Printf("ci: %s %.12s: waited, then: %v", repo, c.Hash, err)
continue
}
if len(pipelines) == 0 {
_ = st.Undefer(repo, c.Hash) // nothing to run after all
continue
}
if need().full() && !r.HasOpenLanding(c.Branch) {
continue
}
if err := launch(e, st, repo, r, c, cloneURL, pipelines, scopePush); err != nil {
log.Printf("ci: %s %.12s: waited, then: %v", repo, c.Hash, err)
continue
}
need().take()
}
}
// pipelinesAt lists a check-in's pipeline files, or the transient
// reason it could not: a store that answers no files at all is not a
// check-in with none, and the watermark must not pass it.
func pipelinesAt(r Repo, c Checkin) ([]string, error) {
files, err := r.Files(c.Hash)
if err != nil {
return nil, transient(err)
}
if len(files) == 0 {
return nil, transient(fmt.Errorf("the store answered no files at all for this check-in"))
}
var pipelines []string
for _, f := range files {
if IsPipeline(f) {
pipelines = append(pipelines, f)
}
}
return pipelines, nil
}
// RunAt launches a check-in's pipelines regardless of the watermark —
// the re-run path (RFC 0009 §triggers). An empty checkin means the tip.
func RunAt(e *Engine, st Store, repo string, r Repo, cloneURL, checkin string) error {
if !e.Enabled() {
return fmt.Errorf("ci: no engine configured")
}
var c Checkin
if checkin == "" {
tip, err := r.NewCheckins("", 1)
if err != nil {
return err
}
if len(tip) == 0 {
return fmt.Errorf("ci: %s has no check-ins", repo)
}
c = tip[len(tip)-1]
} else {
c = Checkin{Hash: checkin}
}
pipelines, err := pipelinesAt(r, c)
if err != nil {
return err
}
return launch(e, st, repo, r, c, cloneURL, pipelines, scopeAll)
}
// RunMissingAt starts, at an opened landing's tip, the pipelines that
// have no verdict there: the "when": "landing" ones a bare push
// skipped, and any whose run was cancelled or pruned. What already
// passed, failed, or is running is left alone, so a failure is never
// buried under a fresh attempt.
func RunMissingAt(e *Engine, st Store, repo string, r Repo, cloneURL string, c Checkin) error {
if !e.Enabled() {
return fmt.Errorf("ci: no engine configured")
}
pipelines, err := pipelinesAt(r, c)
if err != nil {
return err
}
return launch(e, st, repo, r, c, cloneURL, pipelines, scopeMissing)
}
func launch(e *Engine, st Store, repo string, r Repo, c Checkin, cloneURL string, pipelines []string, sc scope) error {
if len(pipelines) == 0 {
return nil // no ci/ directory: the common case, silently fine
}
var ran []string
if sc == scopeMissing {
var err error
if ran, err = st.RunPipelines(repo, c.Hash); err != nil {
return transient(err)
}
}
onLine := c.Branch != "" && (c.Branch == r.DefaultBranch() || r.HasOpenLanding(c.Branch))
for _, p := range pipelines {
if slices.Contains(ran, p) {
continue // a verdict is here or on its way; nothing to read
}
def, err := r.Read(p, c.Hash)
if err != nil {
return transient(fmt.Errorf("read %s: %w", p, err))
}
isJSON := strings.HasSuffix(p, ".json")
if !isJSON && yamlQuarryWhen.Match(def) {
return fmt.Errorf("%s: quarry_when is read from JSON pipelines only; move this pipeline to %s.json or drop the key", p, strings.TrimSuffix(strings.TrimSuffix(p, ".yml"), ".yaml"))
}
if sc == scopePush && isJSON && landingOnly(def) && !onLine {
continue
}
// Registered from the definition AT THIS CHECK-IN, so a pipeline
// edit takes effect for the check-in that made it and not before.
// A ci.yml is named by its file stem — the spec language has no
// name field of its own — which is the same posture JSON
// pipelines already take on a shared engine: the register just
// before every Start keeps each run on its own definition.
// Registered under a name that names the REPOSITORY as well as
// the file, because the engine is shared and a file stem is not
// unique across repositories. See ScopedName.
stem := strings.TrimSuffix(strings.TrimSuffix(path.Base(p), ".yml"), ".yaml")
stem = strings.TrimSuffix(stem, ".json")
name := ScopedName(repo, stem)
var wf Workflow
if strings.HasSuffix(p, ".json") {
wf, err = e.RegisterNamed(name, def)
} else {
wf, err = e.RegisterYAML(name, def)
}
if err != nil {
if engineUnreachable(err) {
return transient(err)
}
return err // the engine answered: a content refusal
}
labels := map[string]string{"repo": repo, "checkin": c.Hash, "pipeline": p}
if c.Branch != "" {
labels["branch"] = c.Branch
// A new check-in on a branch makes the branch's still-running
// builds stale; stop them (dagnats#629, best-effort).
if n, err := e.Supersede(wf.Name, map[string]string{
"repo": repo, "branch": c.Branch, "pipeline": p,
}); err == nil && n > 0 {
log.Printf("ci: %s %s: superseded %d run(s)", repo, c.Branch, n)
}
}
runID, err := e.Start(wf.Name, map[string]any{
"repo": repo,
"checkin": c.Hash,
"branch": c.Branch,
"date": c.Date,
"user": c.User,
"comment": c.Comment,
"cloneUrl": cloneURL,
"pipeline": p,
}, labels)
if err != nil {
if engineUnreachable(err) {
return transient(err)
}
return err
}
if err := st.AddRun(runID, repo, c.Hash, wf.Name, p, "running"); err != nil {
// The run is launched but unrecorded; losing the mapping
// by advancing would orphan it.
return transient(err)
}
log.Printf("ci: %s %.12s -> %s run %s", repo, c.Hash, wf.Name, runID)
}
// Started, by whichever door: the wait, if there was one, is over.
if err := st.Undefer(repo, c.Hash); err != nil {
log.Printf("ci: %s %.12s: forgetting the wait: %v", repo, c.Hash, err)
}
return nil
}