| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | "strings" |
| 8 | "testing" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | func TestPlan14GenerationWorkerStartBoundary(t *testing.T) { |
| 15 | t.Parallel() |
| 16 | for _, claimed := range []bool{false, true} { |
| 17 | for _, staged := range []bool{false, true} { |
| 18 | for _, pinned := range []bool{false, true} { |
| 19 | name := "legacy" |
| 20 | if claimed { |
| 21 | name = "claimed" |
| 22 | } |
| 23 | if staged { |
| 24 | name += "/staged" |
| 25 | } else { |
| 26 | name += "/plain" |
| 27 | } |
| 28 | if pinned { |
| 29 | name += "/pinned" |
| 30 | } else { |
| 31 | name += "/live" |
| 32 | } |
| 33 | t.Run(name, func(t *testing.T) { |
| 34 | a, b, _, reg, _ := stagedActivator(t) |
| 35 | t.Cleanup(func() { a.Shutdown(context.Background()); reg.Close() }) |
| 36 | a.cfg.HostWrites.Enabled = claimed |
| 37 | if !staged { |
| 38 | a.backend = b.fakeBackend |
| 39 | } |
| 40 | a.GenerationsFile = release.GenerationsPath(t.TempDir()) |
| 41 | var onto *release.StagedIdentity |
| 42 | if pinned { |
| 43 | id := fakeIdentity("v2") |
| 44 | onto = &id |
| 45 | } |
| 46 | gates := 0 |
| 47 | p, id, pending, err := a.startWorker(t.Context(), workerRoute("bot.x.internal"), nil, onto, func() error { gates++; return nil }) |
| 48 | if p != nil { |
| 49 | defer p.Stop(context.Background()) |
| 50 | } |
| 51 | if claimed { |
| 52 | if !errors.Is(err, ErrClaimIsolationUnavailable) || p != nil || id != (release.StagedIdentity{}) || pending != 0 || b.workerStarts() != 0 || gates != 0 { |
| 53 | t.Fatalf("unreserved worker reached backend: proc=%v id=%+v pending=%v starts=%d gates=%d err=%v", p, id, pending, b.workerStarts(), gates, err) |
| 54 | } |
| 55 | if _, err := os.Stat(a.GenerationsFile); !os.IsNotExist(err) { |
| 56 | t.Fatalf("refused start published agent generations: %v", err) |
| 57 | } |
| 58 | } else if err != nil || p == nil || b.workerStarts() != 1 || gates == 0 { |
| 59 | t.Fatalf("legacy worker changed: proc=%v starts=%d gates=%d err=%v", p, b.workerStarts(), gates, err) |
| 60 | } |
| 61 | }) |
| 62 | } |
| 63 | } |
| 64 | } |
| 65 | } |
| 66 | |
| 67 | func TestPlan14GenerationWorkerRealBackendRefusal(t *testing.T) { |
| 68 | t.Parallel() |
| 69 | for _, pinned := range []bool{false, true} { |
| 70 | name := "live" |
| 71 | if pinned { |
| 72 | name = "pinned" |
| 73 | } |
| 74 | t.Run(name, func(t *testing.T) { |
| 75 | a, b, rt, output, _ := workerGateService(t, nil) |
| 76 | a.cfg.HostWrites.Enabled = true |
| 77 | var onto *release.StagedIdentity |
| 78 | if pinned { |
| 79 | id, err := release.ResolveLive(b.OptDir, rt.Host) |
| 80 | if err != nil { |
| 81 | t.Fatal(err) |
| 82 | } |
| 83 | onto = &id |
| 84 | } |
| 85 | p, _, _, err := a.startWorker(t.Context(), rt, nil, onto, nil) |
| 86 | if p != nil { |
| 87 | defer p.Stop(context.Background()) |
| 88 | } |
| 89 | if !errors.Is(err, ErrClaimIsolationUnavailable) || p != nil { |
| 90 | t.Fatalf("real backend admitted unreserved worker: %v %v", p, err) |
| 91 | } |
| 92 | if _, err := os.Stat(output); !os.IsNotExist(err) { |
| 93 | t.Fatalf("refused worker executed: %v", err) |
| 94 | } |
| 95 | if _, err := os.Stat(a.GenerationsFile); !os.IsNotExist(err) { |
| 96 | t.Fatalf("refused worker published agent generations: %v", err) |
| 97 | } |
| 98 | }) |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | func TestPlan14GenerationWorkerReplacementBoundary(t *testing.T) { |
| 103 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 20*time.Millisecond) |
| 104 | for _, claimed := range []bool{false, true} { |
| 105 | for _, cause := range []Cause{CauseRelease, CauseRestart, CauseConfig} { |
| 106 | name := "legacy/" + string(cause) |
| 107 | if claimed { |
| 108 | name = "claimed/" + string(cause) |
| 109 | } |
| 110 | t.Run(name, func(t *testing.T) { |
| 111 | a, b, _, reg, _ := stagedActivator(t) |
| 112 | t.Cleanup(func() { a.Shutdown(context.Background()); reg.Close() }) |
| 113 | if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| 114 | t.Fatal(err) |
| 115 | } |
| 116 | old := b.lastWorker() |
| 117 | a.cfg.HostWrites.Enabled = claimed |
| 118 | a.GenerationsFile = release.GenerationsPath(t.TempDir()) |
| 119 | fallback := fakeIdentity("v1") |
| 120 | got := a.ReplaceWorker(t.Context(), workerAttempt(cause, ModeOverlap, fakeIdentity("v2")), &fallback, one) |
| 121 | if claimed { |
| 122 | if got.Outcome != SwapAbandoned || got.Executed || !strings.Contains(got.Reason, ErrClaimIsolationUnavailable.Error()) || b.workerStarts() != 1 { |
| 123 | t.Fatalf("unreserved replacement: %+v starts=%d", got, b.workerStarts()) |
| 124 | } |
| 125 | if exited := old.Exited(); exited { |
| 126 | t.Fatal("refused replacement stopped the serving worker") |
| 127 | } |
| 128 | if a.entryServing("bot.x.internal") != fallback { |
| 129 | t.Fatal("refused replacement changed serving identity") |
| 130 | } |
| 131 | if _, err := os.Stat(a.GenerationsFile); !os.IsNotExist(err) { |
| 132 | t.Fatalf("refused replacement published agent generations: %v", err) |
| 133 | } |
| 134 | } else if got.Outcome != SwapDone || !got.Executed || b.workerStarts() != 2 { |
| 135 | t.Fatalf("legacy replacement changed: %+v starts=%d", got, b.workerStarts()) |
| 136 | } |
| 137 | }) |
| 138 | } |
| 139 | } |
| 140 | } |
| 141 | |
| 142 | func TestPlan14GenerationWorkerSupervisionBoundary(t *testing.T) { |
| 143 | t.Parallel() |
| 144 | for _, claimed := range []bool{false, true} { |
| 145 | for _, cause := range []string{"crash", "max-run", "converge"} { |
| 146 | name := "legacy/" + cause |
| 147 | if claimed { |
| 148 | name = "claimed/" + cause |
| 149 | } |
| 150 | t.Run(name, func(t *testing.T) { |
| 151 | a, b, clk, reg, ver := stagedActivator(t) |
| 152 | t.Cleanup(func() { a.Shutdown(context.Background()); reg.Close() }) |
| 153 | rt := alwaysOnWorkerRoute("bot.x.internal") |
| 154 | if cause == "max-run" { |
| 155 | rt.MaxRunSeconds = 30 |
| 156 | } |
| 157 | if err := a.WakeWorker(rt); err != nil { |
| 158 | t.Fatal(err) |
| 159 | } |
| 160 | old := b.lastWorker() |
| 161 | a.cfg.HostWrites.Enabled = claimed |
| 162 | switch cause { |
| 163 | case "crash": |
| 164 | old.kill() |
| 165 | case "max-run": |
| 166 | clk.advance(31 * time.Second) |
| 167 | case "converge": |
| 168 | ver.Store("v2") |
| 169 | a.sweepOnce() |
| 170 | clk.advance(ConvergeGrace + time.Minute) |
| 171 | } |
| 172 | a.sweepOnce() |
| 173 | a.mu.Lock() |
| 174 | done := a.entries[rt.Host].respawnDone |
| 175 | a.mu.Unlock() |
| 176 | if done != nil { |
| 177 | select { |
| 178 | case <-done: |
| 179 | case <-time.After(5 * time.Second): |
| 180 | t.Fatal("respawn did not settle") |
| 181 | } |
| 182 | } |
| 183 | if claimed { |
| 184 | if b.workerStarts() != 1 { |
| 185 | t.Fatalf("supervision spawned without root reservation: %d", b.workerStarts()) |
| 186 | } |
| 187 | if exited := old.Exited(); exited && cause != "crash" { |
| 188 | t.Fatal("refused recycle stopped the serving worker") |
| 189 | } |
| 190 | } else if b.workerStarts() != 2 { |
| 191 | t.Fatalf("legacy supervision changed: %d", b.workerStarts()) |
| 192 | } |
| 193 | }) |
| 194 | } |
| 195 | } |
| 196 | } |
| 197 | |
amadan