package ci
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"log"
"maps"
"net/url"
"path"
"regexp"
"slices"
"strings"
"sync"
"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
// def is DefHash of the definition the run was started from.
AddRun(runID, repo, checkin, workflow, pipeline, def, 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)
// Refuse records that a pipeline at a check-in will not run, and
// why, as a failing check on subject.
Refuse(repo, subject, pipeline, reason string) error
// RunsOf lists a subject's runs, newest first.
RunsOf(repo, subject string) ([]RunRecord, error)
// Inherit records a pipeline's verdict at a check-in as the one
// run from reached on an equal tree (RFC 0009 §inheritance).
Inherit(repo, checkin, workflow, pipeline, status, from string) error
// Note attaches a run's one-line explanation, shown as its check's
// summary.
Note(runID, note string) error
// Started records what a run token for the run may be minted from
// (RFC 0021 §runs): the event, the starter, each job's declaration.
Started(runID string, start RunStart) error
// PushedBy is the login that authenticated to carry an artifact to
// this forge, "" when none is recorded (a mirror's pull, a push
// from before the record): never the artifact's own U card, which
// any clone can write.
PushedBy(repo, artifact string) string
}
// RunRecord is one recorded run, as RunsOf lists it.
type RunRecord struct {
RunID string
Pipeline string
Status string
Def string // DefHash of the definition it ran; "" when unrecorded
}
// DefHash names a pipeline definition by its bytes.
func DefHash(def []byte) string {
sum := sha256.Sum256(def)
return hex.EncodeToString(sum[:])
}
// 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
// BranchOf names the branch a check-in is on, "" when unknown.
BranchOf(hash string) string
// LandingTarget names the branch an open landing proposes branch
// into, "" when none is open.
LandingTarget(branch string) string
// BranchTip is a branch's newest check-in, "" when unknown.
BranchTip(branch string) string
// MergeState is the store's answer to whether branchTip merges
// into targetTip: clean, up-to-date, unrelated, or conflict and the
// paths in the way.
MergeState(branchTip, targetTip string) (state string, conflicts []string, err error)
// ParentsOf lists a check-in's parents.
ParentsOf(hash string) ([]string, error)
// SameTree reports whether a check-in's tree, every path and its
// content hash, is the tree merging branchTip into targetTip writes.
SameTree(checkin, branchTip, targetTip string) (bool, error)
// PairPipelines is the pipeline files of the tree merging branchTip
// into targetTip writes, by path; ErrNoMerge when it does not merge.
PairPipelines(branchTip, targetTip string) (map[string][]byte, error)
}
// Checkin is the context a pipeline receives.
type Checkin struct {
Hash string
Branch string
Date string
User string
Comment string
// Head is a premerge pair's branch tip, set when Hash is a pair
// rather than a check-in: its refusals are recorded there.
Head string
// Note is why the runs are about this subject, when it is not the
// one they would be: a push tested as its tip because its merge
// does not apply.
Note string
// Tag is the tag whose adding started this launch (scopeTag).
Tag string
// Starter is the person whose act started the launch when it is
// not the check-in's carrier: the tag's carrier, or whoever asked
// for a re-run (RFC 0021 §runs).
Starter string
}
// ErrNoMerge is Repo.PairPipelines' answer for a pair that does not
// merge: a conflict, or unrelated histories. There is no merged tree.
var ErrNoMerge = errors.New("the pair does not merge")
// definitions is the pipelines a launch runs and how to read each one.
type definitions struct {
paths []string
read func(path string) ([]byte, error)
}
// only is d narrowed to paths.
func (d definitions) only(paths []string) definitions {
return definitions{paths: paths, read: d.read}
}
// AcceptKey is what a landing's accept records its account under before
// the merge exists: the branch tip it merges. A merge check-in nobody
// pushed is the forge's own, and its starter is whoever accepted.
func AcceptKey(branchTip string) string { return "accept:" + branchTip }
// pusherOf is a check-in's starter: the login that carried it here, or,
// for a merge the forge wrote itself, the account that accepted it.
func pusherOf(st Store, repo string, r Repo, hash string) string {
if who := st.PushedBy(repo, hash); who != "" {
return who
}
parents, err := r.ParentsOf(hash)
if err != nil || len(parents) != 2 {
return ""
}
for _, p := range parents {
if who := st.PushedBy(repo, AcceptKey(p)); who != "" {
return who
}
}
return ""
}
// nameLocks holds one mutex per engine workflow name: what a launch's
// Register and Start hold together.
var nameLocks sync.Map
func nameLock(name string) *sync.Mutex {
m, _ := nameLocks.LoadOrStore(name, &sync.Mutex{})
return m.(*sync.Mutex)
}
// Pair is the canonical spelling of a premerge subject (RFC 0009): the
// branch tip merged into the target tip.
func Pair(branchTip, targetTip string) string {
return "pair:" + branchTip + ":" + targetTip
}
// splitPair reads a Pair subject back into its two tips.
func splitPair(s string) (branchTip, targetTip string, ok bool) {
rest, ok := strings.CutPrefix(s, "pair:")
if !ok {
return "", "", false
}
branchTip, targetTip, ok = strings.Cut(rest, ":")
return branchTip, targetTip, ok && branchTip != "" && targetTip != ""
}
// 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)
behind := behindInRound(r, fresh)
for _, c := range fresh {
if d, ok := behind[c.Hash]; ok {
// A later check-in of this push is on top of it: the tip
// is what gets tested. Launching this too only for the
// tip's launch to cancel it a moment later earned no
// verdict before #314, and since arrival order can put it
// second it cancelled the tip's run instead.
log.Printf("ci: %s %.12s on %s: not launched; %.12s in the same round is on top of it", repo, c.Hash, c.Branch, d)
if err := st.SetCIWatermark(repo, c.Hash); err != nil {
return err
}
continue
}
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.paths) > 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 := launchPush(e, st, repo, r, c, cloneURL, pipelines); 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.paths) > 0 {
need().take()
}
}
return nil
}
// behindInRound names, for each check-in of a round that another
// check-in of the same round and branch descends from, that descendant.
func behindInRound(r Repo, fresh []Checkin) map[string]string {
out := map[string]string{}
for _, d := range fresh {
if d.Branch == "" {
continue
}
var inLine func(string) bool
for _, c := range fresh {
if c.Hash == d.Hash || c.Branch != d.Branch {
continue
}
if inLine == nil {
inLine = lineOf(r, d.Hash)
}
if inLine(c.Hash) {
out[c.Hash] = d.Hash
}
}
}
return out
}
// maxLine bounds an ancestry walk. A run further back than this is left
// to finish rather than walked to: the builds a launch supersedes are a
// branch's last few, and a round holds at most twenty check-ins.
const maxLine = 500
// lineOf answers whether a check-in is hash or one of its ancestors.
// The walk over ParentsOf goes only as far as each question needs and
// keeps what it saw for the next. A walk that fails or reaches maxLine
// answers for what it saw: a run it cannot place is not cancelled,
// which costs a stale build and never a live one.
func lineOf(r Repo, hash string) func(string) bool {
seen := map[string]bool{hash: true}
queue := []string{hash}
return func(h string) bool {
for !seen[h] && len(queue) > 0 && len(seen) < maxLine {
ps, err := r.ParentsOf(queue[0])
queue = queue[1:]
if err != nil {
queue = nil
break
}
for _, p := range ps {
if !seen[p] {
seen[p] = true
queue = append(queue, p)
}
}
}
return seen[h]
}
}
// scope says which pipelines a launch starts. A JSON pipeline's
// top-level "quarry_when" (quarry's key, namespaced so the engine can
// grow its own "when"; the engine ignores unknown keys) narrows where it
// runs: "landing" on the default branch and in landings' merges only,
// "default-branch" on check-ins of the default branch only.
type scope int
const (
scopePush scope = iota // a new check-in: landing pipelines only on the default branch
scopeMissing // a landing's merge: every pipeline with no run for it yet
scopeAll // an asked-for re-run: everything
scopeTag // a tag added: the workflows whose on.tag matches it, nothing else
scopeSchedule // a scheduled minute: the due workflows, on the default tip
)
// The values "quarry_when" may take.
const (
whenLanding = "landing"
whenDefaultBranch = "default-branch"
)
// quarryWhen reads a JSON pipeline's "quarry_when". Any other value is
// an error: a misspelt "default-branch" silently running everywhere
// would put a deploy on every branch.
func quarryWhen(def []byte) (string, error) {
var p struct {
When string `json:"quarry_when"`
}
if json.Unmarshal(def, &p) != nil {
return "", nil // the engine's registration says what is wrong with it
}
switch p.When {
case "", whenLanding, whenDefaultBranch:
return p.When, nil
}
return "", fmt.Errorf("quarry_when is %q; it may be %q or %q", p.When, whenLanding, whenDefaultBranch)
}
// launchEvent names what started a launch in `on:`'s words. A pair is a
// landing's merge, judged by its target branch, which is the branch the
// target tip is on. A new check-in is a push to its branch, whether its
// tip or its merge is what runs. A run asked
// for by hand is workflow_dispatch; it also runs whatever the check-in's
// push would have (see event.pushed). A tag added is a tag event on the
// tagged check-in.
func launchEvent(c Checkin, sc scope, r Repo) event {
switch {
case sc == scopeTag:
return event{kind: "tag", branch: c.Branch, tag: c.Tag}
case sc == scopeSchedule:
return event{kind: "schedule"}
case c.Head != "" && sc != scopePush:
target := ""
if _, tip, ok := splitPair(c.Hash); ok {
target = r.BranchOf(tip)
}
return event{kind: "landing", branch: target}
case sc == scopePush:
return event{kind: "push", branch: c.Branch}
}
pushed := c.Branch
if pushed == "" {
pushed = r.BranchOf(c.Hash)
}
return event{kind: "workflow_dispatch", pushed: pushed}
}
// 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 releases a repository's waiting check-ins without a new
// check-in to prompt it. Trigger drains only when the store changes, so
// on a quiet repository a wait outlived the room that ended it
// (quarry#296); a timer calls this instead. It asks the workers about
// room only when something is waiting.
func Drain(e *Engine, st Store, repo string, r Repo, cloneURL string) {
if !e.Enabled() {
return
}
var rm *room
drain(e, st, repo, r, cloneURL, func() *room {
if rm == nil {
x := slots(e, st)
rm = &x
}
return rm
})
}
// draining serialises drains of one repository: the fold's Trigger and
// the timer's Drain can meet, and both would launch the same waiter.
var draining sync.Map // repo -> *sync.Mutex
// 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) {
mu, _ := draining.LoadOrStore(repo, &sync.Mutex{})
mu.(*sync.Mutex).Lock()
defer mu.(*sync.Mutex).Unlock()
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.paths) == 0 {
_ = st.Undefer(repo, c.Hash) // nothing to run after all
continue
}
if need().full() && !r.HasOpenLanding(c.Branch) {
continue
}
if err := launchPush(e, st, repo, r, c, cloneURL, pipelines); err != nil {
log.Printf("ci: %s %.12s: waited, then: %v", repo, c.Hash, err)
continue
}
need().take()
}
}
// pipelinesAt is a subject's pipelines, or the transient reason it
// could not read them: a store that answers no files at all is not a
// check-in with none, and the watermark must not pass it.
//
// A check-in's are its own. A pair's are its merged tree's (RFC 0009
// §once): a pair tests what would land, so a pipeline the target added
// or changed runs as the merge holds it, the same definition the
// target's merge check-in will read. A pair that does not merge has no
// tree; its runs read the branch tip's, start, and fail on the
// conflict, which is the verdict.
func pipelinesAt(r Repo, c Checkin) (definitions, error) {
if c.Head != "" {
if _, target, ok := splitPair(c.Hash); ok {
defs, err := r.PairPipelines(c.Head, target)
switch {
case err == nil:
paths := make([]string, 0, len(defs))
for p := range defs {
paths = append(paths, p)
}
slices.Sort(paths)
return definitions{paths: paths, read: func(p string) ([]byte, error) {
if def, ok := defs[p]; ok {
return def, nil
}
return nil, fmt.Errorf("%s is not in the merged tree", p)
}}, nil
case !errors.Is(err, ErrNoMerge):
return definitions{}, transient(err)
}
}
}
at := c.Hash
if c.Head != "" {
at = c.Head
}
files, err := r.Files(at)
if err != nil {
return definitions{}, transient(err)
}
if len(files) == 0 {
return definitions{}, transient(fmt.Errorf("the store answered no files at all for this check-in"))
}
d := definitions{read: func(p string) ([]byte, error) { return r.Read(p, at) }}
for _, f := range files {
if IsPipeline(f) {
d.paths = append(d.paths, f)
}
}
return d, 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.
// by is who asked: the run's starter.
func RunAt(e *Engine, st Store, repo string, r Repo, cloneURL, checkin, by 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}
}
c.Starter = by
pipelines, err := pipelinesAt(r, c)
if err != nil {
return err
}
return launch(e, st, repo, r, c, cloneURL, pipelines, scopeAll)
}
// RunPairAt starts the pipelines a premerge pair has no verdict for:
// the tree accepting branchTip onto targetTip would write, fetched by
// the runner from the forge's archive of the pair (RFC 0009 §premerge).
// The definitions are the merged tree's (pipelinesAt). Every pipeline runs, "quarry_when": "landing" ones included: a pair
// exists only for a proposed branch.
//
// ran is false when the branch tip has no pipelines at all, which is
// the answer for every later ask about that tip too.
func RunPairAt(e *Engine, st Store, repo string, r Repo, cloneURL, branch, branchTip, targetTip string) (ran bool, err error) {
if !e.Enabled() {
return false, fmt.Errorf("ci: no engine configured")
}
c := Checkin{Hash: Pair(branchTip, targetTip), Branch: branch, Head: branchTip}
pipelines, err := pipelinesAt(r, c)
if err != nil {
return false, err
}
if len(pipelines.paths) == 0 {
return false, nil
}
return true, launch(e, st, repo, r, c, cloneURL, pipelines, scopeMissing)
}
// RunTip tests a branch tip as a new check-in would be: the landing
// sweep's answer for a tip whose landing closed with nothing else to
// say about it.
func RunTip(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 launchPush(e, st, repo, r, c, cloneURL, pipelines)
}
// RerunPairAt runs every pipeline on a landing's merge again, whatever
// already ran for it: a Re-run somebody asked for.
func RerunPairAt(e *Engine, st Store, repo string, r Repo, cloneURL, branch, branchTip, targetTip string) error {
if !e.Enabled() {
return fmt.Errorf("ci: no engine configured")
}
c := Checkin{Hash: Pair(branchTip, targetTip), Branch: branch, Head: branchTip}
pipelines, err := pipelinesAt(r, c)
if err != nil {
return err
}
return launch(e, st, repo, r, c, cloneURL, pipelines, scopeAll)
}
// launchPush starts a new check-in's pipelines. A check-in on a branch
// with an open landing is tested as its merge with the landing's target
// (RFC 0009 §once): that merge is what the landing tests and what the
// target will hold, so testing the bare tip as well was the same change
// run twice. A branch nobody has proposed is tested as its tip: it has
// no target to merge into, and a long-lived line (a release branch)
// must be tested as itself. A workflow with an environment: job, or one
// that names a secret, stays on the bare tip: a merge's tree is no
// reviewed commit and is never handed secrets. A merge that does not apply, or a target with no tip
// to read, is tested as the tip, never not at all.
func launchPush(e *Engine, st Store, repo string, r Repo, c Checkin, cloneURL string, pipelines definitions) error {
if c.Branch == "" || c.Branch == r.DefaultBranch() || len(pipelines.paths) == 0 {
return launch(e, st, repo, r, c, cloneURL, pipelines, scopePush)
}
target := r.LandingTarget(c.Branch)
if target == "" {
return launch(e, st, repo, r, c, cloneURL, pipelines, scopePush)
}
targetTip := r.BranchTip(target)
if targetTip == "" || targetTip == c.Hash {
return launch(e, st, repo, r, c, cloneURL, pipelines, scopePush)
}
state, conflicts, err := r.MergeState(c.Hash, targetTip)
if err != nil {
return transient(err)
}
subject := Checkin{Hash: Pair(c.Hash, targetTip), Branch: c.Branch, Date: c.Date, User: c.User,
Comment: c.Comment, Head: c.Hash}
merged := pipelines
if state == "conflict" || state == "unrelated" {
subject = c
subject.Note = fmt.Sprintf("tested as the branch tip: it does not merge into %s at %.12s (%s)", target, targetTip, state)
if len(conflicts) > 0 {
subject.Note += " in " + strings.Join(conflicts, ", ")
}
} else if merged, err = pipelinesAt(r, subject); err != nil {
return err
}
// What needs secrets is judged once, on the merged tree's definition
// (the tip's where the merge has none, or does not apply), so each
// pipeline runs on the merge or on the tip and never falls between.
secret := map[string]bool{}
for _, d := range []definitions{pipelines, merged} {
for _, p := range d.paths {
def, err := d.read(p)
if err != nil {
return transient(fmt.Errorf("read %s: %w", p, err))
}
secret[p] = reviewedOnly(def, strings.HasSuffix(p, ".json"))
}
}
var onTip, onPair []string
for _, p := range pipelines.paths {
if secret[p] {
onTip = append(onTip, p)
}
}
for _, p := range merged.paths {
if !secret[p] {
onPair = append(onPair, p)
}
}
err = launch(e, st, repo, r, subject, cloneURL, merged.only(onPair), scopePush)
if errors.Is(err, errTransient) {
return err // the whole check-in is tried again; nothing ran twice
}
err = errors.Join(err, launch(e, st, repo, r, c, cloneURL, pipelines.only(onTip), scopePush))
// launch ends a wait only for a subject it started something for.
if uerr := st.Undefer(repo, c.Hash); uerr != nil {
log.Printf("ci: %s %.12s: forgetting the wait: %v", repo, c.Hash, uerr)
}
return err
}
func launch(e *Engine, st Store, repo string, r Repo, c Checkin, cloneURL string, pipelines definitions, sc scope) error {
if len(pipelines.paths) == 0 {
return nil // no ci/ directory: the common case, silently fine
}
// A proposed branch's merge, pushed, is the landing's merge: it runs
// what the landing would, as well as what the push would.
proposed := c.Head != "" && sc == scopePush
var ran []string
// Only a re-run asked for runs what already has a verdict. A merge
// a landing asks for (scopeMissing) or a push's, which a landing
// opening in the meantime may already have tested, runs what has
// none; so does a tip, which its landing closing before the trigger
// reached it may already have tested (RunTip). A tag is its own
// event: the check-in's verdict is not the release's.
if sc != scopeAll && sc != scopeTag && sc != scopeSchedule {
var err error
if ran, err = st.RunPipelines(repo, c.Hash); err != nil {
return transient(err)
}
}
// A landing pipeline runs on the default branch and in a landing's
// merge (RunPairAt, scopeMissing): a proposed branch's checks are its
// merge, as a pull_request workflow runs on a pull request's test
// merge, so its tip does not run them a second time.
onLine := c.Branch != "" && c.Branch == r.DefaultBranch()
// A "default-branch" pipeline (a deploy) runs on the default branch
// and nowhere else, by any door: never in a merge, never on a branch
// tip, and a re-run by hand only of a default-branch check-in. That
// re-run carries no Branch (so it supersedes nothing), which is why
// the check-in's branch is looked up rather than read off c.
onDefault := func() bool {
if c.Head != "" {
return false // a pair: a proposal, not the line
}
if c.Branch != "" {
return onLine
}
return r.BranchOf(c.Hash) == r.DefaultBranch()
}
// A launch supersedes the builds of its own check-in's history and
// nothing else: never a descendant's (an older check-in arriving
// after its tip must not cancel the tip's run), never another
// fork's. A pair's run is judged by its branch tip.
head := c.Hash
if c.Head != "" {
head = c.Head
}
inLine := lineOf(r, head)
older := func(run Run) bool {
h := run.Labels["checkin"]
if tip, _, ok := splitPair(h); ok {
h = tip
}
return h != "" && inLine(h)
}
var heirs heir
var held *sync.Mutex
release := func() {
if held != nil {
held.Unlock()
held = nil
}
}
defer release()
// A content refusal is one pipeline's: the rest still start, and it
// is reported after them. A transient failure returns at once so the
// watermark holds and the whole check-in is tried again.
var refused []error
// A refused pipeline is recorded as a failing check where the
// check-in's and the landing's panels read checks, so a workflow
// quarry will not run is never mistaken for one with nothing to
// say ("All checks have passed"). A pair's lands on its branch tip,
// where external checks count for the pair.
refuse := func(p string, err error) {
refused = append(refused, err)
subject := c.Hash
if c.Head != "" {
subject = c.Head
}
if rerr := st.Refuse(repo, subject, p, err.Error()); rerr != nil {
log.Printf("ci: %s %.12s: recording %s's refusal: %v", repo, subject, p, rerr)
}
}
for _, p := range pipelines.paths {
if slices.Contains(ran, p) {
continue // a verdict is here or on its way; nothing to read
}
if (sc == scopeTag || sc == scopeSchedule) && strings.HasSuffix(p, ".json") {
continue // no on: to name a tag; quarry_when speaks of pushes and landings
}
def, err := pipelines.read(p)
if err != nil {
return transient(fmt.Errorf("read %s: %w", p, err))
}
isJSON := strings.HasSuffix(p, ".json")
if !isJSON && yamlQuarryWhen.Match(def) {
refuse(p, 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")))
continue
}
if isJSON {
when, err := quarryWhen(def)
if err != nil {
refuse(p, fmt.Errorf("%s: %w", p, err))
continue
}
if sc == scopePush && when == whenLanding && !onLine && !proposed {
continue
}
if when == whenDefaultBranch && !onDefault() {
continue
}
}
inherit, err := inheritOf(def, isJSON)
if err != nil {
refuse(p, fmt.Errorf("%s: %w", p, err))
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)
// The engine starts a run by name, so this launch's Register and
// Start are one step: another launch's Register between them
// would run its file under this launch's record.
release()
held = nameLock(name)
held.Lock()
var wf Workflow
var on json.RawMessage
if strings.HasSuffix(p, ".json") {
wf, err = e.RegisterNamed(name, def)
} else {
wf, on, err = e.RegisterYAML(name, def, GroupFor(repo))
}
if err != nil {
if engineUnreachable(err) {
return transient(err)
}
refuse(p, err) // the engine answered: this pipeline's refusal alone
continue
}
// A jobs: workflow says when it runs (RFC 0019 §on); the engine
// hands on: back uninterpreted, because only the forge knows what
// a landing or a re-run is.
if len(on) > 0 || sc == scopeTag || sc == scopeSchedule {
ok, err := runsOn(on, launchEvent(c, sc, r))
if err == nil && !ok && proposed {
ok, err = runsOn(on, launchEvent(c, scopeMissing, r))
}
if err != nil {
refuse(p, fmt.Errorf("%s: %w", p, err))
continue
}
if !ok {
continue
}
}
// The default branch's merge of a tested pair inherits the
// pair's verdict rather than running again (RFC 0009
// §inheritance) — never a workflow that deploys or names a
// secret, and never one that asks not to.
if onLine && sc == scopePush && c.Head == "" && inherit && !reviewedOnly(def, isJSON) {
if from, ok := heirs.verdict(st, repo, r, c, p, def); ok {
if err := st.Inherit(repo, c.Hash, wf.Name, p, from.Status, from.RunID); err != nil {
return transient(err)
}
log.Printf("ci: %s %.12s %s inherits run %s (%s)", repo, c.Hash, p, from.RunID, from.Status)
continue
}
}
labels := map[string]string{"repo": repo, "checkin": c.Hash, "pipeline": p}
if sc == scopeTag {
// No branch label: a push supersedes by branch, and a release
// in flight is not a stale build of its branch.
labels["subject"] = "tag"
labels["tag"] = c.Tag
} else if sc == scopeSchedule {
labels["subject"] = "schedule" // a push does not supersede it either
} else 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). That
// includes its pair runs, which judged the old tip, and a
// push's runs on its merge stop the same. A landing's pair
// run stops only older pair runs: the branch's own runs are
// still the verdict on the tip. A merge a push ran is the
// push's (push-pair), one the landing sweep ran the
// landing's (pair): a landing closing cancels only its own.
stale := map[string]string{"repo": repo, "branch": c.Branch, "pipeline": p}
if c.Head != "" {
labels["subject"] = "pair"
if sc == scopePush {
labels["subject"] = "push-pair"
} else {
stale["subject"] = "pair"
}
}
stales := []map[string]string{stale}
if stale["subject"] == "pair" {
// The push's merge of this tip is as stale as the
// landing's older ones once a newer merge runs.
pushed := maps.Clone(stale)
pushed["subject"] = "push-pair"
stales = append(stales, pushed)
}
for _, stale := range stales {
if n, err := e.Supersede(wf.Name, stale, older); err == nil && n > 0 {
log.Printf("ci: %s %s: superseded %d run(s)", repo, c.Branch, n)
}
}
}
input := map[string]any{
"repo": repo,
"checkin": c.Hash,
"branch": c.Branch,
"date": c.Date,
"user": c.User,
"comment": c.Comment,
"cloneUrl": cloneURL,
"pipeline": p,
}
if sc == scopeTag {
input["tag"] = c.Tag // a job's steps see it as QUARRY_TAG
}
runID, err := e.Start(wf.Name, input, labels)
release()
if err != nil {
if engineUnreachable(err) {
return transient(err)
}
refuse(p, err)
continue
}
if err := st.AddRun(runID, repo, c.Hash, wf.Name, p, DefHash(def), "running"); err != nil {
// The run is launched but unrecorded; losing the mapping
// by advancing would orphan it.
return transient(err)
}
start := RunStart{Event: launchEvent(c, sc, r).kind, Starter: c.Starter}
if sc != scopeAll && sc != scopeTag && sc != scopeSchedule {
start.Starter = pusherOf(st, repo, r, c.Hash)
}
if !isJSON {
start.Jobs, _ = permissionsOf(def) // RegisterYAML refused it already if it errs
}
if err := st.Started(runID, start); err != nil {
// The run goes on; without the record it is minted no token.
log.Printf("ci: %s %.12s: recording run %s's start: %v", repo, c.Hash, runID, err)
}
if c.Note != "" {
if err := st.Note(runID, c.Note); err != nil {
log.Printf("ci: %s %.12s: noting run %s: %v", repo, c.Hash, runID, 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.
// A tag's launch is not that door: the check-in still owes its push.
if sc == scopeTag || sc == scopeSchedule {
return errors.Join(refused...)
}
if err := st.Undefer(repo, c.Hash); err != nil {
log.Printf("ci: %s %.12s: forgetting the wait: %v", repo, c.Hash, err)
}
return errors.Join(refused...)
}