internal/store/builds.go

  1package store
  2
  3import (
  4	"database/sql"
  5	"errors"
  6	"os"
  7	"strconv"
  8	"strings"
  9	"time"
 10)
 11
 12// Build is one CI job execution for one commit.
 13type Build struct {
 14	ID         int64
 15	RepoID     int64
 16	Number     int64
 17	Job        string
 18	SHA        string
 19	Ref        string
 20	Steps      string // JSON array of shell commands
 21	Image      string // container image for the steps; "" means the runner default
 22	Tree       string // the commit's tree; "" when not deduplicated by tree
 23	Status     string // pending|running|success|failure
 24	CreatedAt  string
 25	StartedAt  string
 26	FinishedAt string
 27	// LogClosedAt is when the runner's log stream ended; "" while it is
 28	// open or was never opened. Set on a running build only.
 29	LogClosedAt string
 30	// Trusted is false for a merge request head fetched from another
 31	// repository: its steps run without the target's secrets.
 32	Trusted bool
 33	// FailedStep is the 1-based step a failed build stopped at, 0 when it
 34	// stopped before its first step or did not fail. FailedReason is the
 35	// runner's one line: "exit 1", "build timed out after 45m0s".
 36	FailedStep   int
 37	FailedReason string
 38}
 39
 40// MaxBuildLog caps a build's stored log; appends past it are dropped.
 41const MaxBuildLog = 2 << 20
 42
 43// truncNotice is appended once when a log first hits the cap. A log that
 44// simply stops is indistinguishable from a build that died mid-step, which
 45// is the reading that sent people hunting for a nonexistent test failure.
 46var truncNotice = []byte("\n[log truncated: reached the " +
 47	strconv.Itoa(MaxBuildLog>>20) + " MiB cap; earlier output is above]\n")
 48
 49// CreateBuild allocates the per-repo build number in the same transaction
 50// as the insert, like issue and MR numbers.
 51func (s *Store) CreateBuild(repoID int64, job, sha, ref, stepsJSON, image, tree string, trusted bool) (int64, error) {
 52	tx, err := s.DB.Begin()
 53	if err != nil {
 54		return 0, err
 55	}
 56	defer tx.Rollback()
 57	if _, err := tx.Exec("UPDATE repos SET build_counter = build_counter + 1 WHERE id = ?", repoID); err != nil {
 58		return 0, err
 59	}
 60	var n int64
 61	if err := tx.QueryRow("SELECT build_counter FROM repos WHERE id = ?", repoID).Scan(&n); err != nil {
 62		return 0, err
 63	}
 64	// A runner names a build by id in runner log and runner done, so an id
 65	// is never handed out twice, including after a repository's deletion
 66	// takes the newest builds with it (#306). builds is not AUTOINCREMENT
 67	// because rebuilding it would copy every stored log; the high-water
 68	// mark lives in settings instead.
 69	var id int64
 70	if err := tx.QueryRow(`SELECT MAX(
 71			COALESCE((SELECT MAX(id) FROM builds), 0),
 72			COALESCE((SELECT CAST(value AS INTEGER) FROM settings WHERE key = 'build_id_seq'), 0)) + 1`).
 73		Scan(&id); err != nil {
 74		return 0, err
 75	}
 76	if _, err := tx.Exec(
 77		"INSERT INTO builds (id, repo_id, number, job, sha, ref, steps, image, tree, trusted) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
 78		id, repoID, n, job, sha, ref, stepsJSON, image, tree, trusted); err != nil {
 79		return 0, err
 80	}
 81	if _, err := tx.Exec(`INSERT INTO settings (key, value) VALUES ('build_id_seq', ?)
 82		ON CONFLICT (key) DO UPDATE SET value = excluded.value`, strconv.FormatInt(id, 10)); err != nil {
 83		return 0, err
 84	}
 85	return n, tx.Commit()
 86}
 87
 88const buildSelect = `
 89	SELECT id, repo_id, number, job, sha, ref, steps, image, tree, status, created_at, started_at, finished_at, log_closed_at, trusted,
 90	       failed_step, failed_reason
 91	FROM builds`
 92
 93func scanBuild(row interface{ Scan(...any) error }) (Build, error) {
 94	var b Build
 95	var trusted int
 96	err := row.Scan(&b.ID, &b.RepoID, &b.Number, &b.Job, &b.SHA, &b.Ref, &b.Steps, &b.Image, &b.Tree,
 97		&b.Status, &b.CreatedAt, &b.StartedAt, &b.FinishedAt, &b.LogClosedAt, &trusted,
 98		&b.FailedStep, &b.FailedReason)
 99	b.Trusted = trusted != 0
100	return b, err
101}
102
103// ClaimBuild atomically hands the oldest pending build to a runner and
104// marks it running. A non-empty repoIDs restricts the claim to those
105// repositories. Untrusted builds — merge request heads from another
106// repository — are skipped unless untrusted is set: they run a stranger's
107// code, which only a runner that isolates should take.
108func (s *Store) ClaimBuild(repoIDs []int64, untrusted bool) (Build, bool, error) {
109	tx, err := s.DB.Begin()
110	if err != nil {
111		return Build{}, false, err
112	}
113	defer tx.Rollback()
114	query := "SELECT id FROM builds WHERE status = 'pending'"
115	args := []any{}
116	if !untrusted {
117		query += " AND trusted = 1"
118	}
119	if len(repoIDs) > 0 {
120		marks := strings.TrimSuffix(strings.Repeat("?,", len(repoIDs)), ",")
121		query += " AND repo_id IN (" + marks + ")"
122		for _, id := range repoIDs {
123			args = append(args, id)
124		}
125	}
126	query += " ORDER BY id LIMIT 1"
127	var id int64
128	err = tx.QueryRow(query, args...).Scan(&id)
129	if errors.Is(err, sql.ErrNoRows) {
130		return Build{}, false, nil
131	}
132	if err != nil {
133		return Build{}, false, err
134	}
135	if _, err := tx.Exec(
136		"UPDATE builds SET status = 'running', started_at = strftime('%Y-%m-%dT%H:%M:%SZ','now') WHERE id = ?", id); err != nil {
137		return Build{}, false, err
138	}
139	b, err := scanBuild(tx.QueryRow(buildSelect+" WHERE id = ?", id))
140	if err != nil {
141		return Build{}, false, err
142	}
143	return b, true, tx.Commit()
144}
145
146// StaleBuildDeadline is how long a claimed build may stay running before the
147// server gives up on it. Comfortably longer than the runner's own -timeout
148// (45m by default), so this only fires when the runner never reported at all —
149// it was killed, restarted, or lost the network mid-build.
150const StaleBuildDeadline = 90 * time.Minute
151
152// staleBuildDeadline is StaleBuildDeadline unless GITBAY_STALE_BUILD_DEADLINE
153// shortens it, which tests do.
154func staleBuildDeadline() time.Duration {
155	if v := os.Getenv("GITBAY_STALE_BUILD_DEADLINE"); v != "" {
156		if d, err := time.ParseDuration(v); err == nil && d > 0 {
157			return d
158		}
159	}
160	return StaleBuildDeadline
161}
162
163// StaleLogGrace is how long a running build may go on after its log
164// stream ended before it is treated as abandoned. The runner reports the
165// outcome right after closing the stream, retrying for about thirty
166// seconds if the server is unreachable; two minutes outlasts that.
167const StaleLogGrace = 2 * time.Minute
168
169// MarkBuildLogClosed records that the runner's log stream for a build
170// ended, on a build still running. A build that finishes normally is
171// reported moments later and the mark is moot; one that is not has lost
172// its runner, and ReapStaleBuilds fails it after StaleLogGrace rather
173// than at the deadline (#179).
174func (s *Store) MarkBuildLogClosed(id int64) error {
175	_, err := s.DB.Exec(`
176		UPDATE builds SET log_closed_at = strftime('%Y-%m-%dT%H:%M:%SZ','now')
177		WHERE id = ? AND status = 'running' AND log_closed_at = ''`, id)
178	return err
179}
180
181// ReapStaleBuilds fails every running build whose runner is gone and
182// returns them, so the caller can resolve their commit statuses: one whose
183// log stream ended more than StaleLogGrace ago with no outcome reported,
184// or one running past the deadline with no stream ever seen. A runner that
185// dies between claiming a build and reporting it otherwise leaves the row
186// claimed forever, and the commit pending forever with it.
187func (s *Store) ReapStaleBuilds() ([]Build, error) {
188	const layout = "2006-01-02T15:04:05Z"
189	now := time.Now().UTC()
190	cutoff := now.Add(-staleBuildDeadline()).Format(layout)
191	logCutoff := now.Add(-StaleLogGrace).Format(layout)
192	rows, err := s.DB.Query(buildSelect+
193		" WHERE status = 'running' AND ((started_at != '' AND started_at < ?)"+
194		" OR (log_closed_at != '' AND log_closed_at < ?))", cutoff, logCutoff)
195	if err != nil {
196		return nil, err
197	}
198	defer rows.Close()
199	var stale []Build
200	for rows.Next() {
201		b, err := scanBuild(rows)
202		if err != nil {
203			return nil, err
204		}
205		stale = append(stale, b)
206	}
207	if err := rows.Err(); err != nil {
208		return nil, err
209	}
210	for _, b := range stale {
211		if err := s.AppendBuildLog(b.ID, []byte(
212			"\nbuild abandoned: the runner never reported an outcome\n")); err != nil {
213			return nil, err
214		}
215		if err := s.FinishBuild(b.ID, "failure"); err != nil {
216			return nil, err
217		}
218		if _, err := s.DB.Exec(`UPDATE builds SET reaped_at = finished_at WHERE id = ?`, b.ID); err != nil {
219			return nil, err
220		}
221	}
222	return stale, nil
223}
224
225// QueueStats is the state of the build queue: what waits now, and over
226// the last day how long a build waited to be claimed and how many were
227// ended by the reaper rather than by a runner's report (#184).
228type QueueStats struct {
229	Pending       int64 `json:"pending"`
230	Claimed24h    int64 `json:"claimed_24h"`
231	ClaimWaitAvgS int64 `json:"claim_wait_avg_s"`
232	ClaimWaitMaxS int64 `json:"claim_wait_max_s"`
233	Reaped24h     int64 `json:"reaped_24h"`
234}
235
236func (s *Store) QueueStats() (QueueStats, error) {
237	var q QueueStats
238	since := time.Now().UTC().Add(-24 * time.Hour).Format("2006-01-02T15:04:05Z")
239	err := s.DB.QueryRow(`SELECT
240		(SELECT COUNT(*) FROM builds WHERE status = 'pending'),
241		COUNT(*),
242		CAST(COALESCE(AVG(strftime('%s', started_at) - strftime('%s', created_at)), 0) AS INTEGER),
243		CAST(COALESCE(MAX(strftime('%s', started_at) - strftime('%s', created_at)), 0) AS INTEGER),
244		(SELECT COUNT(*) FROM builds WHERE reaped_at >= ?)
245		FROM builds WHERE started_at >= ?`, since, since).
246		Scan(&q.Pending, &q.Claimed24h, &q.ClaimWaitAvgS, &q.ClaimWaitMaxS, &q.Reaped24h)
247	return q, err
248}
249
250// AppendBuildLog adds a chunk to the build's log, dropping bytes past the cap.
251func (s *Store) AppendBuildLog(id int64, chunk []byte) error {
252	res, err := s.DB.Exec(`
253		UPDATE builds SET log = log || ?
254		WHERE id = ? AND length(log) < ?`, chunk, id, MaxBuildLog)
255	if err != nil {
256		return err
257	}
258	if n, _ := res.RowsAffected(); n > 0 {
259		s.wakeBuild(id)
260		return nil
261	}
262	// Over the cap. The bounds match exactly once: appending the notice puts
263	// the log past the upper bound, so later chunks fall through silently.
264	_, err = s.DB.Exec(`
265		UPDATE builds SET log = log || ?
266		WHERE id = ? AND length(log) >= ? AND length(log) < ?`,
267		truncNotice, id, MaxBuildLog, MaxBuildLog+len(truncNotice))
268	if err == nil {
269		s.wakeBuild(id)
270	}
271	return err
272}
273
274// FinishBuild records the outcome of a running build.
275func (s *Store) FinishBuild(id int64, status string) error {
276	res, err := s.DB.Exec(`
277		UPDATE builds SET status = ?, finished_at = strftime('%Y-%m-%dT%H:%M:%SZ','now')
278		WHERE id = ? AND status = 'running'`, status, id)
279	if err != nil {
280		return err
281	}
282	if n, _ := res.RowsAffected(); n == 0 {
283		return ErrNotFound
284	}
285	s.wakeBuild(id)
286	return nil
287}
288
289// SetBuildFailure records where a running build failed. The runner
290// reports it with the outcome; it is written first, so a reader woken
291// by the finish sees both.
292func (s *Store) SetBuildFailure(id int64, step int, reason string) error {
293	res, err := s.DB.Exec(`UPDATE builds SET failed_step = ?, failed_reason = ?
294		WHERE id = ? AND status = 'running'`, step, reason, id)
295	if err != nil {
296		return err
297	}
298	if n, _ := res.RowsAffected(); n == 0 {
299		return ErrNotFound
300	}
301	return nil
302}
303
304func (s *Store) BuildByID(id int64) (Build, error) {
305	b, err := scanBuild(s.DB.QueryRow(buildSelect+" WHERE id = ?", id))
306	if errors.Is(err, sql.ErrNoRows) {
307		return b, ErrNotFound
308	}
309	return b, err
310}
311
312func (s *Store) BuildByNumber(repoID, number int64) (Build, error) {
313	b, err := scanBuild(s.DB.QueryRow(buildSelect+" WHERE repo_id = ? AND number = ?", repoID, number))
314	if errors.Is(err, sql.ErrNoRows) {
315		return b, ErrNotFound
316	}
317	return b, err
318}
319
320// BuildFilter narrows ListBuilds to builds matching every non-empty field.
321// Before is the keyset cursor: only builds numbered below it, which with
322// the newest-first order is the page after the one that ended there.
323type BuildFilter struct {
324	Ref    string
325	Status string
326	Job    string
327	Before int64
328}
329
330func (s *Store) ListBuilds(repoID int64, f BuildFilter, limit int) ([]Build, error) {
331	q := buildSelect + " WHERE repo_id = ?"
332	args := []any{repoID}
333	if f.Ref != "" {
334		q += " AND ref = ?"
335		args = append(args, f.Ref)
336	}
337	if f.Status != "" {
338		q += " AND status = ?"
339		args = append(args, f.Status)
340	}
341	if f.Job != "" {
342		q += " AND job = ?"
343		args = append(args, f.Job)
344	}
345	if f.Before > 0 {
346		q += " AND number < ?"
347		args = append(args, f.Before)
348	}
349	q += " ORDER BY number DESC LIMIT ?"
350	args = append(args, limit)
351	rows, err := s.DB.Query(q, args...)
352	if err != nil {
353		return nil, err
354	}
355	defer rows.Close()
356	var out []Build
357	for rows.Next() {
358		b, err := scanBuild(rows)
359		if err != nil {
360			return nil, err
361		}
362		out = append(out, b)
363	}
364	return out, rows.Err()
365}
366
367// BuildLog returns the stored log bytes.
368func (s *Store) BuildLog(id int64) ([]byte, error) {
369	var log []byte
370	err := s.DB.QueryRow("SELECT log FROM builds WHERE id = ?", id).Scan(&log)
371	if errors.Is(err, sql.ErrNoRows) {
372		return nil, ErrNotFound
373	}
374	return log, err
375}
376
377// BuildLogWait returns a channel closed by the next append to, finish of
378// or cancel of the build. Take it before reading, so a change between the
379// read and the wait still wakes the reader. Only this process's writes
380// wake it.
381func (s *Store) BuildLogWait(id int64) <-chan struct{} {
382	s.logMu.Lock()
383	defer s.logMu.Unlock()
384	if s.logWait == nil {
385		s.logWait = map[int64]chan struct{}{}
386	}
387	ch, ok := s.logWait[id]
388	if !ok {
389		ch = make(chan struct{})
390		s.logWait[id] = ch
391	}
392	return ch
393}
394
395func (s *Store) wakeBuild(id int64) {
396	s.logMu.Lock()
397	defer s.logMu.Unlock()
398	if ch, ok := s.logWait[id]; ok {
399		close(ch)
400		delete(s.logWait, id)
401	}
402}
403
404// BuildLogFrom returns the build's status and its log past offset bytes,
405// read together so a terminal status comes with every byte before it.
406// The cast matters: || stores the log as text, and substr on text counts
407// characters.
408func (s *Store) BuildLogFrom(id, offset int64) (string, []byte, error) {
409	var status string
410	var chunk []byte
411	err := s.DB.QueryRow(`SELECT status, substr(CAST(log AS BLOB), ?) FROM builds WHERE id = ?`,
412		offset+1, id).Scan(&status, &chunk)
413	if errors.Is(err, sql.ErrNoRows) {
414		return "", nil, ErrNotFound
415	}
416	return status, chunk, err
417}
418
419// LatestBuild returns the newest build for a repo, optionally narrowed to
420// one job. It is what a status badge reports.
421func (s *Store) LatestBuild(repoID int64, job string) (Build, error) {
422	q := buildSelect + " WHERE repo_id = ?"
423	args := []any{repoID}
424	if job != "" {
425		q += " AND job = ?"
426		args = append(args, job)
427	}
428	q += " ORDER BY number DESC LIMIT 1"
429	b, err := scanBuild(s.DB.QueryRow(q, args...))
430	if errors.Is(err, sql.ErrNoRows) {
431		return b, ErrNotFound
432	}
433	return b, err
434}
435
436// BuildsForCommit returns the newest build per job for one commit. A merge
437// request's checks are ci/<job> statuses; this is where their timing comes
438// from, in one query rather than one per check.
439func (s *Store) BuildsForCommit(repoID int64, sha string) (map[string]Build, error) {
440	rows, err := s.DB.Query(buildSelect+" WHERE repo_id = ? AND sha = ? ORDER BY number ASC", repoID, sha)
441	if err != nil {
442		return nil, err
443	}
444	defer rows.Close()
445	out := map[string]Build{}
446	for rows.Next() {
447		b, err := scanBuild(rows)
448		if err != nil {
449			return nil, err
450		}
451		out[b.Job] = b // ascending: the last row for a job wins
452	}
453	return out, rows.Err()
454}
455
456// Elapsed reports how long a build ran. Zero until it has both a start and
457// a finish, which is every state but success and failure.
458func (b Build) Elapsed() time.Duration {
459	const layout = "2006-01-02T15:04:05Z"
460	start, err := time.Parse(layout, b.StartedAt)
461	if err != nil {
462		return 0
463	}
464	end, err := time.Parse(layout, b.FinishedAt)
465	if err != nil {
466		return 0
467	}
468	if d := end.Sub(start); d > 0 {
469		return d.Round(time.Second)
470	}
471	return 0
472}
473
474// CancelBuild withdraws a queued or running build. A running one is
475// ended by the runner, which learns of the cancellation when its log
476// session is closed, and whose later report lands on a row that already
477// says cancelled.
478func (s *Store) CancelBuild(id int64) error {
479	res, err := s.DB.Exec(`UPDATE builds SET status = 'cancelled',
480		finished_at = strftime('%Y-%m-%dT%H:%M:%SZ','now') WHERE id = ? AND status IN ('pending', 'running')`, id)
481	if err != nil {
482		return err
483	}
484	if n, _ := res.RowsAffected(); n == 0 {
485		return ErrNotFound
486	}
487	s.wakeBuild(id)
488	return nil
489}
490
491// SuccessBuildForTree finds a passed build of the job for a tree rather
492// than a commit: a rebase that changes nothing in the tree has already
493// been built (#177). Only a trusted build on the image the job names
494// counts: a fork's result, or one from an image the job has left, does
495// not stand for the repository's own (#258). A job naming no image
496// matches builds that named none, whichever default the runner used;
497// the CI wiki page says so. An empty tree never matches.
498func (s *Store) SuccessBuildForTree(repoID int64, tree, job, image string) (Build, bool, error) {
499	if tree == "" {
500		return Build{}, false, nil
501	}
502	b, err := scanBuild(s.DB.QueryRow(buildSelect+
503		" WHERE repo_id = ? AND tree = ? AND job = ? AND image = ? AND trusted = 1 AND status = 'success'"+
504		" ORDER BY number DESC LIMIT 1", repoID, tree, job, image))
505	if errors.Is(err, sql.ErrNoRows) {
506		return Build{}, false, nil
507	}
508	return b, err == nil, err
509}
510
511// SuccessBuildFor finds a passed trusted build of the commit for the job,
512// on any ref: what a cancelled duplicate can point back at.
513func (s *Store) SuccessBuildFor(repoID int64, sha, job string) (Build, bool, error) {
514	b, err := scanBuild(s.DB.QueryRow(buildSelect+
515		" WHERE repo_id = ? AND sha = ? AND job = ? AND trusted = 1 AND status = 'success' ORDER BY number DESC LIMIT 1", repoID, sha, job))
516	if errors.Is(err, sql.ErrNoRows) {
517		return Build{}, false, nil
518	}
519	return b, err == nil, err
520}