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 · 2h ago 4b8c9e1d
history blame raw 5.0 KiB
1package agent
2
3// The slot is one worker: one session, one job at a time. --jobs N is N of them in
4// one process, which is where the failure policy matters: a lost connection belongs
5// to the slot, and a job the runner refuses belongs to the runner.
6
7import (
8 "bufio"
9 "context"
10 "encoding/json"
11 "errors"
12 "fmt"
13 "io"
14 "testing"
15 "time"
16)
17
18func TestNextDelayDoublesAndCaps(t *testing.T) {
19 t.Parallel()
20 for _, tc := range []struct{ from, want time.Duration }{
21 {time.Second, 2 * time.Second},
22 {20 * time.Second, reconnectMax},
23 {reconnectMax, reconnectMax},
24 } {
25 if got := nextDelay(tc.from); got != tc.want {
26 t.Errorf("nextDelay(%s) = %s, want %s", tc.from, got, tc.want)
27 }
28 }
29}
30
31func TestSleepStopsWhenTheContextEnds(t *testing.T) {
32 t.Parallel()
33 ctx, cancel := context.WithCancel(context.Background())
34 cancel()
35 if sleep(ctx, time.Minute) {
36 t.Fatal("a cancelled context slept the whole delay")
37 }
38 if !sleep(context.Background(), time.Millisecond) {
39 t.Fatal("a live context did not finish a short wait")
40 }
41}
42
43// A refused job is the one error a slot must not retry: the runner and the instance
44// disagree about what a job may contain, and reconnecting asks the same question.
45func TestARefusedJobSurvivesWrapping(t *testing.T) {
46 t.Parallel()
47 refused := refusedJob{errors.New("run id is not a run id")}
48 wrapped := fmt.Errorf("slot stopped: %w", error(refused))
49 var got refusedJob
50 if !errors.As(wrapped, &got) {
51 t.Fatal("a wrapped refusal is no longer a refusal")
52 }
53 if got.Error() != "run id is not a run id" {
54 t.Fatalf("the refusal lost its reason: %q", got.Error())
55 }
56}
57
58// pipeSession is a session with no ssh under it: requests go into a pipe, and the
59// answering goroutine writes one reply per request from the list. It is enough to
60// exercise what a slot does before it touches a checkout or a container.
61func pipeSession(t *testing.T, replies ...runnerReply) *runnerSession {
62 t.Helper()
63 reqR, reqW := io.Pipe()
64 repR, repW := io.Pipe()
65 go func() {
66 defer repW.Close()
67 dec := json.NewDecoder(reqR)
68 enc := json.NewEncoder(repW)
69 for _, reply := range replies {
70 var req runnerRequest
71 if err := dec.Decode(&req); err != nil {
72 return
73 }
74 if err := enc.Encode(reply); err != nil {
75 return
76 }
77 }
78 }()
79 t.Cleanup(func() { _ = reqW.Close() })
80 return &runnerSession{
81 in: bufio.NewWriter(reqW),
82 out: bufio.NewReader(repR),
83 enc: json.NewEncoder(reqW),
84 dec: json.NewDecoder(repR),
85 }
86}
87
88// A leased job that does not have the shape this runner documents stops the slot
89// with a refusal, not with a retry loop that would lease it again.
90func TestASlotRefusesAJobItWillNotRun(t *testing.T) {
91 t.Parallel()
92 job := &runnerJob{ID: "not-a-ulid", Repo: "demo", SHA: "ab", Image: "alpine:3.22"}
93 sess := pipeSession(t, runnerReply{Op: "lease", OK: true, Run: job})
94 sl := &slot{sess: sess}
95
96 err := sl.serve(context.Background())
97 var refused refusedJob
98 if !errors.As(err, &refused) {
99 t.Fatalf("serve = %v, want a refused job", err)
100 }
101}
102
103// A lease that answers "nothing to do" is not a failure: a one-shot slot leaves,
104// and a resident one asks again. Either way it is not a refusal.
105func TestAnEmptyLeaseIsNotARefusal(t *testing.T) {
106 t.Parallel()
107 sess := pipeSession(t, runnerReply{Op: "lease", OK: true})
108 sl := &slot{sess: sess, opts: Options{Once: true}}
109
110 if err := sl.serve(context.Background()); err != nil {
111 t.Fatalf("serve = %v, want no error from an empty one-shot lease", err)
112 }
113}
114
115// silentSession is a session whose reader never gets a reply: what a resident runner
116// looks like while it is parked on a lease.
117func silentSession(t *testing.T) *runnerSession {
118 t.Helper()
119 reqR, reqW := io.Pipe()
120 repR, repW := io.Pipe()
121 go func() { _, _ = io.Copy(io.Discard, reqR) }()
122 t.Cleanup(func() {
123 _ = reqW.Close()
124 _ = repW.Close()
125 })
126 return &runnerSession{
127 in: bufio.NewWriter(reqW),
128 out: bufio.NewReader(repR),
129 enc: json.NewEncoder(reqW),
130 dec: json.NewDecoder(repR),
131 }
132}
133
134// A parked lease is the only wait a stopping runner cannot interrupt from inside, so
135// it has to end when the runner is asked to stop — otherwise a shutdown sits out the
136// whole wait window while nothing is running.
137func TestAParkedLeaseEndsWhenTheRunnerStops(t *testing.T) {
138 t.Parallel()
139 sess := silentSession(t)
140 ctx, cancel := context.WithCancel(context.Background())
141 done := make(chan error, 1)
142 go func() {
143 _, err := sess.lease(ctx, runnerRequest{Op: "lease", Wait: 25})
144 done <- err
145 }()
146 cancel()
147 select {
148 case err := <-done:
149 if !errors.Is(err, errStopped) {
150 t.Fatalf("a canceled lease returned %v, want errStopped", err)
151 }
152 case <-time.After(5 * time.Second):
153 t.Fatal("a canceled lease stayed parked")
154 }
155}
156
157// A slot that was told to stop before it leased anything does not lease at all: the
158// shutdown is not a last chance to pick up work.
159func TestAStoppedSlotTakesNoWork(t *testing.T) {
160 t.Parallel()
161 sess := silentSession(t)
162 ctx, cancel := context.WithCancel(context.Background())
163 cancel()
164 if err := (&slot{sess: sess}).serve(ctx); err != nil {
165 t.Fatalf("serve = %v, want no error from a stopped slot", err)
166 }
167}
168