amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package activator
2
3import (
4 "context"
5 "errors"
6 "path/filepath"
7 "strings"
8 "sync"
9 "sync/atomic"
10 "testing"
11 "time"
12
13 "github.com/carlosframework/platform/internal/hostwrite"
14 "github.com/carlosframework/platform/internal/instancectl"
15 "github.com/carlosframework/platform/internal/provision"
16 "github.com/carlosframework/platform/internal/registry"
17)
18
19func (a *Service) stopWhyOf(host string) string {
20 a.mu.Lock()
21 defer a.mu.Unlock()
22 if e := a.entries[host]; e != nil {
23 return e.stopWhy
24 }
25 return ""
26}
27
28func TestStopsRecordTheirReason(t *testing.T) {
29 a, b, _, _ := testActivator(t, time.Second)
30 rt := testRoute("why.example.com")
31 arrangeAwake(t, a, rt)
32 gate := make(chan struct{})
33 b.lastProc().setStopGate(gate) // hold the entry in stopping
34 defer close(gate)
35 a.Bounce(rt.Host)
36 if got := a.stopWhyOf(rt.Host); got != StopRestart {
37 t.Fatalf("Bounce: stopWhy %q, want %q", got, StopRestart)
38 }
39}
40
41// fakeCmdUnit records the stop and whether replication was already closed
42// when it came.
43type fakeCmdUnit struct {
44 mu sync.Mutex
45 stops int
46 repWasClosed bool
47 settleErr error
48 rep func() *fakeRep
49 status instancectl.CmdStatus
50}
51
52func (u *fakeCmdUnit) Invocation() string { return "inv1" }
53func (u *fakeCmdUnit) Poll(context.Context) (instancectl.CmdStatus, error) {
54 u.mu.Lock()
55 defer u.mu.Unlock()
56 return u.status, nil
57}
58func (u *fakeCmdUnit) StopAndSettle(context.Context) error {
59 u.mu.Lock()
60 defer u.mu.Unlock()
61 u.stops++
62 if u.rep != nil {
63 u.repWasClosed = u.rep().isClosed()
64 }
65 return u.settleErr
66}
67func (u *fakeCmdUnit) setSettleErr(err error) {
68 u.mu.Lock()
69 defer u.mu.Unlock()
70 u.settleErr = err
71}
72
73func (u *fakeCmdUnit) stopped() (int, bool) {
74 u.mu.Lock()
75 defer u.mu.Unlock()
76 return u.stops, u.repWasClosed
77}
78
79// cmdFakeBackend is fakeBackend with the run capability and a nested layout.
80type cmdFakeBackend struct {
81 *fakeBackend
82 cmdMu sync.Mutex
83 unit *fakeCmdUnit
84 startErr error
85 cmdStarts int
86 gate chan struct{} // when set, StartCmd blocks until it closes
87 entered chan struct{} // when set, closed as StartCmd is entered
88 // errWithUnit makes a failing start return the unit too: the start
89 // reached the control service and the unit may be running.
90 errWithUnit bool
91 // startCmd, when set, replaces the fake start (a real ExecBackend's).
92 startCmd func(context.Context, registry.Route, CmdSpec) (CmdUnit, error)
93}
94
95func (b *cmdFakeBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) {
96 if b.entered != nil {
97 close(b.entered)
98 }
99 if b.gate != nil {
100 <-b.gate
101 }
102 b.cmdMu.Lock()
103 defer b.cmdMu.Unlock()
104 b.cmdStarts++
105 if b.startCmd != nil {
106 return b.startCmd(ctx, rt, spec)
107 }
108 if b.startErr != nil {
109 if b.errWithUnit {
110 return b.unit, b.startErr
111 }
112 return nil, b.startErr
113 }
114 return b.unit, nil
115}
116
117func testCmdActivator(t *testing.T, cfg Config) (*Service, *cmdFakeBackend, *fakeClock) {
118 t.Helper()
119 fb := &fakeBackend{layoutFn: func(registry.Route) (Layout, string) { return LayoutNested, "" }}
120 cb := &cmdFakeBackend{fakeBackend: fb}
121 cb.unit = &fakeCmdUnit{rep: fb.lastRep}
122 clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
123 reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
124 if err != nil {
125 t.Fatal(err)
126 }
127 if cfg.Bucket == "" {
128 cfg.Bucket = "test-bucket"
129 }
130 if cfg.Idle == 0 {
131 cfg.Idle = time.Second
132 }
133 a := New(cfg, cb, reg)
134 a.now = clk.now
135 return a, cb, clk
136}
137
138func TestRunCmdRefusesWithClaimControlArmed(t *testing.T) {
139 a, cb, _ := testCmdActivator(t, Config{HostWrites: hostwrite.Config{Enabled: true}})
140 if _, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdClaim) {
141 t.Fatalf("err = %v, want ErrCmdClaim", err)
142 }
143 if acquires, _, _ := cb.counts(); acquires != 0 {
144 t.Fatal("a claim-armed refusal woke the instance")
145 }
146}
147
148func TestRunCmdRefusesAFlatInstance(t *testing.T) {
149 a, cb, _ := testCmdActivator(t, Config{})
150 cb.layoutFn = func(registry.Route) (Layout, string) { return LayoutFlat, "" }
151 if _, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdNotNested) {
152 t.Fatalf("err = %v, want ErrCmdNotNested", err)
153 }
154}
155
156func TestRunCmdWakesAHibernatedInstanceFirst(t *testing.T) {
157 a, cb, _ := testCmdActivator(t, Config{})
158 h, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"})
159 if err != nil {
160 t.Fatal(err)
161 }
162 if _, _, wakes := cb.counts(); h.Invocation() != "inv1" || wakes != 1 || cb.cmdStarts != 1 {
163 t.Fatalf("wake starts %d, cmd starts %d", wakes, cb.cmdStarts)
164 }
165 h.Release()
166}
167
168func TestRunCmdWakeFailureIsAWakeRefusal(t *testing.T) {
169 a, cb, _ := testCmdActivator(t, Config{})
170 cb.acquireErr = errors.New("restore failed")
171 _, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"})
172 var we *CmdWakeError
173 if !errors.As(err, &we) || we.Reason == "" {
174 t.Fatalf("err = %v, want a CmdWakeError with a member-safe reason", err)
175 }
176 if cb.cmdStarts != 0 {
177 t.Fatal("a failed wake started the command")
178 }
179}
180
181func TestRunCmdRefusesASecondRun(t *testing.T) {
182 a, _, _ := testCmdActivator(t, Config{})
183 rt := testRoute("c.example.com")
184 h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
185 if err != nil {
186 t.Fatal(err)
187 }
188 defer h.Release()
189 if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_b"}); !errors.Is(err, ErrCmdBusy) {
190 t.Fatalf("err = %v, want ErrCmdBusy", err)
191 }
192}
193
194func TestRunCmdHoldsTheIdleTimer(t *testing.T) {
195 t.Parallel()
196 a, _, clk := testCmdActivator(t, Config{})
197 rt := testRoute("c.example.com")
198 h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
199 if err != nil {
200 t.Fatal(err)
201 }
202 clk.advance(time.Hour)
203 a.sweepOnce()
204 if a.entryCount() != 1 || a.stopWhyOf(rt.Host) != "" {
205 t.Fatal("the idle timer stopped an instance with a run open")
206 }
207 h.Release()
208 clk.advance(time.Hour)
209 a.sweepOnce()
210 waitFor(t, "hibernation after the run", func() bool { return a.entryCount() == 0 })
211}
212
213// Every stop path stops and settles the run before replication closes, and
214// the run learns why (run spec §5.3, codex 1).
215func TestEveryStopPathSettlesTheRunBeforeReplicationCloses(t *testing.T) {
216 t.Parallel()
217 cases := []struct {
218 name string
219 why string
220 stop func(a *Service, cb *cmdFakeBackend, host string)
221 }{
222 {"bounce", StopRestart, func(a *Service, _ *cmdFakeBackend, h string) { a.Bounce(h) }},
223 {"process exit", StopExited, func(a *Service, cb *cmdFakeBackend, _ string) { cb.lastProc().exit(); a.sweepOnce() }},
224 {"lease lost", StopLeaseLost, func(a *Service, cb *cmdFakeBackend, _ string) { cb.lastRep().setErr(errors.New("lost")); a.sweepOnce() }},
225 {"convergence", StopRelease, func(a *Service, _ *cmdFakeBackend, h string) {
226 a.liveVer = func(string) (string, error) { return "v2", nil }
227 a.ConvergeAdopted(h, "v2")
228 }},
229 {"move fence", StopFenced, func(a *Service, _ *cmdFakeBackend, h string) {
230 // FenceAndStop refuses a context without a deadline.
231 ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
232 go func() {
233 defer cancel()
234 a.FenceAndStop(ctx, h, 1) //nolint:errcheck
235 }()
236 }},
237 {"retire", StopRetired, func(a *Service, _ *cmdFakeBackend, _ string) { a.RetireExcept(map[string]bool{}) }},
238 {"agent shutdown", StopShutdown, func(a *Service, _ *cmdFakeBackend, _ string) {
239 ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
240 defer cancel()
241 a.Shutdown(ctx)
242 }},
243 }
244 for _, c := range cases {
245 t.Run(c.name, func(t *testing.T) {
246 a, cb, _ := testCmdActivator(t, Config{})
247 cb.execVerFn = func() string { return "v1" }
248 rt := testRoute("c.example.com")
249 h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
250 if err != nil {
251 t.Fatal(err)
252 }
253 c.stop(a, cb, rt.Host)
254 waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
255 stops, repWasClosed := cb.unit.stopped()
256 if stops != 1 {
257 t.Fatalf("the cmd unit was stopped %d times, want 1", stops)
258 }
259 if repWasClosed {
260 t.Fatal("replication closed before the run settled")
261 }
262 if !cb.lastRep().isClosed() {
263 t.Fatal("replication never closed")
264 }
265 if got, at := h.Interrupted(); got != c.why || at.IsZero() {
266 t.Fatalf("Interrupted() = %q at %v, want %q and when", got, at, c.why)
267 }
268 })
269 }
270}
271
272// The hold stamps the first stop asked for with the activator's clock, so
273// the run can tell an instance stop that came after its command finished.
274func TestRunCmdInterruptedSaysWhenTheStopWasAsked(t *testing.T) {
275 t.Parallel()
276 a, _, clk := testCmdActivator(t, Config{})
277 rt := testRoute("c.example.com")
278 h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
279 if err != nil {
280 t.Fatal(err)
281 }
282 clk.advance(5 * time.Minute)
283 asked := clk.now()
284 a.Bounce(rt.Host)
285 waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
286 clk.advance(time.Minute)
287 if err := h.Stop(context.Background()); err != nil {
288 t.Fatal(err)
289 }
290 if why, at := h.Interrupted(); why != StopRestart || !at.Equal(asked) {
291 t.Fatalf("Interrupted() = %q at %v, want %q at %v (the first stop's)", why, at, StopRestart, asked)
292 }
293}
294
295func TestRunCmdNeverStartsUnderAStoppingEntry(t *testing.T) {
296 a, cb, _ := testCmdActivator(t, Config{})
297 rt := testRoute("c.example.com")
298 _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a", Prepare: func(context.Context) error {
299 a.Bounce(rt.Host) // the instance starts stopping while the files download
300 return nil
301 }})
302 var se *CmdStoppedError
303 if !errors.As(err, &se) || se.Reason != StopRestart {
304 t.Fatalf("err = %v, want CmdStoppedError{%q}", err, StopRestart)
305 }
306 if cb.cmdStarts != 0 {
307 t.Fatal("the cmd unit started under a stopping entry")
308 }
309}
310
311func TestAStopDuringTheStartWaitsForIt(t *testing.T) {
312 t.Parallel()
313 a, cb, _ := testCmdActivator(t, Config{})
314 cb.gate = make(chan struct{})
315 cb.entered = make(chan struct{})
316 rt := testRoute("c.example.com")
317 done := make(chan error, 1)
318 go func() {
319 _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
320 done <- err
321 }()
322 waitFor(t, "the run attached", func() bool {
323 a.mu.Lock()
324 defer a.mu.Unlock()
325 e := a.entries[rt.Host]
326 return e != nil && e.cmd != nil
327 })
328 // The attach comes before RunCmd's last awake check; the stop must land
329 // after that check, with StartCmd in flight.
330 <-cb.entered
331 a.Bounce(rt.Host)
332 time.Sleep(50 * time.Millisecond)
333 if cb.lastRep().isClosed() {
334 t.Fatal("replication closed while the cmd unit's start was still in flight")
335 }
336 close(cb.gate)
337 if err := <-done; err != nil {
338 t.Fatalf("RunCmd: %v", err)
339 }
340 waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 })
341 if stops, repWasClosed := cb.unit.stopped(); stops != 1 || repWasClosed {
342 t.Fatalf("stops %d, replication closed first %v", stops, repWasClosed)
343 }
344}
345
346// A run whose stop cannot be proven keeps the instance's stop pending:
347// the entry, replication and the lease stay, and the run's handle with
348// them. The sweep tries the settle again, and the retry that proves it
349// closes replication and releases the lease (merge ruling M1). That holds
350// with generations off and on.
351func TestAnUnprovenRunStopKeepsReplicationAndTheLease(t *testing.T) {
352 t.Parallel()
353 for _, gens := range []bool{false, true} {
354 t.Run(map[bool]string{false: "generations off", true: "generations on"}[gens], func(t *testing.T) {
355 a, cb, _ := testCmdActivator(t, Config{})
356 a.gens = gens
357 cb.unit.setSettleErr(errors.New("processes remained"))
358 rt := testRoute("c.example.com")
359 if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); err != nil {
360 t.Fatal(err)
361 }
362 a.Bounce(rt.Host)
363 waitFor(t, "the first settle", func() bool { n, _ := cb.unit.stopped(); return n >= 1 })
364 // Every sweep retries, and every retry fails: nothing is given up.
365 for i := 0; i < 3; i++ {
366 before, _ := cb.unit.stopped()
367 a.sweepOnce()
368 waitFor(t, "a retried settle", func() bool { n, _ := cb.unit.stopped(); return n > before })
369 if a.entryCount() != 1 {
370 t.Fatal("the entry went while its run was not proven stopped")
371 }
372 if cb.lastRep().isClosed() {
373 t.Fatal("replication closed, and the lease went, while the run was not proven stopped")
374 }
375 }
376 a.mu.Lock()
377 held := a.entries[rt.Host] != nil && len(cmdHoldsLocked(a.entries[rt.Host])) == 1
378 a.mu.Unlock()
379 if !held {
380 t.Fatal("the entry let go of the run's handle")
381 }
382 if !a.StopUncertain(rt.Host) {
383 t.Fatal("StopUncertain = false with the run unproven")
384 }
385 cb.unit.setSettleErr(nil)
386 waitFor(t, "the retry that proves the run stopped", func() bool {
387 a.sweepOnce()
388 return a.entryCount() == 0
389 })
390 if _, repWasClosed := cb.unit.stopped(); repWasClosed {
391 t.Fatal("replication closed before the run was proven stopped")
392 }
393 if !cb.lastRep().isClosed() {
394 t.Fatal("replication never closed, so the lease was never released")
395 }
396 if a.StopUncertain(rt.Host) {
397 t.Fatal("StopUncertain = true after the run was proven stopped")
398 }
399 })
400 }
401}
402
403// A run released without a proven settle (its Stop failed, or was never
404// called) still belongs to the instance's next stop: that stop settles it
405// before replication closes, and holds replication and the lease until
406// it has.
407func TestAReleasedRunThatNeverSettledStillHoldsTheStop(t *testing.T) {
408 t.Parallel()
409 for _, c := range []struct {
410 name string
411 stopFirst bool
412 settleErr error
413 }{
414 {"stopped and settled", true, nil},
415 {"never stopped", false, nil},
416 {"stop failed", true, errors.New("processes remained")},
417 } {
418 t.Run(c.name, func(t *testing.T) {
419 a, cb, clk := testCmdActivator(t, Config{})
420 cb.unit.setSettleErr(c.settleErr)
421 rt := testRoute("c.example.com")
422 h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"})
423 if err != nil {
424 t.Fatal(err)
425 }
426 if c.stopFirst {
427 if err := h.Stop(context.Background()); (err != nil) != (c.settleErr != nil) {
428 t.Fatalf("Stop: %v", err)
429 }
430 }
431 h.Release()
432 clk.advance(time.Hour)
433 a.sweepOnce() // the release ended the idle hold
434 if c.settleErr != nil {
435 waitFor(t, "the stop's own settle", func() bool { n, _ := cb.unit.stopped(); return n >= 2 })
436 a.sweepOnce()
437 if a.entryCount() != 1 || cb.lastRep().isClosed() {
438 t.Fatal("the stop finished over a run that is not proven stopped")
439 }
440 cb.unit.setSettleErr(nil)
441 }
442 waitFor(t, "the instance stopped", func() bool { a.sweepOnce(); return a.entryCount() == 0 })
443 stops, repWasClosed := cb.unit.stopped()
444 // The failing case: the run's own Stop, the instance stop's
445 // settle, and at least one retry, the last of which proved it.
446 if c.settleErr == nil && stops != 1 || c.settleErr != nil && stops < 3 {
447 t.Fatalf("stops %d", stops)
448 }
449 if repWasClosed {
450 t.Fatal("replication closed before the run settled")
451 }
452 if a.StopUncertain(rt.Host) {
453 t.Fatal("StopUncertain = true after every run was proven stopped")
454 }
455 if got, _ := h.Interrupted(); got != "" && c.stopFirst {
456 t.Fatalf("Interrupted() = %q after the run's own stop", got)
457 }
458 })
459 }
460}
461
462func TestRunCmdRefusesAnythingButAnInstance(t *testing.T) {
463 a, cb, _ := testCmdActivator(t, Config{})
464 rt := testRoute("c.example.com")
465 rt.Kind = "sidecar"
466 if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdNotInstance) {
467 t.Fatalf("err = %v, want ErrCmdNotInstance", err)
468 }
469 if acquires, _, _ := cb.counts(); acquires != 0 {
470 t.Fatal("a refused route was woken")
471 }
472}
473
474// A stop during a slow Prepare (the file downloads) does not wait for it:
475// no unit has started, so the stop is prompt and complete, Prepare's
476// context is cancelled, and the run ends interrupted with no unit started.
477func TestAStopDuringASlowPrepareDoesNotWaitForIt(t *testing.T) {
478 t.Parallel()
479 a, cb, _ := testCmdActivator(t, Config{})
480 rt := testRoute("c.example.com")
481 inPrepare, unblock := make(chan struct{}), make(chan struct{})
482 var cancelled atomic.Bool
483 done := make(chan error, 1)
484 go func() {
485 _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a", Prepare: func(ctx context.Context) error {
486 close(inPrepare)
487 <-unblock // a download that does not watch its context
488 cancelled.Store(ctx.Err() != nil)
489 return nil
490 }})
491 done <- err
492 }()
493 <-inPrepare
494 a.Bounce(rt.Host)
495 waitFor(t, "the instance stopped while Prepare was still running", func() bool { return a.entryCount() == 0 })
496 if a.StopUncertain(rt.Host) {
497 t.Fatal("a stop that no unit outlived was recorded uncertain")
498 }
499 close(unblock)
500 var se *CmdStoppedError
501 if err := <-done; !errors.As(err, &se) || se.Reason != StopRestart {
502 t.Fatalf("err = %v, want CmdStoppedError{%q}", err, StopRestart)
503 }
504 if !cancelled.Load() {
505 t.Error("the stop did not cancel Prepare's context")
506 }
507 if cb.cmdStarts != 0 {
508 t.Fatal("a unit started after the stop")
509 }
510}
511
512// A start that failed after it reached the control service may have left
513// the unit running: RunCmd settles it, and a unit that will not settle
514// keeps the instance from another run and makes its next stop uncertain.
515func TestAFailedStartThatMayHaveRunIsSettled(t *testing.T) {
516 t.Parallel()
517 for _, c := range []struct {
518 name string
519 settleErr error
520 }{{"settles", nil}, {"will not settle", errors.New("processes remained")}} {
521 t.Run(c.name, func(t *testing.T) {
522 a, cb, _ := testCmdActivator(t, Config{})
523 cb.startErr, cb.errWithUnit = errors.New("502 mid-roll"), true
524 cb.unit.setSettleErr(c.settleErr)
525 rt := testRoute("c.example.com")
526 if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); !errors.Is(err, cb.startErr) {
527 t.Fatalf("err = %v, want the start's error", err)
528 }
529 if stops, _ := cb.unit.stopped(); stops != 1 {
530 t.Fatalf("the unit was stopped %d times, want 1", stops)
531 }
532 _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_b"})
533 if busy := errors.Is(err, ErrCmdBusy); busy != (c.settleErr != nil) {
534 t.Fatalf("second run: err = %v, busy = %v", err, busy)
535 }
536 a.Bounce(rt.Host)
537 if c.settleErr != nil {
538 waitFor(t, "the stop's settle", func() bool { n, _ := cb.unit.stopped(); return n >= 2 })
539 a.sweepOnce()
540 if a.entryCount() != 1 || cb.lastRep().isClosed() {
541 t.Fatal("the stop finished over a unit that is not proven stopped")
542 }
543 cb.unit.setSettleErr(nil)
544 }
545 waitFor(t, "the instance stopped", func() bool { a.sweepOnce(); return a.entryCount() == 0 })
546 if _, repWasClosed := cb.unit.stopped(); repWasClosed {
547 t.Fatal("replication closed before the unit settled")
548 }
549 if a.StopUncertain(rt.Host) {
550 t.Fatal("StopUncertain = true after the unit was proven stopped")
551 }
552 })
553 }
554}
555
556// The real ExecBackend: POST /cmd starts the unit and then fails (a lost
557// response). The cmd role is stopped and its cgroup waited for before
558// replication closes. A cgroup that cannot be read empty keeps the stop
559// pending with replication open, and the retry once it reads empty
560// finishes it (merge ruling M1).
561func TestALostCmdResponseStopsTheCmdRoleBeforeReplicationCloses(t *testing.T) {
562 for _, c := range []struct {
563 name string
564 cgroupErr error
565 }{{"cgroup empties", nil}, {"cgroup unreadable", errors.New("cgroup.events: permission denied")}} {
566 t.Run(c.name, func(t *testing.T) {
567 shrinkCmdAcceptGrace(t, 50*time.Millisecond) // the stop drains two streams the fake unit never connected
568 a, cb, _ := testCmdActivator(t, Config{})
569 ctl := &fakeControl{cmdErr: errors.New("502 mid-roll")}
570 unit := &fakeUnit{}
571 unit.set("active", "inv1", true, true) // the POST reached systemd
572 b, rt := nestedBackend(t, ctl, unit)
573 b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
574 var repClosedAtEmpty, unreadable atomic.Bool
575 unreadable.Store(c.cgroupErr != nil)
576 b.CgroupEvents = func(slice, u string) (bool, bool, error) {
577 if unreadable.Load() {
578 return true, true, c.cgroupErr
579 }
580 exists, populated, err := unit.events(slice, u)
581 if !populated {
582 repClosedAtEmpty.Store(cb.lastRep().isClosed())
583 }
584 return exists, populated, err
585 }
586 ctl.stopHook = func(_ string, role instancectl.Role) {
587 if role == instancectl.RoleCmd {
588 go func() {
589 time.Sleep(50 * time.Millisecond)
590 unit.set("inactive", "inv1", false, false)
591 }()
592 }
593 }
594 cb.startCmd = b.StartCmd
595 dir := filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), CmdRunDirName, "run_a")
596 if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a", RunDir: dir}); err == nil || !strings.Contains(err.Error(), "502 mid-roll") {
597 t.Fatalf("err = %v, want the lost response", err)
598 }
599 a.Bounce(rt.Host)
600 if c.cgroupErr != nil {
601 waitFor(t, "the stop kept pending", func() bool {
602 a.mu.Lock()
603 defer a.mu.Unlock()
604 e := a.entries[rt.Host]
605 return e != nil && e.pendingStop
606 })
607 if cb.lastRep().isClosed() {
608 t.Fatal("replication closed while the cmd cgroup could not be read empty")
609 }
610 unreadable.Store(false)
611 }
612 waitFor(t, "the instance stopped", func() bool { a.sweepOnce(); return a.entryCount() == 0 })
613 if !strings.Contains(ctl.joined(), "stop "+rt.Host+" cmd") {
614 t.Fatalf("the cmd role was never stopped:\n%s", ctl.joined())
615 }
616 if repClosedAtEmpty.Load() {
617 t.Fatal("replication closed before the cmd cgroup emptied")
618 }
619 if a.StopUncertain(rt.Host) {
620 t.Fatal("StopUncertain = true after the cmd role was proven stopped")
621 }
622 })
623 }
624}
625
626// A run runs what the instance serves (zero-downtime spec §7.5, as for the
627// legacy sidecar): the current generation's recorded file by path, captured
628// at the last awake check, with generations on or off (settings ruling
629// R-EXEC-SIDECAR: a flag-off wake runs its recorded file by path too, so
630// the live link may name another build). Only a generation that names no
631// file leaves the run the live link. A caller's own setting never
632// survives: RunCmd decides.
633func TestRunCmdRunsTheServingGenerationsBuild(t *testing.T) {
634 t.Parallel()
635 rt := testRoute("c.example.com")
636 staged := &instancectl.StagedExe{Path: "/opt/carlos/versions/x", Version: "v9", Digest: strings.Repeat("0", 64)}
637 want := instancectl.StagedExe{Path: "/opt/carlos/versions/v2/blog", Version: "v2", Digest: strings.Repeat("a", 64)}
638 for _, gens := range []bool{false, true} {
639 t.Run(map[bool]string{false: "generations off", true: "generations on"}[gens], func(t *testing.T) {
640 // The switch is set before any wake, as CARLOS_SWAP=overlap
641 // sets it, before anything spawns.
642 a, cb, _ := testCmdActivator(t, Config{})
643 a.gens = gens
644 var got []*instancectl.StagedExe
645 cb.startCmd = func(_ context.Context, _ registry.Route, spec CmdSpec) (CmdUnit, error) {
646 got = append(got, spec.serving)
647 return cb.unit, nil
648 }
649 run := func(id string, spec CmdSpec) {
650 t.Helper()
651 spec.RunID = id
652 h, err := a.RunCmd(context.Background(), rt, spec)
653 if err != nil {
654 t.Fatalf("%s: %v", id, err)
655 }
656 if err := h.Stop(context.Background()); err != nil {
657 t.Fatal(err)
658 }
659 h.Release()
660 }
661 // A generation that names no file: the live link, whatever the
662 // caller put in the spec.
663 arrangeAwake(t, a, rt)
664 a.mu.Lock()
665 e := a.entries[rt.Host]
666 e.gen.StagedPath, e.gen.Digest = "", ""
667 a.mu.Unlock()
668 run("run_livelink", CmdSpec{serving: staged})
669 a.mu.Lock()
670 e.gen.Build, e.gen.StagedPath, e.gen.Digest = want.Version, want.Path, want.Digest
671 a.mu.Unlock()
672 run("run_staged", CmdSpec{})
673 if len(got) != 2 || got[0] != nil {
674 t.Fatalf("serving builds %v, want the live link for a generation that names no file", got)
675 }
676 if got[1] == nil || *got[1] != want {
677 t.Fatalf("serving build %v, want the current generation's %v", got[1], want)
678 }
679 })
680 }
681}
682
683// StartCmd names a run's serving build to root through CmdExe, and needs no
684// live link for it. A control channel that cannot carry the build refuses
685// the run rather than run the live link, which after a fallback names a
686// build the instance is not serving.
687func TestStartCmdNamesTheServingBuild(t *testing.T) {
688 b, rt := nestedBackend(t, &fakeControl{cmdInv: "inv1"}, &fakeUnit{})
689 b.layoutFn = func(registry.Route) (Layout, string) { return LayoutNested, "" }
690 dir := filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), CmdRunDirName, "run_a")
691 x := instancectl.StagedExe{Path: "/opt/carlos/versions/v2/blog", Version: "v2", Digest: strings.Repeat("a", 64)}
692 spec := CmdSpec{RunID: "run_a", RunDir: dir, Stdout: &syncBuffer{}, Stderr: &syncBuffer{}, serving: &x, genEnv: map[string]string{}}
693 if u, err := b.StartCmd(context.Background(), rt, spec); !errors.Is(err, errNoCmdExe) || u != nil {
694 t.Fatalf("StartCmd without CmdExe = %v, %v; want errNoCmdExe and no unit", u, err)
695 }
696 ctl := &exeControl{}
697 b.Control = ctl
698 u, err := b.StartCmd(context.Background(), rt, spec)
699 if err != nil {
700 t.Fatal(err)
701 }
702 u.(*cmdUnit).out.Close()
703 u.(*cmdUnit).errs.Close()
704 ctl.mu.Lock()
705 defer ctl.mu.Unlock()
706 if len(ctl.cmdExes) != 1 || ctl.cmdExes[0] != x {
707 t.Fatalf("cmd starts %v, want one naming %v", ctl.cmdExes, x)
708 }
709}
710