amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package host
2
3import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "log"
9 "os"
10 "path/filepath"
11 "strings"
12 "sync"
13 "testing"
14 "time"
15
16 "github.com/carlosframework/platform/internal/activator"
17 "github.com/carlosframework/platform/internal/hostclaim"
18 "github.com/carlosframework/platform/internal/provision"
19 "github.com/carlosframework/platform/internal/registry"
20 "github.com/carlosframework/platform/internal/release"
21 "github.com/carlosframework/platform/internal/runs"
22)
23
24const runHost = "acme.example.com"
25
26func runRecord() provision.Record {
27 return provision.Record{Schema: 1, Host: runHost, Slug: "acme", App: "shop", Account: "bab", Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}
28}
29
30func runOwnedRoute(host string) func(string) (*registry.Route, error) {
31 return func(h string) (*registry.Route, error) {
32 if h != host {
33 return nil, nil
34 }
35 return &registry.Route{Host: host, Kind: "instance", Hibernate: true, Socket: "/run/x.sock", Provisioned: true, App: "shop", Account: "bab"}, nil
36 }
37}
38
39func noRoutes(string) (*registry.Route, error) { return nil, nil }
40
41// queueRun writes a record whose run is queued, with its request.json.
42func queueRun(t *testing.T, s release.Store, deadline time.Time) (provision.Record, string) {
43 t.Helper()
44 rec := runRecord()
45 if err := provision.CreateRecord(context.Background(), s, rec); err != nil {
46 t.Fatal(err)
47 }
48 id := "run_abcdefghijklmnopqrstuvwxyz"
49 inc, _ := hostclaimIncarnation(rec)
50 req := runs.Request{Schema: 1, ID: id, Account: rec.Account, App: rec.App, Host: rec.Host, Incarnation: inc, Argv: []string{"import"}, TimeoutS: 60, RequestedAt: time.Now()}
51 putJSONObject(t, s, runs.RequestKey(rec.Account, rec.App, id), req)
52 got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunAdmit(provision.RunAdmission{ID: id, By: "alice@keymail.dev", Deadline: deadline}))
53 if err != nil {
54 t.Fatal(err)
55 }
56 return *got, id
57}
58
59type execLog struct {
60 mu sync.Mutex
61 ids []string
62}
63
64func (l *execLog) exec(_ context.Context, _ provision.Record, req runs.Request, _ registry.Route, _ *runHandle) {
65 l.mu.Lock()
66 l.ids = append(l.ids, req.ID)
67 l.mu.Unlock()
68}
69func (l *execLog) count() int { l.mu.Lock(); defer l.mu.Unlock(); return len(l.ids) }
70
71// testWorker execs inline, so a test counts launches as soon as consider
72// returns.
73func testWorker(s release.Store, routes func(string) (*registry.Route, error), box, boot string, log *execLog) *runsWorker {
74 w := newRunsWorker(s, routes, nil, "/data", box, boot)
75 w.exec = log.exec
76 w.start = func(f func()) { f() }
77 w.booted.Store(true) // a test of the boot gate unsets it
78 return w
79}
80
81func currentRun(t *testing.T, s release.Store) *provision.RunState {
82 t.Helper()
83 rec, err := provision.GetRecord(context.Background(), s, "bab", "shop", runHost)
84 if err != nil {
85 t.Fatal(err)
86 }
87 return rec.ActiveRun
88}
89
90func TestRunsWorkerPicksOnlyRunsOnItsOwnRows(t *testing.T) {
91 s := release.NewDirStore(t.TempDir())
92 rec, id := queueRun(t, s, time.Now().Add(time.Minute))
93 var other, mine execLog
94 testWorker(s, noRoutes, "box-2", "boot-2", &other).consider(context.Background(), rec)
95 if other.count() != 0 || currentRun(t, s).State != provision.RunQueued {
96 t.Fatal("a box that does not own the row acted on the run")
97 }
98 testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &mine).consider(context.Background(), rec)
99 r := currentRun(t, s)
100 if mine.count() != 1 || r.State != provision.RunPicked || r.Box != "box-1" || r.BootID != "boot-1" || r.ID != id {
101 t.Fatalf("pickup: execs %d, run %+v", mine.count(), r)
102 }
103 if d := time.Until(r.Deadline); d < 50*time.Second || d > 61*time.Second {
104 t.Errorf("the picked deadline is the run's timeout (60 s), got %s", d)
105 }
106}
107
108func TestRunsWorkerPickupRaceLaunchesOnce(t *testing.T) {
109 s := release.NewDirStore(t.TempDir())
110 rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
111 var a, b execLog
112 wa := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &a)
113 wb := testWorker(s, runOwnedRoute(runHost), "box-2", "boot-2", &b)
114 var wg sync.WaitGroup
115 for _, w := range []*runsWorker{wa, wb} {
116 wg.Add(1)
117 go func() { defer wg.Done(); w.consider(context.Background(), rec) }()
118 }
119 wg.Wait()
120 if a.count()+b.count() != 1 {
121 t.Fatalf("launched %d times, want once", a.count()+b.count())
122 }
123}
124
125func TestRunsWorkerLeavesARunPastItsPickupDeadline(t *testing.T) {
126 s := release.NewDirStore(t.TempDir())
127 rec, _ := queueRun(t, s, time.Now().Add(-time.Second))
128 var l execLog
129 testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), rec)
130 if l.count() != 0 || currentRun(t, s).State != provision.RunQueued {
131 t.Fatal("a box picked a run past its pickup deadline: that run is the expiry's")
132 }
133}
134
135func TestRunsWorkerNeverPicksACancelledOrExpiredRun(t *testing.T) {
136 for _, mut := range []func(string) func(*provision.Record) error{provision.RunCancelTransition, provision.RunExpire} {
137 s := release.NewDirStore(t.TempDir())
138 rec, id := queueRun(t, s, time.Now().Add(time.Minute))
139 if _, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, mut(id)); err != nil {
140 t.Fatal(err)
141 }
142 var l execLog
143 // rec is the stale read the pass noted, still queued: the CAS refuses.
144 testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), rec)
145 if l.count() != 0 {
146 t.Fatal("a cancelled or expired run was launched (codex R2-6)")
147 }
148 }
149}
150
151func TestRunsWorkerClosesOutARunPickedUnderAnEarlierBoot(t *testing.T) {
152 s := release.NewDirStore(t.TempDir())
153 _, id := queueRun(t, s, time.Now().Add(time.Minute))
154 got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-1", "boot-old", time.Now().Add(time.Hour)))
155 if err != nil {
156 t.Fatal(err)
157 }
158 if _, err := s.Put(context.Background(), runs.ChunkKey("bab", "shop", id, 4, runs.StreamOut), []byte("x"), ""); err != nil {
159 t.Fatal(err)
160 }
161 var l execLog
162 testWorker(s, runOwnedRoute(runHost), "box-1", "boot-new", &l).consider(context.Background(), *got)
163 if l.count() != 0 {
164 t.Fatal("a run picked before an agent restart was run again")
165 }
166 res, err := runs.ReadResult(context.Background(), s, "bab", "shop", id)
167 if err != nil || res == nil || res.Interrupted != runs.InterruptedAgentRestart || res.LastSeq != 4 || res.Box != "box-1" {
168 t.Fatalf("result %+v, %v", res, err)
169 }
170 if r := currentRun(t, s); r.State != provision.RunDone {
171 t.Fatalf("run %+v, want done", r)
172 }
173}
174
175func TestRunsWorkerForwardsACancelToItsRun(t *testing.T) {
176 s := release.NewDirStore(t.TempDir())
177 _, id := queueRun(t, s, time.Now().Add(time.Minute))
178 var l execLog
179 w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
180 h := newRunHandle()
181 w.active[id] = h
182 got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
183 if err != nil {
184 t.Fatal(err)
185 }
186 if got, err = provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunCancelTransition(id)); err != nil {
187 t.Fatal(err)
188 }
189 w.consider(context.Background(), *got)
190 select {
191 case <-h.cancel:
192 default:
193 t.Fatal("the cancel did not reach the run")
194 }
195}
196
197func TestBootPassPicksARunQueuedBeforeStart(t *testing.T) {
198 s := release.NewDirStore(t.TempDir())
199 queueRun(t, s, time.Now().Add(time.Minute))
200 var l execLog
201 w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
202 w.dataDir = t.TempDir()
203 w.bootPass(context.Background())
204 if l.count() != 1 {
205 t.Fatal("the boot pass did not pick a run queued before agent start")
206 }
207}
208
209func TestRunsWorkerNotesAndPasses(t *testing.T) {
210 s := release.NewDirStore(t.TempDir())
211 rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
212 var l execLog
213 w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
214 w.note(rec)
215 w.note(rec)
216 w.pass(context.Background())
217 w.pass(context.Background())
218 if l.count() != 1 {
219 t.Fatalf("execs %d, want 1: a noted record is acted on once", l.count())
220 }
221}
222
223func TestRunDirNamesAgree(t *testing.T) {
224 if runs.DirName != activator.CmdRunDirName {
225 t.Fatalf("runs.DirName %q, activator.CmdRunDirName %q", runs.DirName, activator.CmdRunDirName)
226 }
227}
228
229func hostclaimIncarnation(rec provision.Record) (string, error) {
230 return hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
231}
232
233func putJSONObject(t *testing.T, s release.Store, key string, v any) {
234 t.Helper()
235 raw, err := json.Marshal(v)
236 if err != nil {
237 t.Fatal(err)
238 }
239 if _, err := s.Put(context.Background(), key, raw, ""); err != nil {
240 t.Fatal(err)
241 }
242}
243
244// A run another box picked is that box's: this box neither closes it out
245// nor forwards anything to it, whatever boot picked it.
246func TestRunsWorkerLeavesARunAnotherBoxPicked(t *testing.T) {
247 s := release.NewDirStore(t.TempDir())
248 _, id := queueRun(t, s, time.Now().Add(time.Minute))
249 got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-2", "boot-old", time.Now().Add(time.Hour)))
250 if err != nil {
251 t.Fatal(err)
252 }
253 var l execLog
254 testWorker(s, runOwnedRoute(runHost), "box-1", "boot-new", &l).consider(context.Background(), *got)
255 if res, err := runs.ReadResult(context.Background(), s, "bab", "shop", id); err != nil || res != nil {
256 t.Fatalf("a box wrote a result for a run another box picked: %+v, %v", res, err)
257 }
258 if r := currentRun(t, s); l.count() != 0 || r.State != provision.RunPicked || r.Box != "box-2" {
259 t.Fatalf("execs %d, run %+v", l.count(), r)
260 }
261}
262
263// A run this box picked under its own boot is not an agent restart, even
264// when the worker holds no handle for it (a stale record noted after the
265// run's goroutine let go).
266func TestRunsWorkerLeavesItsOwnRunUnderThisBoot(t *testing.T) {
267 s := release.NewDirStore(t.TempDir())
268 _, id := queueRun(t, s, time.Now().Add(time.Minute))
269 got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
270 if err != nil {
271 t.Fatal(err)
272 }
273 var l execLog
274 testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), *got)
275 if res, err := runs.ReadResult(context.Background(), s, "bab", "shop", id); err != nil || res != nil {
276 t.Fatalf("a run of this boot was closed out as interrupted: %+v, %v", res, err)
277 }
278 if r := currentRun(t, s); l.count() != 0 || r.State != provision.RunPicked {
279 t.Fatalf("execs %d, run %+v", l.count(), r)
280 }
281}
282
283// A run of this boot that is no longer executing and has its result, but is
284// still picked (its done CAS failed past the finish bound), is CASed to done
285// on the next pass, so it does not hold the instance until the lost sweep.
286func TestRunsWorkerFinishesItsOwnStrandedRun(t *testing.T) {
287 for _, state := range []string{provision.RunPicked, provision.RunCancel} {
288 s := release.NewDirStore(t.TempDir())
289 _, id := queueRun(t, s, time.Now().Add(time.Minute))
290 got, err := provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
291 if err != nil {
292 t.Fatal(err)
293 }
294 if state == provision.RunCancel {
295 if got, err = provision.SetRecordRun(context.Background(), s, "bab", "shop", runHost, provision.RunCancelTransition(id)); err != nil {
296 t.Fatal(err)
297 }
298 }
299 if err := runs.WriteResult(context.Background(), s, "bab", "shop", id, runs.Result{Exit: 3, Box: "box-1"}); err != nil {
300 t.Fatal(err)
301 }
302 var l execLog
303 testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), *got)
304 if r := currentRun(t, s); l.count() != 0 || r.State != provision.RunDone {
305 t.Fatalf("%s: execs %d, run %+v, want done", state, l.count(), r)
306 }
307 if res, _ := runs.ReadResult(context.Background(), s, "bab", "shop", id); res == nil || res.Exit != 3 {
308 t.Fatalf("%s: the box's result changed: %+v", state, res)
309 }
310 }
311}
312
313// Only a provisioned row for the record's account and app that the
314// activator manages is this box's to pick for.
315func TestRunsWorkerPicksOnlyForActivatorManagedRowsOfTheRecord(t *testing.T) {
316 for name, mut := range map[string]func(*registry.Route){
317 "unprovisioned": func(rt *registry.Route) { rt.Provisioned = false },
318 "another account": func(rt *registry.Route) { rt.Account = "other" },
319 "another app": func(rt *registry.Route) { rt.App = "other" },
320 "not hibernating": func(rt *registry.Route) { rt.Hibernate = false },
321 "unit-backed": func(rt *registry.Route) { rt.Backing = "unit" },
322 "not an instance row": func(rt *registry.Route) { rt.Kind = "service" },
323 } {
324 s := release.NewDirStore(t.TempDir())
325 rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
326 routes := func(h string) (*registry.Route, error) {
327 rt, err := runOwnedRoute(runHost)(h)
328 if rt != nil {
329 mut(rt)
330 }
331 return rt, err
332 }
333 var l execLog
334 testWorker(s, routes, "box-1", "boot-1", &l).consider(context.Background(), rec)
335 if l.count() != 0 || currentRun(t, s).State != provision.RunQueued {
336 t.Errorf("%s: the box picked a run for a row it does not run", name)
337 }
338 }
339}
340
341// A request.json that is not the run the record names, or whose timeout is
342// outside 1 s to 6 h, is never picked: the run is left for the expiry.
343func TestRunsWorkerLeavesARequestThatDoesNotMatchItsRecord(t *testing.T) {
344 for name, mut := range map[string]func(*runs.Request){
345 "no timeout": func(r *runs.Request) { r.TimeoutS = 0 },
346 "timeout over 6h": func(r *runs.Request) { r.TimeoutS = 6*3600 + 1 },
347 "another id": func(r *runs.Request) { r.ID = "run_bbbbbbbbbbbbbbbbbbbbbbbbbb" },
348 "another account": func(r *runs.Request) { r.Account = "other" },
349 "another app": func(r *runs.Request) { r.App = "other" },
350 "another host": func(r *runs.Request) { r.Host = "other.example.com" },
351 } {
352 s := release.NewDirStore(t.TempDir())
353 rec, id := queueRun(t, s, time.Now().Add(time.Minute))
354 req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
355 if err != nil {
356 t.Fatal(err)
357 }
358 mut(req)
359 key := runs.RequestKey(rec.Account, rec.App, id)
360 if err := s.Delete(context.Background(), key); err != nil {
361 t.Fatal(err)
362 }
363 putJSONObject(t, s, key, req)
364 var l execLog
365 w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
366 w.consider(context.Background(), rec)
367 if l.count() != 0 || currentRun(t, s).State != provision.RunQueued || len(w.active) != 0 {
368 t.Errorf("%s: picked (execs %d, run %+v)", name, l.count(), currentRun(t, s))
369 }
370 }
371}
372
373// A refused request.json is logged once, not on every pass that sees the
374// run still queued.
375func TestRunsWorkerLogsAnUnfitRequestOnce(t *testing.T) {
376 s := release.NewDirStore(t.TempDir())
377 rec, id := queueRun(t, s, time.Now().Add(time.Minute))
378 req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
379 if err != nil {
380 t.Fatal(err)
381 }
382 req.TimeoutS = 0
383 key := runs.RequestKey(rec.Account, rec.App, id)
384 if err := s.Delete(context.Background(), key); err != nil {
385 t.Fatal(err)
386 }
387 putJSONObject(t, s, key, req)
388 var buf bytes.Buffer
389 log.SetOutput(&buf)
390 defer log.SetOutput(os.Stderr)
391 var l execLog
392 w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
393 for range 3 {
394 w.consider(context.Background(), rec)
395 }
396 if n := strings.Count(buf.String(), "not picked"); n != 1 {
397 t.Fatalf("logged %d times, want once:\n%s", n, buf.String())
398 }
399}
400
401// lostAnswer answers every Put of the record with an error; when land is
402// set the write lands first, as a request whose answer was lost does.
403type lostAnswer struct {
404 release.Store
405 land bool
406}
407
408func (l *lostAnswer) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
409 if key != provision.RecordKey("bab", "shop", runHost) {
410 return l.Store.Put(ctx, key, body, ifMatch)
411 }
412 if l.land {
413 if _, err := l.Store.Put(ctx, key, body, ifMatch); err != nil {
414 return "", err
415 }
416 }
417 return "", errors.New("connection reset by peer")
418}
419
420// A pickup CAS whose answer was lost is read back: if the record names
421// this run as this box's under this boot, the box launches it, so the run
422// is not stranded until the sweep calls it lost.
423func TestRunsWorkerLaunchesAPickupWhoseAnswerWasLost(t *testing.T) {
424 dir := release.NewDirStore(t.TempDir())
425 rec, id := queueRun(t, dir, time.Now().Add(time.Minute))
426 var l execLog
427 w := testWorker(&lostAnswer{Store: dir, land: true}, runOwnedRoute(runHost), "box-1", "boot-1", &l)
428 w.consider(context.Background(), rec)
429 if r := currentRun(t, dir); l.count() != 1 || r.State != provision.RunPicked || r.Box != "box-1" || r.ID != id {
430 t.Fatalf("execs %d, run %+v", l.count(), r)
431 }
432}
433
434// A pickup CAS that failed and did not land is forgotten: the run stays
435// queued and nothing is launched.
436func TestRunsWorkerForgetsAPickupThatDidNotLand(t *testing.T) {
437 dir := release.NewDirStore(t.TempDir())
438 rec, _ := queueRun(t, dir, time.Now().Add(time.Minute))
439 var l execLog
440 w := testWorker(&lostAnswer{Store: dir}, runOwnedRoute(runHost), "box-1", "boot-1", &l)
441 w.consider(context.Background(), rec)
442 if l.count() != 0 || currentRun(t, dir).State != provision.RunQueued || len(w.active) != 0 {
443 t.Fatalf("execs %d, run %+v, active %d", l.count(), currentRun(t, dir), len(w.active))
444 }
445}
446
447func TestRunsHookIsOnlyThereWhenArmed(t *testing.T) {
448 armedRuns.Store(nil)
449 if cfg := withRunsHook(provision.ReconcileConfig{}); cfg.RunSeen != nil {
450 t.Fatal("an unarmed box must not hook reconcile")
451 }
452 runRunsPass(context.Background()) // no-op, no panic
453 runRunsBootPass(context.Background()) // ditto
454 var l execLog
455 w := testWorker(release.NewDirStore(t.TempDir()), noRoutes, "box-1", "boot-1", &l)
456 armedRuns.Store(w)
457 t.Cleanup(func() { armedRuns.Store(nil) })
458 cfg := withRunsHook(provision.ReconcileConfig{})
459 if cfg.RunSeen == nil {
460 t.Fatal("an armed box hooks reconcile")
461 }
462 cfg.RunSeen(runRecord())
463 if len(w.noted) != 1 {
464 t.Fatal("the hook does not reach the worker")
465 }
466}
467
468// listGate holds every List until released, so a test can act while the
469// boot pass is part-way through.
470type listGate struct {
471 release.Store
472 entered chan struct{}
473 proceed chan struct{}
474 once sync.Once
475}
476
477func (g *listGate) List(ctx context.Context, prefix string) ([]string, error) {
478 g.once.Do(func() { close(g.entered) })
479 <-g.proceed
480 return g.Store.List(ctx, prefix)
481}
482
483// The boot pass deletes every .carlos-run directory (run spec §5.8), so no
484// reconcile-driven pass may pick or launch anything before it has finished:
485// a run it launched would have its directory swept from under it.
486func TestRunsPassDoesNothingUntilTheBootPassHasFinished(t *testing.T) {
487 dir := release.NewDirStore(t.TempDir())
488 rec, _ := queueRun(t, dir, time.Now().Add(time.Minute))
489 gate := &listGate{Store: dir, entered: make(chan struct{}), proceed: make(chan struct{})}
490 var l execLog
491 w := testWorker(gate, runOwnedRoute(runHost), "box-1", "boot-1", &l)
492 w.booted.Store(false)
493 w.dataDir = t.TempDir()
494 stale := filepath.Join(provision.InstanceDir(w.dataDir, runHost), runs.DirName, "run_old")
495 if err := os.MkdirAll(stale, 0o700); err != nil {
496 t.Fatal(err)
497 }
498 armedRuns.Store(w)
499 t.Cleanup(func() { armedRuns.Store(nil) })
500
501 // Reconcile ran before the boot pass: the hook notes, the pass waits.
502 withRunsHook(provision.ReconcileConfig{}).RunSeen(rec)
503 runRunsPass(context.Background())
504 if l.count() != 0 || currentRun(t, dir).State != provision.RunQueued {
505 t.Fatal("a pass picked a run before the boot pass ran")
506 }
507
508 done := make(chan struct{})
509 go func() { runRunsBootPass(context.Background()); close(done) }()
510 <-gate.entered // the sweep is done, the boot pass is reading records
511 if _, err := os.Stat(stale); !errors.Is(err, os.ErrNotExist) {
512 t.Fatalf("the boot pass did not sweep first: %v", err)
513 }
514 runRunsPass(context.Background())
515 if l.count() != 0 || currentRun(t, dir).State != provision.RunQueued {
516 t.Fatal("a pass picked a run while the boot pass was still running")
517 }
518 close(gate.proceed)
519 <-done
520 if l.count() != 1 {
521 t.Fatalf("execs %d, want 1: the boot pass picks the run queued before start", l.count())
522 }
523}
524
525// A boot pass that cannot read the records still opens the gate: the runs
526// it could not see are the next passes' to pick.
527func TestRunsPassActsAfterABootPassThatFailed(t *testing.T) {
528 dir := release.NewDirStore(t.TempDir())
529 rec, _ := queueRun(t, dir, time.Now().Add(time.Minute))
530 var l execLog
531 w := testWorker(listFails{dir}, runOwnedRoute(runHost), "box-1", "boot-1", &l)
532 w.booted.Store(false)
533 w.dataDir = t.TempDir()
534 w.note(rec)
535 w.pass(context.Background())
536 if l.count() != 0 {
537 t.Fatal("a pass acted before the boot pass")
538 }
539 w.bootPass(context.Background())
540 w.pass(context.Background())
541 if l.count() != 1 {
542 t.Fatalf("execs %d, want 1", l.count())
543 }
544}
545
546type listFails struct{ release.Store }
547
548func (listFails) List(context.Context, string) ([]string, error) {
549 return nil, errors.New("store down")
550}
551