vole
Clone
Read only, no credentials:
https://rodent.software/git/vole.git
Write over SSH with a registered key:
git clone ssh://git@0.0.0.0:2222/vole.git
A free and open-source forge that lives in your Git repo.
|
b44cbfaf
Keenan Nemetz
17d ago
|
1 | package server |
| 2 | ||
| 3 | import ( | |
| 4 | "context" | |
| 5 | "fmt" | |
|
eb07ff56
Keenan Nemetz
8d ago
|
6 | "sort" |
|
b44cbfaf
Keenan Nemetz
17d ago
|
7 | "strings" |
| 8 | "time" | |
| 9 | ||
|
d37d301f
Keenan Nemetz
15h ago
|
10 | "rodent.software/vole/ci" |
| 11 | "rodent.software/vole/ci/runs" | |
| 12 | "rodent.software/vole/core/git" | |
| 13 | "rodent.software/vole/core/id" | |
| 14 | "rodent.software/vole/forge/access" | |
| 15 | "rodent.software/vole/forge/repo" | |
| 16 | "rodent.software/vole/metadata" | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
17 | ) |
| 18 | ||
| 19 | // watchSpool turns accepted pushes into queued CI runs. | |
| 20 | // | |
| 21 | // The hook cannot write to the database — it runs inside a push and must not hold | |
| 22 | // a lock or fail a commit — so it drops one file per ref update and this loop | |
| 23 | // consumes them. Ingestion is idempotent: a spool file read twice enqueues the | |
| 24 | // same (repo, ref, sha, workflow, job) once, because the queue has a unique index | |
| 25 | // on exactly that tuple. | |
| 26 | func (s *Server) watchSpool(ctx context.Context) { | |
| 27 | t := time.NewTicker(time.Second) | |
| 28 | defer t.Stop() | |
| 29 | for { | |
| 30 | select { | |
| 31 | case <-ctx.Done(): | |
| 32 | return | |
| 33 | case <-t.C: | |
| 34 | entries, err := runs.Next(s.Cfg.SpoolDir()) | |
| 35 | if err != nil { | |
| 36 | s.Log.Warn("ci: cannot read the push spool", "err", err) | |
| 37 | continue | |
| 38 | } | |
| 39 | for _, e := range entries { | |
| 40 | if e.Err != nil { | |
| 41 | s.Log.Warn("ci: dropping an unreadable spool entry", "path", e.Path, "err", e.Err) | |
| 42 | } else { | |
| 43 | s.enqueue(e.Event) | |
| 44 | } | |
| 45 | if err := runs.Done(e.Path); err != nil { | |
| 46 | s.Log.Warn("ci: cannot clear a spool entry", "path", e.Path, "err", err) | |
| 47 | } | |
| 48 | } | |
| 49 | } | |
| 50 | } | |
| 51 | } | |
| 52 | ||
|
0f442b1c
Keenan Nemetz
17d ago
|
53 | // enqueue creates one run per job of every workflow that matches the pushed branch |
| 54 | // or tag, pinned to the commit and the workflow blob that were pushed. | |
| 55 | // | |
| 56 | // A tag push is a release build: the tag names what is being released, so a workflow | |
| 57 | // opts into it with on.push.tags. A tag has no branch, and a branch has no tag, which | |
| 58 | // is why the two are carried separately rather than as one "ref". | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
59 | func (s *Server) enqueue(ev runs.Event) { |
|
eb07ff56
Keenan Nemetz
8d ago
|
60 | if ev.Ref == metadata.Ref { |
| 61 | s.processMetadataRange(ev.Repo, ev.Old, ev.New) | |
| 62 | return | |
| 63 | } | |
|
0f442b1c
Keenan Nemetz
17d ago
|
64 | var branch, tag string |
| 65 | switch { | |
| 66 | case strings.HasPrefix(ev.Ref, "refs/heads/"): | |
| 67 | branch = strings.TrimPrefix(ev.Ref, "refs/heads/") | |
| 68 | case strings.HasPrefix(ev.Ref, "refs/tags/"): | |
| 69 | tag = strings.TrimPrefix(ev.Ref, "refs/tags/") | |
| 70 | default: | |
| 71 | return // the metadata ref(s) and refs/vole/* do not run CI | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
72 | } |
| 73 | if ev.New == "" || strings.HasPrefix(ev.New, "0000000000000000000000000000000000000000") { | |
|
0f442b1c
Keenan Nemetz
17d ago
|
74 | return // a deletion |
|
b44cbfaf
Keenan Nemetz
17d ago
|
75 | } |
|
6c84d387
Keenan Nemetz
15d ago
|
76 | // A contributor branch is untrusted code. It does not reach a runner until a |
| 77 | // writer has allowed this exact tip; the approval is keyed by SHA, so a new | |
| 78 | // push is a new decision. This check covers both the spool and reconciliation. | |
|
474a812d
Keenan Nemetz
15d ago
|
79 | if _, contributorBranch := access.ContributorRefName(ev.Ref); contributorBranch { |
|
6c84d387
Keenan Nemetz
15d ago
|
80 | allowed, err := s.Runs.CIAllowed(ev.Repo, ev.Ref, ev.New) |
| 81 | if err != nil { | |
| 82 | s.Log.Warn("ci: cannot read contributor approval", "repo", ev.Repo, "ref", ev.Ref, "err", err) | |
| 83 | return | |
| 84 | } | |
| 85 | if !allowed { | |
| 86 | s.Log.Debug("ci: contributor branch is waiting for a maintainer to allow it", "repo", ev.Repo, "ref", ev.Ref) | |
| 87 | return | |
| 88 | } | |
| 89 | } | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
90 | m, err := s.Repos.Open(ev.Repo) |
| 91 | if err != nil { | |
| 92 | s.Log.Warn("ci: cannot open the pushed repository", "repo", ev.Repo, "err", err) | |
| 93 | return | |
| 94 | } | |
| 95 | // Trust the ref, not the spool: a push that lost a race may have left an event | |
|
0f442b1c
Keenan Nemetz
17d ago
|
96 | // whose new object never became the tip. For a tag that object is the tag, so the |
| 97 | // comparison is against the raw ref, not the commit it peels to. | |
| 98 | tip, err := m.Git().RevParseRaw(ev.Ref) | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
99 | if err != nil || tip != ev.New { |
| 100 | s.Log.Debug("ci: skipping a spool entry whose ref has moved", "repo", ev.Repo, "ref", ev.Ref) | |
| 101 | return | |
| 102 | } | |
|
0f442b1c
Keenan Nemetz
17d ago
|
103 | // A run checks out a commit; a tag object is not one. |
| 104 | commit, err := m.Git().RevParse(ev.Ref) | |
| 105 | if err != nil || commit == "" { | |
| 106 | s.Log.Warn("ci: cannot resolve the pushed ref to a commit", "repo", ev.Repo, "ref", ev.Ref, "err", err) | |
| 107 | return | |
| 108 | } | |
| 109 | annotated := false | |
| 110 | if tag != "" { | |
| 111 | if kind, err := m.Git().ObjectType(ev.Ref); err == nil { | |
| 112 | annotated = kind == "tag" | |
| 113 | } | |
| 114 | } | |
|
eb07ff56
Keenan Nemetz
8d ago
|
115 | changed, err := changedPaths(m.Git(), ev.Old, commit) |
| 116 | if err != nil { | |
| 117 | s.Log.Warn("ci: cannot read the paths a push changed", "repo", ev.Repo, "ref", ev.Ref, "err", err) | |
| 118 | return | |
| 119 | } | |
| 120 | s.queue(m, ev.Repo, queueSpec{ | |
| 121 | ref: ev.Ref, branch: branch, tag: tag, commit: commit, annotated: annotated, | |
| 122 | deployPages: true, push: true, cancelSuperseded: true, | |
| 123 | trigger: "push", action: "pushed", changed: changed, | |
| 124 | }) | |
|
e5478085
Keenan Nemetz
11d ago
|
125 | } |
| 126 | ||
|
eb07ff56
Keenan Nemetz
8d ago
|
127 | // queueSpec is one request to queue the jobs a commit owes. A push and an event |
| 128 | // differ in what selects a workflow and in the event pinned with each run; nothing | |
| 129 | // else about a run depends on how it was requested. | |
| 130 | type queueSpec struct { | |
| 131 | ref string | |
| 132 | branch string | |
| 133 | tag string | |
| 134 | commit string | |
| 135 | annotated bool | |
| 136 | // deployPages says whether a job that declares pages publishes the site. A push | |
| 137 | // does; a combined result does not, and a metadata event does only on its base | |
| 138 | // branch. | |
| 139 | deployPages bool | |
|
727df386
Keenan Nemetz
7d ago
|
140 | // push selects push triggers (branch, tag, path filters). When false, trigger, |
| 141 | // action and event select an event trigger. | |
| 142 | push bool | |
| 143 | trigger string | |
| 144 | action string | |
| 145 | // event carries the filters a non-push trigger is evaluated against. Kind and | |
| 146 | // Action are filled from trigger and action in queue, so a caller only supplies | |
| 147 | // what the event payload says about its body, sender and assignees. | |
| 148 | event ci.EventContext | |
|
eb07ff56
Keenan Nemetz
8d ago
|
149 | payload string |
| 150 | occurrence string | |
| 151 | changed []string | |
| 152 | cancelSuperseded bool | |
| 153 | } | |
| 154 | ||
| 155 | // queue pins and enqueues the jobs a commit owes. The callers are a push, the | |
| 156 | // combined-result reconciler, and the event triggers; they differ in what selects a | |
| 157 | // workflow and in the event pinned with the run. | |
| 158 | func (s *Server) queue(m *repo.Meta, repoName string, spec queueSpec) int { | |
| 159 | wfs, problems := ci.At(m.Git(), spec.commit) | |
| 160 | s.Log.Debug("ci: examined commit", "repo", repoName, "ref", spec.ref, "commit", spec.commit, "workflows", len(wfs), "problems", len(problems)) | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
161 | if len(problems) > 0 { |
| 162 | // pre-receive rejects these, so reaching here means a workflow was already | |
| 163 | // broken: queue nothing rather than guess. | |
|
e5478085
Keenan Nemetz
11d ago
|
164 | s.Log.Warn("ci: workflows in a pushed commit do not validate", "repo", repoName, "problems", len(problems)) |
| 165 | return 0 | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
166 | } |
|
eb07ff56
Keenan Nemetz
8d ago
|
167 | // What this event owes: one run per job of every workflow that applies to it. The |
| 168 | // list is built before anything is queued, because an event that owes nothing must | |
|
6c03e461
Keenan Nemetz
16d ago
|
169 | // not cancel the previous tip's builds. A job's runs-on labels travel beside the |
| 170 | // run rather than inside it: they are lease policy, and Enqueue takes them as its | |
| 171 | // own argument so a caller holding the job cannot forget them. | |
| 172 | type owedRun struct { | |
| 173 | run runs.Run | |
| 174 | labels []string | |
| 175 | } | |
| 176 | var owing []owedRun | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
177 | for _, wf := range wfs { |
|
eb07ff56
Keenan Nemetz
8d ago
|
178 | if spec.push { |
| 179 | if !wf.AppliesToPush(spec.branch, spec.tag, spec.changed) { | |
| 180 | continue | |
| 181 | } | |
|
727df386
Keenan Nemetz
7d ago
|
182 | } else { |
| 183 | event := spec.event | |
| 184 | event.Kind = spec.trigger | |
| 185 | event.Action = spec.action | |
| 186 | if !wf.AppliesToEvent(event) { | |
| 187 | continue | |
| 188 | } | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
189 | } |
| 190 | for _, job := range wf.Jobs { | |
|
e5478085
Keenan Nemetz
11d ago
|
191 | site := "" |
|
eb07ff56
Keenan Nemetz
8d ago
|
192 | if spec.deployPages { |
|
e5478085
Keenan Nemetz
11d ago
|
193 | site = pagesPath(job) |
| 194 | } | |
|
5f6163bb
Keenan Nemetz
1h ago
|
195 | // A landing lands the combined result's own commit, so this tree, this |
| 196 | // workflow and this job are what already ran for this branch before it | |
| 197 | // landed. Queueing them again spends the work twice and shows one commit | |
| 198 | // twice in the run list. | |
| 199 | // | |
| 200 | // Only a push to the branch itself: a tag is a release build of its own, an | |
| 201 | // event run is a new occurrence of it, and the candidate's own run is the | |
| 202 | // one being reused. The exception is a job that deploys the site — a | |
| 203 | // combined result never deploys — because that run is different work. | |
| 204 | if spec.push && spec.ref == "refs/heads/"+spec.branch && site == "" { | |
| 205 | passed, err := s.Runs.CombinedResultPassed(repoName, spec.branch, spec.commit, wf.Path, job.ID) | |
| 206 | if err != nil { | |
| 207 | s.Log.Warn("ci: cannot tell whether a combined result already ran a job", | |
| 208 | "repo", repoName, "job", job.ID, "err", err) | |
| 209 | } else if passed { | |
| 210 | s.Log.Info("ci: not queueing a job the combined result already passed", | |
| 211 | "repo", repoName, "job", job.ID, "sha", spec.commit[:8]) | |
| 212 | continue | |
| 213 | } | |
| 214 | } | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
215 | run := runs.Run{ |
|
1cde8b9e
Keenan Nemetz
8d ago
|
216 | Repo: repoName, |
|
eb07ff56
Keenan Nemetz
8d ago
|
217 | Ref: spec.ref, |
| 218 | Branch: spec.branch, | |
| 219 | Tag: spec.tag, | |
| 220 | TagAnnotated: spec.annotated, | |
| 221 | SHA: spec.commit, | |
|
1cde8b9e
Keenan Nemetz
8d ago
|
222 | Workflow: wf.Path, |
| 223 | WorkflowSHA: wf.BlobSHA, | |
| 224 | WorkflowName: wf.Name, | |
| 225 | WorkflowDescription: wf.Description, | |
|
eb07ff56
Keenan Nemetz
8d ago
|
226 | Trigger: spec.trigger, |
| 227 | TriggerAction: spec.action, | |
| 228 | EventPayload: spec.payload, | |
| 229 | Occurrence: spec.occurrence, | |
|
1cde8b9e
Keenan Nemetz
8d ago
|
230 | Job: job.ID, |
| 231 | JobName: job.Name, | |
| 232 | JobDescription: job.Description, | |
| 233 | Steps: steps(job), | |
| 234 | Image: job.Image, | |
| 235 | ImageFile: job.ImageFile, | |
| 236 | ImageDefinition: job.ImageDefinition, | |
| 237 | ImageSHA: job.ImageSHA, | |
| 238 | Artifacts: artifactPaths(job), | |
|
4ac42680
Keenan Nemetz
7d ago
|
239 | Secrets: job.Secrets, |
|
1cde8b9e
Keenan Nemetz
8d ago
|
240 | Env: job.Env, |
|
c1929d2f
Keenan Nemetz
5d ago
|
241 | Services: services(job), |
|
d7958024
Keenan Nemetz
6d ago
|
242 | Network: job.Network, |
|
e7aa0785
Keenan Nemetz
6d ago
|
243 | Timeout: durationText(job.Timeout), |
|
6e1fca51
Keenan Nemetz
6d ago
|
244 | Bot: job.Bot, |
|
1cde8b9e
Keenan Nemetz
8d ago
|
245 | ReleaseArtifacts: releasePaths(job), |
| 246 | PagesPath: site, | |
| 247 | Created: time.Now().UTC(), | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
248 | } |
| 249 | run.ID, _ = id.New(time.Now()) | |
|
6c03e461
Keenan Nemetz
16d ago
|
250 | owing = append(owing, owedRun{run: run, labels: job.Labels}) |
|
760b523a
Keenan Nemetz
16d ago
|
251 | } |
| 252 | } | |
| 253 | if len(owing) == 0 { | |
|
e5478085
Keenan Nemetz
11d ago
|
254 | return 0 |
|
760b523a
Keenan Nemetz
16d ago
|
255 | } |
| 256 | // A newer tip supersedes the queued builds of the older one: pushing twice in a | |
| 257 | // minute must not build the first commit twice. Running jobs are left alone, and | |
| 258 | // the cancellation is logged because a run that stopped for a reason nobody can | |
|
e5478085
Keenan Nemetz
11d ago
|
259 | // find is the same as a run that vanished. A candidate ref names one commit, so |
|
eb07ff56
Keenan Nemetz
8d ago
|
260 | // this is a no-op for combined results; a metadata event never cancels anything. |
| 261 | if spec.cancelSuperseded { | |
| 262 | if n, err := s.Runs.CancelSuperseded(repoName, spec.ref, spec.commit, time.Now().UTC()); err != nil { | |
| 263 | s.Log.Warn("ci: cannot cancel superseded runs", "repo", repoName, "ref", spec.ref, "err", err) | |
| 264 | } else if n > 0 { | |
| 265 | s.Log.Info("ci: canceled superseded runs", "repo", repoName, "ref", spec.ref, "runs", n) | |
| 266 | } | |
|
760b523a
Keenan Nemetz
16d ago
|
267 | } |
|
e5478085
Keenan Nemetz
11d ago
|
268 | queued := 0 |
|
6c03e461
Keenan Nemetz
16d ago
|
269 | for _, owed := range owing { |
| 270 | run := owed.run | |
| 271 | created, err := s.Runs.Enqueue(run, owed.labels) | |
|
760b523a
Keenan Nemetz
16d ago
|
272 | if err != nil { |
|
e5478085
Keenan Nemetz
11d ago
|
273 | s.Log.Warn("ci: cannot queue a run", "repo", repoName, "job", run.Job, "err", err) |
|
760b523a
Keenan Nemetz
16d ago
|
274 | continue |
| 275 | } | |
| 276 | if created { | |
|
e5478085
Keenan Nemetz
11d ago
|
277 | queued++ |
|
eb07ff56
Keenan Nemetz
8d ago
|
278 | s.Log.Info("ci: queued", "repo", repoName, "job", run.Job, "sha", run.SHA[:8], "run", run.ID, "trigger", run.Trigger) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
279 | } |
| 280 | } | |
|
e5478085
Keenan Nemetz
11d ago
|
281 | return queued |
|
b44cbfaf
Keenan Nemetz
17d ago
|
282 | } |
| 283 | ||
|
eb07ff56
Keenan Nemetz
8d ago
|
284 | // changedPaths lists the paths a ref update touched. A new ref has no old tree, so |
| 285 | // every path in the new commit is new; an existing ref is compared tree to tree, so | |
| 286 | // a force-push is measured by what actually differs. | |
| 287 | func changedPaths(g *git.Repo, old, new string) ([]string, error) { | |
| 288 | if new == "" || new == git.ZeroSHA { | |
| 289 | return nil, nil | |
| 290 | } | |
| 291 | if old == "" || old == git.ZeroSHA { | |
| 292 | entries, err := g.LsTreeR(new) | |
| 293 | if err != nil { | |
| 294 | return nil, err | |
| 295 | } | |
| 296 | paths := make([]string, 0, len(entries)) | |
| 297 | for _, entry := range entries { | |
| 298 | paths = append(paths, entry.Name) | |
| 299 | } | |
| 300 | return paths, nil | |
| 301 | } | |
| 302 | status, err := g.DiffStatus(old, new) | |
| 303 | if err != nil { | |
| 304 | return nil, err | |
| 305 | } | |
| 306 | paths := make([]string, 0, len(status)) | |
| 307 | for p := range status { | |
| 308 | paths = append(paths, p) | |
| 309 | } | |
| 310 | sort.Strings(paths) | |
| 311 | return paths, nil | |
| 312 | } | |
| 313 | ||
|
a945ae55
Keenan Nemetz
16d ago
|
314 | // reconcileSeconds is how often the queue is checked against the refs. The spool |
| 315 | // delivers a push in well under a second; this is the safety net that makes the | |
| 316 | // promise eventually true when the spool file was never written. | |
| 317 | const reconcileSeconds = 30 | |
| 318 | ||
| 319 | // watchRefs reconciles the queue from the refs once at startup and then on a timer. | |
| 320 | // The startup pass is why a queue that was deleted comes back without a 30-second | |
| 321 | // blind spot. | |
| 322 | func (s *Server) watchRefs(ctx context.Context) { | |
| 323 | s.reconcileRefs() | |
| 324 | t := time.NewTicker(reconcileSeconds * time.Second) | |
| 325 | defer t.Stop() | |
| 326 | for { | |
| 327 | select { | |
| 328 | case <-ctx.Done(): | |
| 329 | return | |
| 330 | case <-t.C: | |
| 331 | s.reconcileRefs() | |
| 332 | } | |
| 333 | } | |
| 334 | } | |
| 335 | ||
| 336 | // reconcileRefs queues work for every ref whose tip the queue has never seen. | |
| 337 | // | |
| 338 | // It is what makes the spool a latency optimisation instead of a promise. The hook | |
| 339 | // runs after the refs move and may fail to write its spool file without failing the | |
| 340 | // push (a crash between the two, a full disk), and `runs.db` can be deleted while | |
| 341 | // jobs are queued. Both are answered the same way: a ref tip is the durable | |
| 342 | // statement that a commit is live, so a tip no cursor has recorded is work owed. | |
| 343 | // | |
| 344 | // The trade is explicit: a tip nothing has ever recorded is queued, which includes | |
| 345 | // every branch of an instance whose queue was emptied. The unique index stops a tip | |
| 346 | // that already ran from running again once its row is restored from the records on | |
|
3abbe749
Keenan Nemetz
16d ago
|
347 | // the metadata branch, and `vole admin index rebuild` restores exactly that. |
|
a945ae55
Keenan Nemetz
16d ago
|
348 | func (s *Server) reconcileRefs() { |
| 349 | repos, err := s.Repos.List() | |
| 350 | if err != nil { | |
| 351 | s.Log.Warn("ci: cannot list repositories to reconcile", "err", err) | |
| 352 | return | |
| 353 | } | |
| 354 | now := time.Now().UTC() | |
| 355 | reconciled := 0 | |
| 356 | for _, m := range repos { | |
| 357 | cursors, err := s.Runs.Cursors(m.Name) | |
| 358 | if err != nil { | |
| 359 | s.Log.Warn("ci: cannot read ref cursors", "repo", m.Name, "err", err) | |
| 360 | continue | |
| 361 | } | |
| 362 | refs, err := m.Git().ListRefs("refs/") | |
| 363 | if err != nil { | |
| 364 | s.Log.Warn("ci: cannot list refs", "repo", m.Name, "err", err) | |
| 365 | continue | |
| 366 | } | |
| 367 | for ref, tip := range refs { | |
| 368 | // The metadata branch carries human-authored entities and run records; the | |
| 369 | // runner's own push of a record must not queue a build of the tool's data. | |
|
c5902451
Keenan Nemetz
16d ago
|
370 | if ref == metadata.Ref { |
|
a945ae55
Keenan Nemetz
16d ago
|
371 | continue |
| 372 | } | |
| 373 | if !strings.HasPrefix(ref, "refs/heads/") && !strings.HasPrefix(ref, "refs/tags/") { | |
| 374 | continue | |
| 375 | } | |
| 376 | if held, ok := cursors[ref]; ok && held == tip { | |
| 377 | continue | |
| 378 | } | |
|
6c84d387
Keenan Nemetz
15d ago
|
379 | // A contributor tip without an approval is not owed. Leave the cursor |
| 380 | // untouched: the allow action queues the tip, and if that enqueue hits a | |
| 381 | // transient error the next reconciliation can retry it. | |
|
474a812d
Keenan Nemetz
15d ago
|
382 | if _, contributorBranch := access.ContributorRefName(ref); contributorBranch { |
|
6c84d387
Keenan Nemetz
15d ago
|
383 | allowed, err := s.Runs.CIAllowed(m.Name, ref, tip) |
| 384 | if err != nil { | |
| 385 | s.Log.Warn("ci: cannot read contributor approval", "repo", m.Name, "ref", ref, "err", err) | |
| 386 | continue | |
| 387 | } | |
| 388 | if !allowed { | |
| 389 | s.Log.Debug("ci: contributor branch is waiting for a maintainer to allow it", "repo", m.Name, "ref", ref) | |
| 390 | continue | |
| 391 | } | |
| 392 | } | |
|
a945ae55
Keenan Nemetz
16d ago
|
393 | // enqueue re-checks the tip against the ref, so an event that raced with |
| 394 | // another push is dropped there rather than trusted here. | |
| 395 | s.enqueue(runs.Event{ | |
| 396 | Repo: m.Name, Ref: ref, Old: cursors[ref], New: tip, | |
| 397 | At: now.Format(time.RFC3339Nano), | |
| 398 | }) | |
| 399 | // The cursor moves after the work is queued: it is a note that a tip was | |
| 400 | // handled, not a claim about work that is about to be attempted. | |
| 401 | if err := s.Runs.SetCursor(m.Name, ref, tip, now); err != nil { | |
| 402 | s.Log.Warn("ci: cannot record a ref cursor", "repo", m.Name, "ref", ref, "err", err) | |
| 403 | } | |
| 404 | reconciled++ | |
| 405 | } | |
| 406 | for ref := range cursors { | |
| 407 | if _, ok := refs[ref]; !ok { | |
| 408 | if err := s.Runs.DropCursor(m.Name, ref); err != nil { | |
| 409 | s.Log.Warn("ci: cannot drop a stale cursor", "repo", m.Name, "ref", ref, "err", err) | |
| 410 | } | |
| 411 | } | |
| 412 | } | |
| 413 | } | |
| 414 | if reconciled > 0 { | |
| 415 | s.Log.Info("ci: reconciled refs with the queue", "refs", reconciled) | |
| 416 | } | |
| 417 | } | |
| 418 | ||
|
0f442b1c
Keenan Nemetz
17d ago
|
419 | // releasePaths picks the patterns that also publish to the tag's release. |
|
c5902451
Keenan Nemetz
16d ago
|
420 | func releasePaths(job ci.Job) []string { |
|
0f442b1c
Keenan Nemetz
17d ago
|
421 | var out []string |
| 422 | for _, a := range job.Artifacts { | |
| 423 | if a.Release { | |
| 424 | out = append(out, a.Path) | |
| 425 | } | |
| 426 | } | |
| 427 | return out | |
| 428 | } | |
| 429 | ||
|
c1929d2f
Keenan Nemetz
5d ago
|
430 | // services pins the containers a job asked to run beside it. Like the steps and the |
| 431 | // artifact patterns, the list is read at queue time: a workflow edited afterwards | |
| 432 | // cannot change what an already-queued job starts. | |
| 433 | func services(job ci.Job) []runs.Service { | |
| 434 | out := make([]runs.Service, 0, len(job.Services)) | |
| 435 | for _, svc := range job.Services { | |
| 436 | out = append(out, runs.Service{ID: svc.ID, Image: svc.Image, Env: svc.Env}) | |
| 437 | } | |
| 438 | return out | |
| 439 | } | |
| 440 | ||
|
c5902451
Keenan Nemetz
16d ago
|
441 | func steps(job ci.Job) []runs.Step { |
|
b44cbfaf
Keenan Nemetz
17d ago
|
442 | out := make([]runs.Step, 0, len(job.Steps)) |
| 443 | for _, st := range job.Steps { | |
| 444 | // The step list is pinned at enqueue time, so editing the workflow after the | |
| 445 | // push cannot change what an already-queued run executes. | |
|
e7aa0785
Keenan Nemetz
6d ago
|
446 | out = append(out, runs.Step{Name: st.Name, Run: st.Run, Timeout: durationText(st.Timeout)}) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
447 | } |
| 448 | return out | |
| 449 | } | |
| 450 | ||
|
e7aa0785
Keenan Nemetz
6d ago
|
451 | // durationText is how a pinned timeout is stored: empty when nothing was declared, |
| 452 | // otherwise the duration the reader wrote. Empty is what the runner reads as "no | |
| 453 | // opinion", which is different from a timeout of zero. | |
| 454 | func durationText(d time.Duration) string { | |
| 455 | if d <= 0 { | |
| 456 | return "" | |
| 457 | } | |
| 458 | return d.String() | |
| 459 | } | |
| 460 | ||
|
213f4d6d
Keenan Nemetz
17d ago
|
461 | // artifactPaths pins the declared patterns into the queued run, next to the pinned |
| 462 | // steps: a workflow edited after the push must not change what an already-queued | |
| 463 | // run publishes. | |
|
c5902451
Keenan Nemetz
16d ago
|
464 | func artifactPaths(job ci.Job) []string { |
|
213f4d6d
Keenan Nemetz
17d ago
|
465 | out := make([]string, 0, len(job.Artifacts)) |
| 466 | for _, a := range job.Artifacts { | |
| 467 | out = append(out, a.Path) | |
| 468 | } | |
| 469 | return out | |
| 470 | } | |
| 471 | ||
|
88e980b0
Keenan Nemetz
17d ago
|
472 | // pagesPath pins the site directory into the queued run, like the steps and the |
| 473 | // artifact patterns: a workflow edited afterwards must not change what a queued run | |
| 474 | // publishes. | |
|
c5902451
Keenan Nemetz
16d ago
|
475 | func pagesPath(job ci.Job) string { |
|
88e980b0
Keenan Nemetz
17d ago
|
476 | if job.Pages == nil { |
| 477 | return "" | |
| 478 | } | |
| 479 | return job.Pages.Path | |
| 480 | } | |
| 481 | ||
|
b44cbfaf
Keenan Nemetz
17d ago
|
482 | // watchLeases hands jobs from dead runners back to the queue. |
| 483 | func (s *Server) watchLeases(ctx context.Context) { | |
| 484 | t := time.NewTicker(5 * time.Second) | |
| 485 | defer t.Stop() | |
| 486 | for { | |
| 487 | select { | |
| 488 | case <-ctx.Done(): | |
| 489 | return | |
| 490 | case <-t.C: | |
|
0c14f078
Keenan Nemetz
6d ago
|
491 | back, stopped, failed, err := s.Runs.RequeueExpired(time.Now().UTC()) |
|
b44cbfaf
Keenan Nemetz
17d ago
|
492 | if err != nil { |
| 493 | s.Log.Warn("ci: cannot requeue expired leases", "err", err) | |
| 494 | continue | |
| 495 | } | |
|
0c14f078
Keenan Nemetz
6d ago
|
496 | if back > 0 || stopped > 0 || failed > 0 { |
| 497 | s.Log.Warn(fmt.Sprintf("ci: %d job(s) returned to the queue, %d canceled while stopped, %d failed after %d attempts", | |
| 498 | back, stopped, failed, runs.MaxAttempts)) | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
499 | } |
|
6e1fca51
Keenan Nemetz
6d ago
|
500 | // A run whose runner disappeared never sends a finish, so the key it was |
| 501 | // given is ended by its expiry. Sweeping here keeps the rows bounded and | |
| 502 | // means a credential that is refused is not still sitting in the table. | |
| 503 | if dropped, err := s.Runs.ExpiredCredentials(time.Now().UTC()); err != nil { | |
| 504 | s.Log.Warn("ci: cannot sweep expired run credentials", "err", err) | |
| 505 | } else if dropped > 0 { | |
| 506 | s.Log.Info("ci: swept expired run credentials", "credentials", dropped) | |
| 507 | } | |
|
ac433531
Keenan Nemetz
16d ago
|
508 | // Presence is not history: a runner that was killed leaves a row nothing |
| 509 | // will ever refresh, so it is forgotten rather than shown as connected. | |
| 510 | if n, err := s.Runs.PruneSessions(time.Now().UTC().Add(-10 * time.Minute)); err != nil { | |
| 511 | s.Log.Warn("ci: cannot prune runner presence", "err", err) | |
| 512 | } else if n > 0 { | |
| 513 | s.Log.Debug("ci: forgot silent runners", "runners", n) | |
| 514 | } | |
|
b44cbfaf
Keenan Nemetz
17d ago
|
515 | } |
| 516 | } | |
| 517 | } | |
|
6c84d387
Keenan Nemetz
15d ago
|
518 | |
| 519 | // allowContributorCI is the queue-side half of the "Allow CI" button. The web has | |
| 520 | // already checked that the approver has maintain access; this re-checks the ref is | |
| 521 | // still the contributor branch tip that was approved, records the decision, and | |
| 522 | // queues it. The approval is keyed by SHA, so the next push waits again. | |
| 523 | func (s *Server) allowContributorCI(repo, ref, sha, approver string) error { | |
|
474a812d
Keenan Nemetz
15d ago
|
524 | if _, ok := access.ContributorRefName(ref); !ok { |
|
6c84d387
Keenan Nemetz
15d ago
|
525 | return fmt.Errorf("ci: %s is not a contributor branch", ref) |
| 526 | } | |
| 527 | m, err := s.Repos.Open(repo) | |
| 528 | if err != nil { | |
| 529 | return err | |
| 530 | } | |
| 531 | tip, err := m.Git().RevParseRaw(ref) | |
| 532 | if err != nil || tip == "" { | |
| 533 | return fmt.Errorf("ci: contributor branch %s does not exist", ref) | |
| 534 | } | |
| 535 | if tip != sha { | |
| 536 | return fmt.Errorf("ci: %s moved; approve the new tip", ref) | |
| 537 | } | |
| 538 | now := time.Now().UTC() | |
| 539 | if err := s.Runs.AllowCI(repo, ref, sha, approver, now); err != nil { | |
| 540 | return err | |
| 541 | } | |
| 542 | // Reconciliation may already have recorded this tip while it was unapproved. | |
| 543 | // Dropping the cursor makes the next pass retry if enqueue hits a transient | |
| 544 | // error; enqueue is idempotent, so a successful pass costs only a second no-op. | |
| 545 | if err := s.Runs.DropCursor(repo, ref); err != nil { | |
| 546 | s.Log.Warn("ci: cannot clear a contributor ref cursor", "repo", repo, "ref", ref, "err", err) | |
| 547 | } | |
| 548 | s.enqueue(runs.Event{Repo: repo, Ref: ref, New: sha, At: now.Format(time.RFC3339Nano)}) | |
| 549 | return nil | |
| 550 | } |