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