// 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 "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
);`
// 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 ''`,
}
// 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"`
}
// 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)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, '')`,
r.RunID, r.Repo, r.Checkin, r.Pair, r.Workflow, r.Pipeline, r.Status,
time.Now().UTC().Format(time.RFC3339))
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.
func (d *DB) CompleteCIRun(runID, status, completed string) error {
if completed == "" {
completed = time.Now().UTC().Format(time.RFC3339)
}
_, err := d.w.Exec(`UPDATE ci_run SET status = ?, completed = ? WHERE run_id = ?`,
status, completed, runID)
return err
}
// 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`, repo, pair)
}
// 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`, 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) {
rows, err := d.db.Query(`SELECT run_id, repo, checkin, pair, workflow, pipeline, status, started, completed
FROM ci_run `+where, args...)
if err != nil {
return nil, err
}
defer rows.Close()
var out []CIRun
for rows.Next() {
var r CIRun
if err := rows.Scan(&r.RunID, &r.Repo, &r.Checkin, &r.Pair, &r.Workflow,
&r.Pipeline, &r.Status, &r.Started, &r.Completed); err != nil {
return nil, err
}
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
}
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:])
}