// cideferred.go is the check-ins CI has NOT started yet, on purpose.
//
// The worker runs a few tasks at once and queues the rest, so a burst
// of branch pushes turned two-minute steps into fourteen for everyone,
// landings included. dmestas's rule (029c4730): when the worker is
// full, a check-in on a branch with no open landing waits here rather
// than in the worker's queue, and is launched when a slot frees or a
// landing opens for its branch. A landing therefore never queues
// behind churn, and nothing is dropped: the row is the promise.
package oltp
import (
"database/sql"
"errors"
)
const ciDeferredSchema = `
CREATE TABLE IF NOT EXISTS ci_deferred (
repo TEXT NOT NULL,
checkin TEXT NOT NULL,
branch TEXT NOT NULL DEFAULT '',
date TEXT NOT NULL DEFAULT '',
user TEXT NOT NULL DEFAULT '',
comment TEXT NOT NULL DEFAULT '',
since TEXT NOT NULL,
PRIMARY KEY (repo, checkin)
);`
// CIDeferred is a check-in waiting for a worker slot: the context its
// pipeline will receive when it runs, and when it started waiting.
type CIDeferred struct {
Repo string `json:"repo"`
Checkin string `json:"checkin"`
Branch string `json:"branch"`
Date string `json:"date"`
User string `json:"user"`
Comment string `json:"comment"`
Since string `json:"since"`
}
// DeferCI records a check-in as waiting. Recording it twice keeps the
// first wait, so a re-fold cannot push it to the back of the line.
func (d *DB) DeferCI(c CIDeferred) error {
if c.Repo == "" || c.Checkin == "" {
return errors.New("oltp: a deferred check-in needs a repository and a hash")
}
_, err := d.w.Exec(`INSERT OR IGNORE INTO ci_deferred (repo, checkin, branch, date, user, comment, since)
VALUES (?, ?, ?, ?, ?, ?, ?)`, c.Repo, c.Checkin, c.Branch, c.Date, c.User, c.Comment, now())
return err
}
const ciDeferredCols = `repo, checkin, branch, date, user, comment, since`
func scanCIDeferred(rows interface{ Scan(...any) error }) (CIDeferred, error) {
var c CIDeferred
err := rows.Scan(&c.Repo, &c.Checkin, &c.Branch, &c.Date, &c.User, &c.Comment, &c.Since)
return c, err
}
// CIDeferred lists a repository's waiting check-ins, longest wait first.
func (d *DB) CIDeferred(repo string) ([]CIDeferred, error) {
rows, err := d.db.Query(`SELECT `+ciDeferredCols+` FROM ci_deferred WHERE repo = ? ORDER BY since, checkin`, repo)
if err != nil {
return nil, err
}
defer rows.Close()
var out []CIDeferred
for rows.Next() {
c, err := scanCIDeferred(rows)
if err != nil {
return nil, err
}
out = append(out, c)
}
return out, rows.Err()
}
// CIDeferredRepos lists the repositories with a check-in waiting: the
// drain timer's whole question, answered from this table alone rather
// than by opening any store.
func (d *DB) CIDeferredRepos() ([]string, error) {
rows, err := d.db.Query(`SELECT DISTINCT repo FROM ci_deferred ORDER BY repo`)
if err != nil {
return nil, err
}
defer rows.Close()
var out []string
for rows.Next() {
var r string
if err := rows.Scan(&r); err != nil {
return nil, err
}
out = append(out, r)
}
return out, rows.Err()
}
// CIDeferredFor answers whether one check-in is waiting, for the pages
// that would otherwise show nothing at all for it.
func (d *DB) CIDeferredFor(repo, checkin string) (CIDeferred, bool, error) {
c, err := scanCIDeferred(d.db.QueryRow(`SELECT `+ciDeferredCols+` FROM ci_deferred WHERE repo = ? AND checkin = ?`, repo, checkin))
if errors.Is(err, sql.ErrNoRows) {
return CIDeferred{}, false, nil
}
return c, err == nil, err
}
// UndeferCI forgets a wait: the check-in's run has started, whoever
// started it.
func (d *DB) UndeferCI(repo, checkin string) error {
_, err := d.w.Exec(`DELETE FROM ci_deferred WHERE repo = ? AND checkin = ?`, repo, checkin)
return err
}