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 "context"
5 "crypto/sha256"
6 "encoding/hex"
7 "errors"
8 "io"
9 "os"
10 "strings"
11 "sync"
12 "sync/atomic"
13 "testing"
14 "time"
15
16 "github.com/carlosframework/platform/internal/activator"
17 "github.com/carlosframework/platform/internal/instancectl"
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 "github.com/carlosframework/platform/internal/seeds"
23)
24
25type fakeHandle struct {
26 mu sync.Mutex
27 polls chan instancectl.CmdStatus
28 stops int
29 released bool
30 interrupted string
31 askedAt time.Time // when the interrupting stop was asked for; zero is "long ago"
32 afterStop *instancectl.CmdStatus
33 stopErr error
34}
35
36func (h *fakeHandle) interrupt(why string) {
37 h.mu.Lock()
38 defer h.mu.Unlock()
39 h.interrupted, h.askedAt = why, time.Now()
40}
41
42func (h *fakeHandle) Invocation() string { return "inv1" }
43func (h *fakeHandle) Poll(ctx context.Context) (instancectl.CmdStatus, error) {
44 h.mu.Lock()
45 if h.stops > 0 {
46 st := instancectl.CmdStatus{State: instancectl.CmdGone}
47 if h.afterStop != nil {
48 st = *h.afterStop
49 }
50 h.mu.Unlock()
51 return st, nil
52 }
53 h.mu.Unlock()
54 select {
55 case st := <-h.polls:
56 return st, nil
57 case <-ctx.Done():
58 return instancectl.CmdStatus{}, ctx.Err()
59 }
60}
61func (h *fakeHandle) Interrupted() (string, time.Time) {
62 h.mu.Lock()
63 defer h.mu.Unlock()
64 return h.interrupted, h.askedAt
65}
66func (h *fakeHandle) Stop(context.Context) error {
67 h.mu.Lock()
68 h.stops++
69 err := h.stopErr
70 h.mu.Unlock()
71 return err
72}
73func (h *fakeHandle) Release() { h.mu.Lock(); h.released = true; h.mu.Unlock() }
74
75// fakeLauncher counts the units it starts. With noUnit set it starts none:
76// a test that must start no unit fails at once instead of following one.
77type fakeLauncher struct {
78 mu sync.Mutex
79 calls int
80 units int
81 noUnit bool
82 err error
83 handle *fakeHandle
84 out string
85 errOut string
86}
87
88func (l *fakeLauncher) started() int { l.mu.Lock(); defer l.mu.Unlock(); return l.units }
89
90func (l *fakeLauncher) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
91 l.mu.Lock()
92 l.calls++
93 l.mu.Unlock()
94 if l.err != nil {
95 return nil, l.err
96 }
97 if spec.Prepare != nil {
98 if err := spec.Prepare(ctx); err != nil {
99 return nil, err
100 }
101 }
102 l.mu.Lock()
103 l.units++
104 l.mu.Unlock()
105 if l.noUnit {
106 return nil, errors.New("fake: a unit was started")
107 }
108 io.WriteString(spec.Stdout, l.out)
109 io.WriteString(spec.Stderr, l.errOut)
110 return l.handle, nil
111}
112
113// pickedRun queues and picks a run for box-1/boot-1 and returns what
114// execute is called with.
115func pickedRun(t *testing.T, s release.Store, timeoutS int64) (provision.Record, runs.Request) {
116 t.Helper()
117 rec, id := queueRun(t, s, time.Now().Add(time.Minute))
118 req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
119 if err != nil {
120 t.Fatal(err)
121 }
122 req.TimeoutS = timeoutS
123 got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
124 if err != nil {
125 t.Fatal(err)
126 }
127 return *got, *req
128}
129
130// execWorker is a box-1/boot-1 worker over a data dir that holds the
131// instance's directory, as a real box does.
132func execWorker(t *testing.T, s release.Store, l runLauncher, boot string) *runsWorker {
133 t.Helper()
134 dataDir := t.TempDir()
135 if err := os.MkdirAll(provision.InstanceDir(dataDir, runHost), 0o700); err != nil {
136 t.Fatal(err)
137 }
138 return newRunsWorker(s, runOwnedRoute(runHost), l, dataDir, "box-1", boot)
139}
140
141func runWith(t *testing.T, s release.Store, l *fakeLauncher, rec provision.Record, req runs.Request, h *runHandle) *runs.Result {
142 t.Helper()
143 return runOn(t, execWorker(t, s, l, "boot-1"), s, rec, req, h)
144}
145
146func runOn(t *testing.T, w *runsWorker, s release.Store, rec provision.Record, req runs.Request, h *runHandle) *runs.Result {
147 t.Helper()
148 rt, _ := runOwnedRoute(runHost)(runHost)
149 w.active[req.ID] = h
150 w.execute(context.Background(), rec, req, *rt, h)
151 res, err := runs.ReadResult(context.Background(), s, rec.Account, rec.App, req.ID)
152 if err != nil || res == nil {
153 t.Fatalf("no result: %v", err)
154 }
155 if w.active[req.ID] != nil {
156 t.Error("the run was not forgotten")
157 }
158 if _, err := os.Stat(runDirOf(w, req)); !errors.Is(err, os.ErrNotExist) {
159 t.Errorf("the run directory is still there: %v", err)
160 }
161 return res
162}
163
164func runDirOf(w *runsWorker, req runs.Request) string {
165 return provision.InstanceDir(w.dataDir, runHost) + "/" + runs.DirName + "/" + req.ID
166}
167
168func TestRunExecutesAndWritesTheResultAfterTheLastChunk(t *testing.T) {
169 s := &putOrder{Store: release.NewDirStore(t.TempDir())}
170 rec, req := pickedRun(t, s, 60)
171 h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
172 h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "exit-code", Status: "3", Code: "1"}
173 l := &fakeLauncher{handle: h, out: "hello\n", errOut: "oops\n"}
174 res := runWith(t, s, l, rec, req, newRunHandle())
175 if res.Exit != 3 || res.Signal != "" || res.LastSeq != 2 || res.Box != "box-1" || res.Started.IsZero() || res.Ended.IsZero() {
176 t.Fatalf("result %+v", res)
177 }
178 if h.stops == 0 || !h.released {
179 t.Fatal("the unit was not stopped after done, or the hold not released")
180 }
181 if r := currentRun(t, s.Store); r.State != provision.RunDone {
182 t.Fatalf("run %+v, want done", r)
183 }
184 if !s.resultAfterChunks() {
185 t.Fatalf("puts %q: result.json must come after every chunk", s.keys)
186 }
187}
188
189// putOrder records the order of Puts.
190type putOrder struct {
191 release.Store
192 mu sync.Mutex
193 keys []string
194}
195
196func (p *putOrder) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
197 p.mu.Lock()
198 p.keys = append(p.keys, key)
199 p.mu.Unlock()
200 return p.Store.Put(ctx, key, body, ifMatch)
201}
202
203func (p *putOrder) resultAfterChunks() bool {
204 p.mu.Lock()
205 defer p.mu.Unlock()
206 result := -1
207 lastChunk := -1
208 for i, k := range p.keys {
209 if strings.HasSuffix(k, "/result.json") {
210 result = i
211 }
212 if strings.Contains(k, "/out/") {
213 lastChunk = i
214 }
215 }
216 return result > lastChunk
217}
218
219func TestRunKilledBySignal(t *testing.T) {
220 s := release.NewDirStore(t.TempDir())
221 rec, req := pickedRun(t, s, 60)
222 h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
223 h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "signal", Status: "9", Code: "2"}
224 if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Signal != "9" || res.Exit != 0 {
225 t.Fatalf("result %+v", res)
226 }
227}
228
229// ctxStore refuses a Put whose context has ended, as S3 does; DirStore
230// ignores it.
231type ctxStore struct{ release.Store }
232
233func (c ctxStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
234 if err := ctx.Err(); err != nil {
235 return "", err
236 }
237 return c.Store.Put(ctx, key, body, ifMatch)
238}
239
240func TestRunCancelledAfterLaunch(t *testing.T) {
241 t.Parallel()
242 s := ctxStore{release.NewDirStore(t.TempDir())}
243 rec, req := pickedRun(t, s, 60)
244 done := instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Result: "exit-code", Status: "143", Code: "1"}
245 h := &fakeHandle{polls: make(chan instancectl.CmdStatus), afterStop: &done}
246 rh := newRunHandle()
247 go func() { time.Sleep(50 * time.Millisecond); rh.requestCancel() }()
248 res := runWith(t, s, &fakeLauncher{handle: h, out: "partial\n"}, rec, req, rh)
249 if !res.Cancelled || res.Exit != 143 || res.LastSeq != 1 {
250 t.Fatalf("result %+v (the last chunk lands after a cancel)", res)
251 }
252 if r := currentRun(t, s); r.State != provision.RunDone {
253 t.Fatalf("run %+v, want done", r)
254 }
255}
256
257func TestRunTimesOut(t *testing.T) {
258 s := ctxStore{release.NewDirStore(t.TempDir())}
259 rec, req := pickedRun(t, s, 1)
260 h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
261 start := time.Now()
262 res := runWith(t, s, &fakeLauncher{handle: h, out: "slow\n"}, rec, req, newRunHandle())
263 if !res.TimedOut || time.Since(start) < time.Second || h.stops == 0 {
264 t.Fatalf("result %+v after %s, stops %d", res, time.Since(start), h.stops)
265 }
266 if res.LastSeq != 1 || res.Interrupted != "" {
267 t.Fatalf("result %+v: the last chunk lands after a timeout, and a stopped unit is not interrupted", res)
268 }
269}
270
271// The run's deadline is set at pickup, so slow staging does not buy a full
272// extra timeout: a run whose deadline is nearer than its timeout times out
273// at the deadline, or the console sweep marks a still-running run lost.
274func TestRunTimesOutAtItsDeadlineNotAFullTimeoutLater(t *testing.T) {
275 s := ctxStore{release.NewDirStore(t.TempDir())}
276 rec, id := queueRun(t, s, time.Now().Add(time.Minute))
277 req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
278 if err != nil {
279 t.Fatal(err)
280 }
281 req.TimeoutS = 60
282 got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(1500*time.Millisecond)))
283 if err != nil {
284 t.Fatal(err)
285 }
286 h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
287 start := time.Now()
288 res := runWith(t, s, &fakeLauncher{handle: h, out: "slow\n"}, *got, *req, newRunHandle())
289 if !res.TimedOut || time.Since(start) > 10*time.Second || h.stops == 0 {
290 t.Fatalf("result %+v after %s, stops %d: a run times out at its deadline, not %ds after launch", res, time.Since(start), h.stops, req.TimeoutS)
291 }
292}
293
294// Review Focus 2: a cancel after the pickup CAS and before the unit starts
295// ends the run cancelled with no unit started.
296func TestRunCancelledBeforeLaunchNeverStarts(t *testing.T) {
297 s := release.NewDirStore(t.TempDir())
298 rec, req := pickedRun(t, s, 60)
299 if _, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunCancelTransition(req.ID)); err != nil {
300 t.Fatal(err)
301 }
302 l := &fakeLauncher{handle: &fakeHandle{}, noUnit: true}
303 res := runWith(t, s, l, rec, req, newRunHandle())
304 if !res.Cancelled || l.started() != 0 || !res.Started.IsZero() {
305 t.Fatalf("result %+v, units started %d", res, l.started())
306 }
307 if r := currentRun(t, s); r.State != provision.RunDone {
308 t.Fatalf("run %+v, want done", r)
309 }
310}
311
312// cancelOnInput cancels the run the moment its input is read: a cancel that
313// lands while the files download.
314type cancelOnInput struct {
315 release.Store
316 acct, app, host, id string
317}
318
319func (c *cancelOnInput) Get(ctx context.Context, key string) ([]byte, string, error) {
320 if strings.Contains(key, "/in/") {
321 if _, err := provision.SetRecordRun(ctx, c.Store, c.acct, c.app, c.host, provision.RunCancelTransition(c.id)); err != nil {
322 return nil, "", err
323 }
324 }
325 return c.Store.Get(ctx, key)
326}
327
328func TestRunCancelledWhileFilesDownloadNeverStarts(t *testing.T) {
329 dir := release.NewDirStore(t.TempDir())
330 rec, req := pickedRun(t, dir, 60)
331 body := []byte("a,b\n")
332 sum := sha256.Sum256(body)
333 req.Files = []runs.File{{Name: "in.csv", Size: int64(len(body)), SHA256: hex.EncodeToString(sum[:])}}
334 if _, err := dir.Put(context.Background(), runs.InputKey(rec.Account, rec.App, req.ID, "in.csv"), body, ""); err != nil {
335 t.Fatal(err)
336 }
337 s := &cancelOnInput{Store: dir, acct: rec.Account, app: rec.App, host: rec.Host, id: req.ID}
338 l := &fakeLauncher{handle: &fakeHandle{}, noUnit: true}
339 res := runWith(t, s, l, rec, req, newRunHandle())
340 if !res.Cancelled || l.started() != 0 || !res.Started.IsZero() {
341 t.Fatalf("result %+v, units started %d: the record is read again after the download", res, l.started())
342 }
343}
344
345func TestRunStagesItsFiles(t *testing.T) {
346 s := release.NewDirStore(t.TempDir())
347 rec, req := pickedRun(t, s, 60)
348 body := []byte("a,b\n")
349 sum := sha256.Sum256(body)
350 req.Files = []runs.File{{Name: "in.csv", Size: int64(len(body)), SHA256: hex.EncodeToString(sum[:])}}
351 if _, err := s.Put(context.Background(), runs.InputKey(rec.Account, rec.App, req.ID, "in.csv"), body, ""); err != nil {
352 t.Fatal(err)
353 }
354 w := execWorker(t, s, nil, "boot-1")
355 var staged []byte
356 l := &stagedReader{read: func(spec activator.CmdSpec) {
357 staged, _ = os.ReadFile(spec.RunDir + "/in.csv")
358 if spec.RunDir != runDirOf(w, req) {
359 t.Errorf("RunDir %s, want %s", spec.RunDir, runDirOf(w, req))
360 }
361 }}
362 w.launch = l
363 runOn(t, w, s, rec, req, newRunHandle())
364 if string(staged) != string(body) {
365 t.Fatalf("staged %q, want %q", staged, body)
366 }
367}
368
369// stagedReader prepares, reads what was staged, and refuses to start.
370type stagedReader struct{ read func(activator.CmdSpec) }
371
372func (l *stagedReader) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
373 if err := spec.Prepare(ctx); err != nil {
374 return nil, err
375 }
376 l.read(spec)
377 return nil, errors.New("start cmd unit: 502")
378}
379
380func TestRunRefusesAnIncarnationThatDoesNotMatch(t *testing.T) {
381 s := release.NewDirStore(t.TempDir())
382 rec, req := pickedRun(t, s, 60)
383 req.Incarnation = strings.Repeat("0", 64)
384 l := &fakeLauncher{handle: &fakeHandle{}}
385 if res := runWith(t, s, l, rec, req, newRunHandle()); res.Refused != runs.RefusedIncarnation || l.calls != 0 {
386 t.Fatalf("result %+v, launches %d", res, l.calls)
387 }
388 if r := currentRun(t, s); r.State != provision.RunDone {
389 t.Fatalf("run %+v, want done", r)
390 }
391}
392
393func TestRunRefusesAnImportUncommittedAtLaunch(t *testing.T) {
394 s := release.NewDirStore(t.TempDir())
395 rec, req := pickedRun(t, s, 60)
396 // Admission refuses an uncommitted import, so no walk of the seed
397 // transitions reaches a picked run on one: the launch check is the
398 // backstop. Give the record its seed through the seed writer directly.
399 if _, err := provision.SetRecordSeed(context.Background(), s, rec.Account, rec.App, rec.Host, func(r *provision.Record) error {
400 r.Seed = &provision.SeedSource{ID: "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa", Kind: seeds.KindImport, State: seeds.StatePlaced}
401 return nil
402 }); err != nil {
403 t.Fatal(err)
404 }
405 if res := runWith(t, s, &fakeLauncher{handle: &fakeHandle{}}, rec, req, newRunHandle()); res.Refused != runs.RefusedImport {
406 t.Fatalf("result %+v (codex R2-2: checked again at launch)", res)
407 }
408}
409
410func TestRunLaunchErrorsBecomeResults(t *testing.T) {
411 cases := []struct {
412 err error
413 want func(runs.Result) bool
414 }{
415 {activator.ErrCmdClaim, func(r runs.Result) bool { return r.Refused == runs.RefusedClaim }},
416 {activator.ErrCmdNotNested, func(r runs.Result) bool { return r.Refused == runs.RefusedFlat }},
417 {activator.ErrCmdNotInstance, func(r runs.Result) bool { return r.Refused == runs.RefusedFlat }},
418 {&activator.CmdWakeError{Reason: "wake failed: restore", Err: errors.New("x")}, func(r runs.Result) bool {
419 return r.Refused == runs.RefusedWake && r.Detail == "wake failed: restore"
420 }},
421 {&activator.CmdStoppedError{Reason: activator.StopRestart}, func(r runs.Result) bool { return r.Interrupted == activator.StopRestart }},
422 {activator.ErrCmdBusy, func(r runs.Result) bool { return r.Interrupted == runs.InterruptedStartFailed }},
423 {errors.New("start cmd unit: 502"), func(r runs.Result) bool { return r.Interrupted == runs.InterruptedStartFailed }},
424 }
425 for _, c := range cases {
426 s := release.NewDirStore(t.TempDir())
427 rec, req := pickedRun(t, s, 60)
428 if res := runWith(t, s, &fakeLauncher{err: c.err}, rec, req, newRunHandle()); !c.want(*res) {
429 t.Errorf("%v: result %+v", c.err, res)
430 }
431 if r := currentRun(t, s); r.State != provision.RunDone {
432 t.Errorf("%v: run %+v, want done", c.err, r)
433 }
434 }
435}
436
437func TestRunInterruptedByAnInstanceStop(t *testing.T) {
438 s := release.NewDirStore(t.TempDir())
439 rec, req := pickedRun(t, s, 60)
440 h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1), interrupted: activator.StopRestart}
441 h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "143", Code: "1"}
442 if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != activator.StopRestart {
443 t.Fatalf("result %+v", res)
444 }
445}
446
447// An instance stop that lands after the command's done status arrived, but
448// before the run's own stop, did not end the command: the result is the
449// command's exit, not interrupted.
450func TestRunFinishedBeforeAnInstanceStopKeepsItsExit(t *testing.T) {
451 s := release.NewDirStore(t.TempDir())
452 rec, req := pickedRun(t, s, 60)
453 h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1), interrupted: activator.StopRestart, askedAt: time.Now().Add(time.Hour)}
454 h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "0", Code: "1"}
455 if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != "" || res.Exit != 0 {
456 t.Fatalf("result %+v, want exit 0 and not interrupted", res)
457 }
458}
459
460// A run interrupted by a retired generation while its unit is still
461// running, and whose stop keeps failing, ends interrupted (exit 125) as
462// soon as the interruption is seen: the worker does not follow the command
463// to its own end or its timeout (re-review 6).
464func TestRunInterruptedWhileItsStopIsRetriedEndsAt125(t *testing.T) {
465 s := release.NewDirStore(t.TempDir())
466 rec, req := pickedRun(t, s, 3600)
467 h := &fakeHandle{polls: make(chan instancectl.CmdStatus), stopErr: errors.New("processes remained")}
468 go func() {
469 time.Sleep(100 * time.Millisecond)
470 h.interrupt(activator.StopRelease)
471 }()
472 start := time.Now()
473 res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle())
474 if res.Interrupted != activator.StopRelease || res.TimedOut || res.Cancelled {
475 t.Fatalf("result %+v, want interrupted by the retirement", res)
476 }
477 if took := time.Since(start); took > runInterruptGrace+5*time.Second {
478 t.Fatalf("the result took %s", took)
479 }
480 h.mu.Lock()
481 defer h.mu.Unlock()
482 if h.stops == 0 || !h.released {
483 t.Fatalf("stops %d released %v: the worker must still stop and release the handle", h.stops, h.released)
484 }
485}
486
487func TestRunWhoseUnitWentAway(t *testing.T) {
488 s := release.NewDirStore(t.TempDir())
489 rec, req := pickedRun(t, s, 60)
490 h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
491 h.polls <- instancectl.CmdStatus{State: instancectl.CmdGone, InvocationID: "inv2"}
492 if res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle()); res.Interrupted != runs.InterruptedGone {
493 t.Fatalf("result %+v", res)
494 }
495}
496
497// The record-read backstop: a cancel whose serial bump this box missed is
498// still seen while the run executes.
499func TestRunSeesACancelByRereadingTheRecord(t *testing.T) {
500 defer func(d time.Duration) { runRecheckEvery = d }(runRecheckEvery)
501 runRecheckEvery = 20 * time.Millisecond
502 s := release.NewDirStore(t.TempDir())
503 rec, req := pickedRun(t, s, 60)
504 h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
505 errc := make(chan error, 1)
506 go func() {
507 time.Sleep(50 * time.Millisecond)
508 _, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunCancelTransition(req.ID))
509 errc <- err
510 }()
511 res := runWith(t, s, &fakeLauncher{handle: h}, rec, req, newRunHandle())
512 if err := <-errc; err != nil {
513 t.Fatal(err)
514 }
515 if !res.Cancelled {
516 t.Fatalf("result %+v", res)
517 }
518}
519
520// failingPuts fails every Put whose key holds part.
521type failingPuts struct {
522 release.Store
523 part string
524}
525
526func (f *failingPuts) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
527 if strings.Contains(key, f.part) {
528 return "", errors.New("AccessDenied")
529 }
530 return f.Store.Put(ctx, key, body, ifMatch)
531}
532
533// A store that refuses every chunk cannot hold the run open: the output's
534// bound ends the chunker, and the result says truncated.
535func TestRunOutputIsBoundedWhenEveryChunkFails(t *testing.T) {
536 defer func(d time.Duration) { runOutputGrace = d }(runOutputGrace)
537 runOutputGrace = 300 * time.Millisecond
538 dir := release.NewDirStore(t.TempDir())
539 rec, req := pickedRun(t, dir, 60)
540 s := &failingPuts{Store: dir, part: "/out/"}
541 h := &fakeHandle{polls: make(chan instancectl.CmdStatus, 1)}
542 h.polls <- instancectl.CmdStatus{State: instancectl.CmdDone, InvocationID: "inv1", Status: "0", Code: "1"}
543 start := time.Now()
544 res := runWith(t, s, &fakeLauncher{handle: h, out: "lost\n"}, rec, req, newRunHandle())
545 if !res.Truncated || res.LastSeq != 0 || time.Since(start) > 10*time.Second {
546 t.Fatalf("result %+v after %s", res, time.Since(start))
547 }
548 if r := currentRun(t, dir); r.State != provision.RunDone {
549 t.Fatalf("run %+v, want done", r)
550 }
551}
552
553// A result that cannot be written leaves the run picked, for the sweep to
554// end as lost: done is never set without a result.
555func TestRunWithoutAResultIsNotDone(t *testing.T) {
556 defer func(d time.Duration) { runFinishBound = d }(runFinishBound)
557 runFinishBound = 300 * time.Millisecond
558 dir := release.NewDirStore(t.TempDir())
559 rec, req := pickedRun(t, dir, 60)
560 s := &failingPuts{Store: dir, part: "/result.json"}
561 w := execWorker(t, s, &fakeLauncher{err: activator.ErrCmdClaim}, "boot-1")
562 rt, _ := runOwnedRoute(runHost)(runHost)
563 w.execute(context.Background(), rec, req, *rt, newRunHandle())
564 if r := currentRun(t, dir); r.State != provision.RunPicked {
565 t.Fatalf("run %+v, want still picked", r)
566 }
567}
568
569// The done CAS is only for a run the record names as this box's under this
570// boot (RunFinish does not check), and a box writes no result for a run the
571// record gives to another: whoever holds it ends it.
572func TestRunFinishesOnlyItsOwnPick(t *testing.T) {
573 s := release.NewDirStore(t.TempDir())
574 rec, req := pickedRun(t, s, 60) // picked under boot-1
575 h := &fakeHandle{}
576 w := execWorker(t, s, &fakeLauncher{handle: h}, "boot-2")
577 rt, _ := runOwnedRoute(runHost)(runHost)
578 w.active[req.ID] = newRunHandle()
579 w.execute(context.Background(), rec, req, *rt, w.active[req.ID])
580 if res, err := runs.ReadResult(context.Background(), s, rec.Account, rec.App, req.ID); err != nil || res != nil {
581 t.Fatalf("result %+v, %v: a box wrote a result for another boot's run", res, err)
582 }
583 if r := currentRun(t, s); h.stops != 0 || r.State != provision.RunPicked || r.BootID != "boot-1" {
584 t.Fatalf("run %+v, stops %d: another boot's run was launched or finished", r, h.stops)
585 }
586 if w.active[req.ID] != nil {
587 t.Error("the run was not forgotten")
588 }
589 for _, mut := range []func(*provision.RunState){
590 func(r *provision.RunState) { r.Box = "box-2" },
591 func(r *provision.RunState) { r.BootID = "boot-2" },
592 func(r *provision.RunState) { r.ID = "run_bbbbbbbbbbbbbbbbbbbbbbbbbb" },
593 } {
594 cur := rec
595 r := *rec.ActiveRun
596 mut(&r)
597 cur.ActiveRun = &r
598 if err := runFinishOurs(req.ID, "box-1", "boot-1")(&cur); !errors.Is(err, provision.ErrRunConflict) {
599 t.Errorf("run %+v: err %v, want a conflict", r, err)
600 }
601 }
602 cur := rec
603 if err := runFinishOurs(req.ID, "box-1", "boot-1")(&cur); err != nil || cur.ActiveRun.State != provision.RunDone {
604 t.Fatalf("own run: %v, %+v", err, cur.ActiveRun)
605 }
606}
607
608// stalledInput blocks every read of a run input until its context ends.
609type stalledInput struct{ release.Store }
610
611func (s stalledInput) Get(ctx context.Context, key string) ([]byte, string, error) {
612 if strings.Contains(key, "/in/") {
613 <-ctx.Done()
614 return nil, "", ctx.Err()
615 }
616 return s.Store.Get(ctx, key)
617}
618
619// A download that stalls ends at the run's deadline: the run ends as a
620// failed start, with its result and the done CAS, and does not keep the
621// activator's hold (and every later run on the instance) forever.
622func TestRunStagingEndsAtTheRunsDeadline(t *testing.T) {
623 dir := release.NewDirStore(t.TempDir())
624 rec, id := queueRun(t, dir, time.Now().Add(time.Minute))
625 req, err := runs.ReadRequest(context.Background(), dir, rec.Account, rec.App, id)
626 if err != nil {
627 t.Fatal(err)
628 }
629 req.Files = []runs.File{{Name: "in.csv", Size: 4, SHA256: strings.Repeat("0", 64)}}
630 got, err := provision.SetRecordRun(context.Background(), dir, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(300*time.Millisecond)))
631 if err != nil {
632 t.Fatal(err)
633 }
634 l := &fakeLauncher{handle: &fakeHandle{}, noUnit: true}
635 start := time.Now()
636 res := runWith(t, stalledInput{dir}, l, *got, *req, newRunHandle())
637 if res.Interrupted != runs.InterruptedStartFailed || l.started() != 0 || time.Since(start) > 5*time.Second {
638 t.Fatalf("result %+v after %s, units %d", res, time.Since(start), l.started())
639 }
640 if r := currentRun(t, dir); r.State != provision.RunDone {
641 t.Fatalf("run %+v, want done", r)
642 }
643}
644
645// stoppedDuringPrepare runs Prepare and then reports an instance stop, as
646// RunCmd does when a stop overlapped Prepare: the stop wins over Prepare's
647// own error.
648type stoppedDuringPrepare struct{}
649
650func (stoppedDuringPrepare) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
651 _ = spec.Prepare(ctx)
652 return nil, &activator.CmdStoppedError{Reason: activator.StopRestart}
653}
654
655// A run Prepare found is no longer this box's gets no result, even when a
656// stop overlapping Prepare is what RunCmd reports.
657func TestRunNotOursGetsNoResultWhenAStopOverlapsPrepare(t *testing.T) {
658 s := release.NewDirStore(t.TempDir())
659 rec, req := pickedRun(t, s, 60) // picked under boot-1
660 w := execWorker(t, s, stoppedDuringPrepare{}, "boot-2")
661 rt, _ := runOwnedRoute(runHost)(runHost)
662 w.execute(context.Background(), rec, req, *rt, newRunHandle())
663 if res, err := runs.ReadResult(context.Background(), s, rec.Account, rec.App, req.ID); err != nil || res != nil {
664 t.Fatalf("result %+v, %v: a box wrote a result for a run the record gives another boot", res, err)
665 }
666 if r := currentRun(t, s); r.State != provision.RunPicked || r.BootID != "boot-1" {
667 t.Fatalf("run %+v", r)
668 }
669}
670
671// waitsForSwitch is RunCmd as it waits for a swap's switch after Prepare:
672// it holds until the launch context ends, as the activator's wait does
673// under the run's deadline and cancel, and starts nothing.
674type waitsForSwitch struct {
675 waiting chan struct{}
676 units int
677}
678
679func (l *waitsForSwitch) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
680 if err := spec.Prepare(ctx); err != nil {
681 return nil, err
682 }
683 close(l.waiting)
684 <-ctx.Done()
685 return nil, ctx.Err()
686}
687
688// A run cancelled, or past its deadline, while RunCmd waits for a swap's
689// switch ends with its own cancelled or timed-out result, and no unit
690// starts (re-review 5): the launch runs under the run's deadline and
691// cancel, not the worker's background context.
692func TestRunEndedDuringASwitchWaitGetsItsOwnResult(t *testing.T) {
693 t.Run("cancelled", func(t *testing.T) {
694 s := release.NewDirStore(t.TempDir())
695 rec, req := pickedRun(t, s, 60)
696 l := &waitsForSwitch{waiting: make(chan struct{})}
697 h := newRunHandle()
698 go func() { <-l.waiting; h.requestCancel() }()
699 res := runOn(t, execWorker(t, s, l, "boot-1"), s, rec, req, h)
700 if !res.Cancelled || res.TimedOut || l.units != 0 || !res.Started.IsZero() {
701 t.Fatalf("result %+v", res)
702 }
703 })
704 t.Run("expired", func(t *testing.T) {
705 dir := release.NewDirStore(t.TempDir())
706 rec, id := queueRun(t, dir, time.Now().Add(time.Minute))
707 req, err := runs.ReadRequest(context.Background(), dir, rec.Account, rec.App, id)
708 if err != nil {
709 t.Fatal(err)
710 }
711 got, err := provision.SetRecordRun(context.Background(), dir, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(300*time.Millisecond)))
712 if err != nil {
713 t.Fatal(err)
714 }
715 l := &waitsForSwitch{waiting: make(chan struct{})}
716 start := time.Now()
717 res := runOn(t, execWorker(t, dir, l, "boot-1"), dir, *got, *req, newRunHandle())
718 if !res.TimedOut || res.Cancelled || l.units != 0 || time.Since(start) > 5*time.Second {
719 t.Fatalf("result %+v after %s", res, time.Since(start))
720 }
721 })
722}
723
724// blockOneRead holds the first store read after it is armed until the
725// read's context ends: the post-switch record read, cut short.
726type blockOneRead struct {
727 release.Store
728 armed atomic.Bool
729 reading chan struct{}
730}
731
732func (b *blockOneRead) Get(ctx context.Context, key string) ([]byte, string, error) {
733 if b.armed.CompareAndSwap(true, false) {
734 close(b.reading)
735 <-ctx.Done()
736 return nil, "", ctx.Err()
737 }
738 return b.Store.Get(ctx, key)
739}
740
741// rechecksAfterSwitch is RunCmd past a switch wait: Prepare, then the
742// post-switch Recheck, whose record read the store holds. It starts
743// nothing, and returns the Recheck's answer as RunCmd does.
744type rechecksAfterSwitch struct{ store *blockOneRead }
745
746func (l *rechecksAfterSwitch) RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error) {
747 if err := spec.Prepare(ctx); err != nil {
748 return nil, err
749 }
750 l.store.armed.Store(true)
751 return nil, spec.Recheck(ctx)
752}
753
754// A cancel or the deadline that cuts short the post-switch record read is
755// the run's own end, cancelled (130) or timed out (124), never "start
756// failed" (re-review 7).
757func TestRunEndedDuringThePostSwitchRecheckGetsItsOwnResult(t *testing.T) {
758 t.Parallel()
759 t.Run("cancelled", func(t *testing.T) {
760 s := &blockOneRead{Store: release.NewDirStore(t.TempDir()), reading: make(chan struct{})}
761 rec, req := pickedRun(t, s, 60)
762 h := newRunHandle()
763 go func() { <-s.reading; h.requestCancel() }()
764 res := runOn(t, execWorker(t, s, &rechecksAfterSwitch{store: s}, "boot-1"), s, rec, req, h)
765 if !res.Cancelled || res.TimedOut || res.Interrupted != "" || !res.Started.IsZero() {
766 t.Fatalf("result %+v, want cancelled", res)
767 }
768 })
769 t.Run("expired", func(t *testing.T) {
770 s := &blockOneRead{Store: release.NewDirStore(t.TempDir()), reading: make(chan struct{})}
771 rec, id := queueRun(t, s, time.Now().Add(time.Minute))
772 req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
773 if err != nil {
774 t.Fatal(err)
775 }
776 got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(300*time.Millisecond)))
777 if err != nil {
778 t.Fatal(err)
779 }
780 res := runOn(t, execWorker(t, s, &rechecksAfterSwitch{store: s}, "boot-1"), s, *got, *req, newRunHandle())
781 if !res.TimedOut || res.Cancelled || res.Interrupted != "" || !res.Started.IsZero() {
782 t.Fatalf("result %+v, want timed out", res)
783 }
784 })
785}
786
787func TestExitOfReadsNumbersAndWords(t *testing.T) {
788 for _, c := range []struct {
789 code, status string
790 exit int
791 signal string
792 }{
793 {"1", "3", 3, ""}, {"exited", "3", 3, ""}, {"2", "9", 0, "9"}, {"killed", "15", 0, "15"}, {"3", "6", 0, "6"}, {"dumped", "6", 0, "6"},
794 } {
795 exit, signal := exitOf(instancectl.CmdStatus{Code: c.code, Status: c.status})
796 if exit != c.exit || signal != c.signal {
797 t.Errorf("%s/%s: %d %q", c.code, c.status, exit, signal)
798 }
799 }
800}
801