| index 04048ab17..a8eca9ef2 100644 |
| --- a/internal/activator/cmd_test.go |
| +++ b/internal/activator/cmd_test.go |
| @@ -1,8 +1,16 @@ |
| package activator |
| |
| import ( |
| + "context" |
| + "errors" |
| + "path/filepath" |
| + "sync" |
| "testing" |
| "time" |
| + |
| + "github.com/carlosframework/platform/internal/hostwrite" |
| + "github.com/carlosframework/platform/internal/instancectl" |
| + "github.com/carlosframework/platform/internal/registry" |
| ) |
| |
| func (a *Service) stopWhyOf(host string) string { |
| @@ -26,3 +34,328 @@ func TestStopsRecordTheirReason(t *testing.T) { |
| t.Fatalf("Bounce: stopWhy %q, want %q", got, StopRestart) |
| } |
| } |
| + |
| +// fakeCmdUnit records the stop and whether replication was already closed |
| +// when it came. |
| +type fakeCmdUnit struct { |
| + mu sync.Mutex |
| + stops int |
| + repWasClosed bool |
| + settleErr error |
| + rep func() *fakeRep |
| + status instancectl.CmdStatus |
| +} |
| + |
| +func (u *fakeCmdUnit) Invocation() string { return "inv1" } |
| +func (u *fakeCmdUnit) Poll(context.Context) (instancectl.CmdStatus, error) { |
| + u.mu.Lock() |
| + defer u.mu.Unlock() |
| + return u.status, nil |
| +} |
| +func (u *fakeCmdUnit) StopAndSettle(context.Context) error { |
| + u.mu.Lock() |
| + defer u.mu.Unlock() |
| + u.stops++ |
| + if u.rep != nil { |
| + u.repWasClosed = u.rep().isClosed() |
| + } |
| + return u.settleErr |
| +} |
| +func (u *fakeCmdUnit) stopped() (int, bool) { |
| + u.mu.Lock() |
| + defer u.mu.Unlock() |
| + return u.stops, u.repWasClosed |
| +} |
| + |
| +// cmdFakeBackend is fakeBackend with the run capability and a nested layout. |
| +type cmdFakeBackend struct { |
| + *fakeBackend |
| + cmdMu sync.Mutex |
| + unit *fakeCmdUnit |
| + startErr error |
| + cmdStarts int |
| + gate chan struct{} // when set, StartCmd blocks until it closes |
| + entered chan struct{} // when set, closed as StartCmd is entered |
| +} |
| + |
| +func (b *cmdFakeBackend) StartCmd(ctx context.Context, rt registry.Route, spec CmdSpec) (CmdUnit, error) { |
| + if b.entered != nil { |
| + close(b.entered) |
| + } |
| + if b.gate != nil { |
| + <-b.gate |
| + } |
| + b.cmdMu.Lock() |
| + defer b.cmdMu.Unlock() |
| + b.cmdStarts++ |
| + if b.startErr != nil { |
| + return nil, b.startErr |
| + } |
| + return b.unit, nil |
| +} |
| + |
| +func testCmdActivator(t *testing.T, cfg Config) (*Service, *cmdFakeBackend, *fakeClock) { |
| + t.Helper() |
| + fb := &fakeBackend{layoutFn: func(registry.Route) (Layout, string) { return LayoutNested, "" }} |
| + cb := &cmdFakeBackend{fakeBackend: fb} |
| + cb.unit = &fakeCmdUnit{rep: fb.lastRep} |
| + clk := &fakeClock{t: time.Unix(1_700_000_000, 0)} |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if cfg.Bucket == "" { |
| + cfg.Bucket = "test-bucket" |
| + } |
| + if cfg.Idle == 0 { |
| + cfg.Idle = time.Second |
| + } |
| + a := New(cfg, cb, reg) |
| + a.now = clk.now |
| + return a, cb, clk |
| +} |
| + |
| +func TestRunCmdRefusesWithClaimControlArmed(t *testing.T) { |
| + a, cb, _ := testCmdActivator(t, Config{HostWrites: hostwrite.Config{Enabled: true}}) |
| + if _, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdClaim) { |
| + t.Fatalf("err = %v, want ErrCmdClaim", err) |
| + } |
| + if acquires, _, _ := cb.counts(); acquires != 0 { |
| + t.Fatal("a claim-armed refusal woke the instance") |
| + } |
| +} |
| + |
| +func TestRunCmdRefusesAFlatInstance(t *testing.T) { |
| + a, cb, _ := testCmdActivator(t, Config{}) |
| + cb.layoutFn = func(registry.Route) (Layout, string) { return LayoutFlat, "" } |
| + if _, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"}); !errors.Is(err, ErrCmdNotNested) { |
| + t.Fatalf("err = %v, want ErrCmdNotNested", err) |
| + } |
| +} |
| + |
| +func TestRunCmdWakesAHibernatedInstanceFirst(t *testing.T) { |
| + a, cb, _ := testCmdActivator(t, Config{}) |
| + h, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, _, wakes := cb.counts(); h.Invocation() != "inv1" || wakes != 1 || cb.cmdStarts != 1 { |
| + t.Fatalf("wake starts %d, cmd starts %d", wakes, cb.cmdStarts) |
| + } |
| + h.Release() |
| +} |
| + |
| +func TestRunCmdWakeFailureIsAWakeRefusal(t *testing.T) { |
| + a, cb, _ := testCmdActivator(t, Config{}) |
| + cb.acquireErr = errors.New("restore failed") |
| + _, err := a.RunCmd(context.Background(), testRoute("c.example.com"), CmdSpec{RunID: "run_a"}) |
| + var we *CmdWakeError |
| + if !errors.As(err, &we) || we.Reason == "" { |
| + t.Fatalf("err = %v, want a CmdWakeError with a member-safe reason", err) |
| + } |
| + if cb.cmdStarts != 0 { |
| + t.Fatal("a failed wake started the command") |
| + } |
| +} |
| + |
| +func TestRunCmdRefusesASecondRun(t *testing.T) { |
| + a, _, _ := testCmdActivator(t, Config{}) |
| + rt := testRoute("c.example.com") |
| + h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer h.Release() |
| + if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_b"}); !errors.Is(err, ErrCmdBusy) { |
| + t.Fatalf("err = %v, want ErrCmdBusy", err) |
| + } |
| +} |
| + |
| +func TestRunCmdHoldsTheIdleTimer(t *testing.T) { |
| + a, _, clk := testCmdActivator(t, Config{}) |
| + rt := testRoute("c.example.com") |
| + h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + clk.advance(time.Hour) |
| + a.sweepOnce() |
| + if a.entryCount() != 1 || a.stopWhyOf(rt.Host) != "" { |
| + t.Fatal("the idle timer stopped an instance with a run open") |
| + } |
| + h.Release() |
| + clk.advance(time.Hour) |
| + a.sweepOnce() |
| + waitFor(t, "hibernation after the run", func() bool { return a.entryCount() == 0 }) |
| +} |
| + |
| +// Every stop path stops and settles the run before replication closes, and |
| +// the run learns why (run spec §5.3, codex 1). |
| +func TestEveryStopPathSettlesTheRunBeforeReplicationCloses(t *testing.T) { |
| + cases := []struct { |
| + name string |
| + why string |
| + stop func(a *Service, cb *cmdFakeBackend, host string) |
| + }{ |
| + {"bounce", StopRestart, func(a *Service, _ *cmdFakeBackend, h string) { a.Bounce(h) }}, |
| + {"process exit", StopExited, func(a *Service, cb *cmdFakeBackend, _ string) { cb.lastProc().exit(); a.sweepOnce() }}, |
| + {"lease lost", StopLeaseLost, func(a *Service, cb *cmdFakeBackend, _ string) { cb.lastRep().setErr(errors.New("lost")); a.sweepOnce() }}, |
| + {"convergence", StopRelease, func(a *Service, _ *cmdFakeBackend, h string) { |
| + a.liveVer = func(string) (string, error) { return "v2", nil } |
| + a.ConvergeAdopted(h, "v2") |
| + }}, |
| + {"move fence", StopFenced, func(a *Service, _ *cmdFakeBackend, h string) { |
| + // FenceAndStop refuses a context without a deadline. |
| + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| + go func() { |
| + defer cancel() |
| + a.FenceAndStop(ctx, h, 1) //nolint:errcheck |
| + }() |
| + }}, |
| + {"retire", StopRetired, func(a *Service, _ *cmdFakeBackend, _ string) { a.RetireExcept(map[string]bool{}) }}, |
| + {"agent shutdown", StopShutdown, func(a *Service, _ *cmdFakeBackend, _ string) { |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + a.Shutdown(ctx) |
| + }}, |
| + } |
| + for _, c := range cases { |
| + t.Run(c.name, func(t *testing.T) { |
| + a, cb, _ := testCmdActivator(t, Config{}) |
| + cb.execVerFn = func() string { return "v1" } |
| + rt := testRoute("c.example.com") |
| + h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + c.stop(a, cb, rt.Host) |
| + waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 }) |
| + stops, repWasClosed := cb.unit.stopped() |
| + if stops != 1 { |
| + t.Fatalf("the cmd unit was stopped %d times, want 1", stops) |
| + } |
| + if repWasClosed { |
| + t.Fatal("replication closed before the run settled") |
| + } |
| + if !cb.lastRep().isClosed() { |
| + t.Fatal("replication never closed") |
| + } |
| + if got := h.Interrupted(); got != c.why { |
| + t.Fatalf("Interrupted() = %q, want %q", got, c.why) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestRunCmdNeverStartsUnderAStoppingEntry(t *testing.T) { |
| + a, cb, _ := testCmdActivator(t, Config{}) |
| + rt := testRoute("c.example.com") |
| + _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a", Prepare: func(context.Context) error { |
| + a.Bounce(rt.Host) // the instance starts stopping while the files download |
| + return nil |
| + }}) |
| + var se *CmdStoppedError |
| + if !errors.As(err, &se) || se.Reason != StopRestart { |
| + t.Fatalf("err = %v, want CmdStoppedError{%q}", err, StopRestart) |
| + } |
| + if cb.cmdStarts != 0 { |
| + t.Fatal("the cmd unit started under a stopping entry") |
| + } |
| +} |
| + |
| +func TestAStopDuringTheStartWaitsForIt(t *testing.T) { |
| + a, cb, _ := testCmdActivator(t, Config{}) |
| + cb.gate = make(chan struct{}) |
| + cb.entered = make(chan struct{}) |
| + rt := testRoute("c.example.com") |
| + done := make(chan error, 1) |
| + go func() { |
| + _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}) |
| + done <- err |
| + }() |
| + waitFor(t, "the run attached", func() bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + e := a.entries[rt.Host] |
| + return e != nil && e.cmd != nil |
| + }) |
| + // The attach comes before RunCmd's last awake check; the stop must land |
| + // after that check, with StartCmd in flight. |
| + <-cb.entered |
| + a.Bounce(rt.Host) |
| + time.Sleep(50 * time.Millisecond) |
| + if cb.lastRep().isClosed() { |
| + t.Fatal("replication closed while the cmd unit's start was still in flight") |
| + } |
| + close(cb.gate) |
| + if err := <-done; err != nil { |
| + t.Fatalf("RunCmd: %v", err) |
| + } |
| + waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 }) |
| + if stops, repWasClosed := cb.unit.stopped(); stops != 1 || repWasClosed { |
| + t.Fatalf("stops %d, replication closed first %v", stops, repWasClosed) |
| + } |
| +} |
| + |
| +func TestAnUnsettledRunMakesTheStopUncertain(t *testing.T) { |
| + for _, c := range []struct { |
| + settleErr error |
| + uncertain bool |
| + }{{nil, false}, {errors.New("processes remained"), true}} { |
| + a, cb, _ := testCmdActivator(t, Config{}) |
| + cb.unit.settleErr = c.settleErr |
| + rt := testRoute("c.example.com") |
| + if _, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + a.Bounce(rt.Host) |
| + waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 }) |
| + if got := a.StopUncertain(rt.Host); got != c.uncertain { |
| + t.Fatalf("settle error %v: StopUncertain = %v, want %v", c.settleErr, got, c.uncertain) |
| + } |
| + } |
| +} |
| + |
| +// A run released without a proven settle (its Stop failed, or was never |
| +// called) still belongs to the instance's next stop: that stop settles it |
| +// before replication closes, or is uncertain. |
| +func TestAReleasedRunThatNeverSettledStillHoldsTheStop(t *testing.T) { |
| + for _, c := range []struct { |
| + name string |
| + stopFirst bool |
| + settleErr error |
| + uncertain bool |
| + }{ |
| + {"stopped and settled", true, nil, false}, |
| + {"never stopped", false, nil, false}, |
| + {"stop failed", true, errors.New("processes remained"), true}, |
| + } { |
| + t.Run(c.name, func(t *testing.T) { |
| + a, cb, clk := testCmdActivator(t, Config{}) |
| + cb.unit.settleErr = c.settleErr |
| + rt := testRoute("c.example.com") |
| + h, err := a.RunCmd(context.Background(), rt, CmdSpec{RunID: "run_a"}) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if c.stopFirst { |
| + if err := h.Stop(context.Background()); (err != nil) != (c.settleErr != nil) { |
| + t.Fatalf("Stop: %v", err) |
| + } |
| + } |
| + h.Release() |
| + clk.advance(time.Hour) |
| + a.sweepOnce() // the release ended the idle hold |
| + waitFor(t, "the instance stopped", func() bool { return a.entryCount() == 0 }) |
| + if stops, repWasClosed := cb.unit.stopped(); stops != 1 || repWasClosed { |
| + t.Fatalf("stops %d, replication closed first %v", stops, repWasClosed) |
| + } |
| + if got := a.StopUncertain(rt.Host); got != c.uncertain { |
| + t.Fatalf("StopUncertain = %v, want %v", got, c.uncertain) |
| + } |
| + if got := h.Interrupted(); got != "" && c.stopFirst { |
| + t.Fatalf("Interrupted() = %q after the run's own stop", got) |
| + } |
| + }) |
| + } |
| +} |
amadan