// Package ci is 0001 §6's trigger and display: dagnats is the engine,
// quarry launches runs and shows them.
//
// The division matters. quarry does not define pipelines, schedule
// steps, retry, or run anything — a pipeline is TRACKED CONTENT
// (`ci/*.json` in the repository, dagnats' native workflow format), so
// the pipeline for a historical check-in is the pipeline that ran, and
// there is no forge-side pipeline store to drift from the code. Workers
// belong to the project, reached over dagnats' HTTP bridge in whatever
// language it likes.
//
// What quarry owns is narrow: notice a check-in, find its pipeline at
// THAT check-in, start the run, remember which check-in it was for, and
// show the result.
//
// This package is a data plane; the arch gate restricts its importers.
package ci
import (
"bytes"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"net/url"
"strconv"
"strings"
"sync"
"time"
)
// Dir is where a repository keeps its pipelines. Tracked content, read
// at the check-in under test.
const Dir = "ci/"
// Engine talks to dagnats' control plane.
type Engine struct {
base string
admin string
http *http.Client
// limits caches StepLimits: the run page asks on every render, and
// a workflow's retry policy changes only when it is re-registered.
// misses are the (name, hash) pairs already re-read once since.
mu sync.Mutex
limits map[string]listedLimits
limitsAt time.Time
limitsErr error
misses map[string]bool
}
// listedLimits is one listed workflow: its definition's hash and each
// step's attempt limit.
type listedLimits struct {
hash string
steps map[string]int
}
// SetAdminToken hands the client the engine's admin bearer
// (DAGNATS_BRIDGE_TOKEN), which the worker-token routes require.
// Empty means quarry holds no admin credential, and the token surface
// says so instead of guessing.
func (e *Engine) SetAdminToken(t string) { e.admin = t }
// New returns an engine client, or a disabled one when base is empty
// (no --dagnats configured), whose calls are no-ops.
func New(base string) *Engine {
return &Engine{base: strings.TrimRight(base, "/"),
http: &http.Client{Timeout: 10 * time.Second}}
}
// Enabled reports whether CI is configured at all.
func (e *Engine) Enabled() bool { return e != nil && e.base != "" }
// Workflow is the part of dagnats' workflow document quarry needs to
// read. The rest is passed through untouched: quarry is not a schema
// authority for someone else's format, and validating it here would
// mean rejecting pipelines dagnats accepts.
type Workflow struct {
Name string `json:"name"`
}
// Run is one dagnats run as its API reports it.
type Run struct {
RunID string `json:"run_id"`
Workflow string `json:"workflow_id"`
Labels map[string]string `json:"labels,omitempty"`
Status string `json:"status"`
Steps map[string]Step `json:"steps"`
CreatedAt string `json:"created_at"`
CompletedAt string `json:"completed_at,omitempty"`
// Input echoes what the run was started with — repo, checkin,
// cloneUrl — which is what a repro line needs (RFC 0009 §logs).
Input map[string]any `json:"input,omitempty"`
// DefHash names the workflow definition the run is pinned to
// (dagnats#637); a run from before it has none.
DefHash string `json:"def_hash,omitempty"`
}
// Step is one step's state within a run. The timestamps are dagnats#626
// (engine-stamped dispatch and terminal times); absent on an engine
// that predates it, and the page says nothing rather than zero.
type Step struct {
Status string `json:"status"`
Attempts int `json:"attempts"`
Error string `json:"error,omitempty"`
StartedAt string `json:"started_at,omitempty"`
CompletedAt string `json:"completed_at,omitempty"`
}
// LogChunk is one bounded piece of a step's output — dagnats#624's
// hot-lane contract: {seq, ts, stream: out|err|marker, data}. A
// marker chunk records an event (failed) rather than output.
type LogChunk struct {
Seq int `json:"seq"`
TS string `json:"ts"`
Stream string `json:"stream"`
Data string `json:"data"`
// Step and Attempt are quarry's additions when chunks from several
// steps, and several tries of one step, travel in one drained
// object; the engine's per-(step, attempt) reads leave them zero. A
// record drained before attempts were kept has Attempt 0.
Step string `json:"step,omitempty"`
Attempt int `json:"attempt,omitempty"`
}
// ErrNoLogs reports an engine that does not serve the log lane yet —
// dagnats#624 unshipped, or a binary that predates it. The page shows
// why the pane is empty instead of an empty pane.
var ErrNoLogs = errors.New("ci: the engine does not serve logs yet (dagnats#624)")
// Logs reads one attempt of one step, optionally from the first
// failure (RFC 0009 §logs). attempt is 1-based; 0 is the engine's
// default, the step's latest attempt. follow is the caller's business
// over SSE; this reads the stored chunks one-shot.
func (e *Engine) Logs(runID, step string, attempt, afterSeq int, fromFailure bool) ([]LogChunk, error) {
chunks, err := e.logsPage(runID, step, attempt, afterSeq, fromFailure)
if fromFailure && errors.Is(err, ErrNoLogs) {
// from=failure is strict (#624): a step that ended without a
// failure marker 404s. That is "no failure position", not "no
// log lane" — read the whole step instead.
return e.logsPage(runID, step, attempt, afterSeq, false)
}
return chunks, err
}
// logsPage is one GET against the #624 tail API. The shipped contract
// answers {chunks, next_cursor, eof} with base64 data; a pre-#624
// engine answers the RUN document for any sub-path, and an engine
// speaking this package's interim build answers a bare array. All
// three are read for what they are.
func (e *Engine) logsPage(runID, step string, attempt, afterSeq int, fromFailure bool) ([]LogChunk, error) {
u := e.base + "/runs/" + url.PathEscape(runID) + "/logs?step=" + url.QueryEscape(step)
if attempt > 0 {
u += "&attempt=" + strconv.Itoa(attempt)
}
if afterSeq > 0 {
u += "&cursor=" + strconv.Itoa(afterSeq)
}
if fromFailure {
u += "&from=failure"
}
req, err := http.NewRequest("GET", u, nil)
if err != nil {
return nil, err
}
resp, err := e.http.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusNotFound {
return nil, ErrNoLogs
}
if resp.StatusCode >= 300 {
return nil, fmt.Errorf("ci: logs %s: %d", runID, resp.StatusCode)
}
body, err := io.ReadAll(io.LimitReader(resp.Body, 8<<20))
if err != nil {
return nil, err
}
trimmed := bytes.TrimSpace(body)
if bytes.HasPrefix(trimmed, []byte("[")) {
var chunks []LogChunk
if err := json.Unmarshal(body, &chunks); err != nil {
return nil, err
}
return chunks, nil
}
var page struct {
Chunks []struct {
Seq uint64 `json:"seq"`
TS string `json:"ts"`
Stream string `json:"stream"`
Data []byte `json:"data"`
} `json:"chunks"`
}
if err := json.Unmarshal(body, &page); err != nil || !bytes.Contains(trimmed, []byte(`"chunks"`)) {
// A JSON object without a chunks key is the pre-#624 catch-all
// answering the run document: the honest "no logs yet".
return nil, ErrNoLogs
}
out := make([]LogChunk, 0, len(page.Chunks))
for _, c := range page.Chunks {
out = append(out, LogChunk{Seq: int(c.Seq), TS: c.TS, Stream: c.Stream, Data: string(c.Data)})
}
return out, nil
}
// Workers lists the engine's connected workers (dagnats#628).
type Worker struct {
WorkerID string `json:"worker_id"`
TaskTypes []string `json:"task_types"`
Language string `json:"language"`
Transport string `json:"transport"`
MaxTasks int `json:"max_tasks"`
Hostname string `json:"hostname,omitempty"`
Version string `json:"version,omitempty"`
// WorkerGroups are the groups this worker drains, empty for one on
// the ungrouped queue only. The engine records them at connect
// (dagnats#719, v0.0.20); an older engine reports none, which reads
// as ungrouped.
WorkerGroups []string `json:"worker_groups,omitempty"`
// LastSeen is the worker's last heartbeat. A registration older
// than workerStale is a worker that has since restarted.
LastSeen string `json:"last_seen,omitempty"`
}
// workerStale is how long a silent worker registration still counts
// toward capacity.
const workerStale = 90 * time.Second
func (e *Engine) Workers() ([]Worker, error) {
req, err := http.NewRequest("GET", e.base+"/v1/workers", nil)
if err != nil {
return nil, err
}
resp, err := e.http.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusNotFound {
return nil, fmt.Errorf("ci: the engine does not list workers yet (dagnats#628)")
}
if resp.StatusCode >= 300 {
return nil, fmt.Errorf("ci: workers: %d", resp.StatusCode)
}
var out []Worker
body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
if err != nil {
return nil, err
}
if err := json.Unmarshal(body, &out); err != nil {
var wrapped struct {
Workers []Worker `json:"workers"`
}
if werr := json.Unmarshal(body, &wrapped); werr == nil {
return wrapped.Workers, nil
}
return nil, err
}
return out, nil
}
// WorkerToken is one minted worker credential (dagnats#627). The
// bearer itself appears only in Mint's answer, exactly once.
type WorkerToken struct {
ID string `json:"id"`
Token string `json:"token,omitempty"`
Label string `json:"label"`
TaskTypePrefixes []string `json:"task_type_prefixes"`
// WorkerGroups scopes the token to those groups (dagnats#695).
// Empty means unscoped by group — the fleet-wide credential.
WorkerGroups []string `json:"worker_groups,omitempty"`
CreatedAt string `json:"created_at"`
CreatedBy string `json:"created_by,omitempty"`
RevokedAt string `json:"revoked_at,omitempty"`
}
// tokenRequest is one authenticated call on the engine's admin-gated
// token routes, with the two refusals worded for a settings page: an
// engine with no admin credential configured, and a wrong one here.
func (e *Engine) tokenRequest(method, pathname string, body []byte) (*http.Response, error) {
var rd io.Reader
if body != nil {
rd = bytes.NewReader(body)
}
req, err := http.NewRequest(method, e.base+pathname, rd)
if err != nil {
return nil, err
}
if body != nil {
req.Header.Set("Content-Type", "application/json")
}
if e.admin != "" {
req.Header.Set("Authorization", "Bearer "+e.admin)
}
resp, err := e.http.Do(req)
if err != nil {
return nil, err
}
switch resp.StatusCode {
case http.StatusServiceUnavailable:
resp.Body.Close()
return nil, fmt.Errorf("ci: the engine has no admin credential configured (DAGNATS_BRIDGE_TOKEN), so worker tokens are unavailable there")
case http.StatusUnauthorized:
resp.Body.Close()
return nil, fmt.Errorf("ci: the engine refused quarry's admin token (--dagnats-token)")
case http.StatusNotFound:
resp.Body.Close()
return nil, fmt.Errorf("ci: the engine does not manage worker tokens yet (dagnats#627)")
}
return resp, nil
}
// WorkerTokens lists the engine's minted worker tokens.
func (e *Engine) WorkerTokens() ([]WorkerToken, error) {
resp, err := e.tokenRequest("GET", "/v1/tokens", nil)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return nil, fmt.Errorf("ci: list tokens: %d", resp.StatusCode)
}
var out struct {
Tokens []WorkerToken `json:"tokens"`
}
if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&out); err != nil {
return nil, err
}
return out.Tokens, nil
}
// MintWorkerToken mints a scoped worker credential. The answer carries
// the bearer, shown once and never recoverable afterward.
func (e *Engine) MintWorkerToken(label string, prefixes, groups []string) (WorkerToken, error) {
req := map[string]any{"label": label, "task_type_prefixes": prefixes}
// Sent only when non-empty: absent means "unscoped by group", which
// is the fleet-wide credential, while a present-but-empty list is
// refused at mint so it can never silently widen to every group.
if len(groups) > 0 {
req["worker_groups"] = groups
}
body, err := json.Marshal(req)
if err != nil {
return WorkerToken{}, err
}
resp, err := e.tokenRequest("POST", "/v1/tokens", body)
if err != nil {
return WorkerToken{}, err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
msg, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
return WorkerToken{}, fmt.Errorf("ci: mint token: %s", strings.TrimSpace(string(msg)))
}
var tok WorkerToken
if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&tok); err != nil {
return WorkerToken{}, err
}
return tok, nil
}
// RevokeWorkerToken revokes one, immediately.
func (e *Engine) RevokeWorkerToken(id string) error {
resp, err := e.tokenRequest("DELETE", "/v1/tokens/"+url.PathEscape(id), nil)
if err != nil {
return err
}
resp.Body.Close()
if resp.StatusCode >= 300 {
return fmt.Errorf("ci: revoke token: %d", resp.StatusCode)
}
return nil
}
// Terminal reports whether a run has finished, either way. Anything not
// recognised counts as still running: reporting "done" for a state this
// code has not seen would be the worse mistake.
func (r Run) Terminal() bool {
switch r.Status {
case "completed", "succeeded", "success", "failed", "cancelled", "canceled", "error":
return true
}
return false
}
// Drains reports whether w takes work published to group.
func (w Worker) Drains(group string) bool {
for _, g := range w.WorkerGroups {
if g == group {
return true
}
}
return false
}
// Cancelled reports a run that was stopped rather than finished.
//
// The distinction matters because a cancelled run is not a verdict.
// quarry cancels runs ITSELF: a new check-in on a branch supersedes the
// builds still running against the older one, which is the right thing
// to do with work nobody is waiting for any more. Treating that as a
// failure writes "this check-in failed CI" onto a check-in that was
// never finished being tested, as a tag artifact that clones with the
// repository and stays there.
func (r Run) Cancelled() bool {
switch r.Status {
case "cancelled", "canceled":
return true
}
return false
}
// Unclaimed reports a run nothing has taken up: not over, no step
// running, and at least one still waiting for a worker. That is a run
// whose task type no worker serves — it never dispatches, and with no
// step timeout running it never ends.
func (r Run) Unclaimed() bool {
if r.Terminal() {
return false
}
waiting := false
for _, s := range r.Steps {
switch s.Status {
case "running":
return false
case "queued", "pending", "dispatched", "":
waiting = true
}
}
return waiting
}
// Cancel asks the engine to stop a run. The engine settles the run as
// cancelled; quarry hears of it on its next poll.
func (e *Engine) Cancel(runID string) error {
resp, err := e.post("/runs/"+url.PathEscape(runID)+"/cancel", nil)
if err != nil {
return err
}
resp.Body.Close()
if resp.StatusCode >= 300 {
return fmt.Errorf("ci: cancel %s: %d", runID, resp.StatusCode)
}
return nil
}
// Passed reports a successful terminal run.
func (r Run) Passed() bool {
switch r.Status {
case "completed", "succeeded", "success":
return true
}
return false
}
// Register publishes a workflow definition. dagnats is the authority on
// the document's validity, so the raw bytes go up as they were stored
// in the repository.
func (e *Engine) Register(def []byte) (Workflow, error) {
var wf Workflow
if err := json.Unmarshal(def, &wf); err != nil {
return Workflow{}, fmt.Errorf("ci: pipeline is not JSON: %w", err)
}
if wf.Name == "" {
return Workflow{}, fmt.Errorf("ci: pipeline has no name")
}
resp, err := e.post("/workflows", def)
if err != nil {
return Workflow{}, err
}
resp.Body.Close()
if resp.StatusCode >= 300 {
return Workflow{}, fmt.Errorf("ci: register %q: status %d", wf.Name, resp.StatusCode)
}
return wf, nil
}
// ScopedName is the name a repository's pipeline is registered under.
//
// A pipeline was registered under its file's stem alone -- ci/build.yml
// became "build" -- with nothing naming the repository. Two
// repositories with the same file name therefore shared ONE definition
// on the engine, and quarry re-registers the definition immediately
// before every run, so the last writer won and a run could execute the
// other repository's pipeline. Nothing caught it because only one
// repository here has CI at all.
//
// The trailing hash is what makes this safe rather than merely better.
// A repository name may contain dots (oltp.RepoName allows them) and so
// may a file stem, so "a.b" + "c" and "a" + "b.c" both join to
// "a.b.c" -- the same ambiguity dagnats documents for its own version
// keys, where a workflow named "orders.v" collides with "orders". A
// fixed-length digest of the exact pair removes it: two different pairs
// collide only if the digest does.
//
// The readable part is kept in front because this name is what appears
// in the engine's console, and a name nobody can read is a name nobody
// can debug.
// GroupFor is the worker group a repository's own runners drain, and
// the only group a token minted from its settings page may reach.
//
// It is the repository's name, because repository names are already
// unique on a forge — a repository name always wins — so the group is
// unique for free, with no second namespace to keep consistent.
//
// The safe set is deliberately narrow: letters, digits, dash and
// underscore, and nothing else. Two separate reasons, both load-bearing.
//
// A dot would make the group ambiguous with a dotted task type, which
// is what let a token holding the task type "a.b" drain group "b" of
// type "a" (dagnats#695). And "=" is the sentinel dagnats encodes a
// group with (task.{task}.={group}.{runID}); folding it away means no
// repository name can ever spell a group boundary itself, whatever the
// person who named the repository intended.
//
// A name that had to be folded carries a digest of the ORIGINAL, so two
// names that fold together do not land in one queue.
func GroupFor(repo string) string {
folded := []byte(repo)
changed := false
for i, c := range folded {
switch {
case c >= 'a' && c <= 'z', c >= 'A' && c <= 'Z', c >= '0' && c <= '9', c == '-', c == '_':
default:
folded[i] = '-'
changed = true
}
}
if !changed {
return repo
}
sum := sha256.Sum256([]byte(repo))
return string(folded) + "-" + hex.EncodeToString(sum[:])[:6]
}
func ScopedName(repo, stem string) string {
sum := sha256.Sum256([]byte(repo + "\n" + stem))
return repo + "." + stem + "." + hex.EncodeToString(sum[:])[:6]
}
// RegisterNamed registers a JSON pipeline under a name quarry chooses
// rather than the one inside the document.
//
// The name is replaced in a generic map rather than through the
// Workflow struct, because the struct captures only what quarry reads
// and re-marshalling through it would silently drop every field it does
// not know about -- which is most of a pipeline.
func (e *Engine) RegisterNamed(name string, def []byte) (Workflow, error) {
var raw map[string]any
if err := json.Unmarshal(def, &raw); err != nil {
return Workflow{}, fmt.Errorf("ci: pipeline is not JSON: %w", err)
}
if raw["name"] == nil || raw["name"] == "" {
return Workflow{}, fmt.Errorf("ci: pipeline has no name")
}
raw["name"] = name
attemptsToRetries(raw)
body, err := json.Marshal(raw)
if err != nil {
return Workflow{}, err
}
return e.Register(body)
}
// attemptsToRetries translates a pipeline's retry policies from what
// its author writes to what dagnats counts (#303). An author's
// max_attempts is the total attempts, so max_attempts: 2 runs twice;
// dagnats' max_attempts counts retries after the first run, so the
// value handed over is one less, and 0 and 1 both mean run once. The
// workflow's default_retry and each step's retry are translated; the
// legacy per-step "retries" count is dagnats' own and stays as written.
func attemptsToRetries(raw map[string]any) {
translate := func(v any) {
p, ok := v.(map[string]any)
if !ok {
return
}
if n, ok := p["max_attempts"].(float64); ok {
p["max_attempts"] = max(0, n-1)
}
}
translate(raw["default_retry"])
steps, _ := raw["steps"].([]any)
for _, st := range steps {
if m, ok := st.(map[string]any); ok {
translate(m["retry"])
}
}
}
// RegisterYAML compiles and registers a ci.yml pipeline at the engine
// (dagnats#633). Core dagnats stays spec-agnostic — POST /workflows
// only ever takes its native JSON — so YAML enters through the CI
// door, /v1/ci/compile, and the engine stays the schema authority:
// its diagnostics come back as the error, worded as it wrote them.
//
// namespace is the
// repository's worker group: a jobs: workflow (RFC 0019, dagnats#728)
// compiles each job to task <namespace>.job in that group, and returns
// its on: block, normalised and uninterpreted, for the trigger to judge.
// A checks: spec ignores the namespace and returns no on.
func (e *Engine) RegisterYAML(name string, spec []byte, namespace string) (Workflow, json.RawMessage, error) {
body, err := json.Marshal(map[string]any{"name": name, "spec": string(spec), "register": true, "task_namespace": namespace})
if err != nil {
return Workflow{}, nil, err
}
resp, err := e.post("/v1/ci/compile", body)
if err != nil {
return Workflow{}, nil, err
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusNotFound {
return Workflow{}, nil, fmt.Errorf("ci: the engine does not compile ci.yml yet (dagnats#633)")
}
if resp.StatusCode == http.StatusUnprocessableEntity {
var out struct {
Diagnostics []struct {
Line int `json:"line"`
Column int `json:"column"`
Message string `json:"message"`
} `json:"diagnostics"`
}
if derr := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&out); derr == nil && len(out.Diagnostics) > 0 {
parts := make([]string, 0, len(out.Diagnostics))
for _, d := range out.Diagnostics {
switch {
case d.Line > 0 && d.Column > 0:
parts = append(parts, fmt.Sprintf("line %d, column %d: %s", d.Line, d.Column, d.Message))
case d.Line > 0:
parts = append(parts, fmt.Sprintf("line %d: %s", d.Line, d.Message))
default:
parts = append(parts, d.Message)
}
}
return Workflow{}, nil, fmt.Errorf("ci: %s does not compile: %s", name, strings.Join(parts, "; "))
}
return Workflow{}, nil, fmt.Errorf("ci: %s does not compile", name)
}
if resp.StatusCode >= 300 {
return Workflow{}, nil, fmt.Errorf("ci: compile %q: status %d", name, resp.StatusCode)
}
var out struct {
On json.RawMessage `json:"on"`
}
if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&out); err != nil {
return Workflow{}, nil, fmt.Errorf("ci: compile %q: reading the answer: %v", name, err)
}
return Workflow{Name: name}, out.On, nil
}
// Start launches a run, handing the check-in context to the pipeline as
// input: what to build, where it came from, and how to fetch it.
func (e *Engine) Start(workflow string, input map[string]any, labels map[string]string) (string, error) {
doc := map[string]any{"workflow": workflow, "input": input}
if len(labels) > 0 {
doc["labels"] = labels
}
body, err := json.Marshal(doc)
if err != nil {
return "", err
}
resp, err := e.post("/runs", body)
if err != nil {
return "", err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return "", fmt.Errorf("ci: start %q: status %d", workflow, resp.StatusCode)
}
var out struct {
RunID string `json:"run_id"`
}
if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&out); err != nil {
return "", err
}
return out.RunID, nil
}
// stepLimitsTTL is how long a read of the engine's workflows answers
// StepLimits before it is read again; stepLimitsErrTTL is how long a
// failed read is remembered, so an unreachable engine is not asked on
// every render.
const (
stepLimitsTTL = time.Minute
stepLimitsErrTTL = 10 * time.Second
)
// StepLimits is how many attempts each step of a run may take, the
// first try included (#283). dagnats retries while a step's attempts
// are at most its policy's max_attempts, and resolves the policy as the
// step's retry, then the workflow's default_retry, then the legacy
// retries count; no policy is one attempt.
//
// The engine lists only the latest definition of each workflow, while
// a run is pinned to the one it started under (defHash). Limits are
// answered only when the two are the same definition; otherwise — an
// older definition, a run with no hash, an unknown workflow — the
// answer is empty and the caller must not guess. A hash the cache does
// not know is re-read once, since the run may have started under a
// definition registered after the last read.
func (e *Engine) StepLimits(workflow, defHash string) (map[string]int, error) {
if defHash == "" {
return nil, nil
}
e.mu.Lock()
fresh := e.limits != nil && time.Since(e.limitsAt) <= stepLimitsTTL
if e.limitsErr != nil && time.Since(e.limitsAt) <= stepLimitsErrTTL {
err := e.limitsErr
e.mu.Unlock()
return nil, err
}
key := workflow + "\x00" + defHash
if fresh {
if l := e.limits[workflow]; l.hash == defHash {
e.mu.Unlock()
return l.steps, nil
}
if e.misses[key] {
e.mu.Unlock()
return nil, nil
}
}
e.mu.Unlock()
all, err := e.readStepLimits()
e.mu.Lock()
defer e.mu.Unlock()
e.limitsAt = time.Now()
if err != nil {
e.limitsErr = err
return nil, err
}
if !fresh {
e.misses = nil
}
e.limits, e.limitsErr = all, nil
if l := all[workflow]; l.hash == defHash {
return l.steps, nil
}
if e.misses == nil {
e.misses = map[string]bool{}
}
e.misses[key] = true
return nil, nil
}
func (e *Engine) readStepLimits() (map[string]listedLimits, error) {
req, err := http.NewRequest("GET", e.base+"/workflows", nil)
if err != nil {
return nil, err
}
resp, err := e.http.Do(req)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return nil, fmt.Errorf("ci: workflows: %d", resp.StatusCode)
}
type policy struct {
MaxAttempts int `json:"max_attempts"`
}
var defs []struct {
Name string `json:"name"`
DefHash string `json:"def_hash"`
DefaultRetry *policy `json:"default_retry"`
Steps []struct {
ID string `json:"id"`
Retries int `json:"retries"`
Retry *policy `json:"retry"`
} `json:"steps"`
}
if err := json.NewDecoder(io.LimitReader(resp.Body, 8<<20)).Decode(&defs); err != nil {
return nil, err
}
out := make(map[string]listedLimits, len(defs))
for _, d := range defs {
m := make(map[string]int, len(d.Steps))
for _, st := range d.Steps {
retries := 0
switch {
case st.Retry != nil:
retries = st.Retry.MaxAttempts
case d.DefaultRetry != nil:
retries = d.DefaultRetry.MaxAttempts
default:
retries = st.Retries
}
m[st.ID] = 1 + max(0, retries)
}
out[d.Name] = listedLimits{hash: d.DefHash, steps: m}
}
return out, nil
}
// Status fetches one run.
func (e *Engine) Status(runID string) (Run, error) {
req, err := http.NewRequest("GET", e.base+"/runs/"+url.PathEscape(runID), nil)
if err != nil {
return Run{}, err
}
resp, err := e.http.Do(req)
if err != nil {
return Run{}, err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return Run{}, fmt.Errorf("ci: status %s: %d", runID, resp.StatusCode)
}
var r Run
if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&r); err != nil {
return Run{}, err
}
return r, nil
}
func (e *Engine) post(pathname string, body []byte) (*http.Response, error) {
req, err := http.NewRequest("POST", e.base+pathname, bytes.NewReader(body))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/json")
return e.http.Do(req)
}
// ValidateYAML asks the engine whether a ci.yml compiles, without
// registering anything (dagnats#633). The diagnostics come back
// worded as the engine wrote them, line numbers included; an engine
// predating the CI door is named.
func (e *Engine) ValidateYAML(name string, spec []byte) ([]string, error) {
body, err := json.Marshal(map[string]any{"name": name, "spec": string(spec)})
if err != nil {
return nil, err
}
resp, err := e.post("/v1/ci/validate", body)
if err != nil {
return nil, err
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusNotFound {
return nil, fmt.Errorf("ci: the engine does not compile ci.yml yet (dagnats#633)")
}
if resp.StatusCode >= 300 {
return nil, fmt.Errorf("ci: validate: status %d", resp.StatusCode)
}
var out struct {
Diagnostics []struct {
Line int `json:"line"`
Message string `json:"message"`
} `json:"diagnostics"`
}
if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<20)).Decode(&out); err != nil {
return nil, err
}
var diags []string
for _, d := range out.Diagnostics {
if d.Line > 0 {
diags = append(diags, fmt.Sprintf("line %d: %s", d.Line, d.Message))
} else {
diags = append(diags, d.Message)
}
}
return diags, nil
}
// Supersede cancels still-running runs of a workflow that carry every
// given label — the launch path calls it so a new check-in on a branch
// stops the branch's now-stale builds (dagnats#629). Matching is
// verified against each run's own labels before anything is cancelled:
// an engine predating #629 ignores label filters and echoes no labels
// back, and reading that as "cancel everything" would stop other
// branches' builds. Such an engine supersedes nothing, silently.
func (e *Engine) Supersede(workflow string, labels map[string]string) (int, error) {
// The engine's list is a bounded sample filtered AFTER the cap, so a
// small limit can miss the very runs the labels name once the store
// grows. Ask for the ceiling: exact below it, best-effort above.
q := url.Values{"workflow": {workflow}, "limit": {"10000"}}
for k, v := range labels {
q.Add("label", k+"="+v)
}
req, err := http.NewRequest("GET", e.base+"/runs?"+q.Encode(), nil)
if err != nil {
return 0, err
}
resp, err := e.http.Do(req)
if err != nil {
return 0, err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return 0, fmt.Errorf("ci: list runs: %d", resp.StatusCode)
}
var runs []Run
if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&runs); err != nil {
return 0, err
}
n := 0
for _, r := range runs {
if r.Terminal() || !labelsCover(r.Labels, labels) {
continue
}
c, err := e.post("/runs/"+url.PathEscape(r.RunID)+"/cancel", nil)
if err != nil {
continue // best-effort: the run may have finished in the race
}
c.Body.Close()
if c.StatusCode < 300 {
n++
}
}
return n, nil
}
// labelsCover reports whether got carries every wanted label.
// freshQueue is how long a queued step counts as waiting for THESE
// workers. A step queued longer than this is waiting for a task type no
// worker serves (dantest-sandbox's html/puzzles runs have sat queued
// for days), and counting it would call the workers full forever.
const freshQueue = 15 * time.Minute
// Occupied counts the steps holding, or about to take, a worker slot:
// every step the engine reports running, plus steps queued on runs
// created within freshQueue. The worker's max_tasks bounds concurrent
// STEPS, so this is the number to compare with it — not the count of
// running runs, which includes runs queued forever for task types no
// worker serves (029c4730, measured on 7a5d179697db: 13 "running" runs
// against a capacity of 3, with one step actually running), nor any run
// that is not CI at all.
func (e *Engine) Occupied(now time.Time) (int, error) {
req, err := http.NewRequest("GET", e.base+"/runs?status=running&limit=10000", nil)
if err != nil {
return 0, err
}
resp, err := e.http.Do(req)
if err != nil {
return 0, err
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
return 0, fmt.Errorf("ci: list running runs: %d", resp.StatusCode)
}
var runs []Run
if err := json.NewDecoder(io.LimitReader(resp.Body, 16<<20)).Decode(&runs); err != nil {
return 0, err
}
n := 0
for _, r := range runs {
// Only CI runs hold a CI worker. The engine is shared: the
// notification service's digest workflows (quarry-activity) sit
// "running" for hours with a step running, and counting them
// called an idle worker full (measured on 6db5c9d71693: five
// such steps, zero CI steps). Every run quarry launches carries
// a pipeline label; nothing else does.
if r.Terminal() || r.Labels["pipeline"] == "" {
continue
}
fresh := false
if t, err := time.Parse(time.RFC3339Nano, r.CreatedAt); err == nil {
fresh = now.Sub(t) < freshQueue
}
for _, s := range r.Steps {
switch s.Status {
case "running":
n++
case "queued", "pending", "dispatched":
if fresh {
n++
}
}
}
}
return n, nil
}
func labelsCover(got, want map[string]string) bool {
for k, v := range want {
if got[k] != v {
return false
}
}
return true
}
// IsPipeline reports whether a listed path is one of a repository's
// pipeline definitions. Only files directly under ci/ count — a nested
// path is somebody's fixture, not a pipeline, and launching it would be
// surprising.
// RunnerManifest is the file in Dir that tells quarry runner what each
// task type runs. It is never a pipeline.
const RunnerManifest = "runner.json"
func IsPipeline(name string) bool {
if !strings.HasPrefix(name, Dir) {
return false
}
rest := name[len(Dir):]
if rest == "" || strings.Contains(rest, "/") {
return false // nested files are fixtures, not pipelines
}
// The runner's manifest (quarry runner, f24d78d9) lives beside the
// pipelines and is JSON too, but it is not a workflow: registering
// it failed every trigger with "pipeline has no name".
if rest == RunnerManifest {
return false
}
// JSON is dagnats' native workflow format; ci.yml is the spec the
// engine's CI door compiles (dagnats#633).
return strings.HasSuffix(rest, ".json") || strings.HasSuffix(rest, ".yml") || strings.HasSuffix(rest, ".yaml")
}