1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3package usecase
4
5import (
6 "context"
7 "errors"
8 "runtime/debug"
9 "strconv"
10 "time"
11
12 "bnema.dev/ermite/internal/domain"
13)
14
15const sweepEvery = 24 * time.Hour
16
17// advanceRun plans what can start in the latest attempt of a run.
18func (c *CI) advanceRun(ctx context.Context, runID int64) error {
19 return c.withRun(ctx, runID, 0, func(rs *runState) error {
20 if err := rs.advance(ctx); err != nil {
21 return err
22 }
23 return rs.flush(ctx)
24 })
25}
26
27// resume plans again the runs that were open when the server stopped: jobs
28// whose needs ended while it was down are planned now. Running jobs go on:
29// their tokens are in the store, and the heartbeat deadline counts from the
30// start of this process.
31func (c *CI) resume() {
32 ctx := c.ctx
33 runs, err := c.cfg.Store.ListOpenRuns(ctx)
34 if err != nil {
35 ctxLog(ctx).Error().Err(err).Msg("ci: list the open runs")
36 return
37 }
38 for _, r := range runs {
39 if err := c.advanceRun(ctx, r.ID); err != nil && !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
40 ctxLog(ctx).Error().Err(err).Int64("run_id", r.ID).Msg("ci: plan an open run again")
41 }
42 }
43 c.wakeQueue()
44}
45
46// watchdog looks at the jobs in flight every tick, and sweeps the old runs
47// once a day.
48func (c *CI) watchdog() {
49 defer c.wg.Done()
50 tick := time.NewTicker(c.cfg.Tick)
51 defer tick.Stop()
52 lastSweep := time.Time{}
53 for {
54 select {
55 case <-c.ctx.Done():
56 return
57 case <-tick.C:
58 }
59 c.guard("watchdog", func() {
60 c.checkActive(c.ctx)
61 c.checkQueue(c.ctx)
62 if now := c.now(); now.Sub(lastSweep) >= sweepEvery {
63 lastSweep = now
64 c.sweep(c.ctx)
65 }
66 })
67 }
68}
69
70// guard runs fn and logs a panic instead of letting it end the worker.
71func (c *CI) guard(what string, fn func()) {
72 defer func() {
73 if r := recover(); r != nil {
74 ctxLog(c.ctx).Error().Interface("panic", r).Bytes("stack", debug.Stack()).Str("worker", what).Msg("ci worker panicked")
75 }
76 }()
77 fn()
78}
79
80// checkActive applies the server's deadlines to the jobs given to runners:
81// acknowledgement, heartbeat, time budget, and the grace after a cancel.
82func (c *CI) checkActive(ctx context.Context) {
83 jobs, err := c.cfg.Store.ListActiveJobs(ctx)
84 if err != nil {
85 ctxLog(ctx).Error().Err(err).Msg("ci: list the active jobs")
86 return
87 }
88 now := c.now()
89 for _, j := range jobs {
90 if !c.due(j, now) {
91 continue
92 }
93 err := c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
94 cur := rs.find(j.ID)
95 if cur == nil || !cur.Active() {
96 return nil
97 }
98 if !c.enforce(ctx, rs, cur, c.now()) {
99 return nil
100 }
101 if err := rs.advance(ctx); err != nil {
102 return err
103 }
104 return rs.flush(ctx)
105 })
106 if err != nil && !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
107 ctxLog(ctx).Error().Err(err).Str("job_id", j.ID).Msg("ci: enforce a deadline")
108 }
109 }
110}
111
112// lastBeat is the time of the last sign of life of an active job: the later
113// of its last heartbeat, its acknowledgement and the start of the server.
114func (c *CI) lastBeat(j domain.CIJob) time.Time {
115 c.mu.Lock()
116 beat := c.heartbeats[jobRef(j)]
117 c.mu.Unlock()
118 for _, t := range []time.Time{j.AckedAt, c.boot} {
119 if t.After(beat) {
120 beat = t
121 }
122 }
123 return beat
124}
125
126// due is a cheap test, on the listed copy, of whether a deadline may have
127// passed.
128func (c *CI) due(j domain.CIJob, now time.Time) bool {
129 switch {
130 case j.Phase == domain.CIPhaseAssigned:
131 return now.Sub(c.assignedSince(j)) > c.cfg.AckTimeout
132 case j.CancelReason != domain.CICancelNone:
133 return now.Sub(j.CancelRequestedAt) > c.cfg.CancelGrace
134 }
135 return now.Sub(c.lastBeat(j)) > c.cfg.HeartbeatDeadline || now.After(c.deadline(j))
136}
137
138// assignedSince is when the acknowledgement of an assignment starts to count:
139// its assignment, or the start of this process when that is later, so that a
140// runner that could not reach a server that was down is not punished for it
141// (the same grace as the heartbeat deadline).
142func (c *CI) assignedSince(j domain.CIJob) time.Time {
143 if c.boot.After(j.AssignedAt) {
144 return c.boot
145 }
146 return j.AssignedAt
147}
148
149// deadline is when the job's time budget ends for the server: timeout-minutes
150// from the assignment, plus the slack that lets the runner's own timeout win.
151func (c *CI) deadline(j domain.CIJob) time.Time {
152 return j.AssignedAt.Add(time.Duration(j.TimeoutMinutes) * time.Minute).Add(c.cfg.TimeoutSlack)
153}
154
155// enforce applies the deadlines to a job read under the lock of its run and
156// reports whether it changed.
157func (c *CI) enforce(ctx context.Context, rs *runState, j *domain.CIJob, now time.Time) bool {
158 switch {
159 case j.Phase == domain.CIPhaseAssigned && now.Sub(c.assignedSince(*j)) > c.cfg.AckTimeout:
160 c.cancelAssignment(ctx, rs, j)
161 return true
162 case j.CancelReason != domain.CICancelNone && now.Sub(j.CancelRequestedAt) > c.cfg.CancelGrace:
163 conclusion := domain.CICancelled
164 if j.CancelReason == domain.CICancelTimeout {
165 conclusion = domain.CITimedOut
166 }
167 rs.end(j, conclusion, "The runner "+j.RunnerName+" did not answer the cancellation request in time; the job was ended by the server.")
168 return true
169 case j.Phase == domain.CIPhaseRunning && now.Sub(c.lastBeat(*j)) > c.cfg.HeartbeatDeadline:
170 rs.end(j, domain.CIFailure, domain.CILostMessage(j.RunnerName))
171 return true
172 case j.CancelReason == domain.CICancelNone && now.After(c.deadline(*j)):
173 rs.requestCancel(j, domain.CICancelTimeout, "The job exceeded its timeout of "+itoa(int64(j.TimeoutMinutes))+" minutes")
174 return true
175 }
176 return false
177}
178
179func itoa(n int64) string { return strconv.FormatInt(n, 10) }
180
181// cancelAssignment takes a job back from a runner that did not acknowledge
182// it: the token is revoked, what the assignment stored is dropped, and the job
183// is queued again with a fresh token at its next assignment. Three in a row
184// fail the job.
185func (c *CI) cancelAssignment(ctx context.Context, rs *runState, j *domain.CIJob) {
186 ctxLog(ctx).Warn().Str("job_id", j.ID).Str("runner", j.RunnerName).Msg("job assignment not acknowledged: cancelled")
187 _ = c.cfg.Logs.Reset(ctx, logRef(*j))
188 _ = c.cfg.Summaries.RemoveJob(ctx, logRef(*j))
189 rs.clear = append(rs.clear, logRef(*j))
190 j.AssignCancels++
191 j.TokenHash, j.TokenValidUntil = domain.TokenHash{}, time.Time{}
192 j.RunnerID, j.RunnerName, j.Isolation = "", "", domain.CIIsolationNone
193 j.Steps, j.Outputs, j.AssignedAt = nil, nil, time.Time{}
194 rs.revokeTokens(j) // the next assignment gets new ones
195 c.forgetPoll(jobRef(*j))
196 if j.CancelReason != domain.CICancelNone {
197 // Cancelled (by the owner, fail-fast or the time budget) before the
198 // runner answered: it is over, not queued again for another runner.
199 conclusion := domain.CICancelled
200 if j.CancelReason == domain.CICancelTimeout {
201 conclusion = domain.CITimedOut
202 }
203 rs.end(j, conclusion, "")
204 return
205 }
206 if j.AssignCancels >= domain.CIMaxAssignCancels {
207 rs.end(j, domain.CIFailure, "The job was given to a runner "+itoa(int64(domain.CIMaxAssignCancels))+" times and none of them acknowledged it.")
208 return
209 }
210 j.Phase, j.ReadyAt = domain.CIPhaseReady, c.now()
211 rs.touch(j)
212}
213
214// checkQueue fails the jobs that no runner took within the queue timeout.
215func (c *CI) checkQueue(ctx context.Context) {
216 ready, err := c.cfg.Store.ListReadyJobs(ctx, 0)
217 if err != nil {
218 return
219 }
220 now := c.now()
221 for _, j := range ready {
222 if now.Sub(j.ReadyAt) <= c.cfg.QueueTimeout {
223 continue
224 }
225 err := c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
226 cur := rs.find(j.ID)
227 if cur == nil || cur.Phase != domain.CIPhaseReady || c.now().Sub(cur.ReadyAt) <= c.cfg.QueueTimeout {
228 return nil
229 }
230 rs.end(cur, domain.CIFailure, domain.CIQueueTimeoutMessage)
231 if err := rs.advance(ctx); err != nil {
232 return err
233 }
234 return rs.flush(ctx)
235 })
236 if err != nil && !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
237 ctxLog(ctx).Error().Err(err).Str("job_id", j.ID).Msg("ci: end a job nobody took")
238 }
239 }
240}
241
242// sweep removes the runs older than the retention, and the logs of runs that
243// no longer exist (a repository deleted while the server was down). Every
244// failure is logged and the next step still runs: a log directory that cannot
245// be removed is tried again at the next sweep.
246func (c *CI) sweep(ctx context.Context) {
247 log := ctxLog(ctx)
248 ids, err := c.cfg.Store.Sweep(ctx, c.now().Add(-c.cfg.Retention))
249 if err != nil {
250 log.Error().Err(err).Msg("ci: sweep the old runs")
251 return
252 }
253 if len(ids) > 0 {
254 if err := c.cfg.Logs.RemoveRuns(ctx, ids); err != nil {
255 log.Error().Err(err).Int("runs", len(ids)).Msg("ci: remove the logs of the old runs")
256 }
257 if err := c.cfg.Summaries.RemoveRuns(ctx, ids); err != nil {
258 log.Error().Err(err).Int("runs", len(ids)).Msg("ci: remove the summaries of the old runs")
259 }
260 for _, id := range ids {
261 c.locks.Delete(id)
262 }
263 log.Info().Int("runs", len(ids)).Msg("ci: old runs removed")
264 }
265 c.removeOrphans(ctx, "logs", "log directories", c.cfg.Logs.RunIDs, c.cfg.Logs.RemoveRuns)
266 c.removeOrphans(ctx, "summaries", "summary directories", c.cfg.Summaries.RunIDs, c.cfg.Summaries.RemoveRuns)
267}
268
269// removeOrphans deletes what a file store (logs, summaries) holds for runs that
270// no longer exist, for example a repository deleted while the server was down.
271func (c *CI) removeOrphans(ctx context.Context, what, dirs string, list func(context.Context) ([]int64, error), remove func(context.Context, []int64) error) {
272 log := ctxLog(ctx)
273 onDisk, err := list(ctx)
274 if err != nil {
275 log.Error().Err(err).Msg("ci: list the " + dirs)
276 return
277 }
278 var orphans []int64
279 for _, id := range onDisk {
280 switch _, err := c.cfg.Store.GetRunSource(ctx, id); {
281 case errors.Is(err, domain.ErrNotFound):
282 orphans = append(orphans, id)
283 case err != nil:
284 log.Error().Err(err).Int64("run_id", id).Msg("ci: look up a run with a " + what + " directory")
285 }
286 }
287 if len(orphans) > 0 {
288 if err := remove(ctx, orphans); err != nil {
289 log.Error().Err(err).Int("runs", len(orphans)).Msg("ci: remove the " + what + " of runs that no longer exist")
290 }
291 }
292}
293
294// RunIDs lists the runs of a repository, for the cleanup that follows its
295// deletion.
296func (c *CI) RunIDs(ctx context.Context, repo domain.RepoName) ([]int64, error) {
297 return c.cfg.Store.RunIDs(ctx, repo)
298}
299
300// RemoveLogs deletes the logs and the summaries of runs of a repository that
301// was deleted.
302func (c *CI) RemoveLogs(ctx context.Context, runIDs []int64) error {
303 for _, id := range runIDs {
304 c.locks.Delete(id)
305 }
306 return errors.Join(c.cfg.Logs.RemoveRuns(ctx, runIDs), c.cfg.Summaries.RemoveRuns(ctx, runIDs))
307}