| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "slices" |
| 11 | "sync" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/registry" |
| 16 | "github.com/carlosframework/platform/internal/release" |
| 17 | ) |
| 18 | |
| 19 | // restartEtag reads an app's restart-generation etag straight from the store, |
| 20 | // so a test asserts on what the bucket actually reports rather than a value it |
| 21 | // made up. |
| 22 | func restartEtag(t *testing.T, ctx context.Context, s release.Store, acct, app string) string { |
| 23 | t.Helper() |
| 24 | _, etag, err := s.Get(ctx, release.RestartGenKey(acct, app)) |
| 25 | if err != nil { |
| 26 | t.Fatalf("read restart generation %s/%s: %v", acct, app, err) |
| 27 | } |
| 28 | return etag |
| 29 | } |
| 30 | |
| 31 | // askRestart is a member's `carlos restart` landing in the bucket: the one |
| 32 | // write the whole feature turns on. |
| 33 | func askRestart(t *testing.T, ctx context.Context, s release.Store, acct, app string) { |
| 34 | t.Helper() |
| 35 | if err := release.TouchRestartGen(ctx, s, acct, app); err != nil { |
| 36 | t.Fatalf("TouchRestartGen %s/%s: %v", acct, app, err) |
| 37 | } |
| 38 | } |
| 39 | |
| 40 | // namedUnitRoute is the addr-backed, operator-named-unit shape issue #110 |
| 41 | // added and #113's second case is about: a console. Only the ROOT pass can |
| 42 | // cycle it. |
| 43 | func namedUnitRoute(host, acct, app, unit string) registry.Route { |
| 44 | rt := execRoute(host, acct, app) |
| 45 | rt.Socket = "" |
| 46 | rt.Addr = "127.0.0.1:9000" |
| 47 | rt.Unit = unit |
| 48 | return rt |
| 49 | } |
| 50 | |
| 51 | // recordingRestart is the root pass's dispatch seam: every host it was asked |
| 52 | // to cycle, in order, plus an optional per-host failure. |
| 53 | type recordingRestart struct { |
| 54 | hosts []string |
| 55 | fail map[string]error |
| 56 | } |
| 57 | |
| 58 | func (r *recordingRestart) fn(host string) error { |
| 59 | r.hosts = append(r.hosts, host) |
| 60 | if err, ok := r.fail[host]; ok { |
| 61 | return err |
| 62 | } |
| 63 | return nil |
| 64 | } |
| 65 | |
| 66 | // ---- the agent's sweep ------------------------------------------------------- |
| 67 | |
| 68 | func TestRestartGenPassFirstSightIsSilent(t *testing.T) { |
| 69 | ctx, s := genKit(t) |
| 70 | askRestart(t, ctx, s, "bab", "eleven") // requested before this box ever looked |
| 71 | routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")} |
| 72 | cache := map[[2]string]string{} |
| 73 | |
| 74 | moved, errs := restartGenPass(ctx, routes, s, cache) |
| 75 | if len(moved) != 0 { |
| 76 | t.Fatalf("first sight moved %v — a box that has just started must not cycle everything it serves", moved) |
| 77 | } |
| 78 | if len(errs) != 0 { |
| 79 | t.Fatalf("unexpected read errors: %v", errs) |
| 80 | } |
| 81 | if got, want := cache[[2]string{"bab", "eleven"}], restartEtag(t, ctx, s, "bab", "eleven"); got != want { |
| 82 | t.Fatalf("cache etag = %q, want the store's %q", got, want) |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | func TestRestartGenPassMovementAsksOnce(t *testing.T) { |
| 87 | ctx, s := genKit(t) |
| 88 | askRestart(t, ctx, s, "bab", "eleven") |
| 89 | routes := []registry.Route{ |
| 90 | execRoute("a.example.com", "bab", "eleven"), |
| 91 | execRoute("b.example.com", "bab", "eleven"), // same app, two routes |
| 92 | } |
| 93 | cache := map[[2]string]string{} |
| 94 | restartGenPass(ctx, routes, s, cache) // baseline |
| 95 | |
| 96 | askRestart(t, ctx, s, "bab", "eleven") |
| 97 | moved, errs := restartGenPass(ctx, routes, s, cache) |
| 98 | if len(errs) != 0 { |
| 99 | t.Fatalf("unexpected read errors: %v", errs) |
| 100 | } |
| 101 | if len(moved) != 1 || moved[0] != [2]string{"bab", "eleven"} { |
| 102 | t.Fatalf("moved = %v, want exactly one app entry (one request, one ask)", moved) |
| 103 | } |
| 104 | if moved, _ := restartGenPass(ctx, routes, s, cache); len(moved) != 0 { |
| 105 | t.Fatalf("the pass after a request moved %v — a stale cache would ask forever", moved) |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | func TestRestartGenPassAbsentStaysSilentAndFirstRequestMoves(t *testing.T) { |
| 110 | ctx, s := genKit(t) |
| 111 | routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")} |
| 112 | cache := map[[2]string]string{} |
| 113 | |
| 114 | // An app nobody has ever restarted: absent, legal, forever. |
| 115 | for i := range 2 { |
| 116 | moved, errs := restartGenPass(ctx, routes, s, cache) |
| 117 | if len(moved) != 0 { |
| 118 | t.Fatalf("pass %d: an app with no restart generation moved %v", i, moved) |
| 119 | } |
| 120 | if len(errs) != 0 { |
| 121 | t.Fatalf("pass %d: absent is not a read error: %v", i, errs) |
| 122 | } |
| 123 | } |
| 124 | if got := cache[[2]string{"bab", "eleven"}]; got != genAbsent { |
| 125 | t.Fatalf("cache = %q, want the known-absent sentinel %q", got, genAbsent) |
| 126 | } |
| 127 | |
| 128 | // Its FIRST ever request: absent -> present is movement, or the very first |
| 129 | // `carlos restart` for an app would be the one that never works. |
| 130 | askRestart(t, ctx, s, "bab", "eleven") |
| 131 | moved, _ := restartGenPass(ctx, routes, s, cache) |
| 132 | if len(moved) != 1 { |
| 133 | t.Fatalf("absent -> present moved %v, want the app", moved) |
| 134 | } |
| 135 | } |
| 136 | |
| 137 | func TestRestartGenPassReadErrorLeavesCacheAndKeepsGoing(t *testing.T) { |
| 138 | ctx, s := genKit(t) |
| 139 | askRestart(t, ctx, s, "bab", "eleven") |
| 140 | askRestart(t, ctx, s, "bab", "titogo") |
| 141 | routes := []registry.Route{ |
| 142 | execRoute("a.example.com", "bab", "eleven"), |
| 143 | execRoute("b.example.com", "bab", "titogo"), |
| 144 | } |
| 145 | cache := map[[2]string]string{} |
| 146 | restartGenPass(ctx, routes, s, cache) |
| 147 | before := cache[[2]string{"bab", "eleven"}] |
| 148 | |
| 149 | askRestart(t, ctx, s, "bab", "titogo") // a real request on the app that CAN be read |
| 150 | blip := &failingKeyStore{Store: s, failKey: release.RestartGenKey("bab", "eleven"), err: errors.New("503 slow down")} |
| 151 | moved, errs := restartGenPass(ctx, routes, blip, cache) |
| 152 | if len(errs) != 1 { |
| 153 | t.Fatalf("read errors = %v, want exactly one", errs) |
| 154 | } |
| 155 | if len(moved) != 1 || moved[0] != [2]string{"bab", "titogo"} { |
| 156 | t.Fatalf("moved = %v, want only the readable app — one bad object must not stop the pass", moved) |
| 157 | } |
| 158 | if got := cache[[2]string{"bab", "eleven"}]; got != before { |
| 159 | t.Fatalf("cache for the unreadable app = %q, want it untouched at %q (a cleared entry is a bogus first sight)", got, before) |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | func TestRestartGenPassDropsVanishedApps(t *testing.T) { |
| 164 | ctx, s := genKit(t) |
| 165 | askRestart(t, ctx, s, "bab", "eleven") |
| 166 | both := []registry.Route{ |
| 167 | execRoute("a.example.com", "bab", "eleven"), |
| 168 | execRoute("b.example.com", "bab", "titogo"), |
| 169 | } |
| 170 | cache := map[[2]string]string{} |
| 171 | restartGenPass(ctx, both, s, cache) |
| 172 | |
| 173 | restartGenPass(ctx, both[:1], s, cache) // titogo's route is gone |
| 174 | if _, ok := cache[[2]string{"bab", "titogo"}]; ok { |
| 175 | t.Fatal("a route that left the box kept its cache entry") |
| 176 | } |
| 177 | moved, _ := restartGenPass(ctx, both, s, cache) // and comes back |
| 178 | if len(moved) != 0 { |
| 179 | t.Fatalf("a re-added app moved %v, want a silent first sight", moved) |
| 180 | } |
| 181 | } |
| 182 | |
| 183 | // TestRestartGenPassIgnoresNonAdoptableRoutes: the membership predicate is |
| 184 | // release.Adoptable, the same one the adopt pass moves. A platform alias |
| 185 | // shares its canonical route's process, and a mid-provision route must not be |
| 186 | // disturbed — neither is an app this box should be reading a restart signal |
| 187 | // for. |
| 188 | func TestRestartGenPassIgnoresNonAdoptableRoutes(t *testing.T) { |
| 189 | ctx, s := genKit(t) |
| 190 | alias := execRoute("alias.oncarlos.com", "bab", "eleven") |
| 191 | alias.Platform = true |
| 192 | pending := execRoute("new.example.com", "bab", "titogo") |
| 193 | pending.Status = "creating" |
| 194 | |
| 195 | cache := map[[2]string]string{} |
| 196 | restartGenPass(ctx, []registry.Route{alias, pending}, s, cache) |
| 197 | if len(cache) != 0 { |
| 198 | t.Fatalf("cache = %v, want empty (neither route is adoptable)", cache) |
| 199 | } |
| 200 | } |
| 201 | |
| 202 | // ---- the agent's own children ------------------------------------------------ |
| 203 | |
| 204 | func TestBounceExecRoutesCyclesOnlyWhatRootCannot(t *testing.T) { |
| 205 | moved := [][2]string{{"bab", "eleven"}} |
| 206 | routes := []registry.Route{ |
| 207 | execRoute("exec.example.com", "bab", "eleven"), // the agent's child |
| 208 | unitRoute("unit.example.com", "bab", "eleven"), // carlos-app@ — root's |
| 209 | namedUnitRoute("console.example.com", "bab", "eleven", "console.service"), // #110 — root's |
| 210 | execRoute("other.example.com", "bab", "titogo"), // a different app |
| 211 | } |
| 212 | var bounced []string |
| 213 | n := bounceExecRoutes(routes, moved, func(h string) bool { |
| 214 | bounced = append(bounced, h) |
| 215 | return true |
| 216 | }) |
| 217 | if n != 1 || !slices.Equal(bounced, []string{"exec.example.com"}) { |
| 218 | t.Fatalf("bounced %v (n=%d), want only the exec-backed route of the moved app", bounced, n) |
| 219 | } |
| 220 | } |
| 221 | |
| 222 | // TestBounceExecRoutesReachesAWarmHibernator: a hibernate-flagged route is |
| 223 | // included on purpose. Bounce cannot start anything, so a SLEEPING instance is |
| 224 | // untouched — but a warm one is exactly the wedged instance `carlos restart` |
| 225 | // exists for, and leaving it serving would be the feature failing at its own |
| 226 | // headline case. |
| 227 | func TestBounceExecRoutesReachesAWarmHibernator(t *testing.T) { |
| 228 | rt := execRoute("warm.example.com", "bab", "eleven") |
| 229 | rt.Hibernate = true |
| 230 | var bounced []string |
| 231 | // The activator's own contract: false when not awake, true when it stopped |
| 232 | // something. Here it is awake. |
| 233 | bounceExecRoutes([]registry.Route{rt}, [][2]string{{"bab", "eleven"}}, func(h string) bool { |
| 234 | bounced = append(bounced, h) |
| 235 | return true |
| 236 | }) |
| 237 | if !slices.Equal(bounced, []string{"warm.example.com"}) { |
| 238 | t.Fatalf("bounced %v, want the warm hibernator", bounced) |
| 239 | } |
| 240 | } |
| 241 | |
| 242 | func TestBounceExecRoutesWithoutAnActivator(t *testing.T) { |
| 243 | routes := []registry.Route{execRoute("exec.example.com", "bab", "eleven")} |
| 244 | if n := bounceExecRoutes(routes, [][2]string{{"bab", "eleven"}}, nil); n != 0 { |
| 245 | t.Fatalf("bounced %d with no activator armed, want 0 and no panic", n) |
| 246 | } |
| 247 | } |
| 248 | |
| 249 | // ---- the agent's leg, and what makes it run ---------------------------------- |
| 250 | |
| 251 | // TestRestartRequestLegBouncesAndRelays: one serial move's worth of work — the |
| 252 | // agent's own children cycled in-process, and one marker touched for the units |
| 253 | // only root can reach. |
| 254 | func TestRestartRequestLegBouncesAndRelays(t *testing.T) { |
| 255 | ctx, s := genKit(t) |
| 256 | dd := t.TempDir() |
| 257 | marker := filepath.Join(dd, restartRequestMarker) |
| 258 | routes := []registry.Route{ |
| 259 | execRoute("exec.example.com", "bab", "eleven"), |
| 260 | unitRoute("unit.example.com", "bab", "eleven"), |
| 261 | } |
| 262 | var bounced []string |
| 263 | bounce := func(h string) bool { bounced = append(bounced, h); return true } |
| 264 | mem := newRestartMemory() |
| 265 | |
| 266 | // Baseline (what restartSeed does at arming): silent, no marker. |
| 267 | askRestart(t, ctx, s, "bab", "eleven") |
| 268 | if asked := restartRequestLeg(ctx, routes, s, dd, bounce, nil, mem); asked { |
| 269 | t.Fatal("the baselining leg asked for a root pass — first sight must be silent") |
| 270 | } |
| 271 | if _, err := os.Stat(marker); !errors.Is(err, os.ErrNotExist) { |
| 272 | t.Fatalf("marker written on a first sight (err %v)", err) |
| 273 | } |
| 274 | |
| 275 | askRestart(t, ctx, s, "bab", "eleven") |
| 276 | if asked := restartRequestLeg(ctx, routes, s, dd, bounce, nil, mem); !asked { |
| 277 | t.Fatal("a moved restart generation must ask for a root pass") |
| 278 | } |
| 279 | if !slices.Equal(bounced, []string{"exec.example.com"}) { |
| 280 | t.Fatalf("bounced %v, want the exec child only (the unit route is root's)", bounced) |
| 281 | } |
| 282 | if _, err := os.Stat(marker); err != nil { |
| 283 | t.Fatalf("no marker after a request: %v", err) |
| 284 | } |
| 285 | |
| 286 | // And a leg with nothing moved neither bounces nor re-asks. |
| 287 | bounced = nil |
| 288 | if asked := restartRequestLeg(ctx, routes, s, dd, bounce, nil, mem); asked { |
| 289 | t.Fatal("an unmoved generation asked for a root pass") |
| 290 | } |
| 291 | if len(bounced) != 0 { |
| 292 | t.Fatalf("bounced %v with nothing moved", bounced) |
| 293 | } |
| 294 | } |
| 295 | |
| 296 | // TestFastTickRunsTheRestartLegOnASerialMove pins the wiring the redesign rests |
| 297 | // on: the restart check rides the SERIAL, which is what keeps an idle box at |
| 298 | // one GET per tick. A boot seed must not run it (that would cycle processes for |
| 299 | // bumps that happened while the box was down), an unchanged serial must not run |
| 300 | // it, and a real move must — once, after the pass set. |
| 301 | func TestFastTickRunsTheRestartLegOnASerialMove(t *testing.T) { |
| 302 | env := newFastFixture(t) |
| 303 | var passes, legs int |
| 304 | d := env.deps(func() { passes++ }) |
| 305 | d.restartLeg = func(context.Context) { |
| 306 | legs++ |
| 307 | if legs > passes { |
| 308 | t.Errorf("the restart leg ran before/without the pass set (legs=%d passes=%d)", legs, passes) |
| 309 | } |
| 310 | } |
| 311 | now := time.Unix(1754640000, 0) |
| 312 | ctx := context.Background() |
| 313 | |
| 314 | if err := release.BumpSerial(ctx, env.store, now); err != nil { |
| 315 | t.Fatalf("BumpSerial: %v", err) |
| 316 | } |
| 317 | fastTick(ctx, d, now) // first sight of a present serial: seed only |
| 318 | if legs != 0 { |
| 319 | t.Fatalf("boot seed ran the restart leg %d times, want 0", legs) |
| 320 | } |
| 321 | fastTick(ctx, d, now.Add(10*time.Second)) // unchanged |
| 322 | if legs != 0 { |
| 323 | t.Fatalf("unchanged serial ran the restart leg %d times, want 0", legs) |
| 324 | } |
| 325 | if err := release.BumpSerial(ctx, env.store, now); err != nil { |
| 326 | t.Fatalf("BumpSerial: %v", err) |
| 327 | } |
| 328 | fastTick(ctx, d, now.Add(20*time.Second)) |
| 329 | if legs != 1 { |
| 330 | t.Fatalf("a serial move ran the restart leg %d times, want 1", legs) |
| 331 | } |
| 332 | } |
| 333 | |
| 334 | // ---- the root pass ----------------------------------------------------------- |
| 335 | |
| 336 | func marksOf(t *testing.T, path string) *restartMarks { |
| 337 | t.Helper() |
| 338 | raw, err := os.ReadFile(path) |
| 339 | if err != nil { |
| 340 | t.Fatalf("read marks: %v", err) |
| 341 | } |
| 342 | m := &restartMarks{} |
| 343 | if err := json.Unmarshal(raw, m); err != nil { |
| 344 | t.Fatalf("decode marks %s: %v", raw, err) |
| 345 | } |
| 346 | return m |
| 347 | } |
| 348 | |
| 349 | // TestRestartPassSeedsWithoutReplaying is the expensive lesson this repo has |
| 350 | // already paid for once: a box that has never recorded a baseline records one |
| 351 | // and restarts NOTHING. Every app on a freshly upgraded box has a restart |
| 352 | // generation the moment anyone has ever used the verb, and cycling them all on |
| 353 | // the first pass would take the box out. |
| 354 | func TestRestartPassSeedsWithoutReplaying(t *testing.T) { |
| 355 | ctx, s := genKit(t) |
| 356 | askRestart(t, ctx, s, "bab", "eleven") |
| 357 | askRestart(t, ctx, s, "bab", "titogo") |
| 358 | routes := []registry.Route{ |
| 359 | unitRoute("a.example.com", "bab", "eleven"), |
| 360 | unitRoute("b.example.com", "bab", "titogo"), |
| 361 | } |
| 362 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 363 | rec := &recordingRestart{} |
| 364 | |
| 365 | restartPass(ctx, routes, s, path, rec.fn) |
| 366 | if len(rec.hosts) != 0 { |
| 367 | t.Fatalf("the seeding pass cycled %v, want nothing", rec.hosts) |
| 368 | } |
| 369 | m := marksOf(t, path) |
| 370 | if len(m.Apps) != 2 { |
| 371 | t.Fatalf("marks = %v, want both apps baselined", m.Apps) |
| 372 | } |
| 373 | // And the very next pass is a quiet one, not a replay. |
| 374 | restartPass(ctx, routes, s, path, rec.fn) |
| 375 | if len(rec.hosts) != 0 { |
| 376 | t.Fatalf("the pass after seeding cycled %v, want nothing", rec.hosts) |
| 377 | } |
| 378 | } |
| 379 | |
| 380 | // TestRestartPassCyclesOnMovement is the feature working: baseline, request, |
| 381 | // dispatch — for every non-hibernating route of THAT app and no other. |
| 382 | func TestRestartPassCyclesOnMovement(t *testing.T) { |
| 383 | ctx, s := genKit(t) |
| 384 | askRestart(t, ctx, s, "bab", "eleven") |
| 385 | askRestart(t, ctx, s, "bab", "titogo") |
| 386 | sleeping := unitRoute("sleep.example.com", "bab", "eleven") |
| 387 | sleeping.Hibernate = true |
| 388 | routes := []registry.Route{ |
| 389 | unitRoute("a.example.com", "bab", "eleven"), |
| 390 | namedUnitRoute("console.example.com", "bab", "eleven", "console.service"), |
| 391 | sleeping, |
| 392 | unitRoute("z.example.com", "bab", "titogo"), // a bystander app |
| 393 | } |
| 394 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 395 | rec := &recordingRestart{} |
| 396 | restartPass(ctx, routes, s, path, rec.fn) // baseline |
| 397 | |
| 398 | askRestart(t, ctx, s, "bab", "eleven") |
| 399 | restartPass(ctx, routes, s, path, rec.fn) |
| 400 | |
| 401 | slices.Sort(rec.hosts) |
| 402 | want := []string{"a.example.com", "console.example.com"} |
| 403 | if !slices.Equal(rec.hosts, want) { |
| 404 | t.Fatalf("cycled %v, want %v — the hibernating route must be left asleep and another app's route untouched", |
| 405 | rec.hosts, want) |
| 406 | } |
| 407 | if got, want := marksOf(t, path).Apps[restartMarkKey("bab", "eleven")], restartEtag(t, ctx, s, "bab", "eleven"); got != want { |
| 408 | t.Fatalf("mark = %q, want the etag just acted on %q", got, want) |
| 409 | } |
| 410 | // Idempotent: the same request is not served twice. |
| 411 | rec.hosts = nil |
| 412 | restartPass(ctx, routes, s, path, rec.fn) |
| 413 | if len(rec.hosts) != 0 { |
| 414 | t.Fatalf("a second pass on the same generation cycled %v, want nothing", rec.hosts) |
| 415 | } |
| 416 | } |
| 417 | |
| 418 | // TestRestartPassActsOnAnAppsFirstEverRequest: an app with no object yet is |
| 419 | // recorded as ABSENT, and the object's later appearance is movement away from |
| 420 | // that recorded value. Both halves matter — without the recorded absence, the |
| 421 | // first `carlos restart` for every app is either dropped (if "no mark" seeds) |
| 422 | // or joined by a spurious cycle of every other app in the same state (if "no |
| 423 | // mark" dispatches). |
| 424 | func TestRestartPassActsOnAnAppsFirstEverRequest(t *testing.T) { |
| 425 | ctx, s := genKit(t) |
| 426 | routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")} |
| 427 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 428 | rec := &recordingRestart{} |
| 429 | |
| 430 | restartPass(ctx, routes, s, path, rec.fn) // baseline: the app has no object at all |
| 431 | if got := marksOf(t, path).Apps[restartMarkKey("bab", "eleven")]; got != restartAbsent { |
| 432 | t.Fatalf("mark = %q, want the recorded absent sentinel", got) |
| 433 | } |
| 434 | |
| 435 | askRestart(t, ctx, s, "bab", "eleven") |
| 436 | restartPass(ctx, routes, s, path, rec.fn) |
| 437 | if !slices.Equal(rec.hosts, []string{"a.example.com"}) { |
| 438 | t.Fatalf("cycled %v, want the app's route on its first ever request", rec.hosts) |
| 439 | } |
| 440 | } |
| 441 | |
| 442 | // TestRestartPassSeedsAnAppItHasNeverSeen is the storm fix (review, Critical). |
| 443 | // A route added to this box for an app that was restarted somewhere else weeks |
| 444 | // ago must be RECORDED, not cycled: this box owes nothing to a request it never |
| 445 | // saw, and reading "no mark" as movement is what turned a registry that landed |
| 446 | // after the marks file into a box that try-restarted its entire route table in |
| 447 | // one pass. |
| 448 | func TestRestartPassSeedsAnAppItHasNeverSeen(t *testing.T) { |
| 449 | ctx, s := genKit(t) |
| 450 | askRestart(t, ctx, s, "bab", "eleven") |
| 451 | askRestart(t, ctx, s, "bab", "titogo") // both restarted long ago, elsewhere |
| 452 | first := []registry.Route{unitRoute("a.example.com", "bab", "eleven")} |
| 453 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 454 | rec := &recordingRestart{} |
| 455 | restartPass(ctx, first, s, path, rec.fn) // this box knows about eleven only |
| 456 | |
| 457 | // titogo's route arrives now — a fresh instance, mid-provision until this |
| 458 | // moment. Its app's object has existed for weeks. |
| 459 | both := append(first, unitRoute("b.example.com", "bab", "titogo")) |
| 460 | restartPass(ctx, both, s, path, rec.fn) |
| 461 | if len(rec.hosts) != 0 { |
| 462 | t.Fatalf("a newly-seen app cycled %v, want nothing — this box never saw that request", rec.hosts) |
| 463 | } |
| 464 | if got, want := marksOf(t, path).Apps[restartMarkKey("bab", "titogo")], restartEtag(t, ctx, s, "bab", "titogo"); got != want { |
| 465 | t.Fatalf("mark = %q, want the seeded etag %q", got, want) |
| 466 | } |
| 467 | // And it is live from here: the NEXT request cycles it. |
| 468 | askRestart(t, ctx, s, "bab", "titogo") |
| 469 | restartPass(ctx, both, s, path, rec.fn) |
| 470 | if !slices.Equal(rec.hosts, []string{"b.example.com"}) { |
| 471 | t.Fatalf("cycled %v, want the newly-seeded app's route on the next real request", rec.hosts) |
| 472 | } |
| 473 | } |
| 474 | |
| 475 | // TestRestartPassWithNoRoutesWritesNothing: a box whose registry has not landed |
| 476 | // yet (fresh machine, /data restoring from litestream) must not be able to |
| 477 | // record "I know about no apps" — that file is what makes every route it later |
| 478 | // gains look like movement. |
| 479 | func TestRestartPassWithNoRoutesWritesNothing(t *testing.T) { |
| 480 | ctx, s := genKit(t) |
| 481 | askRestart(t, ctx, s, "bab", "eleven") |
| 482 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 483 | rec := &recordingRestart{} |
| 484 | |
| 485 | restartPass(ctx, nil, s, path, rec.fn) |
| 486 | if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) { |
| 487 | t.Fatalf("a pass with zero adoptable routes wrote %s (err %v) — it knows nothing and may claim nothing", path, err) |
| 488 | } |
| 489 | // The registry lands; the app is seeded, not cycled. |
| 490 | routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")} |
| 491 | restartPass(ctx, routes, s, path, rec.fn) |
| 492 | if len(rec.hosts) != 0 { |
| 493 | t.Fatalf("cycled %v after the registry landed, want a silent baseline", rec.hosts) |
| 494 | } |
| 495 | } |
| 496 | |
| 497 | // TestRestartPassDefersABaselineWithReadErrors: a baseline that omits the apps |
| 498 | // a transient 503 hid is a baseline that turns those apps into movement on the |
| 499 | // next pass. Nothing is written until a pass can read every app. |
| 500 | func TestRestartPassDefersABaselineWithReadErrors(t *testing.T) { |
| 501 | ctx, s := genKit(t) |
| 502 | askRestart(t, ctx, s, "bab", "eleven") |
| 503 | askRestart(t, ctx, s, "bab", "titogo") |
| 504 | routes := []registry.Route{ |
| 505 | unitRoute("a.example.com", "bab", "eleven"), |
| 506 | unitRoute("b.example.com", "bab", "titogo"), |
| 507 | } |
| 508 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 509 | rec := &recordingRestart{} |
| 510 | |
| 511 | blip := &failingKeyStore{Store: s, failKey: release.RestartGenKey("bab", "eleven"), err: errors.New("503 slow down")} |
| 512 | restartPass(ctx, routes, blip, path, rec.fn) |
| 513 | if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) { |
| 514 | t.Fatalf("a baselining pass with a read error wrote %s (err %v)", path, err) |
| 515 | } |
| 516 | |
| 517 | // A clean pass baselines both, and cycles neither. |
| 518 | restartPass(ctx, routes, s, path, rec.fn) |
| 519 | if len(rec.hosts) != 0 { |
| 520 | t.Fatalf("the clean baselining pass cycled %v, want nothing", rec.hosts) |
| 521 | } |
| 522 | if m := marksOf(t, path); len(m.Apps) != 2 { |
| 523 | t.Fatalf("marks = %v, want both apps", m.Apps) |
| 524 | } |
| 525 | } |
| 526 | |
| 527 | // TestRestartPassRefusesWithoutALock: no lock means either a concurrent root |
| 528 | // pass or a /data this pass cannot write. Dispatching under the first |
| 529 | // double-restarts one request; under the second it restarts hourly forever, |
| 530 | // since the mark can never advance. Neither is allowed to happen. |
| 531 | func TestRestartPassRefusesWithoutALock(t *testing.T) { |
| 532 | ctx, s := genKit(t) |
| 533 | askRestart(t, ctx, s, "bab", "eleven") |
| 534 | dir := t.TempDir() |
| 535 | path := filepath.Join(dir, "restart-marks.json") |
| 536 | routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")} |
| 537 | rec := &recordingRestart{} |
| 538 | restartPass(ctx, routes, s, path, rec.fn) // baseline, with a working lock |
| 539 | |
| 540 | // A directory where the lock file needs to be: opening it fails (EISDIR), |
| 541 | // exactly as opening it would fail on a read-only or full /data. |
| 542 | if err := os.Remove(path + ".lock"); err != nil { |
| 543 | t.Fatal(err) |
| 544 | } |
| 545 | if err := os.Mkdir(path+".lock", 0o755); err != nil { |
| 546 | t.Fatal(err) |
| 547 | } |
| 548 | askRestart(t, ctx, s, "bab", "eleven") |
| 549 | restartPass(ctx, routes, s, path, rec.fn) |
| 550 | if len(rec.hosts) != 0 { |
| 551 | t.Fatalf("cycled %v with no lock — an unrecordable restart repeats forever", rec.hosts) |
| 552 | } |
| 553 | } |
| 554 | |
| 555 | // TestRestartPassSerializesConcurrentPasses: three root units can run this at |
| 556 | // once (carlos-restart, carlos-stage, carlos-adopt). Under the lock, one |
| 557 | // request produces ONE dispatch however many passes race for it. |
| 558 | func TestRestartPassSerializesConcurrentPasses(t *testing.T) { |
| 559 | ctx, s := genKit(t) |
| 560 | askRestart(t, ctx, s, "bab", "eleven") |
| 561 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 562 | routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")} |
| 563 | var mu sync.Mutex |
| 564 | var hosts []string |
| 565 | dispatch := func(h string) error { |
| 566 | mu.Lock() |
| 567 | defer mu.Unlock() |
| 568 | hosts = append(hosts, h) |
| 569 | return nil |
| 570 | } |
| 571 | restartPass(ctx, routes, s, path, dispatch) // baseline |
| 572 | |
| 573 | askRestart(t, ctx, s, "bab", "eleven") |
| 574 | var wg sync.WaitGroup |
| 575 | for range 3 { |
| 576 | wg.Add(1) |
| 577 | go func() { |
| 578 | defer wg.Done() |
| 579 | restartPass(ctx, routes, s, path, dispatch) |
| 580 | }() |
| 581 | } |
| 582 | wg.Wait() |
| 583 | if len(hosts) != 1 { |
| 584 | t.Fatalf("three concurrent passes dispatched %v, want exactly one — the flock is what makes a request singular", hosts) |
| 585 | } |
| 586 | } |
| 587 | |
| 588 | // TestRestartPassAdvancesTheMarkOnAFailedDispatch: a restart request is a |
| 589 | // point-in-time event, not a desired state. A named unit that will not come |
| 590 | // back must be reported, not retried on every pass until the end of time. |
| 591 | func TestRestartPassAdvancesTheMarkOnAFailedDispatch(t *testing.T) { |
| 592 | ctx, s := genKit(t) |
| 593 | askRestart(t, ctx, s, "bab", "eleven") |
| 594 | routes := []registry.Route{namedUnitRoute("console.example.com", "bab", "eleven", "wrong.service")} |
| 595 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 596 | rec := &recordingRestart{fail: map[string]error{"console.example.com": errors.New("Unit wrong.service not loaded")}} |
| 597 | restartPass(ctx, routes, s, path, rec.fn) // baseline |
| 598 | |
| 599 | askRestart(t, ctx, s, "bab", "eleven") |
| 600 | restartPass(ctx, routes, s, path, rec.fn) |
| 601 | if len(rec.hosts) != 1 { |
| 602 | t.Fatalf("dispatched %v, want one attempt", rec.hosts) |
| 603 | } |
| 604 | rec.hosts = nil |
| 605 | restartPass(ctx, routes, s, path, rec.fn) |
| 606 | if len(rec.hosts) != 0 { |
| 607 | t.Fatalf("a failed dispatch was retried (%v) — the request must not become a standing order", rec.hosts) |
| 608 | } |
| 609 | } |
| 610 | |
| 611 | // TestRestartPassUnreadableMarksReBaseline: a wiped or corrupt marks file |
| 612 | // fails toward NOT restarting. The cost is one dropped request, logged and |
| 613 | // re-runnable; the alternative is a box that cycles everything it serves the |
| 614 | // moment a state file goes missing. |
| 615 | func TestRestartPassUnreadableMarksReBaseline(t *testing.T) { |
| 616 | ctx, s := genKit(t) |
| 617 | askRestart(t, ctx, s, "bab", "eleven") |
| 618 | routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")} |
| 619 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 620 | if err := os.WriteFile(path, []byte("{not json"), 0o644); err != nil { |
| 621 | t.Fatal(err) |
| 622 | } |
| 623 | rec := &recordingRestart{} |
| 624 | |
| 625 | restartPass(ctx, routes, s, path, rec.fn) |
| 626 | if len(rec.hosts) != 0 { |
| 627 | t.Fatalf("a corrupt marks file cycled %v, want nothing", rec.hosts) |
| 628 | } |
| 629 | if got, want := marksOf(t, path).Apps[restartMarkKey("bab", "eleven")], restartEtag(t, ctx, s, "bab", "eleven"); got != want { |
| 630 | t.Fatalf("re-baselined mark = %q, want %q", got, want) |
| 631 | } |
| 632 | } |
| 633 | |
| 634 | // ---- the handover between the two halves ------------------------------------- |
| 635 | |
| 636 | // TestSeedRestartMarksIsCreateOnly: the agent writes the baseline as it arms, |
| 637 | // so a request made before the root pass has ever run is not silently dropped |
| 638 | // — but it must never overwrite a mark the root pass has already advanced, or |
| 639 | // a served request would be served again. Absent sightings ARE recorded: that |
| 640 | // sentinel is what makes an app's first ever request movement. |
| 641 | func TestSeedRestartMarksIsCreateOnly(t *testing.T) { |
| 642 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 643 | mem := newRestartMemory() |
| 644 | mem.gen = map[[2]string]string{ |
| 645 | {"bab", "eleven"}: "etag-1", |
| 646 | {"bab", "titogo"}: genAbsent, // seen, and it has no object |
| 647 | } |
| 648 | mem.scope = map[[2]string]string{ |
| 649 | {"bab", "eleven"}: release.RestartScopeNone, |
| 650 | } |
| 651 | need, err := seedRestartMarks(path, mem) |
| 652 | if err != nil { |
| 653 | t.Fatalf("seed: %v", err) |
| 654 | } |
| 655 | if need { |
| 656 | // The file this seed just wrote carries both memories, so there is |
| 657 | // nothing for a root pass to fill in. |
| 658 | t.Fatal("a seed that wrote the whole baseline still asked for a root pass") |
| 659 | } |
| 660 | m := marksOf(t, path) |
| 661 | if m.Apps[restartMarkKey("bab", "eleven")] != "etag-1" || |
| 662 | m.Apps[restartMarkKey("bab", "titogo")] != restartAbsent { |
| 663 | t.Fatalf("seeded marks = %v, want the etag and the recorded absence", m.Apps) |
| 664 | } |
| 665 | // The narrowed memory is seeded by the same sweep and in the same file: |
| 666 | // a baseline that recorded one signal and not the other would leave the |
| 667 | // root pass reading the second one as a first sight for a second time. |
| 668 | if m.Scopes[restartMarkKey("bab", "eleven")] != release.RestartScopeNone { |
| 669 | t.Fatalf("seeded scope marks = %v, want the recorded empty log", m.Scopes) |
| 670 | } |
| 671 | |
| 672 | mem.gen[[2]string{"bab", "eleven"}] = "etag-2" |
| 673 | if need, err := seedRestartMarks(path, mem); err != nil { |
| 674 | t.Fatalf("second seed: %v", err) |
| 675 | } else if need { |
| 676 | t.Fatal("a complete baseline on disk still asked for a root pass — every agent start would ask") |
| 677 | } |
| 678 | if got := marksOf(t, path).Apps[restartMarkKey("bab", "eleven")]; got != "etag-1" { |
| 679 | t.Fatalf("mark = %q after a second seed, want the original %q — seeding must never overwrite", got, "etag-1") |
| 680 | } |
| 681 | } |
| 682 | |
| 683 | // TestSeedRestartMarksRefusesAnEmptyBaseline: a sweep that saw nothing is most |
| 684 | // often a box whose registry has not landed yet. A file claiming "this box |
| 685 | // knows about no apps" is exactly what makes the root pass treat every route |
| 686 | // the box later gains as movement. |
| 687 | func TestSeedRestartMarksRefusesAnEmptyBaseline(t *testing.T) { |
| 688 | path := filepath.Join(t.TempDir(), "restart-marks.json") |
| 689 | if need, err := seedRestartMarks(path, newRestartMemory()); err != nil { |
| 690 | t.Fatalf("seed: %v", err) |
| 691 | } else if need { |
| 692 | t.Fatal("a sweep that saw nothing asked for a root pass; it may claim nothing at all") |
| 693 | } |
| 694 | if _, err := os.Stat(path); !errors.Is(err, os.ErrNotExist) { |
| 695 | t.Fatalf("an empty sweep wrote %s (err %v), want no file at all", path, err) |
| 696 | } |
| 697 | } |
| 698 | |
| 699 | // TestSeededBaselineIsWhatTheRootPassTrusts closes the handover: the agent |
| 700 | // seeds, a member then asks, and the root pass — running for the FIRST time, |
| 701 | // with no memory of its own — still cycles. Without the seed that pass would |
| 702 | // baseline instead and the request would vanish having been answered |
| 703 | // "requested". |
| 704 | func TestSeededBaselineIsWhatTheRootPassTrusts(t *testing.T) { |
| 705 | ctx, s := genKit(t) |
| 706 | dd := t.TempDir() |
| 707 | routes := []registry.Route{unitRoute("a.example.com", "bab", "eleven")} |
| 708 | askRestart(t, ctx, s, "bab", "eleven") // an older request, already served |
| 709 | |
| 710 | // The agent's first tick: baseline in memory, baseline on disk. |
| 711 | mem := newRestartMemory() |
| 712 | restartGenPass(ctx, routes, s, mem.gen) |
| 713 | restartScopePass(ctx, routes, s, mem.scope) |
| 714 | if _, err := seedRestartMarks(restartMarksPath(dd), mem); err != nil { |
| 715 | t.Fatalf("seed: %v", err) |
| 716 | } |
| 717 | |
| 718 | askRestart(t, ctx, s, "bab", "eleven") // the member types `carlos restart` |
| 719 | moved, _ := restartGenPass(ctx, routes, s, mem.gen) |
| 720 | if len(moved) != 1 { |
| 721 | t.Fatalf("the agent missed the request: moved = %v", moved) |
| 722 | } |
| 723 | rec := &recordingRestart{} |
| 724 | restartPass(ctx, routes, s, restartMarksPath(dd), rec.fn) |
| 725 | if !slices.Equal(rec.hosts, []string{"a.example.com"}) { |
| 726 | t.Fatalf("the first root pass cycled %v, want the requested route", rec.hosts) |
| 727 | } |
| 728 | } |
| 729 | |
| 730 | // ---- Task 11: sidecars in the restart pass ------------------------------- |
| 731 | |
| 732 | // TestRestartPassCyclesSidecars: release.Adoptable requires Kind == |
| 733 | // "instance" and deliberately stays that way (adopt membership, capability |
| 734 | // convergence, socket/backing advisories, status pages all correctly |
| 735 | // exclude a sidecar) — restartCandidates is the widened, NAMED opt-in |
| 736 | // `carlos restart -app X` needs instead. |
| 737 | func TestRestartPassCyclesSidecars(t *testing.T) { |
| 738 | routes := []registry.Route{ |
| 739 | {Host: "bot.x.sidecar.internal", Kind: "sidecar", App: "eleven", Account: "bdm", Status: "ready"}, |
| 740 | } |
| 741 | if len(restartCandidates(routes)) != 1 { |
| 742 | t.Fatal("carlos restart -app X must cycle that app's sidecars too") |
| 743 | } |
| 744 | } |
| 745 | |
| 746 | // TestRestartCandidatesStillExcludesNonAdoptableNonSidecarRoutes: the |
| 747 | // widening is additive — everything release.Adoptable already excluded for |
| 748 | // a non-sidecar route (a Platform alias, a mid-provision route) stays |
| 749 | // excluded. |
| 750 | func TestRestartCandidatesStillExcludesNonAdoptableNonSidecarRoutes(t *testing.T) { |
| 751 | alias := execRoute("alias.oncarlos.com", "bab", "eleven") |
| 752 | alias.Platform = true |
| 753 | pending := execRoute("new.example.com", "bab", "titogo") |
| 754 | pending.Status = "creating" |
| 755 | if got := restartCandidates([]registry.Route{alias, pending}); len(got) != 0 { |
| 756 | t.Fatalf("restartCandidates = %v, want none", got) |
| 757 | } |
| 758 | } |
| 759 | |
| 760 | // TestRestartAppsIncludesASidecarOnlyApp: an app whose only route on this |
| 761 | // box is a sidecar must still be tracked — restartApps (via |
| 762 | // restartCandidates) is what both restartGenPass and restartPass use to |
| 763 | // decide which apps' restart generation to read at all. Without this, a |
| 764 | // sidecar-only app's `carlos restart` request is never even noticed. |
| 765 | func TestRestartAppsIncludesASidecarOnlyApp(t *testing.T) { |
| 766 | routes := []registry.Route{ |
| 767 | {Host: "bot.x.sidecar.internal", Kind: "sidecar", App: "eleven", Account: "bdm", Status: "ready"}, |
| 768 | } |
| 769 | apps := restartApps(routes) |
| 770 | if len(apps) != 1 || apps[0] != [2]string{"bdm", "eleven"} { |
| 771 | t.Fatalf("restartApps = %v, want [[bdm eleven]]", apps) |
| 772 | } |
| 773 | } |
| 774 | |
| 775 | // ---- Task 11: cycling a sidecar's own child ------------------------------ |
| 776 | |
| 777 | // alwaysOnSidecarRoute and hibernatingSidecarRoute are cycleWorkers' |
| 778 | // fixtures — Hibernate governs nothing in cycleWorkers itself (bounce's own |
| 779 | // awake check does the filtering), but the two shapes exist because |
| 780 | // activator.Service.Bounce's stop half behaves identically for both and |
| 781 | // cycleWorkers must too. |
| 782 | func sidecarRoute(host, acct, app string) registry.Route { |
| 783 | return registry.Route{Host: host, Kind: "sidecar", App: app, Account: acct, Status: "ready"} |
| 784 | } |
| 785 | |
| 786 | // TestCycleWorkersBouncesAndWakesARunningSidecar: nothing dials a worker |
| 787 | // the way a request dials an instance, so Bounce's stop half — reused |
| 788 | // verbatim from bounceExecRoutes' own bounce callback — must be paired |
| 789 | // with an explicit wake here, or the sidecar stays down (see |
| 790 | // activator.Service.Bounce's doc comment). Only a HOST BOUNCE ACTUALLY |
| 791 | // STOPPED gets woken. |
| 792 | func TestCycleWorkersBouncesAndWakesARunningSidecar(t *testing.T) { |
| 793 | routes := []registry.Route{sidecarRoute("bot.x.sidecar.internal", "bdm", "eleven")} |
| 794 | moved := [][2]string{{"bdm", "eleven"}} |
| 795 | var mu sync.Mutex |
| 796 | var bounced, woken []string |
| 797 | bounce := func(h string) bool { mu.Lock(); defer mu.Unlock(); bounced = append(bounced, h); return true } |
| 798 | wake := func(rt registry.Route) error { |
| 799 | mu.Lock() |
| 800 | defer mu.Unlock() |
| 801 | woken = append(woken, rt.Host) |
| 802 | return nil |
| 803 | } |
| 804 | n := cycleWorkers(routes, moved, bounce, wake) |
| 805 | mu.Lock() |
| 806 | bouncedSnap := slices.Clone(bounced) |
| 807 | mu.Unlock() |
| 808 | if n != 1 || !slices.Equal(bouncedSnap, []string{"bot.x.sidecar.internal"}) { |
| 809 | t.Fatalf("bounced %v (n=%d), want exactly the sidecar", bouncedSnap, n) |
| 810 | } |
| 811 | // The wake is dispatched OFF cycleWorkers' return path (fix round 2) — it |
| 812 | // may not have landed yet the instant the call above returns, so this |
| 813 | // waits for it rather than asserting synchronously. |
| 814 | waitForCond(t, "the bounced sidecar to be re-woken", func() bool { |
| 815 | mu.Lock() |
| 816 | defer mu.Unlock() |
| 817 | return slices.Equal(woken, []string{"bot.x.sidecar.internal"}) |
| 818 | }) |
| 819 | } |
| 820 | |
| 821 | // TestCycleWorkersReturnsBeforeASlowWakeCompletes is fix round 2's own |
| 822 | // finding, pinned directly: WakeWorker can block up to 30s per sidecar |
| 823 | // (the settle window), and cycleWorkers' only caller (restartRequestLeg) |
| 824 | // runs synchronously on the box's single per-tick convergence goroutine — |
| 825 | // so cycleWorkers itself must never wait on a wake, or one slow sidecar |
| 826 | // delays the root-restart marker touch and every other app's turn on this |
| 827 | // box's tick. Asserted against a deliberately slow fake wake, the same |
| 828 | // elapsed-time shape Task 8 used for the worker boot pass's own |
| 829 | // bounded-concurrency claim. |
| 830 | func TestCycleWorkersReturnsBeforeASlowWakeCompletes(t *testing.T) { |
| 831 | routes := []registry.Route{sidecarRoute("bot.x.sidecar.internal", "bdm", "eleven")} |
| 832 | moved := [][2]string{{"bdm", "eleven"}} |
| 833 | slow := make(chan struct{}) |
| 834 | bounce := func(h string) bool { return true } |
| 835 | wake := func(rt registry.Route) error { <-slow; return nil } |
| 836 | |
| 837 | // Run under a deadline rather than measuring elapsed time on the test's |
| 838 | // own goroutine (fix round 3): under a REGRESSION cycleWorkers itself |
| 839 | // never returns, so a synchronous call plus an elapsed-time check never |
| 840 | // fails at all — it hangs until Go's 10-minute default test timeout, |
| 841 | // which is exactly what forced fix round 2's mutation check to run |
| 842 | // under an external `timeout` to see anything fail. Wrapping the call |
| 843 | // in its own goroutine and racing it against time.After (the same |
| 844 | // shape TestDispatchSidecarWakesBoundsConcurrency already uses) makes a |
| 845 | // regression here fail cleanly and promptly, in this test alone. |
| 846 | var n int |
| 847 | done := make(chan struct{}) |
| 848 | go func() { |
| 849 | n = cycleWorkers(routes, moved, bounce, wake) |
| 850 | close(done) |
| 851 | }() |
| 852 | select { |
| 853 | case <-done: |
| 854 | close(slow) // release the parked wake goroutine so the test can exit cleanly |
| 855 | case <-time.After(time.Second): |
| 856 | close(slow) |
| 857 | t.Fatal("cycleWorkers did not return within 1s with a wake blocked indefinitely — the wake must be dispatched asynchronously, not awaited") |
| 858 | } |
| 859 | |
| 860 | if n != 1 { |
| 861 | t.Fatalf("cycleWorkers = %d, want 1 (the bounce itself is synchronous)", n) |
| 862 | } |
| 863 | } |
| 864 | |
| 865 | // TestCycleWorkersWakesEveryBouncedSidecarEventually: dispatching wakes |
| 866 | // asynchronously must not mean dropping any of them — every sidecar |
| 867 | // cycleWorkers bounces still gets woken, just not before this function |
| 868 | // returns. |
| 869 | func TestCycleWorkersWakesEveryBouncedSidecarEventually(t *testing.T) { |
| 870 | routes := []registry.Route{ |
| 871 | sidecarRoute("bot.a.sidecar.internal", "bdm", "eleven"), |
| 872 | sidecarRoute("bot.b.sidecar.internal", "bdm", "eleven"), |
| 873 | sidecarRoute("bot.c.sidecar.internal", "bdm", "eleven"), |
| 874 | } |
| 875 | moved := [][2]string{{"bdm", "eleven"}} |
| 876 | var mu sync.Mutex |
| 877 | woken := map[string]bool{} |
| 878 | bounce := func(h string) bool { return true } |
| 879 | wake := func(rt registry.Route) error { |
| 880 | mu.Lock() |
| 881 | defer mu.Unlock() |
| 882 | woken[rt.Host] = true |
| 883 | return nil |
| 884 | } |
| 885 | if n := cycleWorkers(routes, moved, bounce, wake); n != 3 { |
| 886 | t.Fatalf("cycleWorkers = %d, want 3", n) |
| 887 | } |
| 888 | waitForCond(t, "all three sidecars to be re-woken", func() bool { |
| 889 | mu.Lock() |
| 890 | defer mu.Unlock() |
| 891 | return len(woken) == 3 |
| 892 | }) |
| 893 | } |
| 894 | |
| 895 | // TestCycleWorkersFailingWakeDoesNotStopTheOthers: a wake error is logged |
| 896 | // (not asserted here — bounceExecRoutes' own analogous case doesn't pin the |
| 897 | // log line either) and must never prevent another sidecar's wake from |
| 898 | // running, nor cycleWorkers' own return. |
| 899 | func TestCycleWorkersFailingWakeDoesNotStopTheOthers(t *testing.T) { |
| 900 | routes := []registry.Route{ |
| 901 | sidecarRoute("bot.bad.sidecar.internal", "bdm", "eleven"), |
| 902 | sidecarRoute("bot.good.sidecar.internal", "bdm", "eleven"), |
| 903 | } |
| 904 | moved := [][2]string{{"bdm", "eleven"}} |
| 905 | var mu sync.Mutex |
| 906 | woken := map[string]bool{} |
| 907 | bounce := func(h string) bool { return true } |
| 908 | wake := func(rt registry.Route) error { |
| 909 | if rt.Host == "bot.bad.sidecar.internal" { |
| 910 | return errors.New("boom") |
| 911 | } |
| 912 | mu.Lock() |
| 913 | defer mu.Unlock() |
| 914 | woken[rt.Host] = true |
| 915 | return nil |
| 916 | } |
| 917 | if n := cycleWorkers(routes, moved, bounce, wake); n != 2 { |
| 918 | t.Fatalf("cycleWorkers = %d, want 2 — both were bounced regardless of the wake outcome", n) |
| 919 | } |
| 920 | waitForCond(t, "the good sidecar to be re-woken despite the other's wake failing", func() bool { |
| 921 | mu.Lock() |
| 922 | defer mu.Unlock() |
| 923 | return woken["bot.good.sidecar.internal"] |
| 924 | }) |
| 925 | } |
| 926 | |
| 927 | // TestCycleWorkersLeavesASleepingSidecarAsleep: Bounce is a no-op on |
| 928 | // anything not awake, and cycleWorkers must never wake a sleeper just |
| 929 | // because a restart was requested — the same "restarting a process that is |
| 930 | // not running is meaningless" rule restartPass applies to a hibernating |
| 931 | // instance. |
| 932 | func TestCycleWorkersLeavesASleepingSidecarAsleep(t *testing.T) { |
| 933 | routes := []registry.Route{sidecarRoute("bot.x.sidecar.internal", "bdm", "eleven")} |
| 934 | moved := [][2]string{{"bdm", "eleven"}} |
| 935 | var woken []string |
| 936 | bounce := func(h string) bool { return false } // asleep: Bounce's own contract |
| 937 | wake := func(rt registry.Route) error { woken = append(woken, rt.Host); return nil } |
| 938 | if n := cycleWorkers(routes, moved, bounce, wake); n != 0 { |
| 939 | t.Fatalf("cycleWorkers = %d, want 0 — nothing was running", n) |
| 940 | } |
| 941 | if len(woken) != 0 { |
| 942 | t.Fatalf("woken %v, want none — a sleeping sidecar must stay asleep", woken) |
| 943 | } |
| 944 | } |
| 945 | |
| 946 | // TestCycleWorkersOnlyTouchesMovedAppsAndSidecarRoutes: the same |
| 947 | // app-scoping bounceExecRoutes gives an instance, and never an ordinary |
| 948 | // instance route even for a moved app (that is bounceExecRoutes' job, not |
| 949 | // this one's). |
| 950 | func TestCycleWorkersOnlyTouchesMovedAppsAndSidecarRoutes(t *testing.T) { |
| 951 | routes := []registry.Route{ |
| 952 | sidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven"), |
| 953 | sidecarRoute("bot.titogo.sidecar.internal", "bdm", "titogo"), // different app |
| 954 | execRoute("inst.example.com", "bdm", "eleven"), // not a sidecar |
| 955 | } |
| 956 | moved := [][2]string{{"bdm", "eleven"}} |
| 957 | var bounced []string |
| 958 | bounce := func(h string) bool { bounced = append(bounced, h); return true } |
| 959 | wake := func(rt registry.Route) error { return nil } |
| 960 | if n := cycleWorkers(routes, moved, bounce, wake); n != 1 || !slices.Equal(bounced, []string{"bot.eleven.sidecar.internal"}) { |
| 961 | t.Fatalf("bounced %v (n=%d), want only the moved app's sidecar", bounced, n) |
| 962 | } |
| 963 | } |
| 964 | |
| 965 | // TestCycleWorkersWithoutAnActivatorOrWakeIsANoop: a box with no activator |
| 966 | // armed gets nil for both callbacks — a working no-op, not a panic. |
| 967 | func TestCycleWorkersWithoutAnActivatorOrWakeIsANoop(t *testing.T) { |
| 968 | routes := []registry.Route{sidecarRoute("bot.x.sidecar.internal", "bdm", "eleven")} |
| 969 | if n := cycleWorkers(routes, [][2]string{{"bdm", "eleven"}}, nil, nil); n != 0 { |
| 970 | t.Fatalf("cycleWorkers = %d with no activator armed, want 0 and no panic", n) |
| 971 | } |
| 972 | } |
| 973 | |
| 974 | // TestCycleWorkersReturnsPromptlyWithManySlowWakesQueued is review round |
| 975 | // 2's own regression pin: a NARROW earlier version of the fan-out |
| 976 | // (fix round 1) acquired the workerBootConcurrency semaphore in cycleWorkers' |
| 977 | // OWN loop, one route at a time — so once workerBootConcurrency (8) wakes |
| 978 | // were already in flight, the (workerBootConcurrency+1)th route's |
| 979 | // acquisition blocked cycleWorkers itself, and therefore restartRequestLeg |
| 980 | // and fastTick's single per-box convergence goroutine, for up to |
| 981 | // wakeWaitBound (30s) per sidecar past the 8th — a narrow reintroduction of |
| 982 | // exactly the bug TestCycleWorkersReturnsBeforeASlowWakeCompletes already |
| 983 | // pins for a SINGLE sidecar. This drives strictly more than |
| 984 | // workerBootConcurrency sidecars, every wake deliberately parked, and |
| 985 | // asserts cycleWorkers still returns immediately: the shape an app like |
| 986 | // Eleven's (a sidecar per enrolled member) hits routinely, not on rare |
| 987 | // unlucky timing. |
| 988 | func TestCycleWorkersReturnsPromptlyWithManySlowWakesQueued(t *testing.T) { |
| 989 | n := workerBootConcurrency + 5 // strictly more than the bound |
| 990 | routes := make([]registry.Route, n) |
| 991 | for i := range routes { |
| 992 | routes[i] = sidecarRoute(fmt.Sprintf("s%d.x.sidecar.internal", i), "bdm", "eleven") |
| 993 | } |
| 994 | moved := [][2]string{{"bdm", "eleven"}} |
| 995 | bounce := func(h string) bool { return true } |
| 996 | slow := make(chan struct{}) |
| 997 | var mu sync.Mutex |
| 998 | woken := map[string]bool{} |
| 999 | wake := func(rt registry.Route) error { |
| 1000 | <-slow |
| 1001 | mu.Lock() |
| 1002 | woken[rt.Host] = true |
| 1003 | mu.Unlock() |
| 1004 | return nil |
| 1005 | } |
| 1006 | |
| 1007 | // Same deadline-vs-elapsed-time correction as |
| 1008 | // TestCycleWorkersReturnsBeforeASlowWakeCompletes (fix round 3): a |
| 1009 | // synchronous call on the test's own goroutine never fails under a |
| 1010 | // regression, it hangs — this is the test fix round 2's own mutation |
| 1011 | // check needed an external `timeout` to see fail at all. |
| 1012 | var got int |
| 1013 | done := make(chan struct{}) |
| 1014 | go func() { |
| 1015 | got = cycleWorkers(routes, moved, bounce, wake) |
| 1016 | close(done) |
| 1017 | }() |
| 1018 | select { |
| 1019 | case <-done: |
| 1020 | case <-time.After(time.Second): |
| 1021 | close(slow) |
| 1022 | t.Fatalf("cycleWorkers did not return within 1s with %d wakes blocked (bound %d) — the whole bounded fan-out must run off the caller's goroutine, not just each individual wake", n, workerBootConcurrency) |
| 1023 | } |
| 1024 | if got != n { |
| 1025 | t.Fatalf("cycleWorkers = %d, want %d (every route bounced)", got, n) |
| 1026 | } |
| 1027 | |
| 1028 | close(slow) |
| 1029 | waitForCond(t, "every queued sidecar to be woken eventually", func() bool { |
| 1030 | mu.Lock() |
| 1031 | defer mu.Unlock() |
| 1032 | return len(woken) == n |
| 1033 | }) |
| 1034 | } |
| 1035 | |
amadan