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 j.AssignCancels++
189 j.TokenHash, j.TokenValidUntil = domain.TokenHash{}, time.Time{}
190 j.RunnerID, j.RunnerName, j.Isolation = "", "", domain.CIIsolationNone
191 j.Steps, j.Outputs, j.AssignedAt = nil, nil, time.Time{}
192 c.forgetPoll(jobRef(*j))
193 if j.CancelReason != domain.CICancelNone {
194 // Cancelled (by the owner, fail-fast or the time budget) before the
195 // runner answered: it is over, not queued again for another runner.
196 conclusion := domain.CICancelled
197 if j.CancelReason == domain.CICancelTimeout {
198 conclusion = domain.CITimedOut
199 }
200 rs.end(j, conclusion, "")
201 return
202 }
203 if j.AssignCancels >= domain.CIMaxAssignCancels {
204 rs.end(j, domain.CIFailure, "The job was given to a runner "+itoa(int64(domain.CIMaxAssignCancels))+" times and none of them acknowledged it.")
205 return
206 }
207 j.Phase, j.ReadyAt = domain.CIPhaseReady, c.now()
208 rs.touch(j)
209}
210
211// checkQueue fails the jobs that no runner took within the queue timeout.
212func (c *CI) checkQueue(ctx context.Context) {
213 ready, err := c.cfg.Store.ListReadyJobs(ctx, 0)
214 if err != nil {
215 return
216 }
217 now := c.now()
218 for _, j := range ready {
219 if now.Sub(j.ReadyAt) <= c.cfg.QueueTimeout {
220 continue
221 }
222 err := c.withRun(ctx, j.RunID, j.RunAttempt, func(rs *runState) error {
223 cur := rs.find(j.ID)
224 if cur == nil || cur.Phase != domain.CIPhaseReady || c.now().Sub(cur.ReadyAt) <= c.cfg.QueueTimeout {
225 return nil
226 }
227 rs.end(cur, domain.CIFailure, domain.CIQueueTimeoutMessage)
228 if err := rs.advance(ctx); err != nil {
229 return err
230 }
231 return rs.flush(ctx)
232 })
233 if err != nil && !errors.Is(err, domain.ErrNotFound) && ctx.Err() == nil {
234 ctxLog(ctx).Error().Err(err).Str("job_id", j.ID).Msg("ci: end a job nobody took")
235 }
236 }
237}
238
239// sweep removes the runs older than the retention, and the logs of runs that
240// no longer exist (a repository deleted while the server was down). Every
241// failure is logged and the next step still runs: a log directory that cannot
242// be removed is tried again at the next sweep.
243func (c *CI) sweep(ctx context.Context) {
244 log := ctxLog(ctx)
245 ids, err := c.cfg.Store.Sweep(ctx, c.now().Add(-c.cfg.Retention))
246 if err != nil {
247 log.Error().Err(err).Msg("ci: sweep the old runs")
248 return
249 }
250 if len(ids) > 0 {
251 if err := c.cfg.Logs.RemoveRuns(ctx, ids); err != nil {
252 log.Error().Err(err).Int("runs", len(ids)).Msg("ci: remove the logs of the old runs")
253 }
254 for _, id := range ids {
255 c.locks.Delete(id)
256 }
257 log.Info().Int("runs", len(ids)).Msg("ci: old runs removed")
258 }
259 onDisk, err := c.cfg.Logs.RunIDs(ctx)
260 if err != nil {
261 log.Error().Err(err).Msg("ci: list the log directories")
262 return
263 }
264 var orphans []int64
265 for _, id := range onDisk {
266 switch _, err := c.cfg.Store.GetRunSource(ctx, id); {
267 case errors.Is(err, domain.ErrNotFound):
268 orphans = append(orphans, id)
269 case err != nil:
270 log.Error().Err(err).Int64("run_id", id).Msg("ci: look up a run with a log directory")
271 }
272 }
273 if len(orphans) > 0 {
274 if err := c.cfg.Logs.RemoveRuns(ctx, orphans); err != nil {
275 log.Error().Err(err).Int("runs", len(orphans)).Msg("ci: remove the logs of runs that no longer exist")
276 }
277 }
278}
279
280// RunIDs lists the runs of a repository, for the cleanup that follows its
281// deletion.
282func (c *CI) RunIDs(ctx context.Context, repo domain.RepoName) ([]int64, error) {
283 return c.cfg.Store.RunIDs(ctx, repo)
284}
285
286// RemoveLogs deletes the logs of runs of a repository that was deleted.
287func (c *CI) RemoveLogs(ctx context.Context, runIDs []int64) error {
288 for _, id := range runIDs {
289 c.locks.Delete(id)
290 }
291 return c.cfg.Logs.RemoveRuns(ctx, runIDs)
292}