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/vole.git

Write over SSH with a registered key:

git clone ssh://git@rodent.software:2222/vole.git

A free and open-source forge that lives in your Git repo.

refactor

c5902451905acdaa478f8476844ccd4a510d9ca9 · Keenan Nemetz · 17d ago · parent ec2aba4b

ci/agent/agent.go +865

@@ -0,0 +1,865 @@
1 // Package agent implements the Vole CI agent.
2 //
3 // A runner connects to the instance over SSH with the key its account pushes
4 // with, leases one job at a time, runs its steps in a persistent checkout, then
5 // uploads the log and artifacts and publishes the attempt as a signed commit.
6 // This package is the agent; argument parsing and the CLI's messages stay in
7 // cmd/vole.
8 package agent
9
10 import (
11 "bufio"
12 "context"
13 "encoding/json"
14 "errors"
15 "fmt"
16 "net/url"
17 "os"
18 "os/exec"
19 "path"
20 "path/filepath"
21 "runtime"
22 "sort"
23 "strings"
24 "sync"
25 "syscall"
26 "time"
27
28 "github.com/rodent-software/vole/ci"
29 "github.com/rodent-software/vole/ci/runs"
30 "github.com/rodent-software/vole/client"
31 "github.com/rodent-software/vole/core/lfs"
32 "github.com/rodent-software/vole/internal/format"
33 "github.com/rodent-software/vole/metadata"
34 )
35
36 // leaseWait is how long an idle runner parks on the lease before asking again. It is
37 // below the 60-second cap the server applies and long enough that a quiet pool is
38 // invisible in the queue's load.
39 const leaseWait = 25 * time.Second
40
41 type runnerStep struct {
42 Name string `json:"name"`
43 Run string `json:"run"`
44 Timeout string `json:"timeout"`
45 }
46
47 type runnerJob struct {
48 ID string `json:"id"`
49 Repo string `json:"repo"`
50 SHA string `json:"sha"`
51 // Attempt is which try at this job this is. The record lives under it, so the
52 // runner has to know the number it is reporting.
53 Attempt int `json:"attempt"`
54 // Ref, WorkflowSHA and StepsSHA are part of what this runner signs when the
55 // attempt ends: the exact revision, the exact workflow file, and the digest of
56 // the steps that were pinned when the job was queued.
57 Ref string `json:"ref"`
58 WorkflowSHA string `json:"workflow_sha"`
59 StepsSHA string `json:"steps_sha"`
60 Branch string `json:"branch"`
61 Workflow string `json:"workflow"`
62 Job string `json:"job"`
63 Steps []runnerStep `json:"steps"`
64 // Artifacts are path patterns, expanded in the checkout after the steps pass.
65 Artifacts []string `json:"artifacts,omitempty"`
66 // Tag is set for a tag-triggered run; TagAnnotated says whether it carries
67 // notes, which is what makes it a release. ReleaseArtifacts are the patterns
68 // that also publish to that release.
69 Tag string `json:"tag,omitempty"`
70 TagAnnotated bool `json:"tag_annotated,omitempty"`
71 ReleaseArtifacts []string `json:"release_artifacts,omitempty"`
72 // Pages is the site deploy this run performs, if any.
73 Pages *runnerPages `json:"pages,omitempty"`
74 }
75
76 type runnerPages struct {
77 Path string `json:"path"`
78 Ref string `json:"ref"`
79 }
80
81 type runnerArtifact struct {
82 Name string `json:"name"`
83 OID string `json:"oid"`
84 Size int64 `json:"size"`
85 }
86
87 // runnerRequest is the wire. A finish carries nothing but the run id: the result
88 // went up as a git push before this was sent, because the record is the runner's
89 // statement and the push is what the gate checks. This channel only closes the lease.
90 type runnerRequest struct {
91 Op string `json:"op"`
92 Run string `json:"run"`
93 // Repos is the --repos filter, sent with every lease. The server intersects it
94 // with what the account's allowlists grant, so it can only narrow.
95 Repos []string `json:"repos,omitempty"`
96 // Labels are what this machine is, sent with every lease so the server can match
97 // them against the runs-on a job asked for.
98 Labels []string `json:"labels,omitempty"`
99 // Wait is how many seconds an idle lease may park before answering "nothing to
100 // do"; the server caps it. It is what makes an idle runner cost a parked channel
101 // rather than a query every couple of seconds.
102 Wait int `json:"wait,omitempty"`
103 }
104
105 type runnerReply struct {
106 Op string `json:"op"`
107 Error string `json:"error,omitempty"`
108 Run *runnerJob `json:"run,omitempty"`
109 OK bool `json:"ok"`
110 Lease int `json:"lease,omitempty"`
111 // The account this session authenticated as and the fingerprints of its keys,
112 // which is how a runner finds out before it starts work whether the key it will
113 // sign with is one the server will accept.
114 Account string `json:"account,omitempty"`
115 Name string `json:"name,omitempty"`
116 // Repos are the repositories this account may run jobs for. A runner answers the
117 // "ping" with these so a --repos filter can be checked against them at startup:
118 // a filter that matches nothing would otherwise look exactly like a quiet instance.
119 Repos []string `json:"repos,omitempty"`
120 }
121
122 // Options configure one runner process.
123 type Options struct {
124 // Remote is the instance's SSH URL, e.g. ssh://git@host:2222.
125 Remote string
126 // Dir is where checkouts are cached. Empty means the user cache directory.
127 Dir string
128 // Repos narrows this runner to the named repositories. The server intersects
129 // it with the account's runner allowlist, so it can only subtract.
130 Repos []string
131 // Labels describe the machine. A job is only leased to a runner carrying every
132 // label the job asked for with runs-on.
133 Labels []string
134 // Once runs one job and exits, which is what a scheduled container wants.
135 Once bool
136 }
137
138 // Run leases and executes CI jobs until the connection fails, the context ends,
139 // or (with Once) one job has been handled.
140 func Run(ctx context.Context, opts Options) error {
141 if opts.Remote == "" {
142 return errors.New("--remote is required, e.g. --remote ssh://git@example.com:2222")
143 }
144 cache := opts.Dir
145 if cache == "" {
146 base, err := os.UserCacheDir()
147 if err != nil {
148 base = os.TempDir()
149 }
150 cache = filepath.Join(base, "vole-runner")
151 }
152 if err := os.MkdirAll(cache, 0o755); err != nil {
153 return err
154 }
155
156 // A label has to be one the workflow validator would accept, or the machine would
157 // advertise something no job can ask for — a typo with no symptom at all.
158 for _, label := range opts.Labels {
159 if err := ci.ValidLabel(label); err != nil {
160 return fmt.Errorf("--label %w", err)
161 }
162 }
163
164 sess, err := dialRunner(opts.Remote)
165 if err != nil {
166 return err
167 }
168 defer sess.close()
169
170 // Ask who the server thinks this is, so the runner can say whose results it reports
171 // and check its own --repos filter against what the account may run.
172 who, err := sess.call(runnerRequest{Op: "ping"})
173 if err != nil {
174 return err
175 }
176 if who.Error != "" {
177 return errors.New(who.Error)
178 }
179 // --repos narrows the account's grant. Check it against the grant now: a filter that
180 // matches nothing is indistinguishable from a quiet instance, so it would idle
181 // forever while looking healthy — cheaper to find at startup than after a build that
182 // never arrives.
183 if err := checkRepoFilter(opts.Repos, who); err != nil {
184 return err
185 }
186 sess.account, sess.name = who.Account, who.Name
187 sess.repos, sess.labels = opts.Repos, opts.Labels
188 fmt.Fprintf(os.Stderr, "vole runner: %s\n", sess.describe(who.Name))
189 for {
190 // A resident runner parks on the server's lease: the reply arrives when work is
191 // queued, not on a fixed beat, so an idle pool costs one open channel per
192 // runner instead of a query every couple of seconds. A one-shot runner asks
193 // once and leaves.
194 req := runnerRequest{Op: "lease", Repos: sess.repos, Labels: sess.labels}
195 if !opts.Once {
196 req.Wait = int(leaseWait.Seconds())
197 }
198 rep, err := sess.call(req)
199 if err != nil {
200 return err
201 }
202 if rep.Error != "" {
203 return errors.New(rep.Error)
204 }
205 if rep.Run == nil {
206 if opts.Once {
207 return nil
208 }
209 // Either the wait expired with nothing to do, or the server answered at
210 // once because it is older than this client and ignores `wait`. The sleep
211 // keeps the second case at a polite two requests per second instead of a
212 // spin.
213 select {
214 case <-ctx.Done():
215 return nil
216 case <-time.After(2 * time.Second):
217 }
218 continue
219 }
220 lease := time.Duration(rep.Lease) * time.Second
221 if lease <= 0 {
222 lease = 60 * time.Second
223 }
224 sess.run(opts.Remote, cache, rep.Run, lease)
225 if opts.Once {
226 return nil
227 }
228 }
229 }
230
231 // ---------- connection ----------
232
233 type runnerSession struct {
234 cmd *exec.Cmd
235 in *bufio.Writer
236 out *bufio.Reader
237 enc *json.Encoder
238 dec *json.Decoder
239 mu sync.Mutex
240 // account is the account id the server resolved from the connection. A record
241 // names it, and the gate refuses a record whose signer is not the pusher.
242 account string
243 name string
244 // repos is the --repos filter, sent with every lease. It is what the operator
245 // asked for, not the effective list: the server intersects it with the grant on
246 // each request, so an allowlist edit takes effect without a restart.
247 repos []string
248 // labels is what this machine is (--label). A job is only given to a runner that
249 // carries every label the job asked for, so an unlabelled runner only sees jobs
250 // that ask for nothing.
251 labels []string
252 }
253
254 // describe is the one line a runner prints when it starts work, because "why is
255 // nothing running" is answered by reading what it says it will take. Both filters are
256 // named when they are set and omitted when they are not, so the common case stays one
257 // short line.
258 func (s *runnerSession) describe(name string) string {
259 out := "connected as " + name
260 if len(s.labels) > 0 {
261 out += ", taking jobs that ask for " + strings.Join(s.labels, ", ")
262 }
263 if len(s.repos) > 0 {
264 out += ", only in " + strings.Join(s.repos, ", ")
265 }
266 return out + "; waiting for work"
267 }
268
269 // checkRepoFilter refuses a --repos filter that can never match. A name the server
270 // did not grant is dropped by the intersection rather than refused — an operator may
271 // reasonably name a repository before it exists — but a filter whose names are *all*
272 // outside the grant is a typo or a permissions problem, and it would otherwise present
273 // as a runner that connected happily and never did anything.
274 func checkRepoFilter(asked []string, who runnerReply) error {
275 if len(asked) == 0 {
276 if len(who.Repos) == 0 {
277 // Not an error: an allowlist nobody has written yet is a normal state, and it
278 // takes effect without a restart. Say so instead of idling silently.
279 fmt.Fprintf(os.Stderr, "vole runner: %s may not run jobs for any repository yet; "+
280 "waiting (an admin allows it with `vole acl runners <repo> --add %s`)\n",
281 who.Name, who.Name)
282 }
283 return nil
284 }
285 granted := make(map[string]bool, len(who.Repos))
286 for _, r := range who.Repos {
287 granted[r] = true
288 }
289 for _, r := range asked {
290 if granted[r] {
291 return nil
292 }
293 }
294 allowed := "none"
295 if len(who.Repos) > 0 {
296 allowed = strings.Join(who.Repos, ", ")
297 }
298 return fmt.Errorf("--repos %s matches nothing %s may run (allowed: %s); "+
299 "a runner can only narrow what its account is allowed",
300 strings.Join(asked, ", "), who.Name, allowed)
301 }
302
303 func dialRunner(remote string) (*runnerSession, error) {
304 u, err := url.Parse(remote)
305 if err != nil {
306 return nil, fmt.Errorf("%q is not an ssh URL: %w", remote, err)
307 }
308 if u.Scheme != "ssh" || u.Hostname() == "" {
309 return nil, fmt.Errorf("%q is not an ssh URL (expected ssh://user@host:port)", remote)
310 }
311 user := u.User.Username()
312 if user == "" {
313 user = "git"
314 }
315 cmd := client.SSHCommand(user+"@"+u.Hostname(), u.Port(), "vole-runner")
316 cmd.Stderr = os.Stderr
317 stdin, err := cmd.StdinPipe()
318 if err != nil {
319 return nil, err
320 }
321 stdout, err := cmd.StdoutPipe()
322 if err != nil {
323 return nil, err
324 }
325 if err := cmd.Start(); err != nil {
326 return nil, fmt.Errorf("start ssh: %w", err)
327 }
328 s := &runnerSession{cmd: cmd, in: bufio.NewWriter(stdin), out: bufio.NewReader(stdout)}
329 s.enc = json.NewEncoder(s.in)
330 s.dec = json.NewDecoder(s.out)
331 return s, nil
332 }
333
334 func (s *runnerSession) call(req runnerRequest) (runnerReply, error) {
335 var rep runnerReply
336 s.mu.Lock()
337 defer s.mu.Unlock()
338 if err := s.enc.Encode(req); err != nil {
339 return rep, fmt.Errorf("connection lost: %w", err)
340 }
341 if err := s.in.Flush(); err != nil {
342 return rep, fmt.Errorf("connection lost: %w", err)
343 }
344 if err := s.dec.Decode(&rep); err != nil {
345 return rep, fmt.Errorf("connection lost: %w", err)
346 }
347 return rep, nil
348 }
349
350 func (s *runnerSession) close() {
351 s.mu.Lock()
352 _ = s.enc.Encode(runnerRequest{Op: "bye"})
353 _ = s.in.Flush()
354 s.mu.Unlock()
355 _ = s.cmd.Process.Signal(syscall.SIGTERM)
356 _ = s.cmd.Wait()
357 }
358
359 // ---------- execution ----------
360
361 // run executes a leased job and reports the outcome. A lost lease ends the job
362 // without reporting a result: another runner owns it now.
363 func (s *runnerSession) run(remote, cache string, job *runnerJob, lease time.Duration) {
364 log := &strings.Builder{}
365 work, err := prepareWorktree(remote, cache, job.Repo, job.SHA)
366 if err != nil {
367 // No checkout means no branch to commit a record to, so there is nothing to
368 // report: the lease expires and the queue gives the job to a runner that can.
369 fmt.Fprintf(os.Stderr, "vole runner: cannot prepare a checkout of %s at %s: %v\n",
370 job.Repo, format.SHA(job.SHA), err)
371 return
372 }
373 fail := func(format string, args ...any) {
374 msg := fmt.Sprintf(format, args...)
375 fmt.Fprintf(os.Stderr, "vole runner: %s\n", msg)
376 fmt.Fprintf(log, "\n%s\n", msg)
377 s.finish(remote, work, job, log, 1, nil, nil)
378 }
379 fmt.Fprintf(log, "=== %s · %s · %s @ %s\n", job.Repo, job.Job, job.Branch, format.SHA(job.SHA))
380
381 for _, step := range job.Steps {
382 fmt.Fprintf(log, "\n--- %s\n", step.Name)
383 code, chunk, lost := s.execStep(job.ID, work, step, lease)
384 log.WriteString(chunk)
385 if lost {
386 fmt.Fprintf(os.Stderr, "vole runner: lease lost while running %q; stopping\n", step.Name)
387 return // no finish: the queue already gave it to somebody else
388 }
389 if code != 0 {
390 fmt.Fprintf(log, "\nstep %q exited %d\n", step.Name, code)
391 s.finish(remote, work, job, log, code, nil, nil)
392 return
393 }
394 }
395 arts, releaseArts, err := publishArtifacts(remote, work, job, log)
396 if err != nil {
397 fail("%s", err)
398 return
399 }
400 if len(releaseArts) > 0 {
401 // The pointers ride the record's push, but a declaration that cannot name a
402 // release is a mistake worth failing on before the push rather than after it.
403 if err := checkReleaseJob(job); err != nil {
404 fail("%s", err)
405 return
406 }
407 }
408 if err := publishSite(remote, work, job, log); err != nil {
409 fail("%s", err)
410 return
411 }
412 s.finish(remote, work, job, log, 0, arts, releaseArts)
413 }
414
415 // finish uploads this attempt's log, commits its record and pushes it, and only then
416 // tells the server it is done.
417 //
418 // The bytes go in first, the same rule artifacts follow: a record is a set of
419 // pointers, and a pointer to an object the store does not have would fail the push
420 // that carries it. Then one signed commit carries the whole run — the record and any
421 // release pointers the build published — and one push lands it. The server creates
422 // nothing here: `pre-receive` checks the signature and the runner allowlist, and the
423 // finish message only closes the lease.
424 //
425 // A log that cannot be stored does not lose the result: the record then names no log
426 // and the failure is reported on stderr rather than swallowed. A record that cannot be
427 // pushed is not reported at all, because there is nothing for the server to read.
428 func (s *runnerSession) finish(remote, work string, job *runnerJob, log *strings.Builder, exit int, arts, releaseArts []runnerArtifact) {
429 oid, size, err := uploadLog(remote, job.Repo, log.String())
430 if err != nil {
431 fmt.Fprintf(os.Stderr, "vole runner: could not upload the log for %s: %v\n", job.ID, err)
432 }
433 artifacts := make([]runs.Artifact, 0, len(arts))
434 for _, a := range arts {
435 artifacts = append(artifacts, runs.Artifact{Name: a.Name, OID: a.OID, Size: a.Size})
436 }
437 rec := runs.Record{
438 RunID: job.ID, Attempt: job.Attempt, Repo: job.Repo, Ref: job.Ref, SHA: job.SHA,
439 Workflow: job.Workflow, WorkflowSHA: job.WorkflowSHA, Job: job.Job,
440 StepsSHA: job.StepsSHA, Exit: exit, Finished: time.Now().UTC(),
441 Signer: s.account, Artifacts: artifacts,
442 // The machine this attempt ran on. The runner binary's own build target,
443 // which is the honest thing it can say: a live probe would add a subprocess
444 // and a platform quirk for a fact the binary already knows.
445 OS: runtime.GOOS, Arch: runtime.GOARCH,
446 }
447 if oid != "" {
448 rec.Log = &lfs.Pointer{OID: oid, Size: size}
449 }
450
451 // Keep the lease through the push: the record is checked against it, and a slow
452 // upload must not turn a finished job into one that expired mid-report.
453 if rep, err := s.call(runnerRequest{Op: "heartbeat", Run: job.ID}); err != nil {
454 fmt.Fprintf(os.Stderr, "vole runner: lost the connection while finishing %s: %v\n", job.ID, err)
455 return
456 } else if !rep.OK {
457 fmt.Fprintf(os.Stderr, "vole runner: the lease for %s is gone; the result was not published\n", job.ID)
458 return
459 }
460
461 if err := publishRecord(remote, work, job, rec, releaseArts, log); err != nil {
462 fmt.Fprintf(os.Stderr, "vole runner: could not publish the record for %s: %v\n", job.ID, err)
463 return
464 }
465
466 rep, err := s.call(runnerRequest{Op: "finish", Run: job.ID})
467 switch {
468 case err != nil:
469 fmt.Fprintf(os.Stderr, "vole runner: connection lost after publishing %s; it is on the branch\n", job.ID)
470 case rep.Error != "":
471 fmt.Fprintf(os.Stderr, "vole runner: %s\n", rep.Error)
472 case !rep.OK:
473 fmt.Fprintf(os.Stderr, "vole runner: the lease for %s moved; the record is on the branch but the queue did not take it\n", job.ID)
474 }
475 }
476
477 // publishRecord commits the attempt's record, with the release pointers the build
478 // published beside it, and pushes once.
479 //
480 // One push per run is the point: the result and the artifacts it produced are one
481 // commit, so a release cannot carry bytes whose build has no record and a run does not
482 // pay a push per output. A lost race against somebody else's issue is re-applied on
483 // the new tip, the same way every other metadata write is.
484 func publishRecord(remote, work string, job *runnerJob, rec runs.Record, rel []runnerArtifact, log *strings.Builder) error {
485 url, err := client.RepoURL(remote, job.Repo)
486 if err != nil {
487 return err
488 }
489 gitDir, err := client.GitDirOf(work)
490 if err != nil {
491 return err
492 }
493 c, err := client.OpenBranch(gitDir, url, strings.TrimPrefix(metadata.Ref, "refs/heads/"), "vole-runner")
494 if err != nil {
495 return err
496 }
497 add := func(files map[string][]byte) error {
498 for _, a := range rel {
499 files[metadata.ReleasePath(job.Tag, a.Name)] = lfs.Pointer{OID: a.OID, Size: a.Size}.Encode()
500 }
501 return nil
502 }
503 write := func() error {
504 _, err := runs.WriteRecord(c.Repo, metadata.Ref, rec, c.Author, add)
505 return err
506 }
507 if err := write(); err != nil {
508 return err
509 }
510 if err := c.CommitAndPush(false, write); err != nil {
511 return err
512 }
513 if rec.Attempt > 0 {
514 fmt.Fprintf(log, "\nrecord: pushed %s\n", runs.RecordPath(rec.RunID, rec.Attempt))
515 }
516 if len(rel) > 0 {
517 fmt.Fprintf(log, "\nrelease: attached %d artifact(s) to %s\n", len(rel), job.Tag)
518 }
519 return nil
520 }
521
522 // checkReleaseJob refuses a workflow that publishes to a release it cannot name. It
523 // runs before the push so the refusal lands in the job's log, where somebody will see
524 // it, rather than as a push error after a successful build.
525 func checkReleaseJob(job *runnerJob) error {
526 if job.Tag == "" {
527 return fmt.Errorf("this job declares release artifacts but ran on a branch; a release " +
528 "is built by a tag push (add on.push.tags and push a tag)")
529 }
530 if !job.TagAnnotated {
531 return fmt.Errorf("%s is a lightweight tag; a release needs an annotated tag with notes "+
532 "(git tag -a %s -F notes.md)", job.Tag, job.Tag)
533 }
534 return nil
535 }
536
537 // uploadLog puts the log's bytes in the instance's object store over the same SSH
538 // transfer the artifacts use, and returns the id the store computed. An empty log
539 // uploads nothing and reports no object, so a run with no output has no log rather
540 // than an empty file.
541 func uploadLog(remote, repo, text string) (oid string, size int64, err error) {
542 if text == "" {
543 return "", 0, nil
544 }
545 url, err := client.RepoURL(remote, repo)
546 if err != nil {
547 return "", 0, err
548 }
549 tx, err := client.DialTransfer(url, "upload")
550 if err != nil {
551 return "", 0, err
552 }
553 defer tx.Close()
554
555 f, err := os.CreateTemp("", "vole-log-*")
556 if err != nil {
557 return "", 0, err
558 }
559 name := f.Name()
560 defer os.Remove(name)
561 if _, err := f.WriteString(text); err != nil {
562 f.Close()
563 return "", 0, err
564 }
565 if err := f.Close(); err != nil {
566 return "", 0, err
567 }
568 ptr, err := tx.Upload(name)
569 if err != nil {
570 return "", 0, err
571 }
572 return ptr.OID, ptr.Size, nil
573 }
574
575 // publishArtifacts expands the job's declared patterns in the checkout and uploads
576 // what they match through the same SSH object transfer git-lfs uses. The bytes go
577 // in before the finish, so the server can refuse a claim it cannot serve.
578 // publishArtifacts uploads every declared output and returns them twice: once as
579 // the run's outputs, once as the subset that belongs to the tag's release.
580 func publishArtifacts(remote, work string, job *runnerJob, log *strings.Builder) (all, release []runnerArtifact, err error) {
581 if len(job.Artifacts) == 0 {
582 return nil, nil, nil
583 }
584 files, err := collectArtifactFiles(work, job.Artifacts)
585 if err != nil {
586 return nil, nil, err
587 }
588 isRelease := map[string]bool{}
589 for _, pat := range job.ReleaseArtifacts {
590 isRelease[pat] = true
591 }
592 url, err := client.RepoURL(remote, job.Repo)
593 if err != nil {
594 return nil, nil, err
595 }
596 tx, err := client.DialTransfer(url, "upload")
597 if err != nil {
598 return nil, nil, fmt.Errorf("connect to publish %d artifact(s): %w", len(files), err)
599 }
600 defer tx.Close()
601
602 all = make([]runnerArtifact, 0, len(files))
603 for _, f := range files {
604 ptr, err := tx.Upload(filepath.Join(work, filepath.FromSlash(f.Path)))
605 if err != nil {
606 return nil, nil, fmt.Errorf("upload %s: %w", f.Path, err)
607 }
608 art := runnerArtifact{Name: f.Name, OID: ptr.OID, Size: ptr.Size}
609 fmt.Fprintf(log, "\nartifact %s (%s) sha256:%s\n", f.Name, format.Bytes(ptr.Size), ptr.OID)
610 all = append(all, art)
611 if isRelease[f.Pattern] {
612 release = append(release, art)
613 }
614 }
615 return all, release, nil
616 }
617
618 // publishSite deploys the job's declared directory as the repository's site, using
619 // the same snapshot code the pages CLI uses.
620 //
621 // The staleness check first: a site is a snapshot, so a run that finishes after a
622 // newer commit has landed would quietly revert the site to an older build. Refusing
623 // is the honest outcome — the log says why, and the job fails instead of publishing
624 // a stale tree that nobody notices for a week.
625 func publishSite(remote, work string, job *runnerJob, log *strings.Builder) error {
626 if job.Pages == nil || job.Pages.Path == "" {
627 return nil
628 }
629 url, err := client.RepoURL(remote, job.Repo)
630 if err != nil {
631 return err
632 }
633 tip, err := client.LSRemote(work, url, "refs/heads/"+job.Branch)
634 if err != nil {
635 return fmt.Errorf("check %s before deploying the site: %w", job.Branch, err)
636 }
637 switch {
638 case tip == "":
639 return fmt.Errorf("%s no longer exists, so there is no site to deploy to", job.Branch)
640 case tip != job.SHA:
641 return fmt.Errorf("this run built %s but %s is at %s now; the site was not deployed",
642 format.SHA(job.SHA), job.Branch, format.SHA(tip))
643 }
644
645 gitDir, err := client.GitDirOf(work)
646 if err != nil {
647 return err
648 }
649 c, err := client.OpenBranch(gitDir, url, job.Pages.Ref, "vole-runner")
650 if err != nil {
651 return err
652 }
653 dir := filepath.Join(work, filepath.FromSlash(job.Pages.Path))
654 msg := fmt.Sprintf("site: %s from run %s (%s)", job.Branch, format.SHA(job.SHA), job.ID)
655 if len(msg) > 200 {
656 msg = fmt.Sprintf("site: %s from run %s", job.Branch, job.ID)
657 }
658 diff, files, deployed, err := client.DeploySite(c, dir, msg, false)
659 if err != nil {
660 return err
661 }
662 if deployed {
663 fmt.Fprintf(log, "\nsite: published %d file(s) to %s (%s)\n", files, job.Pages.Ref, diff)
664 } else {
665 fmt.Fprintf(log, "\nsite: %s already matches %s\n", job.Pages.Ref, job.Pages.Path)
666 }
667 return nil
668 }
669
670 // artifactFile is one matched output. Pattern is the declaration that matched it,
671 // because a pattern can also mean "publish this to the release".
672 type artifactFile struct {
673 Path string // slash-separated, relative to the checkout
674 Name string // the download name: the file's base name
675 Pattern string // the declaration that matched
676 }
677
678 // collectArtifactFiles expands patterns against a checkout. A pattern that matches
679 // nothing is an error rather than an empty artifact list: a build that stopped
680 // producing the file is the failure a declaration is supposed to catch, and a
681 // silent publish of nothing is how that goes unnoticed for a week.
682 func collectArtifactFiles(root string, patterns []string) ([]artifactFile, error) {
683 matched := make([]int, len(patterns))
684 var files []artifactFile
685 byName := map[string]string{}
686 err := filepath.WalkDir(root, func(p string, d os.DirEntry, err error) error {
687 if err != nil {
688 return err
689 }
690 if d.IsDir() {
691 if d.Name() == ".git" {
692 return filepath.SkipDir
693 }
694 return nil
695 }
696 rel, err := filepath.Rel(root, p)
697 if err != nil {
698 return err
699 }
700 rel = filepath.ToSlash(rel)
701 for i, pat := range patterns {
702 if ok, err := path.Match(pat, rel); err != nil || !ok {
703 continue
704 }
705 matched[i]++
706 info, err := d.Info()
707 if err != nil {
708 return err
709 }
710 if !info.Mode().IsRegular() {
711 return nil // a symlink or device: not an artifact
712 }
713 name := path.Base(rel)
714 if prev, dup := byName[name]; dup {
715 return fmt.Errorf("two files would be published as %q (%s and %s); rename one "+
716 "or narrow the patterns", name, prev, rel)
717 }
718 byName[name] = rel
719 files = append(files, artifactFile{Path: rel, Name: name, Pattern: pat})
720 return nil
721 }
722 return nil
723 })
724 if err != nil {
725 return nil, err
726 }
727 for i, n := range matched {
728 if n == 0 {
729 return nil, fmt.Errorf("artifact pattern %q matched nothing in the checkout", patterns[i])
730 }
731 }
732 sort.Slice(files, func(i, j int) bool { return files[i].Name < files[j].Name })
733 return files, nil
734 }
735
736 // logBuffer collects a step's output. It is a sync buffer because os/exec copies
737 // stdout and stderr in separate goroutines. Nothing is shipped while the step runs:
738 // the whole log is uploaded once, when the attempt is over.
739 type logBuffer struct {
740 mu sync.Mutex
741 buf strings.Builder
742 }
743
744 func (w *logBuffer) Write(p []byte) (int, error) {
745 w.mu.Lock()
746 defer w.mu.Unlock()
747 return w.buf.Write(p)
748 }
749
750 func (w *logBuffer) String() string {
751 w.mu.Lock()
752 defer w.mu.Unlock()
753 return w.buf.String()
754 }
755
756 // execStep runs one shell step, heartbeating while it works. It returns the exit
757 // code, the output produced, and whether the lease was lost.
758 func (s *runnerSession) execStep(runID, work string, step runnerStep, lease time.Duration) (int, string, bool) {
759 timeout := 10 * time.Minute
760 if d, err := time.ParseDuration(step.Timeout); err == nil && d > 0 {
761 timeout = d
762 }
763 ctx, cancel := context.WithTimeout(context.Background(), timeout)
764 defer cancel()
765
766 cmd := exec.CommandContext(ctx, "sh", "-c", step.Run)
767 cmd.Dir = work
768 cmd.Env = append(os.Environ(), "CI=true", "VOLE_CI=1")
769 // Own process group: a step that spawns children must not outlive its timeout.
770 cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
771 w := &logBuffer{}
772 cmd.Stdout, cmd.Stderr = w, w
773
774 if err := cmd.Start(); err != nil {
775 return 1, fmt.Sprintf("cannot start: %v\n", err), false
776 }
777 done := make(chan error, 1)
778 go func() { done <- cmd.Wait() }()
779
780 interval := lease / 3
781 if interval < time.Second {
782 interval = time.Second
783 }
784 tick := time.NewTicker(interval)
785 defer tick.Stop()
786
787 for {
788 select {
789 case err := <-done:
790 chunk := w.String()
791 if ctx.Err() == context.DeadlineExceeded {
792 // The heartbeat will have extended the lease this far; report the
793 // timeout as a failure rather than leaving the job hanging.
794 return 1, chunk + fmt.Sprintf("\nstep timed out after %s\n", timeout), false
795 }
796 if err == nil {
797 return 0, chunk, false
798 }
799 var ee *exec.ExitError
800 if errors.As(err, &ee) {
801 return ee.ExitCode(), chunk, false
802 }
803 return 1, chunk + fmt.Sprintf("\n%v\n", err), false
804 case <-ctx.Done():
805 if cmd.Process != nil {
806 _ = syscall.Kill(-cmd.Process.Pid, syscall.SIGKILL)
807 }
808 // Wait for the reader goroutine before returning the output.
809 <-done
810 return 1, w.String() + fmt.Sprintf("\nstep timed out after %s\n", timeout), false
811 case <-tick.C:
812 // The lease is the only thing a heartbeat carries.
813 rep, err := s.call(runnerRequest{Op: "heartbeat", Run: runID})
814 if err != nil || !rep.OK {
815 return 1, "", true
816 }
817 }
818 }
819 }
820
821 // prepareWorktree fetches the commit to run and checks it out, from a cache that
822 // persists between jobs so a second run of the same repository is cheap.
823 func prepareWorktree(remote, cache, repo, sha string) (string, error) {
824 cloneURL, err := client.RepoURL(remote, repo)
825 if err != nil {
826 return "", err
827 }
828 dir := filepath.Join(cache, repo)
829
830 if _, err := os.Stat(filepath.Join(dir, ".git")); err != nil {
831 if err := os.RemoveAll(dir); err != nil {
832 return "", err
833 }
834 if out, err := exec.Command("git", "clone", "--quiet", cloneURL, dir).CombinedOutput(); err != nil {
835 return "", fmt.Errorf("git clone: %v: %s", err, out)
836 }
837 } else {
838 cmd := exec.Command("git", "-C", dir, "fetch", "--quiet", "--prune", "origin",
839 "+refs/heads/*:refs/remotes/origin/*")
840 if out, err := cmd.CombinedOutput(); err != nil {
841 return "", fmt.Errorf("git fetch: %v: %s", err, out)
842 }
843 }
844 // Start from the exact commit, with no leftovers from a previous run.
845 for _, args := range [][]string{
846 {"checkout", "--quiet", "--force", "--detach", sha},
847 {"clean", "-qffdx"},
848 } {
849 cmd := exec.Command("git", append([]string{"-C", dir}, args...)...)
850 if out, err := cmd.CombinedOutput(); err != nil {
851 // The commit may not be reachable from any branch yet: fetch it by id.
852 if args[0] == "checkout" {
853 if out2, err2 := exec.Command("git", "-C", dir, "fetch", "--quiet", "origin", sha).CombinedOutput(); err2 == nil {
854 if out3, err3 := exec.Command("git", "-C", dir, "checkout", "--quiet", "--force", "--detach", sha).CombinedOutput(); err3 == nil {
855 _ = out2
856 _ = out3
857 continue
858 }
859 }
860 }
861 return "", fmt.Errorf("git %s: %v: %s", strings.Join(args, " "), err, out)
862 }
863 }
864 return dir, nil
865 }