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.

Code

395 commit(s) on main

Compare History
update readme Keenan Nemetz · 38m ago 32290114
history blame raw 6.8 KiB
1package ssh
2
3// A finished run is stored by the server when the runner sends the record with its
4// finish over the authenticated channel. These tests speak the runner protocol over a
5// real SSH channel and check the boundary: the account comes from the key, the record
6// must describe the job the queue handed out, and a retry of the same record is
7// idempotent because the attempt is append-only.
8
9import (
10 "encoding/json"
11 "strings"
12 "testing"
13 "time"
14
15 xssh "golang.org/x/crypto/ssh"
16 "rodent.software/vole/ci/runs"
17 "rodent.software/vole/core/id"
18)
19
20// runnerConn is the wire, as a runner would see it.
21type runnerConn struct {
22 t *testing.T
23 sess *xssh.Session
24 enc *json.Encoder
25 dec *json.Decoder
26}
27
28func (h *harness) runner(t *testing.T, keyPath string) *runnerConn {
29 t.Helper()
30 client := h.dial(t, keyPath)
31 sess, err := client.NewSession()
32 if err != nil {
33 t.Fatalf("new session: %v", err)
34 }
35 t.Cleanup(func() { sess.Close() })
36 stdin, err := sess.StdinPipe()
37 if err != nil {
38 t.Fatalf("stdin: %v", err)
39 }
40 stdout, err := sess.StdoutPipe()
41 if err != nil {
42 t.Fatalf("stdout: %v", err)
43 }
44 if err := sess.Start("vole-runner"); err != nil {
45 t.Fatalf("start vole-runner: %v", err)
46 }
47 return &runnerConn{t: t, sess: sess, enc: json.NewEncoder(stdin), dec: json.NewDecoder(stdout)}
48}
49
50func (c *runnerConn) call(req runnerRequest) runnerReply {
51 c.t.Helper()
52 if err := c.enc.Encode(req); err != nil {
53 c.t.Fatalf("%s: %v", req.Op, err)
54 }
55 var rep runnerReply
56 if err := c.dec.Decode(&rep); err != nil {
57 c.t.Fatalf("%s: reading the reply: %v", req.Op, err)
58 }
59 return rep
60}
61
62// leaseRun enqueues one job and leases it, returning the queue's view of the run and
63// the job the runner was handed.
64func (h *harness) leaseRun(t *testing.T, c *runnerConn, jobName string) (*runs.Run, *runnerJob) {
65 t.Helper()
66 runID, err := id.New(time.Now().UTC())
67 if err != nil {
68 t.Fatal(err)
69 }
70 run := runs.Run{
71 ID: runID,
72 Repo: h.repoName,
73 Ref: "refs/heads/main",
74 Branch: "main",
75 SHA: strings.Repeat("ab", 20),
76 Workflow: ".vole/workflows/ci.yaml",
77 WorkflowSHA: strings.Repeat("cd", 20),
78 Job: jobName,
79 Steps: []runs.Step{{Name: "test", Run: "true"}},
80 Created: time.Now().UTC(),
81 }
82 // The unique index on (repo, ref, sha, workflow, job) is what makes ingestion
83 // idempotent, so each case below is its own job.
84 if _, err := h.runs.Enqueue(run, nil); err != nil {
85 t.Fatal(err)
86 }
87 lease := c.call(runnerRequest{Op: "lease"})
88 if lease.Run == nil {
89 t.Fatalf("no job was leased for %s: %+v", jobName, lease)
90 }
91 got, err := h.runs.Get(lease.Run.ID)
92 if err != nil || got == nil {
93 t.Fatalf("the leased run is not in the queue: %v", err)
94 }
95 return got, lease.Run
96}
97
98// recordFor builds the record a runner would send for a leased job.
99func recordFor(job *runnerJob, signer string) runs.Record {
100 now := time.Now().UTC()
101 rec := runs.Record{
102 RunID: job.ID, Attempt: job.Attempt, Repo: job.Repo, Ref: job.Ref,
103 SHA: job.SHA, Workflow: job.Workflow, WorkflowSHA: job.WorkflowSHA,
104 Job: job.Job, StepsSHA: job.StepsSHA, Exit: 0, Finished: now,
105 Signer: signer,
106 }
107 if len(job.Steps) > 0 {
108 rec.Steps = []runs.StepResult{{
109 Name: job.Steps[0].Name, Started: now.Add(-time.Second), Finished: now, Exit: 0,
110 }}
111 }
112 return rec
113}
114
115func TestAFinishStoresTheRecord(t *testing.T) {
116 t.Parallel()
117 h := newHarness(t)
118 c := h.runner(t, h.keyPath)
119
120 // Who am I: the reply names the account the connection authenticated as, so a runner
121 // can say whose results it will report.
122 who := c.call(runnerRequest{Op: "ping"})
123 if !who.OK || who.Name != h.writerID {
124 t.Fatalf("ping = %+v", who)
125 }
126
127 // The lease has to carry everything the record names: a runner cannot report what
128 // it was never told.
129 run, job := h.leaseRun(t, c, "build")
130 if job.Ref != run.Ref || job.StepsSHA == "" || job.WorkflowSHA != run.WorkflowSHA ||
131 job.Attempt != run.Attempts {
132 t.Fatalf("lease = %+v, run = %+v", job, run)
133 }
134
135 // 1. Nothing sent: refused, and nothing recorded.
136 rep := c.call(runnerRequest{Op: "finish", Run: job.ID})
137 if rep.Error == "" || !strings.Contains(rep.Error, "did not send a record") {
138 t.Fatalf("a finish with no record was accepted: %+v", rep)
139 }
140 if got, _ := h.runs.Get(run.ID); got.Status != runs.StatusRunning {
141 t.Fatalf("a refused finish changed the run: %+v", got)
142 }
143
144 // 2. A record that names another account is refused: the channel decides who is
145 // finishing, and the record has to be that account's.
146 rec := recordFor(job, "01M5READER0000000000000000Z")
147 rep = c.call(runnerRequest{Op: "finish", Run: job.ID, Record: &rec})
148 if rep.Error == "" || !strings.Contains(rep.Error, "reporter") {
149 t.Fatalf("a record naming another account was accepted: %+v", rep)
150 }
151
152 // 3. A record for an attempt the queue is not on is refused. Replay is how a
153 // runner would try to rewrite the history of an attempt it never ran.
154 rec = recordFor(job, h.writerID)
155 rec.Attempt = job.Attempt + 6
156 rep = c.call(runnerRequest{Op: "finish", Run: job.ID, Record: &rec})
157 if rep.Error == "" || !strings.Contains(rep.Error, "attempt") {
158 t.Fatalf("a record for another attempt was accepted: %+v", rep)
159 }
160
161 // 4. A record that describes another commit is refused.
162 rec = recordFor(job, h.writerID)
163 rec.SHA = strings.Repeat("ef", 20)
164 rep = c.call(runnerRequest{Op: "finish", Run: job.ID, Record: &rec})
165 if rep.Error == "" || !strings.Contains(rep.Error, "commit") {
166 t.Fatalf("a record for another commit was accepted: %+v", rep)
167 }
168
169 // 5. The real thing: the server stores the attempt and the run's result.
170 rec = recordFor(job, h.writerID)
171 rep = c.call(runnerRequest{Op: "finish", Run: job.ID, Record: &rec})
172 if rep.Error != "" || !rep.OK {
173 t.Fatalf("a finish with a record was refused: %+v", rep)
174 }
175 got, err := h.runs.Get(run.ID)
176 if err != nil || got == nil {
177 t.Fatal(err)
178 }
179 if got.Status != runs.StatusSuccess || got.ExitCode != 0 {
180 t.Fatalf("run = %+v", got)
181 }
182 att, err := h.runs.Attestation(run.ID)
183 if err != nil || att == nil {
184 t.Fatalf("no attestation: %v", err)
185 }
186 if att.Signer != h.writerID {
187 t.Fatalf("attestation = %+v", att)
188 }
189 stored, err := h.runs.Attempt(run.ID, job.Attempt)
190 if err != nil || stored == nil {
191 t.Fatalf("the attempt was not stored: %v", err)
192 }
193 if stored.Signer != h.writerID || stored.SHA != job.SHA || len(stored.Steps) != 1 {
194 t.Fatalf("stored attempt = %+v", stored)
195 }
196
197 // 6. A retry after a lost reply is the same attempt and is accepted.
198 rep = c.call(runnerRequest{Op: "finish", Run: job.ID, Record: &rec})
199 if rep.Error != "" || !rep.OK {
200 t.Fatalf("a retry of the same record was refused: %+v", rep)
201 }
202
203 // 7. Different bytes for an attempt that already has a record are refused.
204 changed := rec
205 changed.Exit = 1
206 rep = c.call(runnerRequest{Op: "finish", Run: job.ID, Record: &changed})
207 if rep.OK {
208 t.Fatalf("a rewritten record was accepted: %+v", rep)
209 }
210}
211