// 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"
"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
}
// 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"`
}
// 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 is quarry's addition when chunks from several steps travel
// in one drained object; the engine's per-step reads leave it "".
Step string `json:"step,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 step's chunks, optionally from the first failure
// (RFC 0009 §logs). follow is the caller's business over SSE; this
// reads the stored chunks one-shot.
func (e *Engine) Logs(runID, step string, afterSeq int, fromFailure bool) ([]LogChunk, error) {
chunks, err := e.logsPage(runID, step, 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, 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, afterSeq int, fromFailure bool) ([]LogChunk, error) {
u := e.base + "/runs/" + url.PathEscape(runID) + "/logs?step=" + url.QueryEscape(step)
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"`
}
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"`
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 []string) (WorkerToken, error) {
body, err := json.Marshal(map[string]any{"label": label, "task_type_prefixes": prefixes})
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
}
// 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
}
// 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.
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
body, err := json.Marshal(raw)
if err != nil {
return Workflow{}, err
}
return e.Register(body)
}
// 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.
func (e *Engine) RegisterYAML(name string, spec []byte) (Workflow, error) {
body, err := json.Marshal(map[string]any{"name": name, "spec": string(spec), "register": true})
if err != nil {
return Workflow{}, err
}
resp, err := e.post("/v1/ci/compile", body)
if err != nil {
return Workflow{}, err
}
defer resp.Body.Close()
if resp.StatusCode == http.StatusNotFound {
return Workflow{}, 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"`
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 {
if d.Line > 0 {
parts = append(parts, fmt.Sprintf("line %d: %s", d.Line, d.Message))
} else {
parts = append(parts, d.Message)
}
}
return Workflow{}, fmt.Errorf("ci: %s does not compile: %s", name, strings.Join(parts, "; "))
}
return Workflow{}, fmt.Errorf("ci: %s does not compile", name)
}
if resp.StatusCode >= 300 {
return Workflow{}, fmt.Errorf("ci: compile %q: status %d", name, resp.StatusCode)
}
return Workflow{Name: name}, 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
}
// 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.
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.
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
}
// 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")
}