internal/store/builds.go
246 lines · 7211 bytes
1package store
2
3import (
4 "database/sql"
5 "errors"
6 "time"
7)
8
9// Build is one CI job execution for one commit.
10type Build struct {
11 ID int64
12 RepoID int64
13 Number int64
14 Job string
15 SHA string
16 Ref string
17 Steps string // JSON array of shell commands
18 Status string // pending|running|success|failure
19 CreatedAt string
20 StartedAt string
21 FinishedAt string
22}
23
24// MaxBuildLog caps a build's stored log; appends past it are dropped.
25const MaxBuildLog = 2 << 20
26
27// CreateBuild allocates the per-repo build number in the same transaction
28// as the insert, like issue and MR numbers.
29func (s *Store) CreateBuild(repoID int64, job, sha, ref, stepsJSON string) (int64, error) {
30 tx, err := s.DB.Begin()
31 if err != nil {
32 return 0, err
33 }
34 defer tx.Rollback()
35 if _, err := tx.Exec("UPDATE repos SET build_counter = build_counter + 1 WHERE id = ?", repoID); err != nil {
36 return 0, err
37 }
38 var n int64
39 if err := tx.QueryRow("SELECT build_counter FROM repos WHERE id = ?", repoID).Scan(&n); err != nil {
40 return 0, err
41 }
42 if _, err := tx.Exec(
43 "INSERT INTO builds (repo_id, number, job, sha, ref, steps) VALUES (?, ?, ?, ?, ?, ?)",
44 repoID, n, job, sha, ref, stepsJSON); err != nil {
45 return 0, err
46 }
47 return n, tx.Commit()
48}
49
50const buildSelect = `
51 SELECT id, repo_id, number, job, sha, ref, steps, status, created_at, started_at, finished_at
52 FROM builds`
53
54func scanBuild(row interface{ Scan(...any) error }) (Build, error) {
55 var b Build
56 err := row.Scan(&b.ID, &b.RepoID, &b.Number, &b.Job, &b.SHA, &b.Ref, &b.Steps,
57 &b.Status, &b.CreatedAt, &b.StartedAt, &b.FinishedAt)
58 return b, err
59}
60
61// ClaimBuild atomically hands the oldest pending build to a runner.
62func (s *Store) ClaimBuild() (Build, bool, error) {
63 tx, err := s.DB.Begin()
64 if err != nil {
65 return Build{}, false, err
66 }
67 defer tx.Rollback()
68 var id int64
69 err = tx.QueryRow("SELECT id FROM builds WHERE status = 'pending' ORDER BY id LIMIT 1").Scan(&id)
70 if errors.Is(err, sql.ErrNoRows) {
71 return Build{}, false, nil
72 }
73 if err != nil {
74 return Build{}, false, err
75 }
76 if _, err := tx.Exec(
77 "UPDATE builds SET status = 'running', started_at = strftime('%Y-%m-%dT%H:%M:%SZ','now') WHERE id = ?", id); err != nil {
78 return Build{}, false, err
79 }
80 b, err := scanBuild(tx.QueryRow(buildSelect+" WHERE id = ?", id))
81 if err != nil {
82 return Build{}, false, err
83 }
84 return b, true, tx.Commit()
85}
86
87// StaleBuildDeadline is how long a claimed build may stay running before the
88// server gives up on it. Comfortably longer than the runner's own -timeout
89// (45m by default), so this only fires when the runner never reported at all —
90// it was killed, restarted, or lost the network mid-build.
91const StaleBuildDeadline = 90 * time.Minute
92
93// ReapStaleBuilds fails every build that has been running past the deadline and
94// returns them, so the caller can resolve their commit statuses. A runner that
95// dies between claiming a build and reporting it otherwise leaves the row
96// claimed forever, and the commit pending forever with it.
97func (s *Store) ReapStaleBuilds() ([]Build, error) {
98 cutoff := time.Now().UTC().Add(-StaleBuildDeadline).Format("2006-01-02T15:04:05Z")
99 rows, err := s.DB.Query(buildSelect+
100 " WHERE status = 'running' AND started_at != '' AND started_at < ?", cutoff)
101 if err != nil {
102 return nil, err
103 }
104 defer rows.Close()
105 var stale []Build
106 for rows.Next() {
107 b, err := scanBuild(rows)
108 if err != nil {
109 return nil, err
110 }
111 stale = append(stale, b)
112 }
113 if err := rows.Err(); err != nil {
114 return nil, err
115 }
116 for _, b := range stale {
117 if err := s.AppendBuildLog(b.ID, []byte(
118 "\nbuild abandoned: the runner never reported an outcome\n")); err != nil {
119 return nil, err
120 }
121 if err := s.FinishBuild(b.ID, "failure"); err != nil {
122 return nil, err
123 }
124 }
125 return stale, nil
126}
127
128// AppendBuildLog adds a chunk to the build's log, dropping bytes past the cap.
129func (s *Store) AppendBuildLog(id int64, chunk []byte) error {
130 _, err := s.DB.Exec(`
131 UPDATE builds SET log = log || ?
132 WHERE id = ? AND length(log) < ?`, chunk, id, MaxBuildLog)
133 return err
134}
135
136// FinishBuild records the outcome of a running build.
137func (s *Store) FinishBuild(id int64, status string) error {
138 res, err := s.DB.Exec(`
139 UPDATE builds SET status = ?, finished_at = strftime('%Y-%m-%dT%H:%M:%SZ','now')
140 WHERE id = ? AND status = 'running'`, status, id)
141 if err != nil {
142 return err
143 }
144 if n, _ := res.RowsAffected(); n == 0 {
145 return ErrNotFound
146 }
147 return nil
148}
149
150func (s *Store) BuildByID(id int64) (Build, error) {
151 b, err := scanBuild(s.DB.QueryRow(buildSelect+" WHERE id = ?", id))
152 if errors.Is(err, sql.ErrNoRows) {
153 return b, ErrNotFound
154 }
155 return b, err
156}
157
158func (s *Store) BuildByNumber(repoID, number int64) (Build, error) {
159 b, err := scanBuild(s.DB.QueryRow(buildSelect+" WHERE repo_id = ? AND number = ?", repoID, number))
160 if errors.Is(err, sql.ErrNoRows) {
161 return b, ErrNotFound
162 }
163 return b, err
164}
165
166func (s *Store) ListBuilds(repoID int64, limit int) ([]Build, error) {
167 rows, err := s.DB.Query(buildSelect+" WHERE repo_id = ? ORDER BY number DESC LIMIT ?", repoID, limit)
168 if err != nil {
169 return nil, err
170 }
171 defer rows.Close()
172 var out []Build
173 for rows.Next() {
174 b, err := scanBuild(rows)
175 if err != nil {
176 return nil, err
177 }
178 out = append(out, b)
179 }
180 return out, rows.Err()
181}
182
183// BuildLog returns the stored log bytes.
184func (s *Store) BuildLog(id int64) ([]byte, error) {
185 var log []byte
186 err := s.DB.QueryRow("SELECT log FROM builds WHERE id = ?", id).Scan(&log)
187 if errors.Is(err, sql.ErrNoRows) {
188 return nil, ErrNotFound
189 }
190 return log, err
191}
192
193// LatestBuild returns the newest build for a repo, optionally narrowed to
194// one job. It is what a status badge reports.
195func (s *Store) LatestBuild(repoID int64, job string) (Build, error) {
196 q := buildSelect + " WHERE repo_id = ?"
197 args := []any{repoID}
198 if job != "" {
199 q += " AND job = ?"
200 args = append(args, job)
201 }
202 q += " ORDER BY number DESC LIMIT 1"
203 b, err := scanBuild(s.DB.QueryRow(q, args...))
204 if errors.Is(err, sql.ErrNoRows) {
205 return b, ErrNotFound
206 }
207 return b, err
208}
209
210// BuildsForCommit returns the newest build per job for one commit. A merge
211// request's checks are ci/<job> statuses; this is where their timing comes
212// from, in one query rather than one per check.
213func (s *Store) BuildsForCommit(repoID int64, sha string) (map[string]Build, error) {
214 rows, err := s.DB.Query(buildSelect+" WHERE repo_id = ? AND sha = ? ORDER BY number ASC", repoID, sha)
215 if err != nil {
216 return nil, err
217 }
218 defer rows.Close()
219 out := map[string]Build{}
220 for rows.Next() {
221 b, err := scanBuild(rows)
222 if err != nil {
223 return nil, err
224 }
225 out[b.Job] = b // ascending: the last row for a job wins
226 }
227 return out, rows.Err()
228}
229
230// Elapsed reports how long a build ran. Zero until it has both a start and
231// a finish, which is every state but success and failure.
232func (b Build) Elapsed() time.Duration {
233 const layout = "2006-01-02T15:04:05Z"
234 start, err := time.Parse(layout, b.StartedAt)
235 if err != nil {
236 return 0
237 }
238 end, err := time.Parse(layout, b.FinishedAt)
239 if err != nil {
240 return 0
241 }
242 if d := end.Sub(start); d > 0 {
243 return d.Round(time.Second)
244 }
245 return 0
246}