| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "slices" |
| 8 | "sync" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/registry" |
| 13 | "github.com/carlosframework/platform/internal/release" |
| 14 | ) |
| 15 | |
| 16 | // wakeSidecarRoute is sidecarRoute (restart_test.go) with Slug set — the |
| 17 | // field a sidecar route carries its short name on (registry.Route.Slug's |
| 18 | // doc comment; provision's reconciler stamps it from the sidecar record). |
| 19 | // sidecarWakeChanges keys entirely off it, so a fixture that omits it is |
| 20 | // useless here even though restart_test.go's sidecarRoute never needed it. |
| 21 | func wakeSidecarRoute(host, acct, app, slug string) registry.Route { |
| 22 | rt := sidecarRoute(host, acct, app) |
| 23 | rt.Slug = slug |
| 24 | return rt |
| 25 | } |
| 26 | |
| 27 | // touchSidecarWake bumps name's wake generation through the real writer |
| 28 | // (release.TouchSidecarWake) — the same door the console's wake handler |
| 29 | // (Task 12) will call, so these tests exercise the actual signal a box |
| 30 | // sees rather than a hand-rolled stand-in for it. |
| 31 | func touchSidecarWake(t *testing.T, ctx context.Context, s release.Store, acct, app, name string) { |
| 32 | t.Helper() |
| 33 | if err := release.TouchSidecarWake(ctx, s, acct, app, name, "member:bdm", "test wake"); err != nil { |
| 34 | t.Fatalf("TouchSidecarWake %s/%s/%s: %v", acct, app, name, err) |
| 35 | } |
| 36 | } |
| 37 | |
| 38 | // ---- sidecarWakeChanges: the synchronous etag comparison ------------------ |
| 39 | |
| 40 | func TestSidecarWakeChangesEtagMoveReportsExactlyThatSidecar(t *testing.T) { |
| 41 | ctx, s := genKit(t) |
| 42 | routes := []registry.Route{ |
| 43 | wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot"), |
| 44 | wakeSidecarRoute("other.eleven.sidecar.internal", "bdm", "eleven", "other"), |
| 45 | } |
| 46 | mem := newSidecarWakeMemory(false) |
| 47 | sidecarWakeChanges(ctx, routes, s, mem) // seed: first sight of both, silently |
| 48 | |
| 49 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 50 | got, errs := sidecarWakeChanges(ctx, routes, s, mem) |
| 51 | if len(errs) != 0 { |
| 52 | t.Fatalf("unexpected read errors: %v", errs) |
| 53 | } |
| 54 | if len(got) != 1 || got[0].Host != "bot.eleven.sidecar.internal" { |
| 55 | t.Fatalf("sidecarWakeChanges = %v, want exactly the moved sidecar's route", got) |
| 56 | } |
| 57 | } |
| 58 | |
| 59 | func TestSidecarWakeChangesUnchangedEtagReportsNothing(t *testing.T) { |
| 60 | ctx, s := genKit(t) |
| 61 | routes := []registry.Route{wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot")} |
| 62 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 63 | mem := newSidecarWakeMemory(false) |
| 64 | sidecarWakeChanges(ctx, routes, s, mem) // seed |
| 65 | |
| 66 | got, errs := sidecarWakeChanges(ctx, routes, s, mem) |
| 67 | if len(errs) != 0 { |
| 68 | t.Fatalf("unexpected read errors: %v", errs) |
| 69 | } |
| 70 | if len(got) != 0 { |
| 71 | t.Fatalf("unchanged etag reported %v, want none", got) |
| 72 | } |
| 73 | } |
| 74 | |
| 75 | // TestSidecarWakeChangesBaselinePassDoesNotReplay is the no-replay rule at |
| 76 | // the point it applies: the BASELINE pass — arming, i.e. an agent restart — |
| 77 | // must not treat "I have never looked at this before" as movement, or every |
| 78 | // restart would replay every wake this sidecar has ever received, a |
| 79 | // thundering herd on boot. (First sight AFTER the baseline is a different |
| 80 | // fact entirely — a route minted under a running agent — and is woken; see |
| 81 | // TestSidecarWakeChangesRouteMintedAfterTheBaselineIsWoken.) |
| 82 | func TestSidecarWakeChangesBaselinePassDoesNotReplay(t *testing.T) { |
| 83 | ctx, s := genKit(t) |
| 84 | // Wake object already exists BEFORE this box ever looks — the shape a |
| 85 | // long-lived wake request or a box that just (re)started sees. |
| 86 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 87 | routes := []registry.Route{wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot")} |
| 88 | mem := newSidecarWakeMemory(false) |
| 89 | |
| 90 | got, errs := sidecarWakeChanges(ctx, routes, s, mem) |
| 91 | if len(errs) != 0 { |
| 92 | t.Fatalf("unexpected read errors: %v", errs) |
| 93 | } |
| 94 | if len(got) != 0 { |
| 95 | t.Fatalf("first sight reported %v — a box that just started must not replay every past wake", got) |
| 96 | } |
| 97 | want := sidecarWakeStoreEtag(t, ctx, s, "bdm", "eleven", "bot") |
| 98 | if got := mem.etags[[3]string{"bdm", "eleven", "bot"}]; got != want { |
| 99 | t.Fatalf("cache etag = %q, want the store's %q", got, want) |
| 100 | } |
| 101 | |
| 102 | // A SECOND wake request after the seed IS movement. |
| 103 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 104 | got, errs = sidecarWakeChanges(ctx, routes, s, mem) |
| 105 | if len(errs) != 0 { |
| 106 | t.Fatalf("unexpected read errors: %v", errs) |
| 107 | } |
| 108 | if len(got) != 1 || got[0].Host != "bot.eleven.sidecar.internal" { |
| 109 | t.Fatalf("second request reported %v, want exactly the sidecar", got) |
| 110 | } |
| 111 | } |
| 112 | |
| 113 | // TestSidecarWakeChangesRouteMintedAfterTheBaselineIsWoken is the |
| 114 | // create-then-wake flow, which is the feature's PRIMARY one: an operator |
| 115 | // wake is a hibernating sidecar's only first-run mechanism, and |
| 116 | // `carlos sidecars create` + `carlos sidecars wake` routinely land in the |
| 117 | // same tick — the tick that mints the route. Seeding that route's etag |
| 118 | // silently (the baseline rule, over-applied) loses the wake outright while |
| 119 | // the CLI reports success: a bot that never runs. |
| 120 | func TestSidecarWakeChangesRouteMintedAfterTheBaselineIsWoken(t *testing.T) { |
| 121 | ctx, s := genKit(t) |
| 122 | existing := wakeSidecarRoute("old.eleven.sidecar.internal", "bdm", "eleven", "old") |
| 123 | mem := newSidecarWakeMemory(false) |
| 124 | sidecarWakeChanges(ctx, []registry.Route{existing}, s, mem) // arming: this box's baseline |
| 125 | |
| 126 | // The console creates a sidecar and wakes it; this box's next tick mints |
| 127 | // the route AND sees the wake object for the first time, together. |
| 128 | minted := wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot") |
| 129 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 130 | |
| 131 | got, errs := sidecarWakeChanges(ctx, []registry.Route{existing, minted}, s, mem) |
| 132 | if len(errs) != 0 { |
| 133 | t.Fatalf("unexpected read errors: %v", errs) |
| 134 | } |
| 135 | if len(got) != 1 || got[0].Host != "bot.eleven.sidecar.internal" { |
| 136 | t.Fatalf("reported %v, want the just-minted sidecar — a wake standing at a route this pass minted is a wake somebody already asked for", got) |
| 137 | } |
| 138 | } |
| 139 | |
| 140 | // A route minted after the baseline with NO wake object is silent: only a |
| 141 | // PRESENT object makes a first sight a wake, so ordinary provisioning never |
| 142 | // wakes a hibernating sidecar nobody asked to run. |
| 143 | func TestSidecarWakeChangesRouteMintedWithNoWakeObjectIsSilent(t *testing.T) { |
| 144 | ctx, s := genKit(t) |
| 145 | mem := newSidecarWakeMemory(false) |
| 146 | sidecarWakeChanges(ctx, nil, s, mem) // arming on a box with no sidecars |
| 147 | |
| 148 | minted := wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot") |
| 149 | got, errs := sidecarWakeChanges(ctx, []registry.Route{minted}, s, mem) |
| 150 | if len(errs) != 0 { |
| 151 | t.Fatalf("unexpected read errors: %v", errs) |
| 152 | } |
| 153 | if len(got) != 0 { |
| 154 | t.Fatalf("reported %v — a newly minted sidecar nobody has ever asked to wake must stay asleep", got) |
| 155 | } |
| 156 | |
| 157 | // …and it behaves as an ordinary member from then on: the first real |
| 158 | // request moves the etag and wakes it. |
| 159 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 160 | got, _ = sidecarWakeChanges(ctx, []registry.Route{minted}, s, mem) |
| 161 | if len(got) != 1 || got[0].Host != "bot.eleven.sidecar.internal" { |
| 162 | t.Fatalf("reported %v, want the sidecar on its first real wake request", got) |
| 163 | } |
| 164 | } |
| 165 | |
| 166 | // An arming pass whose route LIST failed must not leave the box thinking |
| 167 | // every sidecar it serves is brand new: the memory stays unseeded, so the |
| 168 | // first leg pass establishes the baseline itself instead of replaying every |
| 169 | // historical wake at once. |
| 170 | func TestSidecarWakeSeedListFailureLeavesTheBaselineToTheFirstPass(t *testing.T) { |
| 171 | ctx, s := genKit(t) |
| 172 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") // history, from before this agent |
| 173 | routes := []registry.Route{wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot")} |
| 174 | mem := newSidecarWakeMemory(false) // arming never completed: no List, nothing seeded |
| 175 | |
| 176 | got, errs := sidecarWakeChanges(ctx, routes, s, mem) |
| 177 | if len(errs) != 0 { |
| 178 | t.Fatalf("unexpected read errors: %v", errs) |
| 179 | } |
| 180 | if len(got) != 0 { |
| 181 | t.Fatalf("reported %v — with no baseline established, the first pass IS the baseline, not a replay of every past wake", got) |
| 182 | } |
| 183 | } |
| 184 | |
| 185 | func sidecarWakeStoreEtag(t *testing.T, ctx context.Context, s release.Store, acct, app, name string) string { |
| 186 | t.Helper() |
| 187 | _, etag, err := s.Get(ctx, release.SidecarWakeKey(acct, app, name)) |
| 188 | if err != nil { |
| 189 | t.Fatalf("read sidecar wake: %v", err) |
| 190 | } |
| 191 | return etag |
| 192 | } |
| 193 | |
| 194 | // TestSidecarWakeChangesReadErrorDoesNotSuppressOthers: a transient read |
| 195 | // failure against one sidecar's wake object must not blind the pass to a |
| 196 | // real move on a different one in the same tick. |
| 197 | func TestSidecarWakeChangesReadErrorDoesNotSuppressOthers(t *testing.T) { |
| 198 | ctx, s0 := genKit(t) |
| 199 | routes := []registry.Route{ |
| 200 | wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot"), |
| 201 | wakeSidecarRoute("worker.titogo.sidecar.internal", "bdm", "titogo", "worker"), |
| 202 | } |
| 203 | mem := newSidecarWakeMemory(false) |
| 204 | sidecarWakeChanges(ctx, routes, s0, mem) // seed both, absent |
| 205 | |
| 206 | touchSidecarWake(t, ctx, s0, "bdm", "eleven", "bot") |
| 207 | touchSidecarWake(t, ctx, s0, "bdm", "titogo", "worker") |
| 208 | failing := &failingKeyStore{Store: s0, failKey: release.SidecarWakeKey("bdm", "eleven", "bot"), err: errors.New("bucket blip")} |
| 209 | |
| 210 | got, errs := sidecarWakeChanges(ctx, routes, failing, mem) |
| 211 | if len(errs) != 1 { |
| 212 | t.Fatalf("errs = %v, want exactly one (the failing key)", errs) |
| 213 | } |
| 214 | if len(got) != 1 || got[0].Host != "worker.titogo.sidecar.internal" { |
| 215 | t.Fatalf("reported %v, want only the unaffected sidecar moved", got) |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | // TestSidecarWakeChangesIgnoresUnservedSidecar: a wake object naming a |
| 220 | // sidecar this box does not serve is never even read, let alone reported — |
| 221 | // the pass only ever asks about routes this box actually has. |
| 222 | func TestSidecarWakeChangesIgnoresUnservedSidecar(t *testing.T) { |
| 223 | ctx, s := genKit(t) |
| 224 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "ghost") // no route for "ghost" at all |
| 225 | routes := []registry.Route{wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot")} |
| 226 | mem := newSidecarWakeMemory(false) |
| 227 | sidecarWakeChanges(ctx, routes, s, mem) // seed |
| 228 | |
| 229 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "ghost") // "ghost" moves again |
| 230 | got, errs := sidecarWakeChanges(ctx, routes, s, mem) |
| 231 | if len(errs) != 0 { |
| 232 | t.Fatalf("unexpected read errors: %v", errs) |
| 233 | } |
| 234 | if len(got) != 0 { |
| 235 | t.Fatalf("an unserved sidecar's wake object was reported: %v, want none", got) |
| 236 | } |
| 237 | if _, ok := mem.etags[[3]string{"bdm", "eleven", "ghost"}]; ok { |
| 238 | t.Fatal("a sidecar this box does not serve must never appear in its cache") |
| 239 | } |
| 240 | } |
| 241 | |
| 242 | // TestSidecarWakeChangesIgnoresNonSidecarRouteWithMatchingSlug: Slug is not |
| 243 | // a sidecar-only field (registry.Route.Slug's doc comment — it is the |
| 244 | // app-account short name usable on a root domain, stamped on ordinary |
| 245 | // routes too), so an ordinary instance route that happens to carry the |
| 246 | // same Slug as a real sidecar name must never be mistaken for that |
| 247 | // sidecar. |
| 248 | func TestSidecarWakeChangesIgnoresNonSidecarRouteWithMatchingSlug(t *testing.T) { |
| 249 | ctx, s := genKit(t) |
| 250 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 251 | notSidecar := execRoute("app.example.com", "bdm", "eleven") |
| 252 | notSidecar.Slug = "bot" // same name a sidecar could carry, wrong Kind |
| 253 | routes := []registry.Route{notSidecar} |
| 254 | mem := newSidecarWakeMemory(false) |
| 255 | |
| 256 | got, errs := sidecarWakeChanges(ctx, routes, s, mem) |
| 257 | if len(errs) != 0 { |
| 258 | t.Fatalf("unexpected read errors: %v", errs) |
| 259 | } |
| 260 | if len(got) != 0 { |
| 261 | t.Fatalf("a non-sidecar route with a matching Slug was reported: %v, want none", got) |
| 262 | } |
| 263 | if len(mem.etags) != 0 { |
| 264 | t.Fatalf("cache = %v, want empty — a non-sidecar route must never seed a sidecar cache entry", mem.etags) |
| 265 | } |
| 266 | } |
| 267 | |
| 268 | // ---- dispatchSidecarWakes: the bounded-concurrency, non-blocking half ---- |
| 269 | |
| 270 | // TestDispatchSidecarWakesReturnsBeforeASlowWakeCompletes pins fix round 1's |
| 271 | // finding directly: WakeWorker can block up to 30s per sidecar (the settle |
| 272 | // window), and dispatchSidecarWakes' only caller (sidecarWakeLeg) runs |
| 273 | // synchronously on the box's single per-tick convergence goroutine — so |
| 274 | // this function must never wait on a wake, or one slow sidecar delays every |
| 275 | // other app's turn on that tick. Asserted against a deliberately slow fake |
| 276 | // wake, the same elapsed-time shape Task 8's boot pass and Task 11's |
| 277 | // TestCycleWorkersReturnsBeforeASlowWakeCompletes both use. |
| 278 | func TestDispatchSidecarWakesReturnsBeforeASlowWakeCompletes(t *testing.T) { |
| 279 | moved := []registry.Route{wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot")} |
| 280 | slow := make(chan struct{}) |
| 281 | wake := func(rt registry.Route) error { <-slow; return nil } |
| 282 | |
| 283 | // Run under a deadline rather than measuring elapsed time on the test's |
| 284 | // own goroutine (fix round 3): under a REGRESSION dispatchSidecarWakes |
| 285 | // itself never returns, so a synchronous call plus an elapsed-time |
| 286 | // check never fails — it hangs until Go's 10-minute default test |
| 287 | // timeout. Racing a goroutine against time.After (the same shape |
| 288 | // TestDispatchSidecarWakesBoundsConcurrency already uses) makes a |
| 289 | // regression here fail cleanly and promptly, in this test alone. |
| 290 | done := make(chan struct{}) |
| 291 | go func() { |
| 292 | dispatchSidecarWakes(moved, wake) |
| 293 | close(done) |
| 294 | }() |
| 295 | select { |
| 296 | case <-done: |
| 297 | close(slow) // release the parked wake goroutine so the test can exit cleanly |
| 298 | case <-time.After(time.Second): |
| 299 | close(slow) |
| 300 | t.Fatal("dispatchSidecarWakes did not return within 1s with a wake blocked indefinitely — wakes must be dispatched asynchronously, not awaited") |
| 301 | } |
| 302 | } |
| 303 | |
| 304 | // TestDispatchSidecarWakesWakesEveryMovedSidecarEventually: dispatching |
| 305 | // asynchronously must not mean dropping any of them — every moved sidecar |
| 306 | // still gets woken, just not before this function returns. |
| 307 | func TestDispatchSidecarWakesWakesEveryMovedSidecarEventually(t *testing.T) { |
| 308 | moved := []registry.Route{ |
| 309 | wakeSidecarRoute("bot.a.sidecar.internal", "bdm", "eleven", "a"), |
| 310 | wakeSidecarRoute("bot.b.sidecar.internal", "bdm", "eleven", "b"), |
| 311 | } |
| 312 | var mu sync.Mutex |
| 313 | var woken []string |
| 314 | wake := func(rt registry.Route) error { |
| 315 | mu.Lock() |
| 316 | woken = append(woken, rt.Host) |
| 317 | mu.Unlock() |
| 318 | return nil |
| 319 | } |
| 320 | dispatchSidecarWakes(moved, wake) |
| 321 | waitForCond(t, "both sidecars to be woken", func() bool { |
| 322 | mu.Lock() |
| 323 | defer mu.Unlock() |
| 324 | slices.Sort(woken) |
| 325 | return slices.Equal(woken, []string{"bot.a.sidecar.internal", "bot.b.sidecar.internal"}) |
| 326 | }) |
| 327 | } |
| 328 | |
| 329 | // TestDispatchSidecarWakesBoundsConcurrency pins the carried requirement |
| 330 | // that wakes must not fan out unbounded: N sidecars strictly more than |
| 331 | // workerBootConcurrency drive more in-flight wakes than the bound and prove |
| 332 | // it is actually reached (not stuck at 1, which a serial dispatch would |
| 333 | // show) and never exceeded (which an unbounded one would show) — the same |
| 334 | // shape TestWorkerBootPassWakesConcurrently pins for Task 8's boot pass. |
| 335 | func TestDispatchSidecarWakesBoundsConcurrency(t *testing.T) { |
| 336 | t.Parallel() |
| 337 | n := workerBootConcurrency + 3 |
| 338 | moved := make([]registry.Route, n) |
| 339 | for i := range moved { |
| 340 | moved[i] = wakeSidecarRoute("s"+string(rune('a'+i))+".x.sidecar.internal", "bdm", "eleven", "s"+string(rune('a'+i))) |
| 341 | } |
| 342 | var mu sync.Mutex |
| 343 | inFlight, maxInFlight := 0, 0 |
| 344 | release := make(chan struct{}) |
| 345 | wake := func(registry.Route) error { |
| 346 | mu.Lock() |
| 347 | inFlight++ |
| 348 | if inFlight > maxInFlight { |
| 349 | maxInFlight = inFlight |
| 350 | } |
| 351 | mu.Unlock() |
| 352 | <-release |
| 353 | mu.Lock() |
| 354 | inFlight-- |
| 355 | mu.Unlock() |
| 356 | return nil |
| 357 | } |
| 358 | |
| 359 | // dispatchSidecarWakes' own dispatch loop blocks on sem once the bound |
| 360 | // saturates (it is a bounded fan-out, not a fire-and-forget launcher) — |
| 361 | // exactly runWorkerBootPass's own shape, and its test |
| 362 | // (TestWorkerBootPassWakesConcurrently) runs the call in a goroutine for |
| 363 | // the same reason: the test needs to observe the saturated state and |
| 364 | // then release it, which it cannot do if the call under test is itself |
| 365 | // the thing blocking the test's only goroutine. |
| 366 | done := make(chan struct{}) |
| 367 | go func() { |
| 368 | dispatchSidecarWakes(moved, wake) |
| 369 | close(done) |
| 370 | }() |
| 371 | |
| 372 | waitForCond(t, "wakes to saturate the concurrency bound", func() bool { |
| 373 | mu.Lock() |
| 374 | defer mu.Unlock() |
| 375 | return inFlight == workerBootConcurrency |
| 376 | }) |
| 377 | // A beat with the bound already saturated: an unbounded implementation |
| 378 | // would keep climbing past it here rather than holding still. |
| 379 | time.Sleep(50 * time.Millisecond) |
| 380 | mu.Lock() |
| 381 | got := maxInFlight |
| 382 | mu.Unlock() |
| 383 | if got != workerBootConcurrency { |
| 384 | t.Fatalf("max concurrent wakes = %d, want exactly the bound %d", got, workerBootConcurrency) |
| 385 | } |
| 386 | |
| 387 | close(release) |
| 388 | select { |
| 389 | case <-done: |
| 390 | case <-time.After(5 * time.Second): |
| 391 | t.Fatal("dispatch loop did not finish after its wakes were released") |
| 392 | } |
| 393 | } |
| 394 | |
| 395 | // TestDispatchSidecarWakesWithoutAnActivatorIsANoop: a box with no |
| 396 | // activator armed gets nil for wake — a working no-op, not a panic. |
| 397 | func TestDispatchSidecarWakesWithoutAnActivatorIsANoop(t *testing.T) { |
| 398 | moved := []registry.Route{wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot")} |
| 399 | dispatchSidecarWakes(moved, nil) // must not panic |
| 400 | } |
| 401 | |
| 402 | // ---- sidecarWakeLeg: the wired combination fastConvergeLoop calls -------- |
| 403 | |
| 404 | // TestSidecarWakeLegActsOnBumpedGeneration is the "wired into the |
| 405 | // convergence path" pin: a real TouchSidecarWake write, run through the |
| 406 | // seed-then-leg shape fastConvergeLoop actually uses (sidecarWakeSeed at |
| 407 | // arming, sidecarWakeLeg on every serial move), must reach WakeWorker for |
| 408 | // exactly the sidecar the console asked for. |
| 409 | func TestSidecarWakeLegActsOnBumpedGeneration(t *testing.T) { |
| 410 | t.Parallel() |
| 411 | ctx, s := genKit(t) |
| 412 | routes := []registry.Route{ |
| 413 | wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot"), |
| 414 | wakeSidecarRoute("other.eleven.sidecar.internal", "bdm", "eleven", "other"), |
| 415 | } |
| 416 | reg := restartTestRegistry(t) |
| 417 | for _, rt := range routes { |
| 418 | if err := reg.Upsert(rt); err != nil { |
| 419 | t.Fatalf("upsert %s: %v", rt.Host, err) |
| 420 | } |
| 421 | } |
| 422 | mem := newSidecarWakeMemory(false) |
| 423 | sidecarWakeSeed(ctx, reg, s, mem) |
| 424 | |
| 425 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 426 | var mu sync.Mutex |
| 427 | var woken []string |
| 428 | wake := func(rt registry.Route) error { |
| 429 | mu.Lock() |
| 430 | woken = append(woken, rt.Host) |
| 431 | mu.Unlock() |
| 432 | return nil |
| 433 | } |
| 434 | moved, errs := sidecarWakeLeg(ctx, routes, s, mem, wake) |
| 435 | if len(errs) != 0 { |
| 436 | t.Fatalf("unexpected read errors: %v", errs) |
| 437 | } |
| 438 | if len(moved) != 1 || moved[0].Host != "bot.eleven.sidecar.internal" { |
| 439 | t.Fatalf("sidecarWakeLeg reported %v, want exactly the bumped sidecar", moved) |
| 440 | } |
| 441 | waitForCond(t, "the bumped sidecar to be woken", func() bool { |
| 442 | mu.Lock() |
| 443 | defer mu.Unlock() |
| 444 | return slices.Equal(woken, []string{"bot.eleven.sidecar.internal"}) |
| 445 | }) |
| 446 | } |
| 447 | |
| 448 | // TestSidecarWakeLegCreateThenWakeInOneTickStillWakesIt is the blocker as a |
| 449 | // member would meet it, through the leg fastConvergeLoop actually calls: the |
| 450 | // agent arms with no sidecars, then one serial move carries BOTH the record |
| 451 | // (so this tick mints the route) and the wake object. The wake must land. |
| 452 | func TestSidecarWakeLegCreateThenWakeInOneTickStillWakesIt(t *testing.T) { |
| 453 | t.Parallel() |
| 454 | ctx, s := genKit(t) |
| 455 | reg := restartTestRegistry(t) |
| 456 | mem := newSidecarWakeMemory(false) |
| 457 | sidecarWakeSeed(ctx, reg, s, mem) // arming: no sidecar routes yet |
| 458 | |
| 459 | // create + wake, both before this box's next tick. |
| 460 | minted := wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot") |
| 461 | if err := reg.Upsert(minted); err != nil { |
| 462 | t.Fatalf("upsert: %v", err) |
| 463 | } |
| 464 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 465 | |
| 466 | var mu sync.Mutex |
| 467 | var woken []string |
| 468 | wake := func(rt registry.Route) error { |
| 469 | mu.Lock() |
| 470 | woken = append(woken, rt.Host) |
| 471 | mu.Unlock() |
| 472 | return nil |
| 473 | } |
| 474 | moved, errs := sidecarWakeLeg(ctx, []registry.Route{minted}, s, mem, wake) |
| 475 | if len(errs) != 0 { |
| 476 | t.Fatalf("unexpected read errors: %v", errs) |
| 477 | } |
| 478 | if len(moved) != 1 || moved[0].Host != "bot.eleven.sidecar.internal" { |
| 479 | t.Fatalf("leg reported %v, want the just-created sidecar — the CLI already told the member this wake was accepted", moved) |
| 480 | } |
| 481 | waitForCond(t, "the just-created sidecar to be woken", func() bool { |
| 482 | mu.Lock() |
| 483 | defer mu.Unlock() |
| 484 | return slices.Equal(woken, []string{"bot.eleven.sidecar.internal"}) |
| 485 | }) |
| 486 | } |
| 487 | |
| 488 | // The other side of the same coin, end to end: an agent RESTART on a box |
| 489 | // whose sidecars have been woken many times before wakes nothing at all. |
| 490 | func TestSidecarWakeLegAfterAnAgentRestartReplaysNothing(t *testing.T) { |
| 491 | t.Parallel() |
| 492 | ctx, s := genKit(t) |
| 493 | reg := restartTestRegistry(t) |
| 494 | routes := []registry.Route{ |
| 495 | wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot"), |
| 496 | wakeSidecarRoute("other.eleven.sidecar.internal", "bdm", "eleven", "other"), |
| 497 | } |
| 498 | for _, rt := range routes { |
| 499 | if err := reg.Upsert(rt); err != nil { |
| 500 | t.Fatalf("upsert %s: %v", rt.Host, err) |
| 501 | } |
| 502 | } |
| 503 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") // history |
| 504 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "other") // history |
| 505 | |
| 506 | mem := newSidecarWakeMemory(false) // a fresh process |
| 507 | sidecarWakeSeed(ctx, reg, s, mem) |
| 508 | if len(mem.etags) != 2 { |
| 509 | t.Fatalf("the arming pass recorded %d etags, want 2 — a baseline that records nothing is no baseline at all", len(mem.etags)) |
| 510 | } |
| 511 | |
| 512 | var mu sync.Mutex |
| 513 | var woken []string |
| 514 | wake := func(rt registry.Route) error { |
| 515 | mu.Lock() |
| 516 | woken = append(woken, rt.Host) |
| 517 | mu.Unlock() |
| 518 | return nil |
| 519 | } |
| 520 | moved, errs := sidecarWakeLeg(ctx, routes, s, mem, wake) |
| 521 | if len(errs) != 0 { |
| 522 | t.Fatalf("unexpected read errors: %v", errs) |
| 523 | } |
| 524 | if len(moved) != 0 { |
| 525 | t.Fatalf("an agent restart replayed %v — every wake these sidecars ever received would fire at once", moved) |
| 526 | } |
| 527 | time.Sleep(50 * time.Millisecond) // the dispatch is asynchronous; give it a chance to be wrong |
| 528 | mu.Lock() |
| 529 | defer mu.Unlock() |
| 530 | if len(woken) != 0 { |
| 531 | t.Fatalf("woke %v on an agent restart, want none", woken) |
| 532 | } |
| 533 | } |
| 534 | |
| 535 | // TestSidecarWakeLegReturnsBeforeASlowWakeCompletes: the glued leg itself |
| 536 | // must not block on WakeWorker either — the property dispatchSidecarWakes |
| 537 | // provides must survive being called from inside sidecarWakeLeg. |
| 538 | func TestSidecarWakeLegReturnsBeforeASlowWakeCompletes(t *testing.T) { |
| 539 | ctx, s := genKit(t) |
| 540 | routes := []registry.Route{wakeSidecarRoute("bot.eleven.sidecar.internal", "bdm", "eleven", "bot")} |
| 541 | mem := newSidecarWakeMemory(false) |
| 542 | sidecarWakeChanges(ctx, routes, s, mem) // seed |
| 543 | touchSidecarWake(t, ctx, s, "bdm", "eleven", "bot") |
| 544 | |
| 545 | slow := make(chan struct{}) |
| 546 | wake := func(rt registry.Route) error { <-slow; return nil } |
| 547 | |
| 548 | // Deadline-vs-elapsed-time correction, same as |
| 549 | // TestDispatchSidecarWakesReturnsBeforeASlowWakeCompletes (fix round 3): |
| 550 | // a synchronous call on the test's own goroutine never fails under a |
| 551 | // regression, it hangs. |
| 552 | var moved []registry.Route |
| 553 | done := make(chan struct{}) |
| 554 | go func() { |
| 555 | moved, _ = sidecarWakeLeg(ctx, routes, s, mem, wake) |
| 556 | close(done) |
| 557 | }() |
| 558 | select { |
| 559 | case <-done: |
| 560 | close(slow) |
| 561 | case <-time.After(time.Second): |
| 562 | close(slow) |
| 563 | t.Fatal("sidecarWakeLeg did not return within 1s with a wake blocked indefinitely — the wake must be dispatched asynchronously") |
| 564 | } |
| 565 | |
| 566 | if len(moved) != 1 { |
| 567 | t.Fatalf("moved = %v, want exactly one (detection is synchronous)", moved) |
| 568 | } |
| 569 | } |
| 570 | |
| 571 | // ---- wired into fastTick: rides the same serial move as the restart leg -- |
| 572 | |
| 573 | // TestFastTickCallsWakeLegOnSerialChange pins the wiring itself: fastTick |
| 574 | // must call fastDeps.wakeLeg exactly when it calls restartLeg — on a serial |
| 575 | // move, after fullPass — or `carlos sidecars wake` writes a generation |
| 576 | // object no box ever reads. Same shape as TestFastTickActsOnSerialChangeNotBoot, |
| 577 | // with a call-counting wakeLeg spy standing in for the production closure. |
| 578 | func TestFastTickCallsWakeLegOnSerialChange(t *testing.T) { |
| 579 | env := newFastFixture(t) |
| 580 | var calls int |
| 581 | restart := func(string) error { return nil } |
| 582 | d := &fastDeps{ |
| 583 | reg: env.reg, store: env.store, pub: env.pub, optDir: env.optDir, |
| 584 | restart: restart, |
| 585 | passOpts: fastPassOpts(env.dataDir, func(string, string, error) {}, restart, func(string) error { return nil }), |
| 586 | fullPass: func(context.Context) {}, |
| 587 | wakeLeg: func(context.Context) { calls++ }, |
| 588 | } |
| 589 | now := time.Unix(1754640000, 0) |
| 590 | fastTick(context.Background(), d, now) // no serial object at all: seeds at zero, runs nothing |
| 591 | if calls != 0 { |
| 592 | t.Fatalf("wakeLeg ran on the boot seed, want 0 calls, got %d", calls) |
| 593 | } |
| 594 | if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| 595 | t.Fatalf("BumpSerial: %v", err) |
| 596 | } |
| 597 | fastTick(context.Background(), d, now.Add(2*time.Second)) |
| 598 | if calls != 1 { |
| 599 | t.Fatalf("wakeLeg calls = %d, want exactly 1 on a serial move", calls) |
| 600 | } |
| 601 | fastTick(context.Background(), d, now.Add(10*time.Second)) // past clamp, unchanged serial |
| 602 | if calls != 1 { |
| 603 | t.Fatalf("wakeLeg calls = %d after an unchanged serial, want still 1", calls) |
| 604 | } |
| 605 | } |
| 606 | |
| 607 | // TestFastTickToleratesANilWakeLeg: a box built before this feature (or a |
| 608 | // test fixture that never sets one) must not panic — wakeLeg is optional |
| 609 | // exactly like restartLeg. |
| 610 | func TestFastTickToleratesANilWakeLeg(t *testing.T) { |
| 611 | env := newFastFixture(t) |
| 612 | d := env.deps(nil) // deps() never sets wakeLeg |
| 613 | now := time.Unix(1754640000, 0) |
| 614 | if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| 615 | t.Fatalf("BumpSerial: %v", err) |
| 616 | } |
| 617 | fastTick(context.Background(), d, now) // must not panic |
| 618 | } |
| 619 | |
| 620 | // TestDispatchSidecarWakesReturnsPromptlyWithManySlowWakesQueued is |
| 621 | // dispatchSidecarWakes' half of review round 2's regression pin (see |
| 622 | // restart_test.go's TestCycleWorkersReturnsPromptlyWithManySlowWakesQueued |
| 623 | // for the full shape of the bug this catches): strictly more than |
| 624 | // workerBootConcurrency sidecars, every wake deliberately parked, must |
| 625 | // still return this function immediately — the whole bounded fan-out runs |
| 626 | // inside dispatchBoundedWakes' own goroutine now, not just each individual |
| 627 | // wake dispatched from a loop that can itself stall once the bound fills. |
| 628 | func TestDispatchSidecarWakesReturnsPromptlyWithManySlowWakesQueued(t *testing.T) { |
| 629 | n := workerBootConcurrency + 5 |
| 630 | moved := make([]registry.Route, n) |
| 631 | for i := range moved { |
| 632 | moved[i] = wakeSidecarRoute(fmt.Sprintf("s%d.x.sidecar.internal", i), "bdm", "eleven", fmt.Sprintf("s%d", i)) |
| 633 | } |
| 634 | slow := make(chan struct{}) |
| 635 | var mu sync.Mutex |
| 636 | woken := map[string]bool{} |
| 637 | wake := func(rt registry.Route) error { |
| 638 | <-slow |
| 639 | mu.Lock() |
| 640 | woken[rt.Host] = true |
| 641 | mu.Unlock() |
| 642 | return nil |
| 643 | } |
| 644 | |
| 645 | // Same deadline-vs-elapsed-time correction as |
| 646 | // TestDispatchSidecarWakesReturnsBeforeASlowWakeCompletes (fix round 3). |
| 647 | done := make(chan struct{}) |
| 648 | go func() { |
| 649 | dispatchSidecarWakes(moved, wake) |
| 650 | close(done) |
| 651 | }() |
| 652 | select { |
| 653 | case <-done: |
| 654 | case <-time.After(time.Second): |
| 655 | close(slow) |
| 656 | t.Fatalf("dispatchSidecarWakes did not return within 1s with %d wakes blocked (bound %d) — the whole bounded fan-out must run off the caller's goroutine", n, workerBootConcurrency) |
| 657 | } |
| 658 | |
| 659 | close(slow) |
| 660 | waitForCond(t, "every queued sidecar to be woken eventually", func() bool { |
| 661 | mu.Lock() |
| 662 | defer mu.Unlock() |
| 663 | return len(woken) == n |
| 664 | }) |
| 665 | } |
| 666 | |
amadan