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