| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "crypto/ed25519" |
| 7 | "encoding/base64" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "log" |
| 11 | "net/http" |
| 12 | "net/http/httptest" |
| 13 | "os" |
| 14 | "path/filepath" |
| 15 | "runtime" |
| 16 | "slices" |
| 17 | "sort" |
| 18 | "strings" |
| 19 | "sync" |
| 20 | "sync/atomic" |
| 21 | "testing" |
| 22 | "time" |
| 23 | |
| 24 | "github.com/carlosframework/platform/internal/appconfig" |
| 25 | "github.com/carlosframework/platform/internal/domains" |
| 26 | "github.com/carlosframework/platform/internal/provision" |
| 27 | "github.com/carlosframework/platform/internal/registry" |
| 28 | "github.com/carlosframework/platform/internal/release" |
| 29 | "github.com/carlosframework/platform/internal/status" |
| 30 | ) |
| 31 | |
| 32 | // genKit is the shared setup for the generation-poll tests: a directory |
| 33 | // store standing in for the deployment bucket, and a context. No registry |
| 34 | // — genPollPass takes routes, not a *registry.Registry, precisely so these |
| 35 | // tests need no bolt database (the loop does its own reg.List()). |
| 36 | func genKit(t *testing.T) (context.Context, release.Store) { |
| 37 | t.Helper() |
| 38 | return context.Background(), release.NewDirStore(t.TempDir()) |
| 39 | } |
| 40 | |
| 41 | // execRoute is the socket-exec instance shape appconfig.ExecConfigApps |
| 42 | // accepts (execSyncable): a non-platform, ready instance with a socket |
| 43 | // upstream. Field-for-field the fixture shape syncexec_test.go uses, so a |
| 44 | // change to the predicates fails these tests for the same reason it fails |
| 45 | // appconfig's. |
| 46 | func execRoute(host, acct, app string) registry.Route { |
| 47 | return registry.Route{ |
| 48 | Host: host, Kind: "instance", App: app, Account: acct, |
| 49 | Status: "ready", Socket: "/run/carlos/" + host + ".sock", |
| 50 | } |
| 51 | } |
| 52 | |
| 53 | // unitRoute is the carlos-app@ shape appconfig.UnitConfigApps accepts: the |
| 54 | // same syncable instance, backed by a systemd unit whose env file only the |
| 55 | // root adopt pass can write. The agent's half of a change to one of these |
| 56 | // is the marker file, never a render. |
| 57 | func unitRoute(host, acct, app string) registry.Route { |
| 58 | rt := execRoute(host, acct, app) |
| 59 | rt.Backing = registry.BackingUnit |
| 60 | return rt |
| 61 | } |
| 62 | |
| 63 | // countingStore records how many times each key was read, so a test can |
| 64 | // pin that an app carrying BOTH route shapes still costs ONE generation |
| 65 | // read per tick — the whole economy of the poll set being a union. |
| 66 | type countingStore struct { |
| 67 | release.Store |
| 68 | gets map[string]int |
| 69 | } |
| 70 | |
| 71 | func (c *countingStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 72 | if c.gets == nil { |
| 73 | c.gets = map[string]int{} |
| 74 | } |
| 75 | c.gets[key]++ |
| 76 | return c.Store.Get(ctx, key) |
| 77 | } |
| 78 | |
| 79 | // genEtag reads the app's generation etag straight from the store, so a |
| 80 | // test can assert the cache holds what the bucket actually reports rather |
| 81 | // than a value the test made up. |
| 82 | func genEtag(t *testing.T, ctx context.Context, s release.Store, acct, app string) string { |
| 83 | t.Helper() |
| 84 | _, etag, err := s.Get(ctx, appconfig.GenKey(acct, app)) |
| 85 | if err != nil { |
| 86 | t.Fatalf("read generation %s/%s: %v", acct, app, err) |
| 87 | } |
| 88 | return etag |
| 89 | } |
| 90 | |
| 91 | func touch(t *testing.T, ctx context.Context, s release.Store, acct, app string) { |
| 92 | t.Helper() |
| 93 | if err := appconfig.TouchGeneration(ctx, s, acct, app); err != nil { |
| 94 | t.Fatalf("TouchGeneration %s/%s: %v", acct, app, err) |
| 95 | } |
| 96 | } |
| 97 | |
| 98 | // failingKeyStore fails Get for exactly one key — a transient bucket blip |
| 99 | // against a single app's generation object while every other key in the |
| 100 | // same pass resolves normally (the shape an S3 503 against one object |
| 101 | // takes). Same wrapper appconfig's syncexec_test.go uses, for the same |
| 102 | // reason. |
| 103 | type failingKeyStore struct { |
| 104 | release.Store |
| 105 | failKey string |
| 106 | err error |
| 107 | } |
| 108 | |
| 109 | func (f *failingKeyStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 110 | if key == f.failKey { |
| 111 | return nil, "", f.err |
| 112 | } |
| 113 | return f.Store.Get(ctx, key) |
| 114 | } |
| 115 | |
| 116 | func TestGenPollPassFirstSightIsSilent(t *testing.T) { |
| 117 | ctx, s := genKit(t) |
| 118 | touch(t, ctx, s, "bab", "eleven") |
| 119 | routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")} |
| 120 | cache := newGenCache() |
| 121 | |
| 122 | execChanged, unitChanged, errs := genPollPass(ctx, routes, s, cache) |
| 123 | if execChanged || unitChanged { |
| 124 | t.Fatal("first sight of an app must not trigger a sync (boot would re-bounce the world)") |
| 125 | } |
| 126 | if len(errs) != 0 { |
| 127 | t.Fatalf("unexpected read errors: %v", errs) |
| 128 | } |
| 129 | want := genEtag(t, ctx, s, "bab", "eleven") |
| 130 | if got := cache.etags[[2]string{"bab", "eleven"}]; got != want { |
| 131 | t.Fatalf("cache etag = %q, want the store's %q", got, want) |
| 132 | } |
| 133 | } |
| 134 | |
| 135 | func TestGenPollPassUnchangedIsSilent(t *testing.T) { |
| 136 | ctx, s := genKit(t) |
| 137 | touch(t, ctx, s, "bab", "eleven") |
| 138 | routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")} |
| 139 | cache := newGenCache() |
| 140 | |
| 141 | genPollPass(ctx, routes, s, cache) |
| 142 | execChanged, unitChanged, errs := genPollPass(ctx, routes, s, cache) |
| 143 | if execChanged || unitChanged { |
| 144 | t.Fatal("an unmoved generation etag must not trigger a sync") |
| 145 | } |
| 146 | if len(errs) != 0 { |
| 147 | t.Fatalf("unexpected read errors: %v", errs) |
| 148 | } |
| 149 | } |
| 150 | |
| 151 | func TestGenPollPassTouchedGenerationTriggersSync(t *testing.T) { |
| 152 | ctx, s := genKit(t) |
| 153 | touch(t, ctx, s, "bab", "eleven") |
| 154 | routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")} |
| 155 | cache := newGenCache() |
| 156 | genPollPass(ctx, routes, s, cache) |
| 157 | |
| 158 | touch(t, ctx, s, "bab", "eleven") |
| 159 | execChanged, unitChanged, errs := genPollPass(ctx, routes, s, cache) |
| 160 | if !execChanged { |
| 161 | t.Fatal("a moved generation etag must trigger a sync") |
| 162 | } |
| 163 | if unitChanged { |
| 164 | t.Fatal("an exec-only app must never ask for a root adopt pass") |
| 165 | } |
| 166 | if len(errs) != 0 { |
| 167 | t.Fatalf("unexpected read errors: %v", errs) |
| 168 | } |
| 169 | want := genEtag(t, ctx, s, "bab", "eleven") |
| 170 | if got := cache.etags[[2]string{"bab", "eleven"}]; got != want { |
| 171 | t.Fatalf("cache etag = %q, want the new %q (a stale cache would re-sync forever)", got, want) |
| 172 | } |
| 173 | if execChanged, unitChanged, _ := genPollPass(ctx, routes, s, cache); execChanged || unitChanged { |
| 174 | t.Fatal("the pass after a change must be silent again") |
| 175 | } |
| 176 | } |
| 177 | |
| 178 | func TestGenPollPassAbsentGenerationStaysSilent(t *testing.T) { |
| 179 | ctx, s := genKit(t) |
| 180 | // No touch at all: an app last written by a pre-generation CLI. Legal |
| 181 | // forever — the hourly configSyncLoop is its floor. |
| 182 | routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")} |
| 183 | cache := newGenCache() |
| 184 | |
| 185 | for i := range 2 { |
| 186 | execChanged, unitChanged, errs := genPollPass(ctx, routes, s, cache) |
| 187 | if execChanged || unitChanged { |
| 188 | t.Fatalf("pass %d: an app with no generation object must never trigger a sync", i) |
| 189 | } |
| 190 | if len(errs) != 0 { |
| 191 | t.Fatalf("pass %d: absent is not a read error: %v", i, errs) |
| 192 | } |
| 193 | } |
| 194 | if got := cache.etags[[2]string{"bab", "eleven"}]; got != genAbsent { |
| 195 | t.Fatalf("cache = %q, want the known-absent sentinel %q", got, genAbsent) |
| 196 | } |
| 197 | } |
| 198 | |
| 199 | func TestGenPollPassAbsentToPresentTriggersSync(t *testing.T) { |
| 200 | ctx, s := genKit(t) |
| 201 | routes := []registry.Route{execRoute("a.example.com", "bab", "eleven")} |
| 202 | cache := newGenCache() |
| 203 | genPollPass(ctx, routes, s, cache) // records known-absent |
| 204 | |
| 205 | touch(t, ctx, s, "bab", "eleven") // the app's first ever config write |
| 206 | execChanged, _, errs := genPollPass(ctx, routes, s, cache) |
| 207 | if !execChanged { |
| 208 | t.Fatal("absent -> present is the first write to an old app: it must sync") |
| 209 | } |
| 210 | if len(errs) != 0 { |
| 211 | t.Fatalf("unexpected read errors: %v", errs) |
| 212 | } |
| 213 | } |
| 214 | |
| 215 | func TestGenPollPassOneChangedAppSyncsOnce(t *testing.T) { |
| 216 | ctx, s := genKit(t) |
| 217 | touch(t, ctx, s, "bab", "eleven") |
| 218 | touch(t, ctx, s, "bab", "titogo") |
| 219 | routes := []registry.Route{ |
| 220 | execRoute("a.example.com", "bab", "eleven"), |
| 221 | execRoute("b.example.com", "bab", "titogo"), |
| 222 | } |
| 223 | cache := newGenCache() |
| 224 | genPollPass(ctx, routes, s, cache) |
| 225 | |
| 226 | touch(t, ctx, s, "bab", "titogo") |
| 227 | execChanged, _, errs := genPollPass(ctx, routes, s, cache) |
| 228 | if !execChanged { |
| 229 | t.Fatal("one moved app in the set must report the pass as changed") |
| 230 | } |
| 231 | if len(errs) != 0 { |
| 232 | t.Fatalf("unexpected read errors: %v", errs) |
| 233 | } |
| 234 | // One flag for the whole pass: the caller runs ONE SyncExec, which |
| 235 | // resolves every app itself. The unchanged app's cache entry must be |
| 236 | // carried forward untouched so it doesn't re-fire next tick. |
| 237 | if got, want := cache.etags[[2]string{"bab", "eleven"}], genEtag(t, ctx, s, "bab", "eleven"); got != want { |
| 238 | t.Fatalf("unchanged app's cache = %q, want %q", got, want) |
| 239 | } |
| 240 | if execChanged, _, _ := genPollPass(ctx, routes, s, cache); execChanged { |
| 241 | t.Fatal("the next pass must be silent for both apps") |
| 242 | } |
| 243 | } |
| 244 | |
| 245 | func TestGenPollPassDropsVanishedApps(t *testing.T) { |
| 246 | ctx, s := genKit(t) |
| 247 | touch(t, ctx, s, "bab", "eleven") |
| 248 | touch(t, ctx, s, "bab", "titogo") |
| 249 | both := []registry.Route{ |
| 250 | execRoute("a.example.com", "bab", "eleven"), |
| 251 | execRoute("b.example.com", "bab", "titogo"), |
| 252 | } |
| 253 | one := both[:1] |
| 254 | cache := newGenCache() |
| 255 | genPollPass(ctx, both, s, cache) |
| 256 | |
| 257 | genPollPass(ctx, one, s, cache) // titogo's route is gone |
| 258 | if _, ok := cache.etags[[2]string{"bab", "titogo"}]; ok { |
| 259 | t.Fatal("a route that left the registry must leave no cache entry behind") |
| 260 | } |
| 261 | // The proof that dropping matters: while the app was unrouted its |
| 262 | // config moved. The ETAG half must record that silently (first sight |
| 263 | // again) rather than comparing against a remembered value — a route |
| 264 | // being re-added already gets fresh config from the spawn path. |
| 265 | // |
| 266 | // The tick itself is NOT silent, and that is the route-shape half |
| 267 | // speaking, not the etag one (issue #196): a route entering or leaving |
| 268 | // scope changes what belongs on disk — a file to write, or one to reap |
| 269 | // — so it earns a pass. The pass is cheap and cannot cost a restart: |
| 270 | // SyncExec bounces only a route whose rendered bytes actually changed. |
| 271 | // What this test still pins directly is the cache entry above; the |
| 272 | // etag half's silence is no longer separately observable from here, |
| 273 | // since the shape movement covers it either way. |
| 274 | touch(t, ctx, s, "bab", "titogo") |
| 275 | _, _, errs := genPollPass(ctx, both, s, cache) |
| 276 | if len(errs) != 0 { |
| 277 | t.Fatalf("unexpected read errors: %v", errs) |
| 278 | } |
| 279 | // Settled again: with the route set stable and no bucket movement, the |
| 280 | // next tick must be silent. A remembered etag or a mis-stored digest |
| 281 | // would still be firing here. |
| 282 | execChanged, unitChanged, errs := genPollPass(ctx, both, s, cache) |
| 283 | if execChanged || unitChanged { |
| 284 | t.Fatal("a settled tick after a re-add must be silent") |
| 285 | } |
| 286 | if len(errs) != 0 { |
| 287 | t.Fatalf("unexpected read errors: %v", errs) |
| 288 | } |
| 289 | } |
| 290 | |
| 291 | func TestGenPollPassReadErrorLeavesCacheAndKeepsGoing(t *testing.T) { |
| 292 | ctx, dir := genKit(t) |
| 293 | touch(t, ctx, dir, "bab", "eleven") |
| 294 | touch(t, ctx, dir, "bab", "titogo") |
| 295 | routes := []registry.Route{ |
| 296 | execRoute("a.example.com", "bab", "eleven"), |
| 297 | execRoute("b.example.com", "bab", "titogo"), |
| 298 | } |
| 299 | cache := newGenCache() |
| 300 | genPollPass(ctx, routes, dir, cache) |
| 301 | elevenEtag := cache.etags[[2]string{"bab", "eleven"}] |
| 302 | |
| 303 | // eleven's generation object goes unreadable; titogo's moves. |
| 304 | boom := errors.New("bucket said no") |
| 305 | s := &failingKeyStore{Store: dir, failKey: appconfig.GenKey("bab", "eleven"), err: boom} |
| 306 | touch(t, ctx, dir, "bab", "titogo") |
| 307 | |
| 308 | execChanged, _, errs := genPollPass(ctx, routes, s, cache) |
| 309 | if !execChanged { |
| 310 | t.Fatal("one app failing must not stop the others being evaluated") |
| 311 | } |
| 312 | if len(errs) != 1 { |
| 313 | t.Fatalf("read errors = %v, want exactly one (aggregated into one log line per pass)", errs) |
| 314 | } |
| 315 | if !errors.Is(errs[0], boom) { |
| 316 | t.Fatalf("read error = %v, want it to wrap %v", errs[0], boom) |
| 317 | } |
| 318 | if got := cache.etags[[2]string{"bab", "eleven"}]; got != elevenEtag { |
| 319 | t.Fatalf("unreadable app's cache = %q, want it untouched at %q (a cleared entry would re-fire on recovery)", got, elevenEtag) |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | func TestGenPollPassUnitFirstSightIsSilent(t *testing.T) { |
| 324 | ctx, s := genKit(t) |
| 325 | touch(t, ctx, s, "bab", "unitapp") |
| 326 | routes := []registry.Route{unitRoute("u.example.com", "bab", "unitapp")} |
| 327 | cache := newGenCache() |
| 328 | |
| 329 | execChanged, unitChanged, errs := genPollPass(ctx, routes, s, cache) |
| 330 | if execChanged || unitChanged { |
| 331 | t.Fatal("first sight of a unit-backed app must not request a root pass (boot would restart every tenant)") |
| 332 | } |
| 333 | if len(errs) != 0 { |
| 334 | t.Fatalf("unexpected read errors: %v", errs) |
| 335 | } |
| 336 | want := genEtag(t, ctx, s, "bab", "unitapp") |
| 337 | if got := cache.etags[[2]string{"bab", "unitapp"}]; got != want { |
| 338 | t.Fatalf("cache etag = %q, want the store's %q", got, want) |
| 339 | } |
| 340 | } |
| 341 | |
| 342 | func TestGenPollPassUnitMovementRequestsRootPassOnly(t *testing.T) { |
| 343 | ctx, s := genKit(t) |
| 344 | touch(t, ctx, s, "bab", "unitapp") |
| 345 | touch(t, ctx, s, "bab", "eleven") |
| 346 | routes := []registry.Route{ |
| 347 | unitRoute("u.example.com", "bab", "unitapp"), |
| 348 | execRoute("a.example.com", "bab", "eleven"), |
| 349 | } |
| 350 | cache := newGenCache() |
| 351 | genPollPass(ctx, routes, s, cache) |
| 352 | |
| 353 | touch(t, ctx, s, "bab", "unitapp") |
| 354 | execChanged, unitChanged, errs := genPollPass(ctx, routes, s, cache) |
| 355 | if !unitChanged { |
| 356 | t.Fatal("a moved generation for a unit-backed app must request the root adopt pass") |
| 357 | } |
| 358 | if execChanged { |
| 359 | t.Fatal("the exec half must stay quiet: no exec-config app moved, so SyncExec would find nothing to write") |
| 360 | } |
| 361 | if len(errs) != 0 { |
| 362 | t.Fatalf("unexpected read errors: %v", errs) |
| 363 | } |
| 364 | if execChanged, unitChanged, _ := genPollPass(ctx, routes, s, cache); execChanged || unitChanged { |
| 365 | t.Fatal("the pass after a change must be silent again (a stale cache would re-touch the marker forever)") |
| 366 | } |
| 367 | } |
| 368 | |
| 369 | func TestGenPollPassBothShapesSetBothHalvesOnOneRead(t *testing.T) { |
| 370 | ctx, dir := genKit(t) |
| 371 | touch(t, ctx, dir, "zed", "mixed") |
| 372 | // One app, two routes: the activator spawns one, systemd owns the other. |
| 373 | // Its config lives in ONE generation object, so the union must read that |
| 374 | // object once per tick and report both halves off it. |
| 375 | routes := []registry.Route{ |
| 376 | execRoute("mixed-exec.example.com", "zed", "mixed"), |
| 377 | unitRoute("mixed-unit.example.com", "zed", "mixed"), |
| 378 | } |
| 379 | s := &countingStore{Store: dir} |
| 380 | cache := newGenCache() |
| 381 | genPollPass(ctx, routes, s, cache) |
| 382 | |
| 383 | touch(t, ctx, dir, "zed", "mixed") |
| 384 | execChanged, unitChanged, errs := genPollPass(ctx, routes, s, cache) |
| 385 | if !execChanged || !unitChanged { |
| 386 | t.Fatalf("execChanged=%v unitChanged=%v, want both — the app has both route shapes", execChanged, unitChanged) |
| 387 | } |
| 388 | if len(errs) != 0 { |
| 389 | t.Fatalf("unexpected read errors: %v", errs) |
| 390 | } |
| 391 | if got := s.gets[appconfig.GenKey("zed", "mixed")]; got != 2 { |
| 392 | t.Fatalf("generation object read %d times over 2 ticks, want 2 — an app in both sets must cost one read per tick", got) |
| 393 | } |
| 394 | } |
| 395 | |
| 396 | func TestGenPollPassPrunesUnitAppsToo(t *testing.T) { |
| 397 | ctx, s := genKit(t) |
| 398 | touch(t, ctx, s, "bab", "unitapp") |
| 399 | both := []registry.Route{ |
| 400 | execRoute("a.example.com", "bab", "eleven"), |
| 401 | unitRoute("u.example.com", "bab", "unitapp"), |
| 402 | } |
| 403 | one := both[:1] |
| 404 | cache := newGenCache() |
| 405 | genPollPass(ctx, both, s, cache) |
| 406 | |
| 407 | // The live set the cache is pruned against is the UNION: a unit app whose |
| 408 | // last route left must lose its entry exactly as an exec app does, or its |
| 409 | // remembered etag would fire a bogus root-pass request when the route |
| 410 | // comes back (adoption already delivers fresh config on the way in). |
| 411 | genPollPass(ctx, one, s, cache) |
| 412 | if _, ok := cache.etags[[2]string{"bab", "unitapp"}]; ok { |
| 413 | t.Fatal("a unit-backed route that left the registry must leave no cache entry behind") |
| 414 | } |
| 415 | // As in TestGenPollPassDropsVanishedApps: the re-add tick itself moves |
| 416 | // the route-shape digest (issue #196), so the silence worth asserting |
| 417 | // is the SETTLED tick after it. The dropped cache entry above is what |
| 418 | // pins the etag half directly. |
| 419 | touch(t, ctx, s, "bab", "unitapp") |
| 420 | _, _, errs := genPollPass(ctx, both, s, cache) |
| 421 | if len(errs) != 0 { |
| 422 | t.Fatalf("unexpected read errors: %v", errs) |
| 423 | } |
| 424 | execChanged, unitChanged, errs := genPollPass(ctx, both, s, cache) |
| 425 | if execChanged || unitChanged { |
| 426 | t.Fatal("a settled tick after a unit-backed re-add must be silent") |
| 427 | } |
| 428 | if len(errs) != 0 { |
| 429 | t.Fatalf("unexpected read errors: %v", errs) |
| 430 | } |
| 431 | } |
| 432 | |
| 433 | // ---- cold-wake materialize hook (carlos-deploy spec §5) --------------------- |
| 434 | |
| 435 | // TestMaterializeHookDegradesToNil pins the safe degrade: a box with no |
| 436 | // release store, or no (or a broken) release pubkey, gets a NIL hook — which |
| 437 | // is today's behavior exactly, the ExecBackend's "no live binary" wake |
| 438 | // failure. A non-nil hook built from a half-configured box would turn every |
| 439 | // cold wake into a staging error that names the wrong problem. |
| 440 | func TestMaterializeHookDegradesToNil(t *testing.T) { |
| 441 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 442 | if err != nil { |
| 443 | t.Fatalf("open registry: %v", err) |
| 444 | } |
| 445 | pub, _, err := ed25519.GenerateKey(nil) |
| 446 | if err != nil { |
| 447 | t.Fatalf("generate key: %v", err) |
| 448 | } |
| 449 | pubB64 := base64.StdEncoding.EncodeToString(pub) |
| 450 | |
| 451 | for _, c := range []struct{ name, dir, pub string }{ |
| 452 | {"no store, no key", "", ""}, |
| 453 | {"store but no key", t.TempDir(), ""}, |
| 454 | {"store but unparseable key", t.TempDir(), "not-base64!!"}, |
| 455 | {"key but no store", "", pubB64}, |
| 456 | } { |
| 457 | t.Run(c.name, func(t *testing.T) { |
| 458 | t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "") |
| 459 | t.Setenv("CARLOS_DEPLOYMENT_DIR", c.dir) |
| 460 | t.Setenv("CARLOS_RELEASE_PUBKEY", c.pub) |
| 461 | if hook := materializeHook(reg, t.TempDir()); hook != nil { |
| 462 | t.Fatal("hook must be nil on a box that cannot adopt — nil is today's behavior") |
| 463 | } |
| 464 | }) |
| 465 | } |
| 466 | } |
| 467 | |
| 468 | // TestMaterializeHookStagesAColdRoute exercises the wired path end to end |
| 469 | // through the very function the ExecBackend calls: a fully configured box, a |
| 470 | // route with nothing staged, and after one hook call the live symlink exists |
| 471 | // and names the channel head. This is what retires "no live binary" as a |
| 472 | // wake-failure class (issue #77). |
| 473 | func TestMaterializeHookStagesAColdRoute(t *testing.T) { |
| 474 | dir := t.TempDir() |
| 475 | optDir := t.TempDir() |
| 476 | s := release.NewDirStore(dir) |
| 477 | ctx := context.Background() |
| 478 | plat := runtime.GOOS + "-" + runtime.GOARCH |
| 479 | |
| 480 | pub, priv, err := ed25519.GenerateKey(nil) |
| 481 | if err != nil { |
| 482 | t.Fatalf("generate key: %v", err) |
| 483 | } |
| 484 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 485 | if err != nil { |
| 486 | t.Fatalf("open registry: %v", err) |
| 487 | } |
| 488 | if err := reg.Upsert(registry.Route{ |
| 489 | Host: "cold.example.com", Kind: "instance", App: "coldapp", |
| 490 | Channel: "canary/dev", Status: "ready", |
| 491 | }); err != nil { |
| 492 | t.Fatalf("upsert: %v", err) |
| 493 | } |
| 494 | m := release.Manifest{ |
| 495 | App: "coldapp", Version: "abc1234", Kind: "binary", |
| 496 | Artifacts: map[string]release.Artifact{plat: {}}, |
| 497 | } |
| 498 | if err := release.Ship(ctx, s, "bab", m, map[string][]byte{plat: []byte("cold build")}); err != nil { |
| 499 | t.Fatalf("Ship: %v", err) |
| 500 | } |
| 501 | if err := release.Promote(ctx, s, priv, "bab", "coldapp", "canary/dev", "abc1234"); err != nil { |
| 502 | t.Fatalf("Promote: %v", err) |
| 503 | } |
| 504 | |
| 505 | t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "") |
| 506 | t.Setenv("CARLOS_DEPLOYMENT_DIR", dir) |
| 507 | t.Setenv("CARLOS_RELEASE_PUBKEY", base64.StdEncoding.EncodeToString(pub)) |
| 508 | t.Setenv("CARLOS_OPT_DIR", optDir) |
| 509 | |
| 510 | hook := materializeHook(reg, t.TempDir()) |
| 511 | if hook == nil { |
| 512 | t.Fatal("a fully configured box must get a hook") |
| 513 | } |
| 514 | if err := hook(ctx, "cold.example.com"); err != nil { |
| 515 | t.Fatalf("hook: %v", err) |
| 516 | } |
| 517 | if v, err := release.LiveVersion(optDir, "cold.example.com"); err != nil || v != "abc1234" { |
| 518 | t.Fatalf("live = %q, %v; want abc1234", v, err) |
| 519 | } |
| 520 | } |
| 521 | |
| 522 | // restartTestRegistry is the bolt-backed registry the restart-dispatch tests |
| 523 | // reason over — makeRestartRoute's whole job is to ask the registry what |
| 524 | // backs a host, so a real registry (not a fake) is what the test needs. |
| 525 | func restartTestRegistry(t *testing.T) *registry.Registry { |
| 526 | t.Helper() |
| 527 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 528 | if err != nil { |
| 529 | t.Fatalf("open registry: %v", err) |
| 530 | } |
| 531 | t.Cleanup(func() { |
| 532 | if err := reg.Close(); err != nil { |
| 533 | t.Error(err) |
| 534 | } |
| 535 | }) |
| 536 | return reg |
| 537 | } |
| 538 | |
| 539 | // TestMakeRestartRouteDispatchesOnBacking pins the correction this split |
| 540 | // exists for: a unit tenant is try-restarted, an exec-backed instance is |
| 541 | // Bounced. Before the split both took the systemctl path — and systemctl |
| 542 | // cannot reach a child of the agent process, so a warm exec instance kept |
| 543 | // serving its old binary until it happened to hibernate. |
| 544 | func TestMakeRestartRouteDispatchesOnBacking(t *testing.T) { |
| 545 | reg := restartTestRegistry(t) |
| 546 | if err := reg.Upsert(registry.Route{ |
| 547 | Host: "unit.test", Kind: "instance", App: "a", Status: "ready", |
| 548 | Backing: registry.BackingUnit, |
| 549 | }); err != nil { |
| 550 | t.Fatalf("upsert unit route: %v", err) |
| 551 | } |
| 552 | if err := reg.Upsert(registry.Route{ |
| 553 | Host: "exec.test", Kind: "instance", App: "a", Status: "ready", |
| 554 | }); err != nil { |
| 555 | t.Fatalf("upsert exec route: %v", err) |
| 556 | } |
| 557 | var restarted, bounced []string |
| 558 | restart := makeRestartRoute(reg, |
| 559 | func(h string) { restarted = append(restarted, h) }, |
| 560 | unitOnly(func(u string) error { |
| 561 | t.Fatalf("named-unit arm fired for %q on a route that names no unit", u) |
| 562 | return nil |
| 563 | }), |
| 564 | func(h string) bool { bounced = append(bounced, h); return true }) |
| 565 | if err := restart("unit.test"); err != nil { |
| 566 | t.Fatalf("restart unit.test: %v", err) |
| 567 | } |
| 568 | if err := restart("exec.test"); err != nil { |
| 569 | t.Fatalf("restart exec.test: %v", err) |
| 570 | } |
| 571 | if len(restarted) != 1 || restarted[0] != "unit.test" { |
| 572 | t.Fatalf("unit path: %v", restarted) |
| 573 | } |
| 574 | if len(bounced) != 1 || bounced[0] != "exec.test" { |
| 575 | t.Fatalf("exec path: %v", bounced) |
| 576 | } |
| 577 | } |
| 578 | |
| 579 | // TestMakeRestartRouteUnknownHostFallsBackToUnit: a host the registry cannot |
| 580 | // answer for gets the OLD behavior (try-restart, a safe no-op for a |
| 581 | // nonexistent unit) — never a bounce of a process we might not own. |
| 582 | func TestMakeRestartRouteUnknownHostFallsBackToUnit(t *testing.T) { |
| 583 | reg := restartTestRegistry(t) |
| 584 | var restarted []string |
| 585 | restart := makeRestartRoute(reg, func(h string) { restarted = append(restarted, h) }, |
| 586 | unitOnly(func(u string) error { |
| 587 | t.Fatalf("named-unit arm fired for %q on a host the registry cannot answer for", u) |
| 588 | return nil |
| 589 | }), nil) |
| 590 | if err := restart("ghost.test"); err != nil { |
| 591 | t.Fatalf("restart ghost.test: %v", err) |
| 592 | } |
| 593 | if len(restarted) != 1 || restarted[0] != "ghost.test" { |
| 594 | t.Fatalf("fallback: %v", restarted) |
| 595 | } |
| 596 | } |
| 597 | |
| 598 | // TestMakeRestartRouteExecFallsThroughWhenNothingBounced covers the state |
| 599 | // issue #14 is about (adopt_advice.go's warning): a route claims exec backing |
| 600 | // but a carlos-app@<host> unit is ACTIVE — an unstamped unit tenant. The |
| 601 | // bounce finds nothing of its own to cycle and returns false, and try-restart |
| 602 | // must still run, exactly as it did before the dispatch split. Same for a box |
| 603 | // with no activator at all (nil bounce), which must not panic either. |
| 604 | // |
| 605 | // The cost when the route really is a sleeping exec child is one try-restart |
| 606 | // of a stopped or absent unit — a no-op, at most one log line per version |
| 607 | // change per host, since the callback only fires when the version moves. |
| 608 | func TestMakeRestartRouteExecFallsThroughWhenNothingBounced(t *testing.T) { |
| 609 | t.Parallel() |
| 610 | for _, tc := range []struct { |
| 611 | name string |
| 612 | bounce func(string) bool |
| 613 | }{ |
| 614 | {"no activator armed", nil}, |
| 615 | {"activator armed but nothing awake", func(string) bool { return false }}, |
| 616 | } { |
| 617 | t.Run(tc.name, func(t *testing.T) { |
| 618 | reg := restartTestRegistry(t) |
| 619 | if err := reg.Upsert(registry.Route{ |
| 620 | Host: "exec.test", Kind: "instance", App: "a", Status: "ready", |
| 621 | }); err != nil { |
| 622 | t.Fatalf("upsert exec route: %v", err) |
| 623 | } |
| 624 | var restarted []string |
| 625 | restart := makeRestartRoute(reg, func(h string) { restarted = append(restarted, h) }, |
| 626 | unitOnly(func(u string) error { |
| 627 | t.Fatalf("named-unit arm fired for %q on a route that names no unit", u) |
| 628 | return nil |
| 629 | }), tc.bounce) |
| 630 | if err := restart("exec.test"); err != nil { |
| 631 | t.Fatalf("restart exec.test: %v", err) |
| 632 | } |
| 633 | if len(restarted) != 1 || restarted[0] != "exec.test" { |
| 634 | t.Fatalf("a bounce that cycled nothing must fall through to try-restart: %v", restarted) |
| 635 | } |
| 636 | }) |
| 637 | } |
| 638 | } |
| 639 | |
| 640 | // TestMakeRestartRouteRestartsTheUnitARouteNames pins issue #110's fix: an |
| 641 | // addr-backed route that names its own unit (the consoles, `-addr |
| 642 | // 127.0.0.1:8090` under a bespoke console.service) is cycled BY THAT NAME, |
| 643 | // and neither of the other two arms runs — both would cycle nothing here, |
| 644 | // which is exactly how adoption came to stamp a Version the running process |
| 645 | // did not have. |
| 646 | func TestMakeRestartRouteRestartsTheUnitARouteNames(t *testing.T) { |
| 647 | reg := restartTestRegistry(t) |
| 648 | if err := reg.Upsert(registry.Route{ |
| 649 | Host: "console.carloku.com", Kind: "instance", App: "console", |
| 650 | Addr: "127.0.0.1:8090", Status: "ready", Unit: "console.service", |
| 651 | }); err != nil { |
| 652 | t.Fatalf("upsert console route: %v", err) |
| 653 | } |
| 654 | var units, restarted, bounced []string |
| 655 | restart := makeRestartRoute(reg, |
| 656 | func(h string) { restarted = append(restarted, h) }, |
| 657 | unitOnly(func(u string) error { units = append(units, u); return nil }), |
| 658 | func(h string) bool { bounced = append(bounced, h); return true }) |
| 659 | if err := restart("console.carloku.com"); err != nil { |
| 660 | t.Fatalf("restart console.carloku.com: %v", err) |
| 661 | } |
| 662 | if len(units) != 1 || units[0] != "console.service" { |
| 663 | t.Fatalf("named-unit arm: %v, want [console.service]", units) |
| 664 | } |
| 665 | if len(restarted) != 0 { |
| 666 | t.Fatalf("a route that names its unit must not also be try-restarted as carlos-app@<host>: %v", restarted) |
| 667 | } |
| 668 | if len(bounced) != 0 { |
| 669 | t.Fatalf("a route that names its unit must not be offered to the activator: %v", bounced) |
| 670 | } |
| 671 | } |
| 672 | |
| 673 | // The named-unit arm wins over BACKING, not merely over its absence: a route |
| 674 | // carrying both a unit name and unit backing names where its process is, and |
| 675 | // the name is the more specific answer. Pinned so a later reordering of the |
| 676 | // dispatch has to argue with a test rather than silently cycle |
| 677 | // carlos-app@<host> for a route whose process is somewhere else. |
| 678 | func TestMakeRestartRouteUnitNameBeatsBacking(t *testing.T) { |
| 679 | reg := restartTestRegistry(t) |
| 680 | if err := reg.Upsert(registry.Route{ |
| 681 | Host: "odd.test", Kind: "instance", App: "a", Status: "ready", |
| 682 | Backing: registry.BackingUnit, Unit: "bespoke.service", |
| 683 | }); err != nil { |
| 684 | t.Fatalf("upsert: %v", err) |
| 685 | } |
| 686 | var units, restarted []string |
| 687 | restart := makeRestartRoute(reg, |
| 688 | func(h string) { restarted = append(restarted, h) }, |
| 689 | unitOnly(func(u string) error { units = append(units, u); return nil }), |
| 690 | nil) |
| 691 | if err := restart("odd.test"); err != nil { |
| 692 | t.Fatalf("restart odd.test: %v", err) |
| 693 | } |
| 694 | if len(units) != 1 || units[0] != "bespoke.service" || len(restarted) != 0 { |
| 695 | t.Fatalf("units = %v, host try-restarts = %v; want the named unit alone", units, restarted) |
| 696 | } |
| 697 | } |
| 698 | |
| 699 | // The root pass (RunAdopt, carlos-adopt.service) must carry the named-unit |
| 700 | // arm, because on a real box it is the pass that actually adopts a console: |
| 701 | // the agent runs inside carlos-edge.service's ProtectSystem=strict sandbox |
| 702 | // and cannot write /opt/carlos, so its stage is blocked and relayed here. |
| 703 | // Everything else about it is unchanged — no activator, so every route it |
| 704 | // used to try-restart by host still is. |
| 705 | func TestRootRestartRouteRefusesMissingDeploymentBinding(t *testing.T) { |
| 706 | t.Parallel() |
| 707 | reg := restartTestRegistry(t) |
| 708 | if err := reg.Upsert(registry.Route{Host: "console.carloku.com", Kind: "instance", App: "console", Addr: "127.0.0.1:8090", Status: "ready", Unit: "console.service"}); err != nil { |
| 709 | t.Fatal(err) |
| 710 | } |
| 711 | if err := rootRestartRoute(reg, t.TempDir(), nil)("console.carloku.com"); err == nil { |
| 712 | t.Fatal("restarted without deployment binding") |
| 713 | } |
| 714 | } |
| 715 | |
| 716 | // TestAdoptPassRestartsANamedUnitOnAdoption is the end-to-end: a promote the |
| 717 | // pass adopts must reach the route's own unit, not just repoint a symlink and |
| 718 | // stamp a version. This is the shape of the deploy that produced issue #110. |
| 719 | func TestAdoptPassRestartsANamedUnitOnAdoption(t *testing.T) { |
| 720 | dir := t.TempDir() |
| 721 | optDir := t.TempDir() |
| 722 | s := release.NewDirStore(dir) |
| 723 | ctx := context.Background() |
| 724 | plat := runtime.GOOS + "-" + runtime.GOARCH |
| 725 | |
| 726 | pub, priv, err := ed25519.GenerateKey(nil) |
| 727 | if err != nil { |
| 728 | t.Fatalf("generate key: %v", err) |
| 729 | } |
| 730 | reg := restartTestRegistry(t) |
| 731 | if err := reg.Upsert(registry.Route{ |
| 732 | Host: "console.test", Kind: "instance", App: "console", Channel: "canary/console", |
| 733 | Addr: "127.0.0.1:8090", Status: "ready", Unit: "console.service", |
| 734 | }); err != nil { |
| 735 | t.Fatalf("upsert console.test: %v", err) |
| 736 | } |
| 737 | m := release.Manifest{ |
| 738 | App: "console", Version: "v1", Kind: "binary", |
| 739 | Artifacts: map[string]release.Artifact{plat: {}}, |
| 740 | } |
| 741 | if err := release.Ship(ctx, s, "bab", m, map[string][]byte{plat: []byte("build one")}); err != nil { |
| 742 | t.Fatalf("Ship: %v", err) |
| 743 | } |
| 744 | if err := release.Promote(ctx, s, priv, "bab", "console", "canary/console", "v1"); err != nil { |
| 745 | t.Fatalf("Promote: %v", err) |
| 746 | } |
| 747 | |
| 748 | var units []string |
| 749 | restart := makeRestartRoute(reg, |
| 750 | func(h string) { t.Fatalf("carlos-app@ arm fired for %q — the console has no such unit", h) }, |
| 751 | unitOnly(func(u string) error { units = append(units, u); return nil }), |
| 752 | func(h string) bool { |
| 753 | t.Fatalf("bounce arm fired for %q — the activator owns no console child", h) |
| 754 | return false |
| 755 | }) |
| 756 | |
| 757 | if err := release.Pass(ctx, reg, s, pub, optDir, restart, release.RestartHibernating(restart)); err != nil { |
| 758 | t.Fatalf("Pass: %v", err) |
| 759 | } |
| 760 | if len(units) != 1 || units[0] != "console.service" { |
| 761 | t.Fatalf("adoption cycled %v, want [console.service] — the header would claim v1 over the old process", units) |
| 762 | } |
| 763 | if v, err := release.LiveVersion(optDir, "console.test"); err != nil || v != "v1" { |
| 764 | t.Fatalf("console.test live = %q, %v; want v1", v, err) |
| 765 | } |
| 766 | } |
| 767 | |
| 768 | // TestAdoptPassRefusesToStampAVersionAWrongUnitDidNotStart is the honesty |
| 769 | // half of issue #110, and the reason the named-unit arm returns its error |
| 770 | // where the other two swallow theirs. A well-formed but WRONG unit name |
| 771 | // (`systemctl try-restart` exits 5, "not loaded") restarted nothing — exactly |
| 772 | // the state the column exists to end — and if that were swallowed adoption |
| 773 | // would stamp Version, the edge would serve the new X-Carlos-Version, and |
| 774 | // `carlos deploy` would report live over the old process. |
| 775 | // |
| 776 | // So: no Version stamp, and the failure is in the journal under the unit name |
| 777 | // an operator would grep for. The next pass retries. |
| 778 | func TestAdoptPassRefusesToStampAVersionAWrongUnitDidNotStart(t *testing.T) { |
| 779 | dir := t.TempDir() |
| 780 | optDir := t.TempDir() |
| 781 | s := release.NewDirStore(dir) |
| 782 | ctx := context.Background() |
| 783 | plat := runtime.GOOS + "-" + runtime.GOARCH |
| 784 | |
| 785 | pub, priv, err := ed25519.GenerateKey(nil) |
| 786 | if err != nil { |
| 787 | t.Fatalf("generate key: %v", err) |
| 788 | } |
| 789 | reg := restartTestRegistry(t) |
| 790 | // The stamp is well-formed — ValidUnitName cannot tell a real unit from an |
| 791 | // absent one — but no such unit is loaded on the box. |
| 792 | if err := reg.Upsert(registry.Route{ |
| 793 | Host: "console.test", Kind: "instance", App: "console", Channel: "canary/console", |
| 794 | Addr: "127.0.0.1:8090", Status: "ready", Unit: "consoel.service", |
| 795 | }); err != nil { |
| 796 | t.Fatalf("upsert console.test: %v", err) |
| 797 | } |
| 798 | m := release.Manifest{ |
| 799 | App: "console", Version: "v1", Kind: "binary", |
| 800 | Artifacts: map[string]release.Artifact{plat: {}}, |
| 801 | } |
| 802 | if err := release.Ship(ctx, s, "bab", m, map[string][]byte{plat: []byte("build one")}); err != nil { |
| 803 | t.Fatalf("Ship: %v", err) |
| 804 | } |
| 805 | if err := release.Promote(ctx, s, priv, "bab", "console", "canary/console", "v1"); err != nil { |
| 806 | t.Fatalf("Promote: %v", err) |
| 807 | } |
| 808 | |
| 809 | // A fake systemctl that answers the way the real one does for a unit that |
| 810 | // is not loaded: exit 5, with systemd's own words on stdout. |
| 811 | bin := t.TempDir() |
| 812 | if err := os.WriteFile(filepath.Join(bin, "systemctl"), |
| 813 | []byte("#!/bin/sh\necho \"Failed to try-restart $2: Unit $2 not found.\"\nexit 5\n"), 0o755); err != nil { |
| 814 | t.Fatal(err) |
| 815 | } |
| 816 | t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) |
| 817 | |
| 818 | var journal bytes.Buffer |
| 819 | log.SetOutput(&journal) |
| 820 | t.Cleanup(func() { log.SetOutput(os.Stderr) }) |
| 821 | |
| 822 | restart := makeRestartRoute(reg, systemctlTryRestart, agentCheckedUnitRestart(systemdUnits{}, optDir, os.Getenv, nil), nil) |
| 823 | if err := release.Pass(ctx, reg, s, pub, optDir, restart); err != nil { |
| 824 | t.Fatalf("Pass: %v", err) |
| 825 | } |
| 826 | |
| 827 | rt, err := reg.Get("console.test") |
| 828 | if err != nil || rt == nil { |
| 829 | t.Fatalf("get console.test: %+v, %v", rt, err) |
| 830 | } |
| 831 | if rt.Version != "" { |
| 832 | t.Fatalf("Version = %q after a restart that started nothing — the header would claim a build the console is not running", rt.Version) |
| 833 | } |
| 834 | if !strings.Contains(journal.String(), "consoel.service") { |
| 835 | t.Fatalf("journal = %q, want the failed try-restart logged under the unit name", journal.String()) |
| 836 | } |
| 837 | } |
| 838 | |
| 839 | // TestMakeStartRouteStartsOnlyANamedUnit pins the one-arm shape of the |
| 840 | // rollback-start dispatch: only an AWAKE route that NAMES its unit has |
| 841 | // anything for it to start. An exec route, a unit-backed route with no name, |
| 842 | // a host the registry cannot answer for, and any Hibernate-flagged row all |
| 843 | // start NOTHING and return nil — starting carlos-app@<host>, an activator |
| 844 | // child, or a sleeping row's unit from the rollback path could wake a |
| 845 | // sleeper, which no adopt path may do. |
| 846 | func TestMakeStartRouteStartsOnlyANamedUnit(t *testing.T) { |
| 847 | reg := restartTestRegistry(t) |
| 848 | if err := reg.Upsert(registry.Route{ |
| 849 | Host: "console.test", Kind: "instance", App: "console", |
| 850 | Addr: "127.0.0.1:8090", Status: "ready", Unit: "console.service", |
| 851 | }); err != nil { |
| 852 | t.Fatalf("upsert console.test: %v", err) |
| 853 | } |
| 854 | if err := reg.Upsert(registry.Route{ |
| 855 | Host: "exec.test", Kind: "instance", App: "a", Status: "ready", |
| 856 | }); err != nil { |
| 857 | t.Fatalf("upsert exec.test: %v", err) |
| 858 | } |
| 859 | if err := reg.Upsert(registry.Route{ |
| 860 | Host: "unit.test", Kind: "instance", App: "a", Status: "ready", |
| 861 | Backing: registry.BackingUnit, |
| 862 | }); err != nil { |
| 863 | t.Fatalf("upsert unit.test: %v", err) |
| 864 | } |
| 865 | // A hibernating row carrying a unit name is the misconfiguration |
| 866 | // restart.go documents (provision refuses addr-with-hibernate, so no |
| 867 | // legitimate named-unit route sleeps) — and it is exactly the row the |
| 868 | // STRUCTURAL guard exists for: even here, the rollback-start starts |
| 869 | // nothing. |
| 870 | if err := reg.Upsert(registry.Route{ |
| 871 | Host: "sleeper.test", Kind: "instance", App: "a", Status: "ready", |
| 872 | Unit: "sleeper.service", Hibernate: true, |
| 873 | }); err != nil { |
| 874 | t.Fatalf("upsert sleeper.test: %v", err) |
| 875 | } |
| 876 | var units []string |
| 877 | start := makeStartRoute(reg, func(u string) error { units = append(units, u); return nil }) |
| 878 | for _, h := range []string{"console.test", "exec.test", "unit.test", "ghost.test", "sleeper.test"} { |
| 879 | if err := start(h); err != nil { |
| 880 | t.Fatalf("start %s: %v", h, err) |
| 881 | } |
| 882 | } |
| 883 | if len(units) != 1 || units[0] != "console.service" { |
| 884 | t.Fatalf("started %v, want [console.service] alone", units) |
| 885 | } |
| 886 | } |
| 887 | |
| 888 | // TestAdoptPassStartsTheNamedUnitAgainAfterAFailedRestart is the end-to-end |
| 889 | // for the rollback-start, through the real release.Pass, the real root |
| 890 | // dispatch, and the production systemctl path: a named unit whose |
| 891 | // try-restart fails (the new build would not come back) is rolled back to |
| 892 | // the version it was running — and STARTED on it, because a failed |
| 893 | // try-restart leaves the unit down and try-restart is a no-op on a stopped |
| 894 | // unit. Without the start the box keeps a correct symlink over no process, |
| 895 | // and the NEXT pass's exit-0 no-op stamps the never-run build. |
| 896 | func TestAdoptPassStartsTheNamedUnitAgainAfterAFailedRestart(t *testing.T) { |
| 897 | dir := t.TempDir() |
| 898 | optDir := t.TempDir() |
| 899 | s := release.NewDirStore(dir) |
| 900 | ctx := context.Background() |
| 901 | plat := runtime.GOOS + "-" + runtime.GOARCH |
| 902 | |
| 903 | pub, priv, err := ed25519.GenerateKey(nil) |
| 904 | if err != nil { |
| 905 | t.Fatalf("generate key: %v", err) |
| 906 | } |
| 907 | reg := restartTestRegistry(t) |
| 908 | if err := reg.Upsert(registry.Route{ |
| 909 | Host: "console.test", Kind: "instance", App: "console", Channel: "canary/console", |
| 910 | Addr: "127.0.0.1:8090", Status: "ready", Unit: "console.service", |
| 911 | }); err != nil { |
| 912 | t.Fatalf("upsert console.test: %v", err) |
| 913 | } |
| 914 | for _, v := range []string{"old", "new"} { |
| 915 | m := release.Manifest{ |
| 916 | App: "console", Version: v, Kind: "binary", |
| 917 | Artifacts: map[string]release.Artifact{plat: {}}, |
| 918 | } |
| 919 | if err := release.Ship(ctx, s, "bab", m, map[string][]byte{plat: []byte("build " + v)}); err != nil { |
| 920 | t.Fatalf("Ship %s: %v", v, err) |
| 921 | } |
| 922 | } |
| 923 | if err := release.StageArtifact(optDir, "bab", "old", "console", []byte("build old")); err != nil { |
| 924 | t.Fatal(err) |
| 925 | } |
| 926 | if err := release.PointLive(optDir, "console.test", "bab", "old", "console"); err != nil { |
| 927 | t.Fatal(err) |
| 928 | } |
| 929 | if err := release.Promote(ctx, s, priv, "bab", "console", "canary/console", "new"); err != nil { |
| 930 | t.Fatalf("Promote new: %v", err) |
| 931 | } |
| 932 | |
| 933 | // A fake systemctl whose try-restart fails the way a service that will |
| 934 | // not come back does, and whose start succeeds — recording the argv of |
| 935 | // both, so this covers systemctlStartUnit's production exec.Command and |
| 936 | // the ordering, not just the constructors. |
| 937 | bin := t.TempDir() |
| 938 | callLog := filepath.Join(t.TempDir(), "systemctl-calls") |
| 939 | if err := os.WriteFile(filepath.Join(bin, "systemctl"), |
| 940 | []byte("#!/bin/sh\necho \"$@\" >> "+callLog+"\ncase \"$1\" in\ntry-restart) echo \"Job for $2 failed.\"; exit 1;;\nesac\nexit 0\n"), 0o755); err != nil { |
| 941 | t.Fatal(err) |
| 942 | } |
| 943 | t.Setenv("PATH", bin+string(os.PathListSeparator)+os.Getenv("PATH")) |
| 944 | |
| 945 | var journal bytes.Buffer |
| 946 | log.SetOutput(&journal) |
| 947 | t.Cleanup(func() { log.SetOutput(os.Stderr) }) |
| 948 | |
| 949 | if err := release.Pass(ctx, reg, s, pub, optDir, makeRestartRoute(reg, systemctlTryRestart, agentCheckedUnitRestart(systemdUnits{}, optDir, os.Getenv, nil), nil), |
| 950 | release.WithStartUnit(makeStartRoute(reg, systemctlStartUnit))); err != nil { |
| 951 | t.Fatalf("Pass: %v", err) |
| 952 | } |
| 953 | |
| 954 | got, err := os.ReadFile(callLog) |
| 955 | if err != nil { |
| 956 | t.Fatalf("no systemctl calls at all: %v", err) |
| 957 | } |
| 958 | // The checked arm's show comes first (its empty fake answer verifies no |
| 959 | // binding, so the restart falls back to the plain try-restart). |
| 960 | if want := "show console.service -p Id -p FragmentPath -p DropInPaths -p ActiveState -p SubState -p InvocationID -p MainPID -p NRestarts -p ExecStart -p ActiveEnterTimestamp\ntry-restart console.service\nstart console.service\n"; string(got) != want { |
| 961 | t.Fatalf("systemctl calls = %q, want %q — the rolled-back unit must be STARTED (try-restart would no-op on it)", got, want) |
| 962 | } |
| 963 | if v, err := release.LiveVersion(optDir, "console.test"); err != nil || v != "old" { |
| 964 | t.Fatalf("live = %q, %v; want old", v, err) |
| 965 | } |
| 966 | rt, err := reg.Get("console.test") |
| 967 | if err != nil || rt == nil { |
| 968 | t.Fatalf("get console.test: %+v, %v", rt, err) |
| 969 | } |
| 970 | if rt.Version == "new" { |
| 971 | t.Fatal("a build whose restart failed was stamped") |
| 972 | } |
| 973 | } |
| 974 | |
| 975 | // TestAdoptPassBouncesAWarmHibernatingInstance is the end-to-end this task |
| 976 | // exists for, through the real release.Pass with the real closure: an exec |
| 977 | // instance minted the way provision does it (Hibernate: true, Backing "") is |
| 978 | // warm when a new version lands, and it BOUNCES onto that version instead of |
| 979 | // serving the old binary until it happens to hibernate. |
| 980 | // |
| 981 | // Before the RestartHibernating option this was dead code in production: |
| 982 | // release.Pass split on the Hibernate FLAG and handed every flagged route |
| 983 | // restart=nil, so the dispatch never ran for the routes it was written for. |
| 984 | func TestAdoptPassBouncesAWarmHibernatingInstance(t *testing.T) { |
| 985 | t.Parallel() |
| 986 | dir := t.TempDir() |
| 987 | optDir := t.TempDir() |
| 988 | s := release.NewDirStore(dir) |
| 989 | ctx := context.Background() |
| 990 | plat := runtime.GOOS + "-" + runtime.GOARCH |
| 991 | |
| 992 | pub, priv, err := ed25519.GenerateKey(nil) |
| 993 | if err != nil { |
| 994 | t.Fatalf("generate key: %v", err) |
| 995 | } |
| 996 | reg := restartTestRegistry(t) |
| 997 | // warm.test is awake; sleepy.test is not. Both are exec-backed hibernators |
| 998 | // — the shape provision/reconcile.go mints. |
| 999 | for _, h := range []string{"warm.test", "sleepy.test"} { |
| 1000 | if err := reg.Upsert(registry.Route{ |
| 1001 | Host: h, Kind: "instance", App: "warmapp", Channel: "canary/dev", |
| 1002 | Status: "ready", Hibernate: true, |
| 1003 | }); err != nil { |
| 1004 | t.Fatalf("upsert %s: %v", h, err) |
| 1005 | } |
| 1006 | } |
| 1007 | m := release.Manifest{ |
| 1008 | App: "warmapp", Version: "v1", Kind: "binary", |
| 1009 | Artifacts: map[string]release.Artifact{plat: {}}, |
| 1010 | } |
| 1011 | if err := release.Ship(ctx, s, "bab", m, map[string][]byte{plat: []byte("build one")}); err != nil { |
| 1012 | t.Fatalf("Ship: %v", err) |
| 1013 | } |
| 1014 | if err := release.Promote(ctx, s, priv, "bab", "warmapp", "canary/dev", "v1"); err != nil { |
| 1015 | t.Fatalf("Promote: %v", err) |
| 1016 | } |
| 1017 | |
| 1018 | var tried, bounced []string |
| 1019 | // The activator's answer: only warm.test has a running child. This is |
| 1020 | // exactly Service.Bounce's contract — false when the host is not awake, and |
| 1021 | // it never starts anything. |
| 1022 | restart := makeRestartRoute(reg, |
| 1023 | func(h string) { tried = append(tried, h) }, |
| 1024 | unitOnly(func(u string) error { |
| 1025 | t.Fatalf("named-unit arm fired for %q on a route that names no unit", u) |
| 1026 | return nil |
| 1027 | }), |
| 1028 | func(h string) bool { |
| 1029 | bounced = append(bounced, h) |
| 1030 | return h == "warm.test" |
| 1031 | }) |
| 1032 | |
| 1033 | if err := release.Pass(ctx, reg, s, pub, optDir, restart, release.RestartHibernating(restart)); err != nil { |
| 1034 | t.Fatalf("Pass: %v", err) |
| 1035 | } |
| 1036 | |
| 1037 | // The warm instance was cycled onto the new binary. |
| 1038 | if len(bounced) != 2 { |
| 1039 | t.Fatalf("both hibernators must be offered to the callback, got %v", bounced) |
| 1040 | } |
| 1041 | if len(tried) != 1 || tried[0] != "sleepy.test" { |
| 1042 | t.Fatalf("only the host with nothing awake falls through to try-restart: %v", tried) |
| 1043 | } |
| 1044 | // And both adopted: the sleeper gets the repoint and execs the new binary |
| 1045 | // at its natural next wake — it is never started by the pass. |
| 1046 | for _, h := range []string{"warm.test", "sleepy.test"} { |
| 1047 | if v, err := release.LiveVersion(optDir, h); err != nil || v != "v1" { |
| 1048 | t.Fatalf("%s live = %q, %v; want v1", h, v, err) |
| 1049 | } |
| 1050 | } |
| 1051 | } |
| 1052 | |
| 1053 | // ---- fastConvergeLoop: serial poll + fast set (carlos-deploy spec §3-§4) ---- |
| 1054 | |
| 1055 | // fastFixture is the shared shape of the fast-converge tick tests: a dir |
| 1056 | // store standing in for the deployment bucket, a bolt registry, a keypair, |
| 1057 | // and fresh opt/data dirs — internal/release's matFixture adapted to this |
| 1058 | // package's conventions (restartTestRegistry, real release helpers). |
| 1059 | type fastFixture struct { |
| 1060 | store release.Store |
| 1061 | pub ed25519.PublicKey |
| 1062 | priv ed25519.PrivateKey |
| 1063 | reg *registry.Registry |
| 1064 | optDir string |
| 1065 | dataDir string |
| 1066 | } |
| 1067 | |
| 1068 | func newFastFixture(t *testing.T) *fastFixture { |
| 1069 | t.Helper() |
| 1070 | pub, priv, err := ed25519.GenerateKey(nil) |
| 1071 | if err != nil { |
| 1072 | t.Fatalf("generate key: %v", err) |
| 1073 | } |
| 1074 | return &fastFixture{ |
| 1075 | store: release.NewDirStore(t.TempDir()), |
| 1076 | pub: pub, |
| 1077 | priv: priv, |
| 1078 | reg: restartTestRegistry(t), |
| 1079 | optDir: t.TempDir(), |
| 1080 | dataDir: t.TempDir(), |
| 1081 | } |
| 1082 | } |
| 1083 | |
| 1084 | // deps builds one tick's fastDeps with every seam pointed at the fixture. |
| 1085 | // fullPass, when non-nil, replaces the production reconcile→adopt→config-sync |
| 1086 | // set with a counter hook — the serial tests pin WHEN the set runs, not what |
| 1087 | // it does. The pass options come from fastPassOpts, the same builder |
| 1088 | // production uses, so a test exercising the sweep exercises the real option |
| 1089 | // set. |
| 1090 | func (f *fastFixture) deps(fullPass func()) *fastDeps { |
| 1091 | restart := func(string) error { return nil } |
| 1092 | d := &fastDeps{ |
| 1093 | reg: f.reg, |
| 1094 | store: f.store, |
| 1095 | pub: f.pub, |
| 1096 | optDir: f.optDir, |
| 1097 | restart: restart, |
| 1098 | passOpts: fastPassOpts(f.dataDir, func(host, ver string, err error) {}, restart, nil), |
| 1099 | fullPass: func(context.Context) {}, |
| 1100 | } |
| 1101 | if fullPass != nil { |
| 1102 | d.fullPass = func(context.Context) { fullPass() } |
| 1103 | } |
| 1104 | return d |
| 1105 | } |
| 1106 | |
| 1107 | // addRoute mints an adoptable instance route on the given channel — the |
| 1108 | // shape release.Pass sweeps (kind instance, ready, non-platform). |
| 1109 | func (f *fastFixture) addRoute(t *testing.T, host, channel string) { |
| 1110 | t.Helper() |
| 1111 | if err := f.reg.Upsert(registry.Route{ |
| 1112 | Host: host, Kind: "instance", App: "fastapp", Channel: channel, |
| 1113 | Status: "ready", |
| 1114 | }); err != nil { |
| 1115 | t.Fatalf("upsert %s: %v", host, err) |
| 1116 | } |
| 1117 | } |
| 1118 | |
| 1119 | // stageLive stages ver and points host's live symlink at it, so |
| 1120 | // release.LiveVersion answers ver — "this host has nothing outstanding". |
| 1121 | func (f *fastFixture) stageLive(t *testing.T, host, ver string) { |
| 1122 | t.Helper() |
| 1123 | if err := release.StageArtifact(f.optDir, "bab", ver, "fastapp", []byte("staged build")); err != nil { |
| 1124 | t.Fatalf("StageArtifact: %v", err) |
| 1125 | } |
| 1126 | if err := release.PointLive(f.optDir, host, "bab", ver, "fastapp"); err != nil { |
| 1127 | t.Fatalf("PointLive: %v", err) |
| 1128 | } |
| 1129 | } |
| 1130 | |
| 1131 | // mutexProbeStore fails the test if any store call arrives while |
| 1132 | // reconcileMu is NOT held — the deterministic pin that reconcileSerialized |
| 1133 | // really runs provision.Reconcile under the package mutex (TryLock succeeds |
| 1134 | // only when nobody holds it). No goroutines, no sleeps: holding the mutex |
| 1135 | // for the duration of the store work IS the serialization property. |
| 1136 | type mutexProbeStore struct { |
| 1137 | release.Store |
| 1138 | t *testing.T |
| 1139 | hits *int |
| 1140 | } |
| 1141 | |
| 1142 | func (s *mutexProbeStore) probe() { |
| 1143 | *s.hits++ |
| 1144 | if reconcileMu.TryLock() { |
| 1145 | reconcileMu.Unlock() |
| 1146 | s.t.Error("store consulted without reconcileMu held — a concurrent Reconcile pass could interleave and mass-retire fresh routes") |
| 1147 | } |
| 1148 | } |
| 1149 | |
| 1150 | func (s *mutexProbeStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 1151 | s.probe() |
| 1152 | return s.Store.Get(ctx, key) |
| 1153 | } |
| 1154 | |
| 1155 | func (s *mutexProbeStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 1156 | s.probe() |
| 1157 | return s.Store.List(ctx, prefix) |
| 1158 | } |
| 1159 | |
| 1160 | // TestReconcileSerializedHoldsTheMutex pins finding-1's fix: Reconcile's |
| 1161 | // removal phase deletes any Provisioned in-scope route absent from its own |
| 1162 | // top-of-pass record snapshot, so two overlapping passes (instancesLoop's |
| 1163 | // 5-minute tick vs fastConvergeLoop's serial-triggered full set) could |
| 1164 | // delete a route the other pass had just minted — and the re-mint comes |
| 1165 | // back with capability grants revoked. Both loops therefore run Reconcile |
| 1166 | // only through reconcileSerialized, which must hold reconcileMu across the |
| 1167 | // whole pass. |
| 1168 | func TestReconcileSerializedHoldsTheMutex(t *testing.T) { |
| 1169 | var hits int |
| 1170 | s := &mutexProbeStore{Store: release.NewDirStore(t.TempDir()), t: t, hits: &hits} |
| 1171 | reg := restartTestRegistry(t) |
| 1172 | cfg := provision.ReconcileConfig{DataDir: t.TempDir()} |
| 1173 | if err := reconcileSerialized(context.Background(), s, reg, cfg); err != nil { |
| 1174 | t.Fatalf("reconcileSerialized: %v", err) |
| 1175 | } |
| 1176 | if hits == 0 { |
| 1177 | t.Fatal("the probe store was never consulted — this test pinned nothing") |
| 1178 | } |
| 1179 | } |
| 1180 | |
| 1181 | // TestReconcileSerializedRunsBothRecordFamilies pins the wiring: the sidecar |
| 1182 | // pass runs from the SAME serialized entry point as the instance pass, under |
| 1183 | // the same mutex hold. Wired anywhere else it would either race the instance |
| 1184 | // pass over one registry, or (worse) be forgotten by one of the two loops and |
| 1185 | // converge only every five minutes — or never. |
| 1186 | func TestReconcileSerializedRunsBothRecordFamilies(t *testing.T) { |
| 1187 | ctx := context.Background() |
| 1188 | s := release.NewDirStore(t.TempDir()) |
| 1189 | sqid := domains.Sqid(2) |
| 1190 | manifest := []byte(`{"id":2,"sqid":"` + sqid + `","name":"Eleven","created_at":"2026-08-02T09:00:00Z"}`) |
| 1191 | if _, err := s.Put(ctx, "accounts/"+sqid+"/account.json", manifest, ""); err != nil { |
| 1192 | t.Fatal(err) |
| 1193 | } |
| 1194 | if err := provision.PutSidecarManifest(ctx, s, sqid, "eleven", provision.SidecarManifest{ |
| 1195 | Schema: 1, App: "eleven", Args: []string{"sidecar", "run"}, Channel: "stable"}); err != nil { |
| 1196 | t.Fatal(err) |
| 1197 | } |
| 1198 | host := provision.SidecarHost("bot", "eleven", sqid) |
| 1199 | if err := provision.CreateSidecarRecord(ctx, s, provision.SidecarRecord{ |
| 1200 | Schema: 1, Name: "bot", Host: host, App: "eleven", Account: sqid, Channel: "stable"}); err != nil { |
| 1201 | t.Fatal(err) |
| 1202 | } |
| 1203 | |
| 1204 | reg := restartTestRegistry(t) |
| 1205 | if err := reconcileSerialized(ctx, s, reg, provision.ReconcileConfig{DataDir: t.TempDir()}); err != nil { |
| 1206 | t.Fatalf("reconcileSerialized: %v", err) |
| 1207 | } |
| 1208 | |
| 1209 | rt, err := reg.Get(host) |
| 1210 | if err != nil { |
| 1211 | t.Fatal(err) |
| 1212 | } |
| 1213 | if rt == nil { |
| 1214 | t.Fatal("no sidecar route minted — the sidecar pass must run from reconcileSerialized, beside the instance pass and inside its mutex") |
| 1215 | } |
| 1216 | } |
| 1217 | |
| 1218 | // TestFastTickActsOnSerialChangeNotBoot pins seed-don't-replay, in both of |
| 1219 | // its halves. An ABSENT serial object seeds at zero — absence is not |
| 1220 | // history, there is nothing to replay, and every production bucket starts |
| 1221 | // without the object, so a deployment's FIRST-EVER bump (0 → 1) must fire |
| 1222 | // rather than being swallowed as boot noise. A PRESENT object seeds at its |
| 1223 | // current value without replaying it (boot must not replay history — |
| 1224 | // adoptLoop's 1-minute first pass covers boot). After either seed, a CHANGE |
| 1225 | // runs the full pass set exactly once and an unchanged serial runs nothing. |
| 1226 | func TestFastTickActsOnSerialChangeNotBoot(t *testing.T) { |
| 1227 | t.Parallel() |
| 1228 | env := newFastFixture(t) |
| 1229 | var passes int |
| 1230 | d := env.deps(func() { passes++ }) |
| 1231 | now := time.Unix(1754640000, 0) |
| 1232 | fastTick(context.Background(), d, now) // no serial object at all: seeds at zero, runs nothing |
| 1233 | if passes != 0 { |
| 1234 | t.Fatalf("an absent serial object must not run a pass, got %d", passes) |
| 1235 | } |
| 1236 | if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| 1237 | t.Fatalf("BumpSerial: %v", err) |
| 1238 | } |
| 1239 | fastTick(context.Background(), d, now.Add(2*time.Second)) // the deployment's first-ever bump |
| 1240 | if passes != 1 { |
| 1241 | t.Fatalf("the first-ever bump must fire (absence seeds at zero), got %d", passes) |
| 1242 | } |
| 1243 | fastTick(context.Background(), d, now.Add(10*time.Second)) // past clamp, unchanged serial |
| 1244 | if passes != 1 { |
| 1245 | t.Fatalf("unchanged serial must not re-run, got %d", passes) |
| 1246 | } |
| 1247 | |
| 1248 | // The boot rule for a PRESENT object: a restarted box (fresh state) |
| 1249 | // seeds at the current value and must not replay it as a change. |
| 1250 | var reboots int |
| 1251 | d2 := env.deps(func() { reboots++ }) |
| 1252 | fastTick(context.Background(), d2, now.Add(12*time.Second)) // first sight of serial=1: SEED only |
| 1253 | if reboots != 0 { |
| 1254 | t.Fatalf("boot seed of a present object must not run a pass, got %d", reboots) |
| 1255 | } |
| 1256 | if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| 1257 | t.Fatalf("BumpSerial: %v", err) |
| 1258 | } |
| 1259 | fastTick(context.Background(), d2, now.Add(14*time.Second)) |
| 1260 | if reboots != 1 { |
| 1261 | t.Fatalf("post-boot serial change must run the pass set once, got %d", reboots) |
| 1262 | } |
| 1263 | } |
| 1264 | |
| 1265 | // TestFastTickClampsFullPasses pins the 5s clamp: a bump landing inside the |
| 1266 | // window is DEFERRED (the acted serial is not recorded), and the next tick |
| 1267 | // past the window fires it — a burst of console writes costs one pass set |
| 1268 | // per 5s, never one per write, and the last write is never lost. |
| 1269 | func TestFastTickClampsFullPasses(t *testing.T) { |
| 1270 | env := newFastFixture(t) |
| 1271 | var passes int |
| 1272 | d := env.deps(func() { passes++ }) |
| 1273 | now := time.Unix(1754640000, 0) |
| 1274 | if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| 1275 | t.Fatalf("BumpSerial: %v", err) |
| 1276 | } |
| 1277 | fastTick(context.Background(), d, now) // seed |
| 1278 | if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| 1279 | t.Fatalf("BumpSerial: %v", err) |
| 1280 | } |
| 1281 | fastTick(context.Background(), d, now.Add(2*time.Second)) // pass 1 |
| 1282 | if err := release.BumpSerial(context.Background(), env.store, now); err != nil { |
| 1283 | t.Fatalf("BumpSerial: %v", err) |
| 1284 | } |
| 1285 | fastTick(context.Background(), d, now.Add(4*time.Second)) // inside 5s clamp: deferred |
| 1286 | if passes != 1 { |
| 1287 | t.Fatalf("clamp: got %d passes", passes) |
| 1288 | } |
| 1289 | fastTick(context.Background(), d, now.Add(8*time.Second)) // clamp expired: fires |
| 1290 | if passes != 2 { |
| 1291 | t.Fatalf("deferred bump must fire after clamp, got %d", passes) |
| 1292 | } |
| 1293 | } |
| 1294 | |
| 1295 | // TestFastSweepMembership pins who the fast sweep is FOR: canary/* routes |
| 1296 | // always (a direct-bucket operator promote never bumps the serial), unstaged |
| 1297 | // routes until they stage, staged non-canary routes never. |
| 1298 | func TestFastSweepMembership(t *testing.T) { |
| 1299 | env := newFastFixture(t) |
| 1300 | // three routes: a canary (always member), a staged edge route (not a |
| 1301 | // member), an unstaged edge route (member until staged, then drops out). |
| 1302 | env.addRoute(t, "canary.test", "canary/x") |
| 1303 | env.addRoute(t, "staged.test", "edge") |
| 1304 | env.stageLive(t, "staged.test", "abc1234") // mint its live symlink |
| 1305 | env.addRoute(t, "cold.test", "edge") |
| 1306 | members := env.deps(nil).fastMembers(time.Unix(1754640000, 0)) |
| 1307 | sort.Strings(members) |
| 1308 | if want := []string{"canary.test", "cold.test"}; !slices.Equal(members, want) { |
| 1309 | t.Fatalf("members = %v, want %v", members, want) |
| 1310 | } |
| 1311 | } |
| 1312 | |
| 1313 | // TestFastSweepBackoffDoubles pins the per-host backoff that keeps a |
| 1314 | // permanently-unstageable route (nothing promoted, or a held head) from |
| 1315 | // costing store reads every 2s forever: 2s doubling to a 5-minute cap. |
| 1316 | func TestFastSweepBackoffDoubles(t *testing.T) { |
| 1317 | env := newFastFixture(t) |
| 1318 | env.addRoute(t, "broken.test", "edge") // nothing promoted: staging can never succeed |
| 1319 | d := env.deps(nil) |
| 1320 | now := time.Unix(1754640000, 0) |
| 1321 | d.noteUnstaged("broken.test", now) |
| 1322 | if m := d.fastMembers(now.Add(1 * time.Second)); len(m) != 0 { |
| 1323 | t.Fatalf("inside backoff window, want no members, got %v", m) |
| 1324 | } |
| 1325 | if m := d.fastMembers(now.Add(3 * time.Second)); len(m) != 1 { |
| 1326 | t.Fatalf("past first backoff, want member, got %v", m) |
| 1327 | } |
| 1328 | for i := 0; i < 20; i++ { // drive to the cap |
| 1329 | d.noteUnstaged("broken.test", now) |
| 1330 | } |
| 1331 | if d.backoffFor("broken.test") != 5*time.Minute { |
| 1332 | t.Fatalf("backoff cap = %v, want 5m", d.backoffFor("broken.test")) |
| 1333 | } |
| 1334 | } |
| 1335 | |
| 1336 | // TestFastSweepBouncesAWarmHibernatingCanary pins, through the real option |
| 1337 | // builder (fastPassOpts) and the real closure (makeRestartRoute), that the |
| 1338 | // fast sweep's pass carries BOTH the positional restart and |
| 1339 | // release.RestartHibernating — dropping the option would silently restore |
| 1340 | // the dead-branch bug the option exists for (a warm exec hibernator keeps |
| 1341 | // serving its old binary until it happens to sleep) and no other test would |
| 1342 | // fail. warm.test (Hibernate: true) is only reachable via the option; |
| 1343 | // plain.test (Hibernate: false) is only reachable via the positional |
| 1344 | // restart; both must bounce. |
| 1345 | func TestFastSweepBouncesAWarmHibernatingCanary(t *testing.T) { |
| 1346 | t.Parallel() |
| 1347 | env := newFastFixture(t) |
| 1348 | ctx := context.Background() |
| 1349 | plat := runtime.GOOS + "-" + runtime.GOARCH |
| 1350 | |
| 1351 | m := release.Manifest{ |
| 1352 | App: "fastapp", Version: "v1", Kind: "binary", |
| 1353 | Artifacts: map[string]release.Artifact{plat: {}}, |
| 1354 | } |
| 1355 | if err := release.Ship(ctx, env.store, "bab", m, map[string][]byte{plat: []byte("fast build")}); err != nil { |
| 1356 | t.Fatalf("Ship: %v", err) |
| 1357 | } |
| 1358 | if err := release.Promote(ctx, env.store, env.priv, "bab", "fastapp", "canary/dev", "v1"); err != nil { |
| 1359 | t.Fatalf("Promote: %v", err) |
| 1360 | } |
| 1361 | if err := env.reg.Upsert(registry.Route{ |
| 1362 | Host: "warm.test", Kind: "instance", App: "fastapp", Channel: "canary/dev", |
| 1363 | Status: "ready", Hibernate: true, |
| 1364 | }); err != nil { |
| 1365 | t.Fatalf("upsert warm.test: %v", err) |
| 1366 | } |
| 1367 | env.addRoute(t, "plain.test", "canary/dev") |
| 1368 | |
| 1369 | var bounced []string |
| 1370 | restart := makeRestartRoute(env.reg, |
| 1371 | func(string) {}, |
| 1372 | unitOnly(func(u string) error { |
| 1373 | t.Fatalf("named-unit arm fired for %q on a route that names no unit", u) |
| 1374 | return nil |
| 1375 | }), |
| 1376 | func(h string) bool { bounced = append(bounced, h); return true }) |
| 1377 | d := env.deps(nil) |
| 1378 | d.restart = restart |
| 1379 | d.passOpts = fastPassOpts(env.dataDir, func(host, ver string, err error) {}, restart, nil) |
| 1380 | |
| 1381 | fastTick(ctx, d, time.Unix(1754640000, 0)) |
| 1382 | |
| 1383 | sort.Strings(bounced) |
| 1384 | if want := []string{"plain.test", "warm.test"}; !slices.Equal(bounced, want) { |
| 1385 | t.Fatalf("bounced = %v, want %v (warm.test missing = RestartHibernating dropped; plain.test missing = positional restart dropped)", bounced, want) |
| 1386 | } |
| 1387 | for _, h := range []string{"warm.test", "plain.test"} { |
| 1388 | if v, err := release.LiveVersion(env.optDir, h); err != nil || v != "v1" { |
| 1389 | t.Fatalf("%s live = %q, %v; want v1", h, v, err) |
| 1390 | } |
| 1391 | } |
| 1392 | } |
| 1393 | |
| 1394 | // TestRestartHoldGrowsCapsAndResetsOnANewVersion pins the memo arithmetic |
| 1395 | // behind the flap bound: same version failing again doubles the hold (to a |
| 1396 | // cap), a different version starts fresh — a recovery promote is new work, |
| 1397 | // not a continuation of the old failure. |
| 1398 | func TestRestartHoldGrowsCapsAndResetsOnANewVersion(t *testing.T) { |
| 1399 | d := &fastDeps{} |
| 1400 | t0 := time.Now() |
| 1401 | d.noteRestartFailed("h", "v2", t0) |
| 1402 | if !d.restartHeld("h", t0.Add(fastRestartBackoffBase-time.Second)) { |
| 1403 | t.Fatal("first failure must hold for the base step") |
| 1404 | } |
| 1405 | if d.restartHeld("h", t0.Add(fastRestartBackoffBase+time.Second)) { |
| 1406 | t.Fatal("the hold must expire after its step") |
| 1407 | } |
| 1408 | d.noteRestartFailed("h", "v2", t0) |
| 1409 | if !d.restartHeld("h", t0.Add(2*fastRestartBackoffBase-time.Second)) { |
| 1410 | t.Fatal("a second failure on the SAME version must double the step") |
| 1411 | } |
| 1412 | for i := 0; i < 12; i++ { |
| 1413 | d.noteRestartFailed("h", "v2", t0) |
| 1414 | } |
| 1415 | if d.restartHeld("h", t0.Add(fastRestartBackoffCap+time.Second)) { |
| 1416 | t.Fatalf("the step must cap at %v", fastRestartBackoffCap) |
| 1417 | } |
| 1418 | d.noteRestartFailed("h", "v3", t0) |
| 1419 | if d.restartHeld("h", t0.Add(fastRestartBackoffBase+time.Second)) { |
| 1420 | t.Fatal("a DIFFERENT version must reset the hold to the base step") |
| 1421 | } |
| 1422 | } |
| 1423 | |
| 1424 | // TestFastSweepBoundsARestartFlap is the incident this bound exists for: a |
| 1425 | // canary head whose restart persistently fails, on a box whose fast loop can |
| 1426 | // actually reach the restart. Before the rollback-start fix that flap |
| 1427 | // self-terminated after two ticks — by stamping the lie (try-restart exits 0 |
| 1428 | // on the unit the failed restart left down). With the fix, each sweep pass |
| 1429 | // is a REAL cycle of the running old service: repoint, restart (kills it), |
| 1430 | // fail, roll back, start it again. Unbounded, that is a service cycle every |
| 1431 | // 2 seconds for as long as the broken head stands. The hold bounds it: the |
| 1432 | // host leaves the sweep for a growing window, a post-hold retry is one cycle |
| 1433 | // (not one per tick), and a restart that finally succeeds clears the hold at |
| 1434 | // once. |
| 1435 | func TestFastSweepBoundsARestartFlap(t *testing.T) { |
| 1436 | t.Parallel() |
| 1437 | env := newFastFixture(t) |
| 1438 | ctx := context.Background() |
| 1439 | plat := runtime.GOOS + "-" + runtime.GOARCH |
| 1440 | |
| 1441 | for _, v := range []string{"old", "new"} { |
| 1442 | m := release.Manifest{ |
| 1443 | App: "fastapp", Version: v, Kind: "binary", |
| 1444 | Artifacts: map[string]release.Artifact{plat: {}}, |
| 1445 | } |
| 1446 | if err := release.Ship(ctx, env.store, "bab", m, map[string][]byte{plat: []byte("build " + v)}); err != nil { |
| 1447 | t.Fatalf("Ship %s: %v", v, err) |
| 1448 | } |
| 1449 | } |
| 1450 | env.addRoute(t, "flap.test", "canary/dev") |
| 1451 | env.stageLive(t, "flap.test", "old") |
| 1452 | if err := release.Promote(ctx, env.store, env.priv, "bab", "fastapp", "canary/dev", "new"); err != nil { |
| 1453 | t.Fatalf("Promote: %v", err) |
| 1454 | } |
| 1455 | |
| 1456 | calls := 0 |
| 1457 | fail := true |
| 1458 | base := func(string) error { |
| 1459 | calls++ |
| 1460 | if fail { |
| 1461 | return errors.New("unit would not come back") |
| 1462 | } |
| 1463 | return nil |
| 1464 | } |
| 1465 | d := env.deps(nil) |
| 1466 | restart := d.observeRestart(base) |
| 1467 | d.restart = restart |
| 1468 | d.passOpts = fastPassOpts(env.dataDir, func(host, ver string, err error) {}, restart, nil) |
| 1469 | |
| 1470 | // Tick 1: the sweep attempts the head, the restart fails, the pass rolls |
| 1471 | // back — and the failure is memo'd. |
| 1472 | t0 := time.Now() |
| 1473 | fastTick(ctx, d, t0) |
| 1474 | if calls != 1 { |
| 1475 | t.Fatalf("tick 1: restart calls = %d, want 1", calls) |
| 1476 | } |
| 1477 | if v, _ := release.LiveVersion(env.optDir, "flap.test"); v != "old" { |
| 1478 | t.Fatalf("live after the failed tick = %q, want old (rolled back)", v) |
| 1479 | } |
| 1480 | |
| 1481 | // Ticks inside the hold: the host is outside the sweep, the service is |
| 1482 | // NOT cycled again. Without the hold every one of these is a real kill. |
| 1483 | for i := 1; i <= 5; i++ { |
| 1484 | fastTick(ctx, d, t0.Add(time.Duration(i)*fastTickEvery)) |
| 1485 | } |
| 1486 | if calls != 1 { |
| 1487 | t.Fatalf("held ticks: restart calls = %d, want still 1 — the sweep must not cycle a failing service every tick", calls) |
| 1488 | } |
| 1489 | |
| 1490 | // Past the hold: exactly one more attempt, which fails again and doubles |
| 1491 | // the step. |
| 1492 | fastTick(ctx, d, t0.Add(fastRestartBackoffBase+2*time.Second)) |
| 1493 | if calls != 2 { |
| 1494 | t.Fatalf("post-hold tick: restart calls = %d, want 2", calls) |
| 1495 | } |
| 1496 | |
| 1497 | // The head is fixed now (a recovery promote, say): the next allowed |
| 1498 | // retry succeeds, adopts, and clears the hold at once. |
| 1499 | fail = false |
| 1500 | fastTick(ctx, d, t0.Add(3*fastRestartBackoffBase+4*time.Second)) |
| 1501 | if calls != 3 { |
| 1502 | t.Fatalf("recovery tick: restart calls = %d, want 3", calls) |
| 1503 | } |
| 1504 | if v, _ := release.LiveVersion(env.optDir, "flap.test"); v != "new" { |
| 1505 | t.Fatalf("live after recovery = %q, want new", v) |
| 1506 | } |
| 1507 | if d.restartHeld("flap.test", t0.Add(24*time.Hour)) { |
| 1508 | t.Fatal("a SUCCESSFUL restart must clear the hold") |
| 1509 | } |
| 1510 | } |
| 1511 | |
| 1512 | // ---- store selection: bucket, dir, or the fleet store door (spec F5) ------- |
| 1513 | |
| 1514 | // doorHit is one request the fake door saw, reduced to the parts these tests |
| 1515 | // reason about: which call it was, and the two headers that prove WHICH box |
| 1516 | // made it (fleetapi sends the tunnel token and channel name on every request). |
| 1517 | type doorHit struct{ Method, Path, Auth, Name string } |
| 1518 | |
| 1519 | // doorFake stands in for the edge's store door. It answers the single GET |
| 1520 | // these tests drive and records every request, so a test can assert both |
| 1521 | // halves of the F5 rule: that a bucketless fleet box's store really does |
| 1522 | // speak to the door under its channel identity, and — the regression pin that |
| 1523 | // matters more — that a bucket-configured box never sends it anything. |
| 1524 | type doorFake struct { |
| 1525 | url string |
| 1526 | mu sync.Mutex |
| 1527 | hits []doorHit |
| 1528 | } |
| 1529 | |
| 1530 | func newDoorFake(t *testing.T) *doorFake { |
| 1531 | t.Helper() |
| 1532 | d := &doorFake{} |
| 1533 | srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 1534 | d.mu.Lock() |
| 1535 | d.hits = append(d.hits, doorHit{r.Method, r.URL.Path, r.Header.Get("Authorization"), r.Header.Get("Carlos-Tunnel-Name")}) |
| 1536 | d.mu.Unlock() |
| 1537 | if r.Method == http.MethodGet && r.URL.Path == "/store/o/control/serial" { |
| 1538 | w.Header().Set("ETag", `"e1"`) |
| 1539 | _, _ = w.Write([]byte("42")) |
| 1540 | return |
| 1541 | } |
| 1542 | http.NotFound(w, r) |
| 1543 | })) |
| 1544 | t.Cleanup(srv.Close) |
| 1545 | d.url = srv.URL |
| 1546 | return d |
| 1547 | } |
| 1548 | |
| 1549 | func (d *doorFake) seen() []doorHit { |
| 1550 | d.mu.Lock() |
| 1551 | defer d.mu.Unlock() |
| 1552 | return append([]doorHit(nil), d.hits...) |
| 1553 | } |
| 1554 | |
| 1555 | // fleetDoorEnv is the env a fleet box that has been CUT OVER carries: fleet |
| 1556 | // identity plus tunnel address/credential, and deliberately no bucket and no |
| 1557 | // AWS credentials at all — the whole point of the door. |
| 1558 | func fleetDoorEnv(doorURL string) map[string]string { |
| 1559 | return map[string]string{ |
| 1560 | "CARLOS_FLEET": "pi", |
| 1561 | "CARLOS_FLEET_BOX": "pi-box", |
| 1562 | "CARLOS_TUNNEL_URL": doorURL, |
| 1563 | "CARLOS_TUNNEL_TOKEN": "tok-123", |
| 1564 | } |
| 1565 | } |
| 1566 | |
| 1567 | // awsEnvForS3Store makes release.NewS3Store constructible in a test without |
| 1568 | // network or a developer's real profile — the same shape internal/release's |
| 1569 | // own S3 tests use. Nothing is ever called on the resulting client here; these |
| 1570 | // tests only assert which store TYPE the selection built. |
| 1571 | func awsEnvForS3Store(t *testing.T) { |
| 1572 | t.Helper() |
| 1573 | t.Setenv("AWS_ACCESS_KEY_ID", "k") |
| 1574 | t.Setenv("AWS_SECRET_ACCESS_KEY", "s") |
| 1575 | t.Setenv("AWS_SESSION_TOKEN", "") |
| 1576 | t.Setenv("AWS_REGION", "eu-west-1") |
| 1577 | t.Setenv("AWS_EC2_METADATA_DISABLED", "true") |
| 1578 | } |
| 1579 | |
| 1580 | // TestOpenDeployStoreSelection pins spec F5's whole rule for the store every |
| 1581 | // converge loop on the box reads through. The migration fallback is the |
| 1582 | // load-bearing half: a bucket or dir still set must win OUTRIGHT, before fleet |
| 1583 | // identity is even consulted, or rolling this branch out would silently move |
| 1584 | // every already-working fleet box onto a door it may not be able to reach. |
| 1585 | func TestOpenDeployStoreSelection(t *testing.T) { |
| 1586 | ctx := context.Background() |
| 1587 | dir := t.TempDir() |
| 1588 | |
| 1589 | t.Run("fleet box with no bucket speaks the door", func(t *testing.T) { |
| 1590 | door := newDoorFake(t) |
| 1591 | s, err := openDeployStoreEnv(ctx, envFunc(fleetDoorEnv(door.url))) |
| 1592 | if err != nil { |
| 1593 | t.Fatalf("openDeployStoreEnv: %v", err) |
| 1594 | } |
| 1595 | body, etag, err := s.Get(ctx, "control/serial") |
| 1596 | if err != nil { |
| 1597 | t.Fatalf("Get through the door: %v", err) |
| 1598 | } |
| 1599 | if string(body) != "42" || etag != `"e1"` { |
| 1600 | t.Fatalf("Get = %q, %q; want \"42\", \"e1\"", body, etag) |
| 1601 | } |
| 1602 | hits := door.seen() |
| 1603 | if len(hits) != 1 { |
| 1604 | t.Fatalf("door saw %d requests, want 1: %+v", len(hits), hits) |
| 1605 | } |
| 1606 | if hits[0].Path != "/store/o/control/serial" { |
| 1607 | t.Fatalf("door path = %q, want /store/o/control/serial", hits[0].Path) |
| 1608 | } |
| 1609 | // The identity on the wire is the channel identity the edge dials — |
| 1610 | // "<fleet>/<label>", never the hashed status pseudonym. |
| 1611 | if hits[0].Name != "pi/pi-box" { |
| 1612 | t.Fatalf("Carlos-Tunnel-Name = %q, want pi/pi-box", hits[0].Name) |
| 1613 | } |
| 1614 | if hits[0].Auth != "Bearer tok-123" { |
| 1615 | t.Fatalf("Authorization = %q, want the tunnel token", hits[0].Auth) |
| 1616 | } |
| 1617 | }) |
| 1618 | |
| 1619 | t.Run("dir set wins on a fleet box, door untouched", func(t *testing.T) { |
| 1620 | door := newDoorFake(t) |
| 1621 | env := fleetDoorEnv(door.url) |
| 1622 | env["CARLOS_DEPLOYMENT_DIR"] = dir |
| 1623 | s, err := openDeployStoreEnv(ctx, envFunc(env)) |
| 1624 | if err != nil { |
| 1625 | t.Fatalf("openDeployStoreEnv: %v", err) |
| 1626 | } |
| 1627 | if _, ok := s.(*release.DirStore); !ok { |
| 1628 | t.Fatalf("store = %T, want *release.DirStore — a configured dir must win", s) |
| 1629 | } |
| 1630 | if _, _, err := s.Get(ctx, "control/serial"); !errors.Is(err, os.ErrNotExist) { |
| 1631 | t.Fatalf("Get = %v, want the dir store's not-exist", err) |
| 1632 | } |
| 1633 | if hits := door.seen(); len(hits) != 0 { |
| 1634 | t.Fatalf("door saw %+v, want zero requests — a dir-configured box must never reach it", hits) |
| 1635 | } |
| 1636 | }) |
| 1637 | |
| 1638 | t.Run("bucket set wins on a fleet box, door untouched", func(t *testing.T) { |
| 1639 | awsEnvForS3Store(t) |
| 1640 | door := newDoorFake(t) |
| 1641 | env := fleetDoorEnv(door.url) |
| 1642 | env["CARLOS_DEPLOYMENT_BUCKET"] = "carlos-deploy" |
| 1643 | s, err := openDeployStoreEnv(ctx, envFunc(env)) |
| 1644 | if err != nil { |
| 1645 | t.Fatalf("openDeployStoreEnv: %v", err) |
| 1646 | } |
| 1647 | if _, ok := s.(*release.S3Store); !ok { |
| 1648 | t.Fatalf("store = %T, want *release.S3Store — a configured bucket must win", s) |
| 1649 | } |
| 1650 | if hits := door.seen(); len(hits) != 0 { |
| 1651 | t.Fatalf("door saw %+v, want zero requests — a bucket-configured box must never reach it", hits) |
| 1652 | } |
| 1653 | }) |
| 1654 | |
| 1655 | t.Run("platform box is unchanged", func(t *testing.T) { |
| 1656 | s, err := openDeployStoreEnv(ctx, envFunc(map[string]string{"CARLOS_DEPLOYMENT_DIR": dir})) |
| 1657 | if err != nil { |
| 1658 | t.Fatalf("openDeployStoreEnv: %v", err) |
| 1659 | } |
| 1660 | if _, ok := s.(*release.DirStore); !ok { |
| 1661 | t.Fatalf("store = %T, want *release.DirStore", s) |
| 1662 | } |
| 1663 | _, err = openDeployStoreEnv(ctx, envFunc(map[string]string{})) |
| 1664 | if err == nil || !strings.Contains(err.Error(), "neither CARLOS_DEPLOYMENT_BUCKET nor CARLOS_DEPLOYMENT_DIR is set") { |
| 1665 | t.Fatalf("unconfigured platform box err = %v, want today's message verbatim", err) |
| 1666 | } |
| 1667 | }) |
| 1668 | |
| 1669 | t.Run("fleet box with no bucket and no tunnel env fails loudly", func(t *testing.T) { |
| 1670 | for _, missing := range []string{"CARLOS_TUNNEL_URL", "CARLOS_TUNNEL_TOKEN"} { |
| 1671 | env := fleetDoorEnv("https://tunnel.example.com") |
| 1672 | delete(env, missing) |
| 1673 | _, err := openDeployStoreEnv(ctx, envFunc(env)) |
| 1674 | if err == nil { |
| 1675 | t.Fatalf("missing %s: want an error, got a store", missing) |
| 1676 | } |
| 1677 | if !strings.Contains(err.Error(), "pi/pi-box") || !strings.Contains(err.Error(), missing) { |
| 1678 | t.Fatalf("missing %s: err = %v, want it to name the box and the variable", missing, err) |
| 1679 | } |
| 1680 | // The token must never be quoted back into an error a log line or |
| 1681 | // a status page could carry. |
| 1682 | if strings.Contains(err.Error(), "tok-123") { |
| 1683 | t.Fatalf("err leaks the tunnel token: %v", err) |
| 1684 | } |
| 1685 | } |
| 1686 | }) |
| 1687 | |
| 1688 | t.Run("unusable fleet identity with no bucket names the identity", func(t *testing.T) { |
| 1689 | _, err := openDeployStoreEnv(ctx, envFunc(map[string]string{ |
| 1690 | "CARLOS_FLEET": "pi", |
| 1691 | "CARLOS_TUNNEL_NAME": "legacy-box", // no "pi/" prefix: underivable label |
| 1692 | "CARLOS_TUNNEL_URL": "https://tunnel.example.com", |
| 1693 | "CARLOS_TUNNEL_TOKEN": "tok-123", |
| 1694 | })) |
| 1695 | if err == nil || !strings.Contains(err.Error(), "CARLOS_TUNNEL_NAME") { |
| 1696 | t.Fatalf("err = %v, want the identity error, not the generic no-bucket line", err) |
| 1697 | } |
| 1698 | }) |
| 1699 | } |
| 1700 | |
| 1701 | // TestOpenStatusStoreSelection pins the same rule for the status/metrics |
| 1702 | // store, including the ledger call it makes deliberately: the door serves the |
| 1703 | // DEPLOYMENT store, so a fleet box that names its own CARLOS_STATUS_BUCKET |
| 1704 | // keeps direct S3 for status — bucket-set-wins, on purpose. |
| 1705 | func TestOpenStatusStoreSelection(t *testing.T) { |
| 1706 | ctx := context.Background() |
| 1707 | |
| 1708 | t.Run("fleet box with no bucket speaks the door", func(t *testing.T) { |
| 1709 | door := newDoorFake(t) |
| 1710 | s, target, err := openStatusStore(ctx, envFunc(fleetDoorEnv(door.url))) |
| 1711 | if err != nil { |
| 1712 | t.Fatalf("openStatusStore: %v", err) |
| 1713 | } |
| 1714 | if _, _, err := s.Get(ctx, "control/serial"); err != nil { |
| 1715 | t.Fatalf("Get through the door: %v", err) |
| 1716 | } |
| 1717 | if hits := door.seen(); len(hits) != 1 || hits[0].Name != "pi/pi-box" { |
| 1718 | t.Fatalf("door hits = %+v, want one request as pi/pi-box", hits) |
| 1719 | } |
| 1720 | if !strings.Contains(target, "pi/pi-box") { |
| 1721 | t.Fatalf("armed-log target = %q, want it to name the door identity", target) |
| 1722 | } |
| 1723 | }) |
| 1724 | |
| 1725 | t.Run("status bucket set on a fleet box keeps direct S3", func(t *testing.T) { |
| 1726 | awsEnvForS3Store(t) |
| 1727 | door := newDoorFake(t) |
| 1728 | env := fleetDoorEnv(door.url) |
| 1729 | env["CARLOS_STATUS_BUCKET"] = "carlos-status" |
| 1730 | s, target, err := openStatusStore(ctx, envFunc(env)) |
| 1731 | if err != nil { |
| 1732 | t.Fatalf("openStatusStore: %v", err) |
| 1733 | } |
| 1734 | if _, ok := s.(*release.S3Store); !ok { |
| 1735 | t.Fatalf("store = %T, want *release.S3Store", s) |
| 1736 | } |
| 1737 | if target != "bucket carlos-status" { |
| 1738 | t.Fatalf("target = %q, want %q", target, "bucket carlos-status") |
| 1739 | } |
| 1740 | if hits := door.seen(); len(hits) != 0 { |
| 1741 | t.Fatalf("door saw %+v, want zero requests", hits) |
| 1742 | } |
| 1743 | }) |
| 1744 | |
| 1745 | t.Run("deployment bucket fallback still applies", func(t *testing.T) { |
| 1746 | awsEnvForS3Store(t) |
| 1747 | door := newDoorFake(t) |
| 1748 | env := fleetDoorEnv(door.url) |
| 1749 | env["CARLOS_DEPLOYMENT_BUCKET"] = "carlos-deploy" |
| 1750 | s, target, err := openStatusStore(ctx, envFunc(env)) |
| 1751 | if err != nil { |
| 1752 | t.Fatalf("openStatusStore: %v", err) |
| 1753 | } |
| 1754 | if _, ok := s.(*release.S3Store); !ok { |
| 1755 | t.Fatalf("store = %T, want *release.S3Store", s) |
| 1756 | } |
| 1757 | if target != "bucket carlos-deploy" { |
| 1758 | t.Fatalf("target = %q, want %q", target, "bucket carlos-deploy") |
| 1759 | } |
| 1760 | if hits := door.seen(); len(hits) != 0 { |
| 1761 | t.Fatalf("door saw %+v, want zero requests", hits) |
| 1762 | } |
| 1763 | }) |
| 1764 | |
| 1765 | t.Run("platform box with no bucket stays unconfigured", func(t *testing.T) { |
| 1766 | _, _, err := openStatusStore(ctx, envFunc(map[string]string{"CARLOS_DEPLOYMENT_DIR": t.TempDir()})) |
| 1767 | if !errors.Is(err, errNoStatusStore) { |
| 1768 | t.Fatalf("err = %v, want errNoStatusStore — a DIR deployment publishes no status", err) |
| 1769 | } |
| 1770 | }) |
| 1771 | |
| 1772 | t.Run("fleet box with no bucket and no tunnel env fails loudly", func(t *testing.T) { |
| 1773 | env := fleetDoorEnv("https://tunnel.example.com") |
| 1774 | delete(env, "CARLOS_TUNNEL_TOKEN") |
| 1775 | _, _, err := openStatusStore(ctx, envFunc(env)) |
| 1776 | if err == nil || errors.Is(err, errNoStatusStore) { |
| 1777 | t.Fatalf("err = %v, want a loud construction error, not the unconfigured sentinel", err) |
| 1778 | } |
| 1779 | }) |
| 1780 | } |
| 1781 | |
| 1782 | // TestResolveStatusPrefixDefaultsOnTheDoorBranch pins the fix for a fleet box |
| 1783 | // that would otherwise 404 forever: through the door, the write ACL admits |
| 1784 | // only fleets/<own fleet>/**, so a door-backed box with an EMPTY |
| 1785 | // CARLOS_STATUS_PREFIX must rebase itself rather than write at the shared |
| 1786 | // platform root. Bucket-backed boxes keep today's semantics exactly — an |
| 1787 | // empty prefix there IS the correct shared root. |
| 1788 | func TestResolveStatusPrefixDefaultsOnTheDoorBranch(t *testing.T) { |
| 1789 | for _, c := range []struct { |
| 1790 | name string |
| 1791 | env map[string]string |
| 1792 | want string |
| 1793 | }{ |
| 1794 | {"platform box, no prefix", map[string]string{"CARLOS_STATUS_BUCKET": "b"}, ""}, |
| 1795 | {"platform box, explicit prefix", map[string]string{"CARLOS_STATUS_BUCKET": "b", "CARLOS_STATUS_PREFIX": "x/"}, "x/"}, |
| 1796 | {"bucket-backed fleet box keeps its own setting", map[string]string{ |
| 1797 | "CARLOS_FLEET": "pi", "CARLOS_FLEET_BOX": "pi-box", |
| 1798 | "CARLOS_STATUS_BUCKET": "b", "CARLOS_STATUS_PREFIX": "fleets/pi/", |
| 1799 | }, "fleets/pi/"}, |
| 1800 | {"bucket-backed fleet box with no prefix is left alone", map[string]string{ |
| 1801 | "CARLOS_FLEET": "pi", "CARLOS_FLEET_BOX": "pi-box", "CARLOS_STATUS_BUCKET": "b", |
| 1802 | }, ""}, |
| 1803 | {"door-backed fleet box defaults", map[string]string{ |
| 1804 | "CARLOS_FLEET": "pi", "CARLOS_FLEET_BOX": "pi-box", |
| 1805 | "CARLOS_TUNNEL_URL": "https://t.example.com", "CARLOS_TUNNEL_TOKEN": "tok", |
| 1806 | }, "fleets/pi/"}, |
| 1807 | {"door-backed fleet box with an explicit prefix keeps it", map[string]string{ |
| 1808 | "CARLOS_FLEET": "pi", "CARLOS_FLEET_BOX": "pi-box", |
| 1809 | "CARLOS_TUNNEL_URL": "https://t.example.com", "CARLOS_TUNNEL_TOKEN": "tok", |
| 1810 | "CARLOS_STATUS_PREFIX": "custom/", |
| 1811 | }, "custom/"}, |
| 1812 | {"no fleet, no bucket", map[string]string{}, ""}, |
| 1813 | } { |
| 1814 | if got := resolveStatusPrefix(envFunc(c.env)); got != c.want { |
| 1815 | t.Errorf("%s: resolveStatusPrefix = %q, want %q", c.name, got, c.want) |
| 1816 | } |
| 1817 | } |
| 1818 | } |
| 1819 | |
| 1820 | // TestDoorBackedStatusWriteLandsInTheFleetCorner is the behavior half: the two |
| 1821 | // lines statusLoop and metricsLoop actually run (openStatusStore + |
| 1822 | // NewPrefixStore over resolveStatusPrefix) must put a heartbeat at |
| 1823 | // fleets/<fleet>/status/boxes/..., the one place the door will accept it and |
| 1824 | // the exact layout the console's fleet page reads. |
| 1825 | func TestDoorBackedStatusWriteLandsInTheFleetCorner(t *testing.T) { |
| 1826 | ctx := context.Background() |
| 1827 | door := newDoorFake(t) |
| 1828 | env := envFunc(fleetDoorEnv(door.url)) // no bucket, no CARLOS_STATUS_PREFIX |
| 1829 | |
| 1830 | raw, _, err := openStatusStore(ctx, env) |
| 1831 | if err != nil { |
| 1832 | t.Fatalf("openStatusStore: %v", err) |
| 1833 | } |
| 1834 | store := release.NewPrefixStore(raw, resolveStatusPrefix(env)) |
| 1835 | // The door fake 404s anything but the serial, so the Put errors — the |
| 1836 | // assertion is the KEY that went on the wire, not the response. |
| 1837 | _, _ = store.Put(ctx, "status/boxes/pi-box.json", []byte("{}"), "") |
| 1838 | |
| 1839 | hits := door.seen() |
| 1840 | if len(hits) != 1 { |
| 1841 | t.Fatalf("door saw %d requests, want 1: %+v", len(hits), hits) |
| 1842 | } |
| 1843 | if want := "/store/o/fleets/pi/status/boxes/pi-box.json"; hits[0].Path != want { |
| 1844 | t.Fatalf("door path = %q, want %q — an unrebased key is outside the door's write ACL and 404s forever", hits[0].Path, want) |
| 1845 | } |
| 1846 | } |
| 1847 | |
| 1848 | // TestRetentionLoopIsAPlatformBoxJob pins spec F6. A customer's box pruning |
| 1849 | // the PLATFORM's releases was wrong under direct S3 and is refused outright |
| 1850 | // through the door (its write ACL scopes a fleet to fleets/<fleet>/**), so a |
| 1851 | // fleet box must not arm the loop at all — one log line, then out, before it |
| 1852 | // ever opens a store. |
| 1853 | func TestRetentionLoopIsAPlatformBoxJob(t *testing.T) { |
| 1854 | const skipLine = "agent: retention loop is a platform-box job — skipped on a fleet box (spec F6)" |
| 1855 | |
| 1856 | t.Run("fleet box skips before arming", func(t *testing.T) { |
| 1857 | logs := captureLog(t) |
| 1858 | door := newDoorFake(t) |
| 1859 | for k, v := range fleetDoorEnv(door.url) { |
| 1860 | t.Setenv(k, v) |
| 1861 | } |
| 1862 | t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "") |
| 1863 | t.Setenv("CARLOS_DEPLOYMENT_DIR", "") |
| 1864 | |
| 1865 | retentionLoop(testRegistry(t), noBuildStates) // must return; a fleet box never reaches the loop |
| 1866 | |
| 1867 | if got := logs(); !strings.Contains(got, skipLine) { |
| 1868 | t.Fatalf("log = %q, want the F6 skip line verbatim", got) |
| 1869 | } |
| 1870 | if hits := door.seen(); len(hits) != 0 { |
| 1871 | t.Fatalf("door saw %+v, want zero — the skip must land BEFORE the store opens", hits) |
| 1872 | } |
| 1873 | }) |
| 1874 | |
| 1875 | t.Run("platform box still arms", func(t *testing.T) { |
| 1876 | pub, _, err := ed25519.GenerateKey(nil) |
| 1877 | if err != nil { |
| 1878 | t.Fatalf("generate key: %v", err) |
| 1879 | } |
| 1880 | t.Setenv("CARLOS_FLEET", "") |
| 1881 | t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "") |
| 1882 | t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir()) |
| 1883 | t.Setenv("CARLOS_RELEASE_PUBKEY", base64.StdEncoding.EncodeToString(pub)) |
| 1884 | // A directory-backed deployment publishes no status: it is its own |
| 1885 | // sole housekeeper and reads this box's own build-state report |
| 1886 | // (ruling R-T9-SOLE), so it still arms. |
| 1887 | t.Setenv("CARLOS_STATUS_BUCKET", "") |
| 1888 | |
| 1889 | // retentionLoop never returns once armed, so the arming log line is |
| 1890 | // what proves it: run it in its own goroutine (it parks in a |
| 1891 | // time.Sleep for a minute, long past the end of this test) against a |
| 1892 | // synchronized sink, and wait for the line to appear. |
| 1893 | var mu sync.Mutex |
| 1894 | var buf bytes.Buffer |
| 1895 | old, flags := log.Writer(), log.Flags() |
| 1896 | log.SetOutput(writerFunc(func(p []byte) (int, error) { |
| 1897 | mu.Lock() |
| 1898 | defer mu.Unlock() |
| 1899 | return buf.Write(p) |
| 1900 | })) |
| 1901 | log.SetFlags(0) |
| 1902 | t.Cleanup(func() { log.SetOutput(old); log.SetFlags(flags) }) |
| 1903 | |
| 1904 | // A REAL registry, never nil: this goroutine outlives the test and |
| 1905 | // runs retentionPass a minute from now, which lists routes — nil |
| 1906 | // would panic the whole test binary long after the test that started |
| 1907 | // it has passed. |
| 1908 | go retentionLoop(testRegistry(t), noBuildStates) |
| 1909 | |
| 1910 | deadline := time.Now().Add(5 * time.Second) |
| 1911 | for { |
| 1912 | mu.Lock() |
| 1913 | got := buf.String() |
| 1914 | mu.Unlock() |
| 1915 | if strings.Contains(got, "agent: retention loop armed") { |
| 1916 | if strings.Contains(got, skipLine) { |
| 1917 | t.Fatalf("platform box took the fleet skip: %q", got) |
| 1918 | } |
| 1919 | return |
| 1920 | } |
| 1921 | if time.Now().After(deadline) { |
| 1922 | t.Fatalf("retention loop never armed on a platform box; log = %q", got) |
| 1923 | } |
| 1924 | time.Sleep(10 * time.Millisecond) |
| 1925 | } |
| 1926 | }) |
| 1927 | } |
| 1928 | |
| 1929 | // noBuildStates is a build-state collector with nothing to report. |
| 1930 | func noBuildStates() (map[string]status.BuildStatePins, error) { return nil, nil } |
| 1931 | |
| 1932 | // writerFunc adapts a func to io.Writer — the log sink above needs its own |
| 1933 | // mutex because the loop under test keeps running (and may keep logging) |
| 1934 | // after the test returns, which a bare bytes.Buffer would race on. |
| 1935 | type writerFunc func(p []byte) (int, error) |
| 1936 | |
| 1937 | func (f writerFunc) Write(p []byte) (int, error) { return f(p) } |
| 1938 | |
| 1939 | // ---- stamped units for the metrics sweep (#110 column) ---------------------- |
| 1940 | |
| 1941 | // TestStampedUnitNamesDedupesAndSkipsUnstamped pins the metrics sweep's one |
| 1942 | // registry read. Route.Unit is a per-ROUTE stamp, so two routes fronted by |
| 1943 | // the same bespoke service (a console and its alias) must yield that unit |
| 1944 | // ONCE — a duplicate would make the box read the same cgroup twice per tick |
| 1945 | // for the same map key. Routes with no unit contribute nothing: the carlos- |
| 1946 | // convention walk owns those, and a "" name would match every unnamed |
| 1947 | // directory the walk compares against. |
| 1948 | func TestStampedUnitNamesDedupesAndSkipsUnstamped(t *testing.T) { |
| 1949 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 1950 | if err != nil { |
| 1951 | t.Fatalf("open registry: %v", err) |
| 1952 | } |
| 1953 | for _, rt := range []registry.Route{ |
| 1954 | {Host: "console.example.com", Kind: "instance", App: "console", Unit: "console.service"}, |
| 1955 | {Host: "dashboard.example.com", Kind: "instance", App: "console", Unit: "console.service"}, |
| 1956 | {Host: "vicky.example.com", Kind: "instance", App: "titogo", Unit: "titogo-canary@vicky.service"}, |
| 1957 | {Host: "plain.example.com", Kind: "instance", App: "plain"}, // unstamped |
| 1958 | } { |
| 1959 | if err := reg.Upsert(rt); err != nil { |
| 1960 | t.Fatalf("upsert %s: %v", rt.Host, err) |
| 1961 | } |
| 1962 | } |
| 1963 | |
| 1964 | got := stampedUnitNames(reg) |
| 1965 | want := []string{"console.service", "titogo-canary@vicky.service"} |
| 1966 | if !slices.Equal(got, want) { |
| 1967 | t.Fatalf("stampedUnitNames = %v, want %v (sorted, deduped, unstamped routes skipped)", got, want) |
| 1968 | } |
| 1969 | } |
| 1970 | |
| 1971 | // TestStampedUnitNamesEmptyRegistry pins the state every box is in today: |
| 1972 | // nothing stamped -> nil, which listStampedUnits turns into no extra scan at |
| 1973 | // all. The sweep must cost nothing on a fleet that has never used the column. |
| 1974 | func TestStampedUnitNamesEmptyRegistry(t *testing.T) { |
| 1975 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 1976 | if err != nil { |
| 1977 | t.Fatalf("open registry: %v", err) |
| 1978 | } |
| 1979 | if got := stampedUnitNames(reg); got != nil { |
| 1980 | t.Fatalf("stampedUnitNames on an empty registry = %v, want nil", got) |
| 1981 | } |
| 1982 | } |
| 1983 | |
| 1984 | // --- issue #196: the registry side of "did anything change" --- |
| 1985 | |
| 1986 | // boundExecRoute is execRoute wired to a named config environment — the |
| 1987 | // binding whose whole point is that it changes which bundle merges into |
| 1988 | // the route's delivered config. |
| 1989 | func boundExecRoute(host, acct, app, env string) registry.Route { |
| 1990 | rt := execRoute(host, acct, app) |
| 1991 | rt.Environment = env |
| 1992 | return rt |
| 1993 | } |
| 1994 | |
| 1995 | // TestGenPollPassSyncsOnRouteRebinding is issue #196: binding a route to an |
| 1996 | // environment must converge from the FAST loop, not an hour later on |
| 1997 | // configSyncLoop's floor. The bucket does not move at all here — only the |
| 1998 | // registry column does, which is exactly why the etag poll could not see |
| 1999 | // it and titogo lost an hour to a symptom reading as a bad credential. |
| 2000 | func TestGenPollPassSyncsOnRouteRebinding(t *testing.T) { |
| 2001 | ctx, s := genKit(t) |
| 2002 | touch(t, ctx, s, "bab", "titogo") |
| 2003 | cache := newGenCache() |
| 2004 | genPollPass(ctx, []registry.Route{execRoute("connect.go.tito.io", "bab", "titogo")}, s, cache) |
| 2005 | |
| 2006 | rebound := []registry.Route{boundExecRoute("connect.go.tito.io", "bab", "titogo", "relay")} |
| 2007 | execChanged, unitChanged, errs := genPollPass(ctx, rebound, s, cache) |
| 2008 | if len(errs) != 0 { |
| 2009 | t.Fatalf("unexpected read errors: %v", errs) |
| 2010 | } |
| 2011 | if !execChanged { |
| 2012 | t.Fatal("binding a route to an environment must make the exec half sync") |
| 2013 | } |
| 2014 | if unitChanged { |
| 2015 | t.Fatal("an exec route's rebinding must not ask for a root config pass") |
| 2016 | } |
| 2017 | // And the pass after it is quiet again: a digest that re-fired every |
| 2018 | // tick would run SyncExec forever. |
| 2019 | if execChanged, _, _ := genPollPass(ctx, rebound, s, cache); execChanged { |
| 2020 | t.Fatal("the pass after a rebinding must be silent again (a stale digest would re-sync forever)") |
| 2021 | } |
| 2022 | } |
| 2023 | |
| 2024 | // TestGenPollPassSyncsOnUnitRouteRebinding: the unit half converges per |
| 2025 | // environment too (appconfig.Sync's group is (acct, app, environment)), so |
| 2026 | // a unit-backed route's rebinding must ask for the root pass — and must |
| 2027 | // not make the agent render anything itself. |
| 2028 | func TestGenPollPassSyncsOnUnitRouteRebinding(t *testing.T) { |
| 2029 | ctx, s := genKit(t) |
| 2030 | touch(t, ctx, s, "bae", "keymail") |
| 2031 | cache := newGenCache() |
| 2032 | genPollPass(ctx, []registry.Route{unitRoute("keymail.dev", "bae", "keymail")}, s, cache) |
| 2033 | |
| 2034 | rebound := []registry.Route{unitRoute("keymail.dev", "bae", "keymail")} |
| 2035 | rebound[0].Environment = "staging" |
| 2036 | execChanged, unitChanged, errs := genPollPass(ctx, rebound, s, cache) |
| 2037 | if len(errs) != 0 { |
| 2038 | t.Fatalf("unexpected read errors: %v", errs) |
| 2039 | } |
| 2040 | if !unitChanged { |
| 2041 | t.Fatal("binding a unit-backed route to an environment must request a root config pass") |
| 2042 | } |
| 2043 | if execChanged { |
| 2044 | t.Fatal("a unit route's rebinding must not run a SyncExec pass") |
| 2045 | } |
| 2046 | } |
| 2047 | |
| 2048 | // TestGenPollPassFirstSightOfRouteShapeIsSilent: a box coming up has never |
| 2049 | // seen the digest before, and must not treat that as a change — every |
| 2050 | // instance on the box would sync at once on boot. configSyncLoop's ~1min |
| 2051 | // first pass owns boot convergence, exactly as it does for etags. |
| 2052 | func TestGenPollPassFirstSightOfRouteShapeIsSilent(t *testing.T) { |
| 2053 | ctx, s := genKit(t) |
| 2054 | touch(t, ctx, s, "bab", "titogo") |
| 2055 | routes := []registry.Route{boundExecRoute("connect.go.tito.io", "bab", "titogo", "relay")} |
| 2056 | execChanged, unitChanged, errs := genPollPass(ctx, routes, s, newGenCache()) |
| 2057 | if len(errs) != 0 { |
| 2058 | t.Fatalf("unexpected read errors: %v", errs) |
| 2059 | } |
| 2060 | if execChanged || unitChanged { |
| 2061 | t.Fatal("the first pass must record the route shape silently") |
| 2062 | } |
| 2063 | } |
| 2064 | |
| 2065 | // TestGenPollPassRouteShapeCostsNoBucketReads: the registry half must not |
| 2066 | // add a single fetch to the tick — the whole economy of this loop is that |
| 2067 | // it reads generation objects and nothing else. |
| 2068 | func TestGenPollPassRouteShapeCostsNoBucketReads(t *testing.T) { |
| 2069 | ctx, _ := genKit(t) |
| 2070 | cs := &countingStore{Store: release.NewDirStore(t.TempDir())} |
| 2071 | touch(t, ctx, cs, "bab", "titogo") |
| 2072 | routes := []registry.Route{execRoute("connect.go.tito.io", "bab", "titogo")} |
| 2073 | cache := newGenCache() |
| 2074 | genPollPass(ctx, routes, cs, cache) |
| 2075 | key := appconfig.GenKey("bab", "titogo") |
| 2076 | before := cs.gets[key] |
| 2077 | rebound := []registry.Route{boundExecRoute("connect.go.tito.io", "bab", "titogo", "relay")} |
| 2078 | genPollPass(ctx, rebound, cs, cache) |
| 2079 | if got := cs.gets[key] - before; got != 1 { |
| 2080 | t.Fatalf("the rebinding tick made %d bucket reads, want exactly 1 (the app's generation object, as before)", got) |
| 2081 | } |
| 2082 | } |
| 2083 | |
| 2084 | // ---- Task 8: the agent socket, the boot pass, and the version write ------- |
| 2085 | |
| 2086 | // waitForCond polls until cond holds — the shared shape every wait-for-a- |
| 2087 | // goroutine test in this package needs (fastConvergeLoop's own tests have |
| 2088 | // no such helper because none of them race a goroutine against a channel). |
| 2089 | func waitForCond(t *testing.T, what string, cond func() bool) { |
| 2090 | t.Helper() |
| 2091 | deadline := time.Now().Add(5 * time.Second) |
| 2092 | for time.Now().Before(deadline) { |
| 2093 | if cond() { |
| 2094 | return |
| 2095 | } |
| 2096 | time.Sleep(5 * time.Millisecond) |
| 2097 | } |
| 2098 | t.Fatalf("timed out waiting for %s", what) |
| 2099 | } |
| 2100 | |
| 2101 | // TestWorkerBootPassStartsOnlyNonHibernatingSidecars is Task 8's own test |
| 2102 | // (brief step 1, verbatim): a sidecar that does not hibernate must be |
| 2103 | // running whenever the agent is; a hibernating one waits for its due time, |
| 2104 | // an operator command, or a delivery signal — the whole point of the class. |
| 2105 | // Non-sidecar routes and a non-ready sidecar are excluded too: the boot pass |
| 2106 | // exists to start THIS box's own always-on sidecars, not to touch instances |
| 2107 | // (which have their own convergence) or a row still mid-provisioning. |
| 2108 | func TestWorkerBootPassStartsOnlyNonHibernatingSidecars(t *testing.T) { |
| 2109 | routes := []registry.Route{ |
| 2110 | {Host: "a.x.sidecar.internal", Kind: "sidecar", Hibernate: false, Status: "ready"}, |
| 2111 | {Host: "b.x.sidecar.internal", Kind: "sidecar", Hibernate: true, Status: "ready"}, |
| 2112 | {Host: "c.example.com", Kind: "instance", Hibernate: false, Status: "ready"}, |
| 2113 | {Host: "d.x.sidecar.internal", Kind: "sidecar", Hibernate: false, Status: "draining"}, |
| 2114 | } |
| 2115 | got := workersToStart(routes) |
| 2116 | if len(got) != 1 || got[0].Host != "a.x.sidecar.internal" { |
| 2117 | t.Fatalf("workersToStart = %v — only a ready, non-hibernating sidecar starts at boot; a hibernating one waits for its timer or a command", got) |
| 2118 | } |
| 2119 | } |
| 2120 | |
| 2121 | // TestWorkerBootPassWakesEveryNonHibernatingSidecar pins the end-to-end |
| 2122 | // wiring shape: given N candidate routes, the boot pass calls wake exactly |
| 2123 | // once per candidate, and a wake that errors does not stop the rest from |
| 2124 | // being tried (one bad sidecar must not strand every other one on the box |
| 2125 | // at boot). |
| 2126 | func TestWorkerBootPassWakesEveryNonHibernatingSidecar(t *testing.T) { |
| 2127 | routes := []registry.Route{ |
| 2128 | {Host: "a.x.sidecar.internal", Kind: "sidecar", Status: "ready"}, |
| 2129 | {Host: "b.x.sidecar.internal", Kind: "sidecar", Status: "ready"}, |
| 2130 | {Host: "c.x.sidecar.internal", Kind: "sidecar", Hibernate: true, Status: "ready"}, // excluded |
| 2131 | } |
| 2132 | var mu sync.Mutex |
| 2133 | woken := map[string]int{} |
| 2134 | wake := func(rt registry.Route) error { |
| 2135 | mu.Lock() |
| 2136 | woken[rt.Host]++ |
| 2137 | mu.Unlock() |
| 2138 | if rt.Host == "a.x.sidecar.internal" { |
| 2139 | return errors.New("boom") |
| 2140 | } |
| 2141 | return nil |
| 2142 | } |
| 2143 | runWorkerBootPass(routes, wake) |
| 2144 | |
| 2145 | mu.Lock() |
| 2146 | defer mu.Unlock() |
| 2147 | if woken["a.x.sidecar.internal"] != 1 || woken["b.x.sidecar.internal"] != 1 { |
| 2148 | t.Fatalf("woken = %v, want exactly one call each for a and b", woken) |
| 2149 | } |
| 2150 | if _, ok := woken["c.x.sidecar.internal"]; ok { |
| 2151 | t.Fatalf("a hibernating sidecar was woken at boot: %v", woken) |
| 2152 | } |
| 2153 | } |
| 2154 | |
| 2155 | // TestWorkerBootPassWakesConcurrently pins the carried requirement that the |
| 2156 | // boot pass must NOT wake sidecars one at a time: WakeWorker blocks for the |
| 2157 | // settle window (10s) when nothing answers /v1/ready, so a serial pass over |
| 2158 | // N sidecars would cost 10N seconds of agent startup. This drives more |
| 2159 | // in-flight wakes than the concurrency bound and proves the bound is |
| 2160 | // actually reached (not stuck at 1, which a serial implementation would |
| 2161 | // show) and never exceeded (which an unbounded implementation would show). |
| 2162 | func TestWorkerBootPassWakesConcurrently(t *testing.T) { |
| 2163 | t.Parallel() |
| 2164 | n := workerBootConcurrency + 3 // strictly more than the bound |
| 2165 | routes := make([]registry.Route, n) |
| 2166 | for i := range routes { |
| 2167 | routes[i] = registry.Route{ |
| 2168 | Host: fmt.Sprintf("s%d.x.sidecar.internal", i), Kind: "sidecar", Status: "ready", |
| 2169 | } |
| 2170 | } |
| 2171 | |
| 2172 | var mu sync.Mutex |
| 2173 | inFlight, maxInFlight := 0, 0 |
| 2174 | release := make(chan struct{}) |
| 2175 | wake := func(registry.Route) error { |
| 2176 | mu.Lock() |
| 2177 | inFlight++ |
| 2178 | if inFlight > maxInFlight { |
| 2179 | maxInFlight = inFlight |
| 2180 | } |
| 2181 | mu.Unlock() |
| 2182 | <-release |
| 2183 | mu.Lock() |
| 2184 | inFlight-- |
| 2185 | mu.Unlock() |
| 2186 | return nil |
| 2187 | } |
| 2188 | |
| 2189 | done := make(chan struct{}) |
| 2190 | go func() { |
| 2191 | runWorkerBootPass(routes, wake) |
| 2192 | close(done) |
| 2193 | }() |
| 2194 | |
| 2195 | waitForCond(t, "the boot pass to saturate its concurrency bound", func() bool { |
| 2196 | mu.Lock() |
| 2197 | defer mu.Unlock() |
| 2198 | return inFlight == workerBootConcurrency |
| 2199 | }) |
| 2200 | // A beat with the bound already saturated: an unbounded implementation |
| 2201 | // would keep climbing past it here rather than holding still. |
| 2202 | time.Sleep(50 * time.Millisecond) |
| 2203 | mu.Lock() |
| 2204 | got := maxInFlight |
| 2205 | mu.Unlock() |
| 2206 | if got != workerBootConcurrency { |
| 2207 | t.Fatalf("max concurrent wakes = %d, want exactly the bound %d", got, workerBootConcurrency) |
| 2208 | } |
| 2209 | |
| 2210 | close(release) |
| 2211 | select { |
| 2212 | case <-done: |
| 2213 | case <-time.After(5 * time.Second): |
| 2214 | t.Fatal("boot pass did not finish after its wakes were released") |
| 2215 | } |
| 2216 | } |
| 2217 | |
| 2218 | // ---- the agent socket: retry/backoff on a bind that lost the race -------- |
| 2219 | |
| 2220 | // TestServeAgentSocketReturnsOnCleanShutdown pins the non-error path: a nil |
| 2221 | // return (agentsock.Serve's contract once ctx is cancelled and the listener |
| 2222 | // shut down cleanly) must end the loop immediately, with no retry. |
| 2223 | func TestServeAgentSocketReturnsOnCleanShutdown(t *testing.T) { |
| 2224 | var calls int32 |
| 2225 | serve := func(context.Context, string) error { |
| 2226 | atomic.AddInt32(&calls, 1) |
| 2227 | return nil |
| 2228 | } |
| 2229 | serveAgentSocket(context.Background(), "/tmp/does-not-matter.sock", serve) |
| 2230 | if n := atomic.LoadInt32(&calls); n != 1 { |
| 2231 | t.Fatalf("serve calls = %d, want exactly 1 — a clean nil return must never retry", n) |
| 2232 | } |
| 2233 | } |
| 2234 | |
| 2235 | // TestServeAgentSocketRetriesWithBackoffOnBusySocket pins the carried |
| 2236 | // requirement that a failed bind — agentsock.Serve deliberately refusing a |
| 2237 | // socket another (likely still-shutting-down) incarnation still owns — must |
| 2238 | // NEVER be fatal: the loop retries with backoff instead of returning the |
| 2239 | // error up to a log.Fatal that would take the whole agent, and every route |
| 2240 | // on the box, down over a transient restart race. |
| 2241 | func TestServeAgentSocketRetriesWithBackoffOnBusySocket(t *testing.T) { |
| 2242 | oldBase, oldCap := agentSocketRetryBase, agentSocketRetryCap |
| 2243 | agentSocketRetryBase = time.Millisecond |
| 2244 | agentSocketRetryCap = 4 * time.Millisecond |
| 2245 | defer func() { agentSocketRetryBase, agentSocketRetryCap = oldBase, oldCap }() |
| 2246 | |
| 2247 | var calls int32 |
| 2248 | succeeded := make(chan struct{}) |
| 2249 | ctx, cancel := context.WithCancel(context.Background()) |
| 2250 | defer cancel() |
| 2251 | serve := func(c context.Context, path string) error { |
| 2252 | n := atomic.AddInt32(&calls, 1) |
| 2253 | if n < 4 { |
| 2254 | return errors.New("agentsock: /run/carlos/agent.sock is already served by a live listener") |
| 2255 | } |
| 2256 | close(succeeded) |
| 2257 | <-c.Done() // the fourth "bind" holds until shutdown, like the real Serve |
| 2258 | return nil |
| 2259 | } |
| 2260 | |
| 2261 | done := make(chan struct{}) |
| 2262 | go func() { |
| 2263 | serveAgentSocket(ctx, "/run/carlos/agent.sock", serve) |
| 2264 | close(done) |
| 2265 | }() |
| 2266 | |
| 2267 | select { |
| 2268 | case <-succeeded: |
| 2269 | case <-time.After(2 * time.Second): |
| 2270 | t.Fatal("a busy socket was never retried to success") |
| 2271 | } |
| 2272 | cancel() |
| 2273 | select { |
| 2274 | case <-done: |
| 2275 | case <-time.After(2 * time.Second): |
| 2276 | t.Fatal("serveAgentSocket did not return after its context was cancelled") |
| 2277 | } |
| 2278 | } |
| 2279 | |
| 2280 | // TestServeAgentSocketStopsRetryingOnContextCancel pins the other half of |
| 2281 | // "never fatal, never stuck": a permanently busy socket must not wedge the |
| 2282 | // goroutine forever — cancelling ctx ends the retry loop promptly even |
| 2283 | // mid-backoff. |
| 2284 | func TestServeAgentSocketStopsRetryingOnContextCancel(t *testing.T) { |
| 2285 | oldBase, oldCap := agentSocketRetryBase, agentSocketRetryCap |
| 2286 | agentSocketRetryBase = time.Millisecond |
| 2287 | agentSocketRetryCap = 2 * time.Millisecond |
| 2288 | defer func() { agentSocketRetryBase, agentSocketRetryCap = oldBase, oldCap }() |
| 2289 | |
| 2290 | ctx, cancel := context.WithCancel(context.Background()) |
| 2291 | var calls int32 |
| 2292 | serve := func(context.Context, string) error { |
| 2293 | atomic.AddInt32(&calls, 1) |
| 2294 | return errors.New("still busy") |
| 2295 | } |
| 2296 | done := make(chan struct{}) |
| 2297 | go func() { |
| 2298 | serveAgentSocket(ctx, "/run/carlos/agent.sock", serve) |
| 2299 | close(done) |
| 2300 | }() |
| 2301 | |
| 2302 | waitForCond(t, "at least one retry to have happened", func() bool { |
| 2303 | return atomic.LoadInt32(&calls) >= 2 |
| 2304 | }) |
| 2305 | cancel() |
| 2306 | select { |
| 2307 | case <-done: |
| 2308 | case <-time.After(2 * time.Second): |
| 2309 | t.Fatal("serveAgentSocket did not stop retrying after context cancellation") |
| 2310 | } |
| 2311 | } |
| 2312 | |
| 2313 | // The edge-local door is dormant unless CARLOS_EDGE_DOOR names a path, and |
| 2314 | // the box's public address is read only when it does: off EC2 that read |
| 2315 | // can block for its whole timeout. |
| 2316 | func TestEdgeDoorOptions(t *testing.T) { |
| 2317 | reads := 0 |
| 2318 | publicAddr := func() string { reads++; return "203.0.113.10" } |
| 2319 | for _, unset := range []string{"", " "} { |
| 2320 | env := func(k string) string { |
| 2321 | if k == "CARLOS_EDGE_DOOR" { |
| 2322 | return unset |
| 2323 | } |
| 2324 | return "" |
| 2325 | } |
| 2326 | if opts := edgeDoorOptions(env, publicAddr); len(opts) != 0 { |
| 2327 | t.Fatalf("CARLOS_EDGE_DOOR=%q: %d option(s), want none", unset, len(opts)) |
| 2328 | } |
| 2329 | } |
| 2330 | if reads != 0 { |
| 2331 | t.Fatalf("the public address was read %d time(s) with the door off", reads) |
| 2332 | } |
| 2333 | env := func(k string) string { |
| 2334 | if k == "CARLOS_EDGE_DOOR" { |
| 2335 | return " /run/carlos/edge-local.sock " |
| 2336 | } |
| 2337 | return "" |
| 2338 | } |
| 2339 | if opts := edgeDoorOptions(env, publicAddr); len(opts) != 1 { |
| 2340 | t.Fatalf("CARLOS_EDGE_DOOR set: %d option(s), want one", len(opts)) |
| 2341 | } |
| 2342 | if reads != 1 { |
| 2343 | t.Fatalf("the public address was read %d time(s), want once", reads) |
| 2344 | } |
| 2345 | } |
| 2346 | |
amadan