// ci.go records which dagnats run belongs to which check-in.
//
// The run itself lives in dagnats — quarry does not duplicate its event
// log. What quarry must remember is the mapping, because dagnats knows
// only that some workflow ran with some input, and the forge is what
// knows that input was a check-in in a repository (0001 §6).
package oltp
import (
"database/sql"
"time"
)
const ciSchema = `
CREATE TABLE IF NOT EXISTS ci_run (
run_id TEXT PRIMARY KEY, -- dagnats run id
repo TEXT NOT NULL,
checkin TEXT NOT NULL,
pair TEXT NOT NULL DEFAULT '', -- premerge pair, RFC 0009
workflow TEXT NOT NULL,
pipeline TEXT NOT NULL, -- the ci/*.json path this came from
status TEXT NOT NULL, -- last status seen; dagnats remains authoritative
started TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS ci_run_by_checkin ON ci_run (repo, checkin);
CREATE INDEX IF NOT EXISTS ci_run_by_repo ON ci_run (repo, started DESC);
-- The watermark is how quarry avoids launching a run per restart: it
-- remembers the newest check-in it has already considered, per repo.
CREATE TABLE IF NOT EXISTS ci_watermark (
repo TEXT PRIMARY KEY,
checkin TEXT NOT NULL,
seen TEXT NOT NULL
);
-- The tags the CI trigger has considered (RFC 0020's tag event), by
-- key: a set rather than a watermark, because a tag arrives whenever its
-- author pushes it, whatever its date, and sometimes before its check-in.
-- The row with an empty key marks a repository the walk has looked at.
CREATE TABLE IF NOT EXISTS ci_tag_seen (
repo TEXT NOT NULL,
src TEXT NOT NULL,
checkin TEXT NOT NULL,
name TEXT NOT NULL,
seen TEXT NOT NULL,
PRIMARY KEY (repo, src, checkin, name)
);`
// ciMigrations widen older stores (RFC 0009 §premerge): a run may be
// about a pair of tips rather than one check-in, and a branch rule may
// ask for premerge testing.
var ciMigrations = []string{
`ALTER TABLE ci_run ADD COLUMN pair TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE branch_required_checks ADD COLUMN mode TEXT NOT NULL DEFAULT ''`,
// After the column exists, never beside CREATE TABLE: on a store
// that already has ci_run the CREATE is a no-op, so an index over
// `pair` in the schema block names a column this ALTER has not
// added yet and the open fails outright.
`CREATE INDEX IF NOT EXISTS ci_run_by_pair ON ci_run (repo, pair)`,
// The list shows durations (RFC 0009): terminal runs record when.
`ALTER TABLE ci_run ADD COLUMN completed TEXT NOT NULL DEFAULT ''`,
// Why a run was ended by hand or by expiry (#297); "" for a run that
// finished on its own.
`ALTER TABLE ci_run ADD COLUMN reason TEXT NOT NULL DEFAULT ''`,
// RFC 0009 §once: why a run is about the subject it is (a push
// tested as its tip because its merge does not apply), and the pair
// run a default-branch check-in's verdict was inherited from.
`ALTER TABLE ci_run ADD COLUMN note TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE ci_run ADD COLUMN inherited_from TEXT NOT NULL DEFAULT ''`,
// The hash of the definition a run started from, so a verdict is
// inherited only by a check-in holding the same definition.
`ALTER TABLE ci_run ADD COLUMN def_hash TEXT NOT NULL DEFAULT ''`,
}
// CIRun is one launched pipeline.
type CIRun struct {
RunID string `json:"runId"`
Repo string `json:"repo"`
Checkin string `json:"checkin"`
// Pair marks a premerge run: the verdict belongs to this pair of
// tips, never to a check-in (RFC 0009 §premerge). Empty for tip runs.
Pair string `json:"pair,omitempty"`
Workflow string `json:"workflow"`
Pipeline string `json:"pipeline"`
Status string `json:"status"`
Started string `json:"started"`
// Completed is set once, at the terminal transition; "" while the
// run is going, and forever for runs that predate the column.
Completed string `json:"completed,omitempty"`
// RerunOf and RerunAs are the run before and after this one for the
// same check-in and pipeline (#302): a launch that repeats a subject
// is a Re-run somebody asked for, so the chain is read off the
// recorded runs and needs no column of its own. "" at either end.
// Reason says why the forge ended the run: cancelled by an admin, or
// expired unclaimed (#297). "" for a run that finished on its own.
Reason string `json:"reason,omitempty"`
RerunOf string `json:"rerunOf,omitempty"`
RerunAs string `json:"rerunAs,omitempty"`
// Note says why the run is about its subject, when that needs
// saying. InheritedFrom names the pair run whose verdict this row
// is, on an equal tree; such a row ran nothing itself.
Note string `json:"note,omitempty"`
InheritedFrom string `json:"inheritedFrom,omitempty"`
// DefHash names the pipeline definition the run started from; ""
// for runs recorded before it was.
DefHash string `json:"defHash,omitempty"`
}
// AddCIRun records a launch.
func (d *DB) AddCIRun(r CIRun) error {
_, err := d.w.Exec(`INSERT OR REPLACE INTO ci_run
(run_id, repo, checkin, pair, workflow, pipeline, status, started, completed, def_hash)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, '', ?)`,
r.RunID, r.Repo, r.Checkin, r.Pair, r.Workflow, r.Pipeline, r.Status,
time.Now().UTC().Format(time.RFC3339), r.DefHash)
return err
}
// InheritCIRun records a check-in's verdict for a pipeline as the one a
// pair run reached on the same tree (RFC 0009 §inheritance). Its status
// is terminal, so the poller never asks the engine about it; it is
// completed once its verdict tag is written (InheritedOwingVerdict).
// Keyed by the source run and the check-in, so recording it again
// changes nothing.
func (d *DB) InheritCIRun(repo, checkin, workflow, pipeline, status, from string) error {
_, err := d.w.Exec(`INSERT OR IGNORE INTO ci_run
(run_id, repo, checkin, pair, workflow, pipeline, status, started, completed, inherited_from)
VALUES (?, ?, ?, '', ?, ?, ?, ?, '', ?)`,
from+"@"+checkin, repo, checkin, workflow, pipeline, status,
time.Now().UTC().Format(time.RFC3339), from)
return err
}
// InheritedOwingVerdict lists inherited verdicts whose tag is not
// written yet: the poller's second work list, beside UnfinishedCIRuns.
func (d *DB) InheritedOwingVerdict() ([]CIRun, error) {
return d.ciRuns(`WHERE inherited_from != '' AND completed = ''`)
}
// NoteCIRun records why a run is about its subject.
func (d *DB) NoteCIRun(runID, note string) error {
_, err := d.w.Exec(`UPDATE ci_run SET note = ? WHERE run_id = ?`, note, runID)
return err
}
// SetCIStatus updates the cached status of a run.
func (d *DB) SetCIStatus(runID, status string) error {
_, err := d.w.Exec(`UPDATE ci_run SET status = ? WHERE run_id = ?`, status, runID)
return err
}
// CompleteCIRun records the terminal transition: the final status and
// when the run finished. completed empty falls back to now, so a run
// whose engine reported no stamp still gets a duration.
//
// The run's tokens end in the same transaction (RFC 0021 §runs).
func (d *DB) CompleteCIRun(runID, status, completed string) error {
if completed == "" {
completed = time.Now().UTC().Format(time.RFC3339)
}
tx, err := d.w.Begin()
if err != nil {
return err
}
defer tx.Rollback()
if _, err := tx.Exec(`UPDATE ci_run SET status = ?, completed = ? WHERE run_id = ?`,
status, completed, runID); err != nil {
return err
}
if _, err := revokeRunTokensIn(tx, runID); err != nil {
return err
}
return tx.Commit()
}
// SetCIReason records why the forge is ending a run. The status stays as
// it is: the poller settles the run once the engine reports it over.
func (d *DB) SetCIReason(runID, reason string) error {
_, err := d.w.Exec(`UPDATE ci_run SET reason = ? WHERE run_id = ?`, reason, runID)
return err
}
// EndCIRun ends a run the engine never ran, in one step: the terminal
// status, the reason, and now as the finish. It touches only a run that
// is still going, so a verdict already in is never rewritten, and
// reports whether it ended one.
func (d *DB) EndCIRun(runID, status, reason string) (bool, error) {
tx, err := d.w.Begin()
if err != nil {
return false, err
}
defer tx.Rollback()
res, err := tx.Exec(`UPDATE ci_run SET status = ?, reason = ?, completed = ?
WHERE run_id = ? AND status NOT IN `+ciTerminal,
status, reason, time.Now().UTC().Format(time.RFC3339), runID)
if err != nil {
return false, err
}
if _, err := revokeRunTokensIn(tx, runID); err != nil {
return false, err
}
n, _ := res.RowsAffected()
return n > 0, tx.Commit()
}
// CIRunByID reads one run, bound to its repository — the run page's
// auth question is "is this run about a repo you may read".
func (d *DB) CIRunByID(repo, runID string) (CIRun, bool, error) {
rows, err := d.ciRuns(`WHERE repo = ? AND run_id = ?`, repo, runID)
if err != nil || len(rows) == 0 {
return CIRun{}, false, err
}
return rows[0], true, nil
}
// CIRunsForPair lists a premerge pair's runs, newest first.
func (d *DB) CIRunsForPair(repo, pair string) ([]CIRun, error) {
return d.ciRuns(`WHERE repo = ? AND pair = ? ORDER BY started DESC, rowid DESC`, repo, pair)
}
// LatestPairFor names the newest premerge pair judging a branch tip,
// whatever target tip it was merged into, or "" when none has run. A
// pair whose every run was cancelled judged nothing (a closing landing
// cancels its merge), so the newest is the newest run not cancelled. A
// range over the (repo, pair) index: every pair for the tip sorts
// between "pair:<tip>:" and "pair:<tip>;", ';' following ':'.
func (d *DB) LatestPairFor(repo, branchTip string) (string, error) {
rows, err := d.ciRuns(`WHERE repo = ? AND pair >= ? AND pair < ? AND status NOT IN ('cancelled','canceled')
ORDER BY started DESC, rowid DESC LIMIT 1`,
repo, "pair:"+branchTip+":", "pair:"+branchTip+";")
if err != nil || len(rows) == 0 {
return "", err
}
return rows[0].Pair, nil
}
// CIRunsForTip lists the runs of every premerge pair judging a branch
// tip, newest first: the same range LatestPairFor reads.
func (d *DB) CIRunsForTip(repo, branchTip string) ([]CIRun, error) {
return d.ciRuns(`WHERE repo = ? AND pair >= ? AND pair < ? ORDER BY started DESC, rowid DESC`,
repo, "pair:"+branchTip+":", "pair:"+branchTip+";")
}
// CIRunsFor lists the runs launched for one check-in, newest first.
func (d *DB) CIRunsFor(repo, checkin string) ([]CIRun, error) {
return d.ciRuns(`WHERE repo = ? AND checkin = ? ORDER BY started DESC, rowid DESC`, repo, checkin)
}
// CIRuns lists a repository's recent runs.
func (d *DB) CIRuns(repo string, limit int) ([]CIRun, error) {
if limit <= 0 {
limit = 50
}
return d.ciRuns(`WHERE repo = ? ORDER BY started DESC LIMIT `+itoa(limit), repo)
}
// CIRunsPage is CIRuns from an offset, newest first.
func (d *DB) CIRunsPage(repo string, limit, offset int) ([]CIRun, error) {
if limit <= 0 {
limit = 50
}
if offset < 0 {
offset = 0
}
return d.ciRuns(`WHERE repo = ? ORDER BY started DESC LIMIT `+itoa(limit)+` OFFSET `+itoa(offset), repo)
}
// ciTerminal is the status vocabulary a run stops moving in. ONE
// definition, because the poller's work list and the retention sweep
// are the same question asked from opposite sides: what is still
// running, and what will never run again. Two copies of this list would
// drift, and the drift would either strand a run nobody polls or prune
// one somebody is waiting on.
const ciTerminal = `('completed','succeeded','success','failed','cancelled','canceled','error')`
// UnfinishedCIRuns lists runs whose cached status is not terminal, so a
// poller knows what to ask dagnats about.
func (d *DB) UnfinishedCIRuns() ([]CIRun, error) {
return d.ciRuns(`WHERE status NOT IN ` + ciTerminal)
}
// CIRunKeep is how long a finished run's mapping is worth keeping in
// full. Beyond it only the newest run per subject survives — see
// PruneCIRuns for why that exception exists and why one window is
// enough.
const CIRunKeep = 90 * 24 * time.Hour
// PruneCIRuns drops finished runs older than the cutoff, keeping the
// newest one per subject, and reports how many went.
//
// Every CI launch inserted a row and nothing ever removed one, so a
// repository committing steadily accumulated rows for ever — in the
// OLTP database 0001 §8 describes as small by construction and
// litestream-replicates. This was the one table that broke that.
//
// Two rules, and both are load-bearing:
//
// An UNFINISHED run is never pruned, whatever its age. A row whose
// status is not terminal IS the poller's work list; deleting it strands
// a run whose verdict nobody will ever record. An ancient unfinished
// row is a bug to find, not a row to hide.
//
// The NEWEST finished run per subject survives the window. quarry does
// not store the run, only the mapping from run to check-in, and that
// mapping is what makes a verdict explicable: a ci-fail badge that
// cannot name the run that produced it is a verdict with no evidence
// behind it. So the badge stays traceable for as long as the check-in
// does, and what the window actually removes is the RETRIES — the
// several runs a check-in accumulates from re-runs and supersessions,
// which is where the unbounded growth was.
//
// Subject is (repo, checkin, pair), not (repo, checkin): a premerge run
// is about a pair of tips rather than one check-in (RFC 0009), so a
// check-in's own run and its premerge runs each keep their newest.
//
// One window, not two. A second window to eventually retire the last
// run per check-in would bound the residue, but the residue is one tiny
// row per check-in — the same order as the DAG the forge already holds,
// and tiny beside it — so the second window would buy nothing and cost
// a reader a second rule to hold.
func (d *DB) PruneCIRuns(before time.Time) (int, error) {
cut := before.UTC().Format(time.RFC3339)
// A row goes when it is finished, old, and NOT the newest finished
// run for its subject. Expressed as "started < the subject's newest
// start" rather than with a window function: the newest row is the
// one for which that is false, ties included, and the correlated
// lookup rides ci_run_by_checkin.
res, err := d.w.Exec(`DELETE FROM ci_run
WHERE status IN `+ciTerminal+`
AND started < ?
AND started < (SELECT max(o.started) FROM ci_run o
WHERE o.repo = ci_run.repo AND o.checkin = ci_run.checkin
AND o.pair = ci_run.pair AND o.status IN `+ciTerminal+`)`, cut)
if err != nil {
return 0, err
}
n, _ := res.RowsAffected()
return int(n), nil
}
func (d *DB) ciRuns(where string, args ...any) ([]CIRun, error) {
// rowid is launch order and is unmoved by the retention sweep, which
// started is not; the two subqueries ride ci_run_by_checkin.
const sibling = `(SELECT o.run_id FROM ci_run o WHERE o.repo = ci_run.repo AND o.checkin = ci_run.checkin
AND o.pair = ci_run.pair AND o.pipeline = ci_run.pipeline AND o.rowid `
rows, err := d.db.Query(`SELECT run_id, repo, checkin, pair, workflow, pipeline, status, started, completed, reason, note, inherited_from, def_hash,
`+sibling+`< ci_run.rowid ORDER BY o.rowid DESC LIMIT 1),
`+sibling+`> ci_run.rowid ORDER BY o.rowid LIMIT 1)
FROM ci_run `+where, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []CIRun
for rows.Next() {
var r CIRun
var prev, next sql.NullString
if err := rows.Scan(&r.RunID, &r.Repo, &r.Checkin, &r.Pair, &r.Workflow,
&r.Pipeline, &r.Status, &r.Started, &r.Completed, &r.Reason, &r.Note, &r.InheritedFrom, &r.DefHash, &prev, &next); err != nil {
return nil, err
}
r.RerunOf, r.RerunAs = prev.String, next.String
out = append(out, r)
}
return out, rows.Err()
}
// CIWatermark returns the newest check-in already considered for a repo.
func (d *DB) CIWatermark(repo string) (string, bool) {
var c string
err := d.db.QueryRow(`SELECT checkin FROM ci_watermark WHERE repo = ?`, repo).Scan(&c)
return c, err == nil
}
// SetCIWatermark records how far CI has looked.
func (d *DB) SetCIWatermark(repo, checkin string) error {
_, err := d.w.Exec(`INSERT OR REPLACE INTO ci_watermark (repo, checkin, seen)
VALUES (?, ?, ?)`, repo, checkin, time.Now().UTC().Format(time.RFC3339))
return err
}
// CITagKey names one tag card: the artifact, the check-in, the tag.
type CITagKey struct {
Src, Checkin, Name string
}
// CITagsSeen returns the tag keys already considered for a repo, and
// whether the walk has looked at the repo at all.
func (d *DB) CITagsSeen(repo string) (map[CITagKey]bool, bool, error) {
rows, err := d.db.Query(`SELECT src, checkin, name FROM ci_tag_seen WHERE repo = ?`, repo)
if err != nil {
return nil, false, err
}
defer rows.Close()
seen, primed := map[CITagKey]bool{}, false
for rows.Next() {
var k CITagKey
if err := rows.Scan(&k.Src, &k.Checkin, &k.Name); err != nil {
return nil, false, err
}
primed = true
if k != (CITagKey{}) {
seen[k] = true
}
}
return seen, primed, rows.Err()
}
// MarkCITagsSeen records keys as considered, and the repo as looked at.
func (d *DB) MarkCITagsSeen(repo string, keys ...CITagKey) error {
tx, err := d.w.Begin()
if err != nil {
return err
}
defer tx.Rollback()
now := time.Now().UTC().Format(time.RFC3339)
for _, k := range append(keys, CITagKey{}) {
if _, err := tx.Exec(`INSERT OR IGNORE INTO ci_tag_seen (repo, src, checkin, name, seen)
VALUES (?, ?, ?, ?, ?)`, repo, k.Src, k.Checkin, k.Name, now); err != nil {
return err
}
}
return tx.Commit()
}
func itoa(n int) string {
if n == 0 {
return "0"
}
var b [20]byte
i := len(b)
for n > 0 {
i--
b[i] = byte('0' + n%10)
n /= 10
}
return string(b[i:])
}