Sign in

Search issue titles and bodies, change requests and comments. Private repositories you cannot read are not included.
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.

Code

392 commit(s) on main

Compare History
Merge change request 01M70WES5ZZDJ6RP226WB11BBZ: Let people rename and remove their own passkeys keenan · 19m ago 4b8c9e1d
history blame raw 19.9 KiB
1package server
2
3import (
4 "context"
5 "fmt"
6 "sort"
7 "strings"
8 "time"
9
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"
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.
26func (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
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".
59func (s *Server) enqueue(ev runs.Event) {
60 if ev.Ref == metadata.Ref {
61 s.processMetadataRange(ev.Repo, ev.Old, ev.New)
62 return
63 }
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
72 }
73 if ev.New == "" || strings.HasPrefix(ev.New, "0000000000000000000000000000000000000000") {
74 return // a deletion
75 }
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.
79 if _, contributorBranch := access.ContributorRefName(ev.Ref); contributorBranch {
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 }
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
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)
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 }
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 }
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 })
125}
126
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.
130type 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
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
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.
158func (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))
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.
164 s.Log.Warn("ci: workflows in a pushed commit do not validate", "repo", repoName, "problems", len(problems))
165 return 0
166 }
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
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
177 for _, wf := range wfs {
178 if spec.push {
179 if !wf.AppliesToPush(spec.branch, spec.tag, spec.changed) {
180 continue
181 }
182 } else {
183 event := spec.event
184 event.Kind = spec.trigger
185 event.Action = spec.action
186 if !wf.AppliesToEvent(event) {
187 continue
188 }
189 }
190 for _, job := range wf.Jobs {
191 site := ""
192 if spec.deployPages {
193 site = pagesPath(job)
194 }
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 }
215 run := runs.Run{
216 Repo: repoName,
217 Ref: spec.ref,
218 Branch: spec.branch,
219 Tag: spec.tag,
220 TagAnnotated: spec.annotated,
221 SHA: spec.commit,
222 Workflow: wf.Path,
223 WorkflowSHA: wf.BlobSHA,
224 WorkflowName: wf.Name,
225 WorkflowDescription: wf.Description,
226 Trigger: spec.trigger,
227 TriggerAction: spec.action,
228 EventPayload: spec.payload,
229 Occurrence: spec.occurrence,
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),
239 Secrets: job.Secrets,
240 Env: job.Env,
241 Services: services(job),
242 Network: job.Network,
243 Timeout: durationText(job.Timeout),
244 Bot: job.Bot,
245 ReleaseArtifacts: releasePaths(job),
246 PagesPath: site,
247 Created: time.Now().UTC(),
248 }
249 run.ID, _ = id.New(time.Now())
250 owing = append(owing, owedRun{run: run, labels: job.Labels})
251 }
252 }
253 if len(owing) == 0 {
254 return 0
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
259 // find is the same as a run that vanished. A candidate ref names one commit, so
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 }
267 }
268 queued := 0
269 for _, owed := range owing {
270 run := owed.run
271 created, err := s.Runs.Enqueue(run, owed.labels)
272 if err != nil {
273 s.Log.Warn("ci: cannot queue a run", "repo", repoName, "job", run.Job, "err", err)
274 continue
275 }
276 if created {
277 queued++
278 s.Log.Info("ci: queued", "repo", repoName, "job", run.Job, "sha", run.SHA[:8], "run", run.ID, "trigger", run.Trigger)
279 }
280 }
281 return queued
282}
283
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.
287func 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
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.
317const 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.
322func (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
347// the metadata branch, and `vole admin index rebuild` restores exactly that.
348func (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.
370 if ref == metadata.Ref {
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 }
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.
382 if _, contributorBranch := access.ContributorRefName(ref); contributorBranch {
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 }
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
419// releasePaths picks the patterns that also publish to the tag's release.
420func releasePaths(job ci.Job) []string {
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
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.
433func 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
441func steps(job ci.Job) []runs.Step {
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.
446 out = append(out, runs.Step{Name: st.Name, Run: st.Run, Timeout: durationText(st.Timeout)})
447 }
448 return out
449}
450
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.
454func durationText(d time.Duration) string {
455 if d <= 0 {
456 return ""
457 }
458 return d.String()
459}
460
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.
464func artifactPaths(job ci.Job) []string {
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
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.
475func pagesPath(job ci.Job) string {
476 if job.Pages == nil {
477 return ""
478 }
479 return job.Pages.Path
480}
481
482// watchLeases hands jobs from dead runners back to the queue.
483func (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:
491 back, stopped, failed, err := s.Runs.RequeueExpired(time.Now().UTC())
492 if err != nil {
493 s.Log.Warn("ci: cannot requeue expired leases", "err", err)
494 continue
495 }
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))
499 }
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 }
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 }
515 }
516 }
517}
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.
523func (s *Server) allowContributorCI(repo, ref, sha, approver string) error {
524 if _, ok := access.ContributorRefName(ref); !ok {
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}
551