| 1 | //go:build linux |
| 2 | |
| 3 | package host |
| 4 | |
| 5 | // The large-box startup, end to end (amadan #281, ask 2). On 2026-10-13 a |
| 6 | // box with 142 instances failed the 20 s update gate twice: the updater |
| 7 | // restarts carlos-instance-control just before the new edge, and the |
| 8 | // agent's startup sweep waited on that cold control service before the |
| 9 | // edge listened. The hotfix (6dfbff1) moved the sweep, layout recovery, |
| 10 | // hold restoration, Start and the key pass off the listen path, behind a |
| 11 | // startup latch that instance admission waits on. |
| 12 | // |
| 13 | // TestStartupServesWhileSweepAndRecoveryWait proves the latch with the |
| 14 | // steps stubbed. This test runs them for real on a box larger than the |
| 15 | // one that failed: |
| 16 | // |
| 17 | // - the agent is a child process (this test binary, re-executed), so |
| 18 | // what is timed is a process start up to the update gate's own probe; |
| 19 | // - instance control is the real instancectl.Server over a unix socket, |
| 20 | // made cold by its systemctl seam: every call waits until the test |
| 21 | // warms it, then each of the box's 300 leftover units takes a moment |
| 22 | // to stop, as a restarted control service does; |
| 23 | // - the sweep is the real client call, layout recovery and hold |
| 24 | // restoration are the real pass over real journals, hold markers and |
| 25 | // files in a temp data dir, Start is the real activator's, and the |
| 26 | // edge is edge.RunWithActivatorReady in its production (TLS) mode; |
| 27 | // - every kind of admission the latch gates arrives during startup: an |
| 28 | // instance request, a sidecar-marked request, a worker wake (from |
| 29 | // runAgent's boot pass and from a direct call), and a scheduled |
| 30 | // delivery through the real schedule runner and deliverer; |
| 31 | // - the activator's backend is a fake that serves an app on the row's |
| 32 | // socket and starts workers that report ready. It and the key pass's |
| 33 | // writer are the only fakes in the agent, and neither is a startup |
| 34 | // step. |
| 35 | // |
| 36 | // The child's startup is runAgent's legacy-box sequence: the same |
| 37 | // activatorStart, with runAgent's Sweep and Recover closures. runAgent |
| 38 | // itself cannot run in a test (its control socket path is a constant, |
| 39 | // it builds AWS clients, and it log.Fatals), so the child assembles the |
| 40 | // same pieces in the same order, and TestRunAgentStartupWiring reads |
| 41 | // runAgent's source to fail if the two drift apart. Sharing one |
| 42 | // orchestration function between them is a follow-up (#281). Two box |
| 43 | // facts differ from production and are set on the pass: its socket |
| 44 | // directory is the fixture's, not /run/carlos, and it has no |
| 45 | // carlos-sockets group. |
| 46 | // |
| 47 | // The child reports what it does as timestamped lines in an events file; |
| 48 | // the test reads them to order "a request reached the activator", "the |
| 49 | // latch opened" and "the backend acquired a host" across the two |
| 50 | // processes. |
| 51 | |
| 52 | import ( |
| 53 | "bufio" |
| 54 | "context" |
| 55 | "crypto/ecdsa" |
| 56 | "crypto/elliptic" |
| 57 | "crypto/rand" |
| 58 | "crypto/tls" |
| 59 | "crypto/x509" |
| 60 | "crypto/x509/pkix" |
| 61 | "encoding/json" |
| 62 | "encoding/pem" |
| 63 | "errors" |
| 64 | "fmt" |
| 65 | "go/ast" |
| 66 | "go/parser" |
| 67 | "go/token" |
| 68 | "go/types" |
| 69 | "io" |
| 70 | "log" |
| 71 | "math/big" |
| 72 | "net" |
| 73 | "net/http" |
| 74 | "os" |
| 75 | "os/exec" |
| 76 | "path/filepath" |
| 77 | "runtime" |
| 78 | "strconv" |
| 79 | "strings" |
| 80 | "sync" |
| 81 | "sync/atomic" |
| 82 | "syscall" |
| 83 | "testing" |
| 84 | "time" |
| 85 | |
| 86 | "github.com/carlosframework/platform/internal/activator" |
| 87 | "github.com/carlosframework/platform/internal/edge" |
| 88 | "github.com/carlosframework/platform/internal/instancectl" |
| 89 | "github.com/carlosframework/platform/internal/instancefs" |
| 90 | "github.com/carlosframework/platform/internal/layout" |
| 91 | "github.com/carlosframework/platform/internal/provision" |
| 92 | "github.com/carlosframework/platform/internal/registry" |
| 93 | "github.com/carlosframework/platform/internal/release" |
| 94 | "github.com/carlosframework/platform/internal/schedule" |
| 95 | "github.com/carlosframework/platform/internal/version" |
| 96 | ) |
| 97 | |
| 98 | const ( |
| 99 | // largeBoxAgentEnv names the child's config file. Set, the test binary |
| 100 | // is the agent and nothing else. |
| 101 | largeBoxAgentEnv = "CARLOS_TEST_LARGEBOX_AGENT" |
| 102 | |
| 103 | largeBoxRoutes = 300 // instance routes; the box that failed had 142 |
| 104 | largeBoxJournals = 8 // hosts 0..7: a layout move was in flight at the restart |
| 105 | largeBoxHolds = 4 // hosts 8..11: held down after a refused move |
| 106 | largeBoxWakeHost = 12 // an ordinary sleeping instance, woken by a request |
| 107 | largeBoxDueHost = 13 // a sleeping instance with a scheduled delivery due |
| 108 | largeBoxSidecarHost = 14 // a sleeping instance its sidecar calls |
| 109 | largeBoxWorker = "worker-000.example.com" |
| 110 | largeBoxDuePath = "/jobs/tick" |
| 111 | |
| 112 | // largeBoxListenBound is how soon the gate's probe must be answered |
| 113 | // after the agent process starts. The gate allows 20 s; the edge's own |
| 114 | // startup on this box takes a fraction of a second. |
| 115 | largeBoxListenBound = 3 * time.Second |
| 116 | // largeBoxCold is how long instance control stays cold in the scenario |
| 117 | // where it warms: longer than the listen bound, so the gate is answered |
| 118 | // while the sweep is provably still waiting. |
| 119 | largeBoxCold = 4 * time.Second |
| 120 | // largeBoxUnitStop is one leftover unit's stop once control is warm. |
| 121 | largeBoxUnitStop = 5 * time.Millisecond |
| 122 | // largeBoxLatchBound is the activator's wakeWaitBound, which the |
| 123 | // startup latch shares: a wake held longer is answered 503. |
| 124 | largeBoxLatchBound = 30 * time.Second |
| 125 | // largeBoxJoinBound is runAgent's join on the startup work after the |
| 126 | // edge returns, and largeBoxDrain the edge's drain of in-flight |
| 127 | // requests before it returns. |
| 128 | largeBoxJoinBound = 5 * time.Second |
| 129 | largeBoxDrain = 10 * time.Second |
| 130 | // largeBoxBindAttempts is how often the agent is started on fresh ports |
| 131 | // when one it was handed is taken before it binds (see startAgent). |
| 132 | largeBoxBindAttempts = 3 |
| 133 | ) |
| 134 | |
| 135 | func largeBoxHost(i int) string { return fmt.Sprintf("app-%03d.example.com", i) } |
| 136 | |
| 137 | // largeBoxConfig is what the child agent is told. |
| 138 | type largeBoxConfig struct { |
| 139 | DataDir, SocketDir, StoreDir, AcmeDir string |
| 140 | Control, Events, HTTP, HTTPS string |
| 141 | } |
| 142 | |
| 143 | func TestStartupLargeBoxEndToEnd(t *testing.T) { |
| 144 | if path := os.Getenv(largeBoxAgentEnv); path != "" { |
| 145 | os.Exit(runLargeBoxAgent(path)) |
| 146 | } |
| 147 | // The parent stays serial on purpose. awaitListening holds each child |
| 148 | // agent to a 3 s startup budget, so running beside the package's |
| 149 | // parallel tests would put their load into that measurement. The |
| 150 | // scenarios still run beside each other. |
| 151 | t.Run("cold instance control", func(t *testing.T) { |
| 152 | t.Parallel() |
| 153 | testLargeBoxColdControl(t) |
| 154 | }) |
| 155 | t.Run("startup overruns the latch", func(t *testing.T) { |
| 156 | t.Parallel() |
| 157 | testLargeBoxOverrun(t) |
| 158 | }) |
| 159 | t.Run("shutdown mid-startup", func(t *testing.T) { |
| 160 | t.Parallel() |
| 161 | testLargeBoxShutdownMidStartup(t) |
| 162 | }) |
| 163 | } |
| 164 | |
| 165 | // testLargeBoxColdControl is the 2026-10-13 shape: control restarted just |
| 166 | // before the agent, and slow to answer. The edge must answer the gate's |
| 167 | // probe while the sweep waits, hold every admission at the latch until |
| 168 | // sweep, recovery and hold restoration are done, and then serve each host |
| 169 | // only where it is safe to. |
| 170 | func testLargeBoxColdControl(t *testing.T) { |
| 171 | b := newLargeBox(t) |
| 172 | a := b.startAgent(t) |
| 173 | t.Logf("gate probe answered %s after the agent started, %d routes, instance control cold", a.listen, largeBoxRoutes) |
| 174 | if _, ok := a.event("ready"); ok { |
| 175 | t.Fatal("the startup latch opened while instance control was still cold") |
| 176 | } |
| 177 | |
| 178 | wake, held, moved, due, sidecar := largeBoxHost(largeBoxWakeHost), largeBoxHost(largeBoxJournals), largeBoxHost(0), largeBoxHost(largeBoxDueHost), largeBoxHost(largeBoxSidecarHost) |
| 179 | flatMoved := filepath.Join(b.socketDir, moved+".sock") |
| 180 | wakeAns, heldAns, movedAns := b.fire(wake, false), b.fire(held, false), b.fire(moved, false) |
| 181 | sidecarAns := b.fire(sidecar, true) |
| 182 | enterWake := a.waitEvent(t, 10*time.Second, "enter", wake) |
| 183 | a.waitEvent(t, 10*time.Second, "enter", held) |
| 184 | enterSidecar := a.waitEvent(t, 10*time.Second, "enter", sidecar) |
| 185 | enterMoved := a.waitEvent(t, 10*time.Second, "enter", moved) |
| 186 | if enterMoved.arg(1) != flatMoved { |
| 187 | t.Fatalf("the edge routed %s to %q before recovery, want its pre-move socket %q", moved, enterMoved.arg(1), flatMoved) |
| 188 | } |
| 189 | // The agent's own admissions: a direct worker wake and the scheduled |
| 190 | // tick, both started before the edge listened. |
| 191 | workerAsked := a.waitEvent(t, 10*time.Second, "worker-direct-begin") |
| 192 | ticked := a.waitEvent(t, 10*time.Second, "tick-done") |
| 193 | // Every admission is now waiting, and startup is not done. |
| 194 | time.Sleep(200 * time.Millisecond) |
| 195 | for host, ch := range map[string]chan largeBoxAnswer{wake: wakeAns, held: heldAns, moved: movedAns, sidecar: sidecarAns} { |
| 196 | select { |
| 197 | case got := <-ch: |
| 198 | t.Fatalf("%s answered before startup finished: %d %q %v", host, got.status, got.body, got.err) |
| 199 | default: |
| 200 | } |
| 201 | } |
| 202 | for _, kind := range []string{"acquire", "start", "worker-start", "served", "worker-direct", "worker-boot", "activator-start"} { |
| 203 | if evs := a.events(kind); len(evs) != 0 { |
| 204 | t.Fatalf("%s while instance control was cold: %v", kind, evs) |
| 205 | } |
| 206 | } |
| 207 | |
| 208 | // Instance control warms: the sweep stops the 300 leftovers, then |
| 209 | // recovery, holds, Start, and the latch. |
| 210 | if d := time.Until(a.began.Add(largeBoxCold)); d > 0 { |
| 211 | time.Sleep(d) |
| 212 | } |
| 213 | close(b.control.warm) |
| 214 | ready := a.waitEvent(t, 30*time.Second, "ready") |
| 215 | t.Logf("startup latch opened %s after the agent started (sweep stopped %d units)", ready.at.Sub(a.began), b.control.stops.Load()) |
| 216 | if got := b.control.stops.Load(); got != largeBoxRoutes { |
| 217 | t.Errorf("the sweep stopped %d units, want %d", got, largeBoxRoutes) |
| 218 | } |
| 219 | for what, at := range map[string]time.Time{"wake": enterWake.at, "sidecar request": enterSidecar.at, "worker wake": workerAsked.at, "scheduled tick": ticked.at} { |
| 220 | if !at.Before(ready.at) { |
| 221 | t.Fatalf("the %s did not arrive before the latch opened; the test proves nothing about its latch", what) |
| 222 | } |
| 223 | } |
| 224 | |
| 225 | // Start ran only once the held hosts were fenced again and every |
| 226 | // journal was complete. |
| 227 | st := a.waitEvent(t, time.Second, "activator-start") |
| 228 | if st.arg(0) != "held="+strconv.Itoa(largeBoxHolds) || st.arg(1) != "fenced="+strconv.Itoa(largeBoxHolds) || st.arg(2) != "journals=0" || st.at.After(ready.at) { |
| 229 | t.Fatalf("at Start: %v", st) |
| 230 | } |
| 231 | |
| 232 | // The wake was held, then served. |
| 233 | got := b.answer(t, wakeAns) |
| 234 | if got.status != http.StatusOK || got.body != "host="+wake+" socket="+filepath.Join(b.socketDir, wake+".sock") { |
| 235 | t.Fatalf("held wake after startup: %d %q %v", got.status, got.body, got.err) |
| 236 | } |
| 237 | a.assertAfter(t, ready, "acquire", wake) |
| 238 | |
| 239 | // The sidecar's call was held too, then answered as the activator |
| 240 | // answers a sidecar whose instance sleeps: it does not wake it. |
| 241 | got = b.answer(t, sidecarAns) |
| 242 | if got.status != http.StatusServiceUnavailable || got.body != "instance is asleep; sidecar requests do not wake it\n" || got.at.Before(ready.at) { |
| 243 | t.Fatalf("sidecar request held through startup: %d %q %v at %s, latch at %s", got.status, got.body, got.err, got.at, ready.at) |
| 244 | } |
| 245 | |
| 246 | // The worker: both wakes waited, then coalesced on one child. |
| 247 | a.assertAfter(t, ready, "worker-direct", largeBoxWorker, "true") |
| 248 | a.assertAfter(t, ready, "worker-boot", largeBoxWorker, "true") |
| 249 | a.assertAfter(t, ready, "worker-start", largeBoxWorker) |
| 250 | if n := len(a.events("worker-start")); n != 1 { |
| 251 | t.Errorf("worker started %d times, want once", n) |
| 252 | } |
| 253 | |
| 254 | // The scheduled delivery waited, then woke its instance and ran. |
| 255 | a.assertAfter(t, ready, "served", due, largeBoxDuePath) |
| 256 | a.assertAfter(t, ready, "acquire", due) |
| 257 | |
| 258 | // The held host was not started: its restored hold answers maintenance. |
| 259 | got = b.answer(t, heldAns) |
| 260 | if got.status != http.StatusServiceUnavailable || got.body != "instance failed to wake\n" || got.header.Get("Cache-Control") != "no-store" { |
| 261 | t.Fatalf("held host after startup: %d %q %v %v", got.status, got.body, got.header, got.err) |
| 262 | } |
| 263 | |
| 264 | // The moved host was not served at the socket recovery moved it from: |
| 265 | // the request answers as a wake in progress, and its retry is routed |
| 266 | // afresh to where recovery put it. |
| 267 | got = b.answer(t, movedAns) |
| 268 | if got.status != http.StatusServiceUnavailable || got.body != "instance is waking; retry shortly\n" || got.header.Get("Retry-After") != "5" { |
| 269 | t.Fatalf("host whose recovery moved it, held through startup: %d %q %v", got.status, got.body, got.err) |
| 270 | } |
| 271 | nested, err := layout.PlanFor(b.dataDir, b.socketDir, moved, layout.ModeNested) |
| 272 | if err != nil { |
| 273 | t.Fatal(err) |
| 274 | } |
| 275 | got = b.get(moved, false) |
| 276 | if got.status != http.StatusOK || got.body != "host="+moved+" socket="+nested.To.Socket { |
| 277 | t.Fatalf("retry of the moved host: %d %q %v", got.status, got.body, got.err) |
| 278 | } |
| 279 | got = b.get(held, false) |
| 280 | if got.status != http.StatusServiceUnavailable || got.body != "instance failed to wake\n" { |
| 281 | t.Fatalf("retry of the held host: %d %q %v", got.status, got.body, got.err) |
| 282 | } |
| 283 | |
| 284 | // The background key pass ran after Start, and its slot refused the |
| 285 | // held hosts. |
| 286 | a.waitEvent(t, 30*time.Second, "startup-done") |
| 287 | keys := a.events("key") |
| 288 | if len(keys) != largeBoxRoutes-largeBoxHolds { |
| 289 | t.Errorf("key pass wrote %d rows, want %d", len(keys), largeBoxRoutes-largeBoxHolds) |
| 290 | } |
| 291 | for _, e := range keys { |
| 292 | if e.at.Before(ready.at) { |
| 293 | t.Errorf("key pass reached %s before Start", e.arg(0)) |
| 294 | } |
| 295 | } |
| 296 | |
| 297 | b.assertNoWrongService(t, a) |
| 298 | b.assertRecovered(t) |
| 299 | |
| 300 | a.signal(t, syscall.SIGTERM) |
| 301 | a.waitExit(t, largeBoxJoinBound+3*time.Second) |
| 302 | a.assertCleanShutdown(t) |
| 303 | } |
| 304 | |
| 305 | // testLargeBoxOverrun: instance control never warms, so startup never |
| 306 | // finishes. A wake must be answered with the wake-timeout 503 at the |
| 307 | // latch's bound: not an error, not a hang, and not a wake. |
| 308 | func testLargeBoxOverrun(t *testing.T) { |
| 309 | b := newLargeBox(t) |
| 310 | a := b.startAgent(t) |
| 311 | t.Logf("gate probe answered %s after the agent started, instance control never warms", a.listen) |
| 312 | |
| 313 | wake := largeBoxHost(largeBoxWakeHost) |
| 314 | began := time.Now() |
| 315 | got := b.get(wake, false) |
| 316 | took := time.Since(began) |
| 317 | t.Logf("wake held %s by a startup that does not finish", took) |
| 318 | if got.err != nil { |
| 319 | t.Fatalf("wake during an overrunning startup: %v", got.err) |
| 320 | } |
| 321 | if got.status != http.StatusServiceUnavailable || got.body != "instance is waking; retry shortly\n" || got.header.Get("Retry-After") != "5" { |
| 322 | t.Fatalf("wake during an overrunning startup: %d %q %v", got.status, got.body, got.header) |
| 323 | } |
| 324 | if took < largeBoxLatchBound-time.Second || took > largeBoxLatchBound+5*time.Second { |
| 325 | t.Fatalf("wake answered after %s, want the latch bound %s", took, largeBoxLatchBound) |
| 326 | } |
| 327 | a.assertNeverStarted(t) |
| 328 | b.assertNoWrongService(t, a) |
| 329 | |
| 330 | // Shutting down with the sweep still waiting: nothing in flight, so the |
| 331 | // edge returns at once and runAgent's join bounds the rest. |
| 332 | at := time.Now() |
| 333 | a.signal(t, syscall.SIGTERM) |
| 334 | a.waitExit(t, largeBoxJoinBound+3*time.Second) |
| 335 | t.Logf("agent exited %s after SIGTERM with the sweep still waiting", time.Since(at)) |
| 336 | a.assertNeverStarted(t) |
| 337 | a.assertCleanShutdown(t) |
| 338 | } |
| 339 | |
| 340 | // testLargeBoxShutdownMidStartup: SIGTERM with startup still waiting on |
| 341 | // the sweep and a wake held at the latch. The held request is answered |
| 342 | // 503 (never proxied), the background work is joined within runAgent's |
| 343 | // bound, and nothing of startup outlives it. |
| 344 | func testLargeBoxShutdownMidStartup(t *testing.T) { |
| 345 | b := newLargeBox(t) |
| 346 | a := b.startAgent(t) |
| 347 | |
| 348 | wake := largeBoxHost(largeBoxWakeHost) |
| 349 | ans := b.fire(wake, false) |
| 350 | a.waitEvent(t, 10*time.Second, "enter", wake) |
| 351 | |
| 352 | at := time.Now() |
| 353 | a.signal(t, syscall.SIGTERM) |
| 354 | // The edge drains in-flight requests for up to largeBoxDrain before it |
| 355 | // returns, and the held wake is in flight until the activator shuts |
| 356 | // down after the join. |
| 357 | a.waitExit(t, largeBoxDrain+largeBoxJoinBound+3*time.Second) |
| 358 | t.Logf("agent exited %s after SIGTERM with a wake held at the latch", time.Since(at)) |
| 359 | got := b.answer(t, ans) |
| 360 | if got.err != nil || got.status != http.StatusServiceUnavailable { |
| 361 | t.Fatalf("wake held through shutdown: %d %q %v", got.status, got.body, got.err) |
| 362 | } |
| 363 | a.assertNeverStarted(t) |
| 364 | b.assertNoWrongService(t, a) |
| 365 | a.assertCleanShutdown(t) |
| 366 | } |
| 367 | |
| 368 | // --- runAgent's wiring ------------------------------------------------------------ |
| 369 | |
| 370 | // TestRunAgentStartupWiring reads the startup's source and fails if it |
| 371 | // drifts from what runLargeBoxAgent rebuilds, or if anything on runAgent's |
| 372 | // path to the edge waits for startup to finish. Calls are matched on the |
| 373 | // callee's own name (a selector's Sel), and order means execution order on |
| 374 | // one goroutine: a step inside a go or defer statement, or inside a |
| 375 | // function literal that is not invoked on the spot, does not count as |
| 376 | // taken. Where the hotfix does mean a step to run in the background, the |
| 377 | // one goroutine it runs on is found structurally and its order checked |
| 378 | // inside it. |
| 379 | func TestRunAgentStartupWiring(t *testing.T) { |
| 380 | t.Parallel() |
| 381 | fset := token.NewFileSet() |
| 382 | decls := map[string]*ast.FuncDecl{} // runAgent, startActivator, recoverLayoutAtStart, activatorStart.Run |
| 383 | files, err := filepath.Glob("*.go") |
| 384 | if err != nil { |
| 385 | t.Fatal(err) |
| 386 | } |
| 387 | for _, name := range files { |
| 388 | if strings.HasSuffix(name, "_test.go") { |
| 389 | continue |
| 390 | } |
| 391 | f, err := parser.ParseFile(fset, name, nil, 0) |
| 392 | if err != nil { |
| 393 | t.Fatal(err) |
| 394 | } |
| 395 | for _, d := range f.Decls { |
| 396 | fd, ok := d.(*ast.FuncDecl) |
| 397 | if !ok { |
| 398 | continue |
| 399 | } |
| 400 | key := fd.Name.Name |
| 401 | if fd.Recv != nil && len(fd.Recv.List) == 1 { |
| 402 | key = types.ExprString(fd.Recv.List[0].Type) + "." + key |
| 403 | } |
| 404 | switch key { |
| 405 | case "runAgent", "startActivator", "recoverLayoutAtStart", "activatorStart.Run", "joinAgentShutdown": |
| 406 | decls[key] = fd |
| 407 | } |
| 408 | } |
| 409 | } |
| 410 | for _, key := range []string{"runAgent", "startActivator", "recoverLayoutAtStart", "activatorStart.Run", "joinAgentShutdown"} { |
| 411 | if decls[key] == nil { |
| 412 | t.Fatalf("%s not found in package host", key) |
| 413 | } |
| 414 | } |
| 415 | str := types.ExprString |
| 416 | |
| 417 | // runAgent: the latch is set, the start sequence is handed the startup |
| 418 | // context, the edge runs, then cancel, join and Shutdown, all on |
| 419 | // runAgent's own goroutine. |
| 420 | runAgent := decls["runAgent"].Body |
| 421 | sync := syncCalls(runAgent) |
| 422 | var run, edgeCall *ast.CallExpr |
| 423 | var lit *ast.CompositeLit |
| 424 | for _, c := range sync { |
| 425 | if sel, ok := c.Fun.(*ast.SelectorExpr); ok && sel.Sel.Name == "Run" { |
| 426 | if cl, ok := sel.X.(*ast.CompositeLit); ok && str(cl.Type) == "activatorStart" { |
| 427 | run, lit = c, cl |
| 428 | } |
| 429 | } |
| 430 | if calleeName(c) == "RunWithActivatorReady" { |
| 431 | edgeCall = c |
| 432 | } |
| 433 | } |
| 434 | if run == nil || edgeCall == nil { |
| 435 | t.Fatalf("runAgent: activatorStart{...}.Run called synchronously: %v; edge.RunWithActivatorReady called synchronously: %v", run != nil, edgeCall != nil) |
| 436 | } |
| 437 | if len(run.Args) != 1 || str(run.Args[0]) != "startupCtx" { |
| 438 | t.Errorf("activatorStart.Run is given %v, want startupCtx", run.Args) |
| 439 | } |
| 440 | var latch token.Pos |
| 441 | ast.Inspect(runAgent, func(n ast.Node) bool { |
| 442 | if a, ok := n.(*ast.AssignStmt); ok && len(a.Lhs) == 1 && len(a.Rhs) == 1 && selName(a.Lhs[0]) == "Startup" && str(a.Rhs[0]) == "started" { |
| 443 | latch = a.Pos() |
| 444 | } |
| 445 | return true |
| 446 | }) |
| 447 | if latch == token.NoPos || latch > run.Pos() { |
| 448 | t.Error("runAgent: svc.Startup = started is not set before activatorStart.Run") |
| 449 | } |
| 450 | after := func(name string, arg string) token.Pos { |
| 451 | for _, c := range sync { |
| 452 | if c.Pos() > edgeCall.Pos() && calleeName(c) == name && (arg == "" || (len(c.Args) > 0 && str(c.Args[len(c.Args)-1]) == arg)) { |
| 453 | return c.Pos() |
| 454 | } |
| 455 | } |
| 456 | return token.NoPos |
| 457 | } |
| 458 | prev, prevName := edgeCall.Pos(), "edge.RunWithActivatorReady" |
| 459 | for _, step := range []struct{ name, arg string }{{"joinAgentShutdown", "dutiesDone"}, {"Shutdown", ""}} { |
| 460 | p := after(step.name, step.arg) |
| 461 | if p == token.NoPos || p < prev { |
| 462 | t.Errorf("runAgent: %s is not called synchronously after %s", step.name, prevName) |
| 463 | continue |
| 464 | } |
| 465 | prev, prevName = p, step.name |
| 466 | } |
| 467 | |
| 468 | // Follow the synchronous helper without weakening cancel/join ordering or |
| 469 | // allowing the caller to pass different startup channels or cancellation. |
| 470 | for _, call := range sync { |
| 471 | if calleeName(call) != "joinAgentShutdown" { |
| 472 | continue |
| 473 | } |
| 474 | want := []string{"seedCtx", "seedCancel", "seedServer", "cancelStartup", "startupDone", "dutiesDone"} |
| 475 | if len(call.Args) != len(want) { |
| 476 | t.Fatal("shutdown helper argument count", len(call.Args)) |
| 477 | } |
| 478 | for i, arg := range call.Args { |
| 479 | if str(arg) != want[i] { |
| 480 | t.Errorf("shutdown argument %d: %s, want %s", i, str(arg), want[i]) |
| 481 | } |
| 482 | } |
| 483 | } |
| 484 | helper := syncCalls(decls["joinAgentShutdown"].Body) |
| 485 | previous := token.NoPos |
| 486 | for _, step := range []struct{ name, arg string }{{"cancelStartup", ""}, {"joinAgentStartup", "startupDone"}, {"joinAgentStartup", "dutiesDone"}} { |
| 487 | found := token.NoPos |
| 488 | for _, call := range helper { |
| 489 | if calleeName(call) == step.name && (step.arg == "" || len(call.Args) > 0 && str(call.Args[len(call.Args)-1]) == step.arg) { |
| 490 | found = call.Pos() |
| 491 | break |
| 492 | } |
| 493 | } |
| 494 | if found == token.NoPos || found <= previous { |
| 495 | t.Fatalf("shutdown helper lost synchronous order at %s(%s)", step.name, step.arg) |
| 496 | } |
| 497 | previous = found |
| 498 | } |
| 499 | |
| 500 | // The activatorStart fields the child copies, and the steps inside |
| 501 | // them. Run calls each field on its background goroutine, so inside a |
| 502 | // field each step must be synchronous. |
| 503 | fields := map[string]ast.Expr{} |
| 504 | for _, el := range lit.Elts { |
| 505 | if kv, ok := el.(*ast.KeyValueExpr); ok { |
| 506 | fields[str(kv.Key)] = kv.Value |
| 507 | } |
| 508 | } |
| 509 | for field, want := range map[string]string{ |
| 510 | "Start": "Start", "Publish": "Store", "Slot": "PrepareScheduledToken", |
| 511 | "Routes": "List", "Route": "Get", "Claims": "Enabled", |
| 512 | } { |
| 513 | if got := selName(fields[field]); got != want { |
| 514 | t.Errorf("activatorStart.%s is %s, want a %s method value", field, exprOrNil(fields[field]), want) |
| 515 | } |
| 516 | } |
| 517 | funcField := func(name string) ast.Node { |
| 518 | fl, _ := fields[name].(*ast.FuncLit) |
| 519 | if fl == nil { |
| 520 | t.Errorf("activatorStart.%s is not a func literal", name) |
| 521 | return nil |
| 522 | } |
| 523 | return fl.Body |
| 524 | } |
| 525 | if body := funcField("Recover"); body != nil { |
| 526 | requireSteps(t, "activatorStart.Recover", body, "newLayoutPass", "startupRecoveryIntent", "recoverLayoutAtStart") |
| 527 | } |
| 528 | if body := funcField("Sweep"); body != nil { |
| 529 | requireSteps(t, "activatorStart.Sweep", body, "Sweep") |
| 530 | } |
| 531 | if body := funcField("Ready"); body != nil { |
| 532 | closes := false |
| 533 | for _, c := range syncCalls(body) { |
| 534 | closes = closes || (calleeName(c) == "close" && len(c.Args) == 1 && str(c.Args[0]) == "started") |
| 535 | } |
| 536 | if !closes { |
| 537 | t.Error("activatorStart.Ready does not close started synchronously") |
| 538 | } |
| 539 | } |
| 540 | |
| 541 | // activatorStart.Run: on a legacy box, one goroutine runs the start |
| 542 | // sequence and then the key pass, in that order; start itself is |
| 543 | // Start, then Ready. |
| 544 | runBody := decls["activatorStart.Run"].Body |
| 545 | var background []*ast.FuncLit |
| 546 | ast.Inspect(runBody, func(n ast.Node) bool { |
| 547 | if g, ok := n.(*ast.GoStmt); ok { |
| 548 | if fl, ok := g.Call.Fun.(*ast.FuncLit); ok && hasCall(syncCalls(fl.Body), "startActivator") { |
| 549 | background = append(background, fl) |
| 550 | } |
| 551 | } |
| 552 | return true |
| 553 | }) |
| 554 | if len(background) != 1 { |
| 555 | t.Errorf("activatorStart.Run: %d goroutines run startActivator, want the one background start", len(background)) |
| 556 | } else { |
| 557 | requireSteps(t, "activatorStart.Run's background goroutine", background[0].Body, "startActivator", "backfillInstanceKeys") |
| 558 | } |
| 559 | var startBody ast.Node |
| 560 | ast.Inspect(runBody, func(n ast.Node) bool { |
| 561 | if a, ok := n.(*ast.AssignStmt); ok && len(a.Lhs) == 1 && len(a.Rhs) == 1 && str(a.Lhs[0]) == "start" { |
| 562 | if fl, ok := a.Rhs[0].(*ast.FuncLit); ok { |
| 563 | startBody = fl.Body |
| 564 | } |
| 565 | } |
| 566 | return true |
| 567 | }) |
| 568 | if startBody == nil { |
| 569 | t.Error("activatorStart.Run: no start closure") |
| 570 | } else { |
| 571 | requireSteps(t, "activatorStart.Run's start", startBody, "Start", "Ready") |
| 572 | } |
| 573 | |
| 574 | // startActivator: its sweep, recoverLayout and start parameters, called |
| 575 | // in that order, synchronously (matched by position, not name). |
| 576 | sa := decls["startActivator"] |
| 577 | var params []string |
| 578 | for _, f := range sa.Type.Params.List { |
| 579 | for _, n := range f.Names { |
| 580 | params = append(params, n.Name) |
| 581 | } |
| 582 | } |
| 583 | if len(params) != 4 { |
| 584 | t.Errorf("startActivator takes %v, want (ctx, sweep, recoverLayout, start)", params) |
| 585 | } else { |
| 586 | requireSteps(t, "startActivator", sa.Body, params[1], params[2], params[3]) |
| 587 | } |
| 588 | |
| 589 | // recoverLayoutAtStart: journal recovery, then hold restoration. |
| 590 | requireSteps(t, "recoverLayoutAtStart", decls["recoverLayoutAtStart"].Body, "Recover", "RestoreHolds") |
| 591 | |
| 592 | // The worker boot pass waits on the latch or the startup context, then |
| 593 | // lists and wakes. |
| 594 | var bootPass *ast.FuncLit |
| 595 | ast.Inspect(runAgent, func(n ast.Node) bool { |
| 596 | if g, ok := n.(*ast.GoStmt); ok { |
| 597 | if fl, ok := g.Call.Fun.(*ast.FuncLit); ok && hasCall(syncCalls(fl.Body), "runWorkerBootPass") { |
| 598 | bootPass = fl |
| 599 | } |
| 600 | } |
| 601 | return true |
| 602 | }) |
| 603 | if bootPass == nil { |
| 604 | t.Error("runAgent: no goroutine runs runWorkerBootPass") |
| 605 | } else if !receivesFrom(bootPass.Body, "started") || !hasCall(syncCalls(bootPass.Body), "Done") { |
| 606 | t.Error("runAgent: the worker boot pass does not wait on started and startupCtx.Done") |
| 607 | } |
| 608 | |
| 609 | // Nothing before the edge waits for startup. On runAgent's own |
| 610 | // goroutine (closures invoked on the spot included): no receive at all, |
| 611 | // no join, and a closure that waits on startup is only called directly |
| 612 | // on a claim box, where Run has already started the activator before |
| 613 | // it returns. |
| 614 | waiters := map[string]bool{} |
| 615 | ast.Inspect(runAgent, func(n ast.Node) bool { |
| 616 | if a, ok := n.(*ast.AssignStmt); ok && len(a.Lhs) == 1 && len(a.Rhs) == 1 { |
| 617 | if fl, ok := a.Rhs[0].(*ast.FuncLit); ok && (receivesFrom(fl.Body, "started") || receivesFrom(fl.Body, "startupDone") || hasCall(syncCalls(fl.Body), "joinAgentStartup")) { |
| 618 | waiters[str(a.Lhs[0])] = true |
| 619 | } |
| 620 | } |
| 621 | return true |
| 622 | }) |
| 623 | var walk func(n ast.Node, claimBranch bool) |
| 624 | walk = func(n ast.Node, claimBranch bool) { |
| 625 | ast.Inspect(n, func(n ast.Node) bool { |
| 626 | if n == nil || n.Pos() >= edgeCall.Pos() { |
| 627 | return false |
| 628 | } |
| 629 | switch n := n.(type) { |
| 630 | case *ast.FuncLit, *ast.GoStmt, *ast.DeferStmt: |
| 631 | return false |
| 632 | case *ast.IfStmt: |
| 633 | if n.Init != nil { |
| 634 | walk(n.Init, claimBranch) |
| 635 | } |
| 636 | walk(n.Cond, claimBranch) |
| 637 | walk(n.Body, claimBranch || selName(n.Cond) == "Enabled") |
| 638 | if n.Else != nil { |
| 639 | walk(n.Else, claimBranch) |
| 640 | } |
| 641 | return false |
| 642 | case *ast.UnaryExpr: |
| 643 | if n.Op == token.ARROW { |
| 644 | t.Errorf("runAgent waits on %s at %s, before the edge listens", str(n.X), fset.Position(n.Pos())) |
| 645 | } |
| 646 | case *ast.CallExpr: |
| 647 | name := calleeName(n) |
| 648 | if name == "joinAgentStartup" { |
| 649 | t.Errorf("runAgent joins startup at %s, before the edge listens", fset.Position(n.Pos())) |
| 650 | } |
| 651 | if waiters[str(n.Fun)] && !claimBranch { |
| 652 | t.Errorf("runAgent calls %s, which waits on startup, at %s, before the edge listens", str(n.Fun), fset.Position(n.Pos())) |
| 653 | } |
| 654 | if fl, ok := n.Fun.(*ast.FuncLit); ok { // invoked on the spot: its body runs here |
| 655 | walk(fl.Body, claimBranch) |
| 656 | for _, a := range n.Args { |
| 657 | walk(a, claimBranch) |
| 658 | } |
| 659 | return false |
| 660 | } |
| 661 | } |
| 662 | return true |
| 663 | }) |
| 664 | } |
| 665 | walk(runAgent, false) |
| 666 | } |
| 667 | |
| 668 | // syncCalls lists, in source order, the calls body makes on its own |
| 669 | // goroutine: none inside a go or defer statement, and none inside a |
| 670 | // function literal unless the literal is invoked right there. |
| 671 | func syncCalls(body ast.Node) []*ast.CallExpr { |
| 672 | var out []*ast.CallExpr |
| 673 | ast.Inspect(body, func(n ast.Node) bool { |
| 674 | switch n := n.(type) { |
| 675 | case *ast.GoStmt, *ast.DeferStmt, *ast.FuncLit: |
| 676 | return false |
| 677 | case *ast.CallExpr: |
| 678 | out = append(out, n) |
| 679 | if fl, ok := n.Fun.(*ast.FuncLit); ok { |
| 680 | for _, a := range n.Args { |
| 681 | out = append(out, syncCalls(a)...) |
| 682 | } |
| 683 | out = append(out, syncCalls(fl.Body)...) |
| 684 | return false |
| 685 | } |
| 686 | } |
| 687 | return true |
| 688 | }) |
| 689 | return out |
| 690 | } |
| 691 | |
| 692 | // requireSteps fails unless body calls each step synchronously, in order, |
| 693 | // and calls none of them anywhere else (in a go or defer statement, or a |
| 694 | // literal that is not invoked on the spot). |
| 695 | func requireSteps(t *testing.T, where string, body ast.Node, steps ...string) { |
| 696 | t.Helper() |
| 697 | sync := syncCalls(body) |
| 698 | inSync := map[*ast.CallExpr]bool{} |
| 699 | for _, c := range sync { |
| 700 | inSync[c] = true |
| 701 | } |
| 702 | ast.Inspect(body, func(n ast.Node) bool { |
| 703 | if c, ok := n.(*ast.CallExpr); ok && !inSync[c] { |
| 704 | for _, s := range steps { |
| 705 | if calleeName(c) == s { |
| 706 | t.Errorf("%s: %s is called off its goroutine (go, defer, or a literal not invoked there)", where, s) |
| 707 | } |
| 708 | } |
| 709 | } |
| 710 | return true |
| 711 | }) |
| 712 | i := 0 |
| 713 | for _, c := range sync { |
| 714 | if i < len(steps) && calleeName(c) == steps[i] { |
| 715 | i++ |
| 716 | } |
| 717 | } |
| 718 | if i < len(steps) { |
| 719 | t.Errorf("%s: want %v called in that order on its goroutine; %s is missing or out of order", where, steps, steps[i]) |
| 720 | } |
| 721 | } |
| 722 | |
| 723 | // calleeName is the name a call is matched on: the function's own name, |
| 724 | // whatever the receiver or package it is reached through. |
| 725 | func calleeName(c *ast.CallExpr) string { |
| 726 | switch f := c.Fun.(type) { |
| 727 | case *ast.Ident: |
| 728 | return f.Name |
| 729 | case *ast.SelectorExpr: |
| 730 | return f.Sel.Name |
| 731 | } |
| 732 | return "" |
| 733 | } |
| 734 | |
| 735 | func selName(e ast.Expr) string { |
| 736 | switch e := e.(type) { |
| 737 | case *ast.SelectorExpr: |
| 738 | return e.Sel.Name |
| 739 | case *ast.Ident: |
| 740 | return e.Name |
| 741 | } |
| 742 | return "" |
| 743 | } |
| 744 | |
| 745 | func hasCall(calls []*ast.CallExpr, name string) bool { |
| 746 | for _, c := range calls { |
| 747 | if calleeName(c) == name { |
| 748 | return true |
| 749 | } |
| 750 | } |
| 751 | return false |
| 752 | } |
| 753 | |
| 754 | // receivesFrom reports whether body receives from the channel named ch |
| 755 | // (in a select case or a bare receive). |
| 756 | func receivesFrom(body ast.Node, ch string) bool { |
| 757 | found := false |
| 758 | ast.Inspect(body, func(n ast.Node) bool { |
| 759 | if u, ok := n.(*ast.UnaryExpr); ok && u.Op == token.ARROW && types.ExprString(u.X) == ch { |
| 760 | found = true |
| 761 | } |
| 762 | return !found |
| 763 | }) |
| 764 | return found |
| 765 | } |
| 766 | |
| 767 | func exprOrNil(e ast.Expr) string { |
| 768 | if e == nil { |
| 769 | return "unset" |
| 770 | } |
| 771 | return types.ExprString(e) |
| 772 | } |
| 773 | |
| 774 | // --- the box ------------------------------------------------------------------ |
| 775 | |
| 776 | type largeBox struct { |
| 777 | dataDir, socketDir, storeDir, acmeDir string |
| 778 | reg *registry.Registry |
| 779 | control *largeBoxControl |
| 780 | httpAddr, httpsAddr string |
| 781 | client *http.Client |
| 782 | } |
| 783 | |
| 784 | // largeBoxSocketDir makes the box's run directory. It must be short: a |
| 785 | // nested instance's socket is <dir>/instances/<instance id>/app.sock, the |
| 786 | // activator binds a ".control" socket beside it, and the whole path must |
| 787 | // fit a unix socket address (provision.MaxUnixPath). The default temp dir |
| 788 | // is tried first, then a directory of the test's own directly under /tmp; |
| 789 | // when neither is short enough the test is skipped, not failed, since |
| 790 | // nothing about the code under test is wrong. |
| 791 | func largeBoxSocketDir(t *testing.T) string { |
| 792 | t.Helper() |
| 793 | fits := func(dir string) bool { |
| 794 | return len(filepath.Join(dir, "instances", provision.InstanceID(largeBoxHost(0)), "app.sock.control")) <= provision.MaxUnixPath |
| 795 | } |
| 796 | var tried []string |
| 797 | for _, base := range []string{os.TempDir(), "/tmp"} { |
| 798 | dir, err := os.MkdirTemp(base, "lb") |
| 799 | if err != nil { |
| 800 | tried = append(tried, base+": "+err.Error()) |
| 801 | continue |
| 802 | } |
| 803 | if fits(dir) { |
| 804 | t.Cleanup(func() { os.RemoveAll(dir) }) |
| 805 | return dir |
| 806 | } |
| 807 | os.RemoveAll(dir) |
| 808 | tried = append(tried, dir+": too long") |
| 809 | } |
| 810 | t.Skipf("no directory short enough for a nested instance's control socket (%d bytes at most): %s", provision.MaxUnixPath, strings.Join(tried, "; ")) |
| 811 | return "" |
| 812 | } |
| 813 | |
| 814 | // newLargeBox builds the box the agent restarts on: largeBoxRoutes |
| 815 | // sleeping instances on flat paths, largeBoxJournals of them with a layout |
| 816 | // move cut short at three different points, largeBoxHolds held down after |
| 817 | // a refused move (the reason, a second name for the database, persists), |
| 818 | // a scheduled delivery due on one instance, an always-on worker the boot |
| 819 | // pass starts, a nested layout intent in the deployment store, a cached |
| 820 | // certificate for every host the test asks for, and a cold instance |
| 821 | // control. |
| 822 | func newLargeBox(t *testing.T) *largeBox { |
| 823 | t.Helper() |
| 824 | sd := largeBoxSocketDir(t) |
| 825 | b := &largeBox{dataDir: t.TempDir(), storeDir: t.TempDir(), acmeDir: t.TempDir(), socketDir: sd} |
| 826 | |
| 827 | var err error |
| 828 | b.reg, err = registry.Open(filepath.Join(b.dataDir, "registry.db")) |
| 829 | if err != nil { |
| 830 | t.Fatal(err) |
| 831 | } |
| 832 | t.Cleanup(func() { b.reg.Close() }) |
| 833 | var hosts []string |
| 834 | for i := 0; i < largeBoxRoutes; i++ { |
| 835 | h := largeBoxHost(i) |
| 836 | hosts = append(hosts, h) |
| 837 | if err := b.reg.Upsert(registry.Route{ |
| 838 | Host: h, Kind: "instance", Account: "acct", App: "shop", Status: "ready", Hibernate: true, |
| 839 | Socket: filepath.Join(sd, h+".sock"), DBPath: provision.DefaultDBPath(b.dataDir, h), |
| 840 | }); err != nil { |
| 841 | t.Fatal(err) |
| 842 | } |
| 843 | } |
| 844 | if err := b.reg.Upsert(registry.Route{ |
| 845 | Host: largeBoxWorker, Kind: "sidecar", Account: "acct", App: "shop", Status: "ready", Args: `["serve"]`, |
| 846 | }); err != nil { |
| 847 | t.Fatal(err) |
| 848 | } |
| 849 | due := time.Now().Unix() - 5 |
| 850 | if err := b.reg.UpsertTimer(registry.Timer{ |
| 851 | Host: largeBoxHost(largeBoxDueHost), Name: "tick", Kind: schedule.KindSchedule, Spec: "every 6h0m0s", |
| 852 | Path: largeBoxDuePath, NextAt: due, DueAt: due, |
| 853 | }); err != nil { |
| 854 | t.Fatal(err) |
| 855 | } |
| 856 | for i := 0; i < largeBoxJournals; i++ { |
| 857 | crash := "after-journal" |
| 858 | switch { |
| 859 | case i >= 6: |
| 860 | crash = "before-remove" |
| 861 | case i >= 4: |
| 862 | crash = "mid-files" |
| 863 | } |
| 864 | b.crashedMove(t, largeBoxHost(i), crash) |
| 865 | } |
| 866 | for i := largeBoxJournals; i < largeBoxJournals+largeBoxHolds; i++ { |
| 867 | b.heldMove(t, largeBoxHost(i), i) |
| 868 | } |
| 869 | appendIntent(t, release.NewDirStore(b.storeDir), release.LayoutModeNested, "") |
| 870 | |
| 871 | served := []string{largeBoxHost(largeBoxWakeHost), largeBoxHost(largeBoxSidecarHost)} |
| 872 | for i := 0; i < largeBoxJournals+largeBoxHolds; i++ { |
| 873 | served = append(served, largeBoxHost(i)) |
| 874 | } |
| 875 | pool := writeLargeBoxCerts(t, b.acmeDir, served) |
| 876 | |
| 877 | b.control = startLargeBoxControl(t, filepath.Join(sd, "control.sock"), hosts) |
| 878 | b.client = &http.Client{ |
| 879 | Timeout: largeBoxLatchBound + 15*time.Second, |
| 880 | Transport: &http.Transport{ |
| 881 | Proxy: nil, |
| 882 | DisableKeepAlives: true, |
| 883 | TLSClientConfig: &tls.Config{RootCAs: pool}, |
| 884 | DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { |
| 885 | return (&net.Dialer{}).DialContext(ctx, "tcp", b.httpsAddr) |
| 886 | }, |
| 887 | }, |
| 888 | CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, |
| 889 | } |
| 890 | return b |
| 891 | } |
| 892 | |
| 893 | func (b *largeBox) seedFlat(t *testing.T, h string) { |
| 894 | t.Helper() |
| 895 | db := provision.DefaultDBPath(b.dataDir, h) |
| 896 | for path, body := range map[string]string{db: "db " + h, db + "-wal": "wal", filepath.Join(b.dataDir, h+".env"): "CARLOS_ADMIN_TOKEN=t\n"} { |
| 897 | if err := os.WriteFile(path, []byte(body), 0o600); err != nil { |
| 898 | t.Fatal(err) |
| 899 | } |
| 900 | } |
| 901 | } |
| 902 | |
| 903 | // crashedMove leaves h as an agent killed mid-migration to nested leaves |
| 904 | // it: the instance directories made and the journal written, then |
| 905 | // nothing moved (after-journal), the database moved (mid-files), or |
| 906 | // everything moved and the row flipped with only the journal's removal |
| 907 | // left (before-remove). |
| 908 | func (b *largeBox) crashedMove(t *testing.T, h, crash string) { |
| 909 | t.Helper() |
| 910 | b.seedFlat(t, h) |
| 911 | if err := layout.EnsureDirs(b.dataDir, b.socketDir, h, -1); err != nil { |
| 912 | t.Fatal(err) |
| 913 | } |
| 914 | plan, err := layout.PlanFor(b.dataDir, b.socketDir, h, layout.ModeNested) |
| 915 | if err != nil { |
| 916 | t.Fatal(err) |
| 917 | } |
| 918 | present := make([]bool, len(plan.Artifacts)) |
| 919 | for i, a := range plan.Artifacts { |
| 920 | if a.Name == "socket" { |
| 921 | continue |
| 922 | } |
| 923 | _, err := os.Lstat(a.Src) |
| 924 | present[i] = err == nil |
| 925 | } |
| 926 | j := plan.Journal(present) |
| 927 | if err := layout.WriteJournal(b.dataDir, j); err != nil { |
| 928 | t.Fatal(err) |
| 929 | } |
| 930 | move := func(a layout.Artifact) { |
| 931 | if err := os.Rename(a.Src, a.Dst); err != nil { |
| 932 | t.Fatal(err) |
| 933 | } |
| 934 | } |
| 935 | switch crash { |
| 936 | case "mid-files": |
| 937 | move(plan.Artifacts[0]) |
| 938 | case "before-remove": |
| 939 | for i, a := range plan.Artifacts { |
| 940 | if present[i] { |
| 941 | move(a) |
| 942 | } |
| 943 | } |
| 944 | if err := b.reg.SetInstancePaths(h, plan.From.DB, plan.From.Socket, plan.To.DB, plan.To.Socket); err != nil { |
| 945 | t.Fatal(err) |
| 946 | } |
| 947 | if err := layout.UpdatePhase(b.dataDir, j, layout.PhaseDone); err != nil { |
| 948 | t.Fatal(err) |
| 949 | } |
| 950 | } |
| 951 | } |
| 952 | |
| 953 | // heldMove leaves h held down after a refused move to nested: its hold |
| 954 | // marker, and the reason, which persists across the restart (a second |
| 955 | // name for its database, which a flat tenant can plant). |
| 956 | func (b *largeBox) heldMove(t *testing.T, h string, i int) { |
| 957 | t.Helper() |
| 958 | b.seedFlat(t, h) |
| 959 | if err := os.Link(provision.DefaultDBPath(b.dataDir, h), filepath.Join(b.dataDir, ".x"+strconv.Itoa(i))); err != nil { |
| 960 | t.Fatal(err) |
| 961 | } |
| 962 | dir := filepath.Join(b.dataDir, "layout") |
| 963 | if err := os.MkdirAll(dir, 0o700); err != nil { |
| 964 | t.Fatal(err) |
| 965 | } |
| 966 | marker, _ := json.Marshal(map[string]any{"host": h, "reason": "db has 2 names", "since": time.Now().UTC()}) |
| 967 | if err := os.WriteFile(filepath.Join(dir, h+".hold"), marker, 0o600); err != nil { |
| 968 | t.Fatal(err) |
| 969 | } |
| 970 | } |
| 971 | |
| 972 | // assertRecovered checks the files: every journal completed and removed, |
| 973 | // each journalled host's row and database at the nested layout, and every |
| 974 | // hold marker kept. |
| 975 | func (b *largeBox) assertRecovered(t *testing.T) { |
| 976 | t.Helper() |
| 977 | if hosts, err := layout.JournalHosts(b.dataDir); err != nil || len(hosts) != 0 { |
| 978 | t.Errorf("journals after startup: %v %v", hosts, err) |
| 979 | } |
| 980 | for i := 0; i < largeBoxJournals; i++ { |
| 981 | h := largeBoxHost(i) |
| 982 | plan, err := layout.PlanFor(b.dataDir, b.socketDir, h, layout.ModeNested) |
| 983 | if err != nil { |
| 984 | t.Fatal(err) |
| 985 | } |
| 986 | rt, err := b.reg.Get(h) |
| 987 | if err != nil || rt == nil || rt.DBPath != plan.To.DB || rt.Socket != plan.To.Socket { |
| 988 | t.Errorf("%s row after recovery: %+v %v", h, rt, err) |
| 989 | } |
| 990 | if body, err := os.ReadFile(plan.Artifacts[0].Dst); err != nil || string(body) != "db "+h { |
| 991 | t.Errorf("%s database after recovery: %q %v", h, body, err) |
| 992 | } |
| 993 | } |
| 994 | held, err := layout.HoldMarkers(b.dataDir) |
| 995 | if err != nil || len(held) != largeBoxHolds { |
| 996 | t.Errorf("hold markers after startup: %v %v", held, err) |
| 997 | } |
| 998 | } |
| 999 | |
| 1000 | // assertNoWrongService reads the backend's record: no wake admitted before |
| 1001 | // the latch opened, no held host started, and no journalled host started |
| 1002 | // at its pre-recovery socket. |
| 1003 | func (b *largeBox) assertNoWrongService(t *testing.T, a *largeBoxAgent) { |
| 1004 | t.Helper() |
| 1005 | if v := a.events("violation"); len(v) != 0 { |
| 1006 | t.Errorf("backend violations: %v", v) |
| 1007 | } |
| 1008 | for _, e := range append(a.events("acquire"), a.events("start")...) { |
| 1009 | for i := largeBoxJournals; i < largeBoxJournals+largeBoxHolds; i++ { |
| 1010 | if e.arg(0) == largeBoxHost(i) { |
| 1011 | t.Errorf("held host %s reached the backend: %v", e.arg(0), e) |
| 1012 | } |
| 1013 | } |
| 1014 | } |
| 1015 | for _, e := range a.events("start") { |
| 1016 | for i := 0; i < largeBoxJournals; i++ { |
| 1017 | if h := largeBoxHost(i); e.arg(0) == h && e.arg(1) == filepath.Join(b.socketDir, h+".sock") { |
| 1018 | t.Errorf("%s started at its pre-recovery socket", h) |
| 1019 | } |
| 1020 | } |
| 1021 | } |
| 1022 | } |
| 1023 | |
| 1024 | type largeBoxAnswer struct { |
| 1025 | status int |
| 1026 | body string |
| 1027 | header http.Header |
| 1028 | err error |
| 1029 | at time.Time // when the answer was read |
| 1030 | } |
| 1031 | |
| 1032 | // get asks the edge for host's root over TLS. sidecar marks the request |
| 1033 | // as the host's sidecar's (activator.SidecarHeader). |
| 1034 | func (b *largeBox) get(host string, sidecar bool) largeBoxAnswer { |
| 1035 | req, err := http.NewRequest(http.MethodGet, "https://"+host+"/", nil) |
| 1036 | if err != nil { |
| 1037 | return largeBoxAnswer{err: err, at: time.Now()} |
| 1038 | } |
| 1039 | if sidecar { |
| 1040 | req.Header.Set(activator.SidecarHeader, "1") |
| 1041 | } |
| 1042 | resp, err := b.client.Do(req) |
| 1043 | if err != nil { |
| 1044 | return largeBoxAnswer{err: err, at: time.Now()} |
| 1045 | } |
| 1046 | defer resp.Body.Close() |
| 1047 | body, err := io.ReadAll(resp.Body) |
| 1048 | return largeBoxAnswer{status: resp.StatusCode, body: string(body), header: resp.Header, err: err, at: time.Now()} |
| 1049 | } |
| 1050 | |
| 1051 | func (b *largeBox) fire(host string, sidecar bool) chan largeBoxAnswer { |
| 1052 | ch := make(chan largeBoxAnswer, 1) |
| 1053 | go func() { ch <- b.get(host, sidecar) }() |
| 1054 | return ch |
| 1055 | } |
| 1056 | |
| 1057 | func (b *largeBox) answer(t *testing.T, ch chan largeBoxAnswer) largeBoxAnswer { |
| 1058 | t.Helper() |
| 1059 | select { |
| 1060 | case got := <-ch: |
| 1061 | return got |
| 1062 | case <-time.After(10 * time.Second): |
| 1063 | t.Fatal("request did not finish") |
| 1064 | return largeBoxAnswer{} |
| 1065 | } |
| 1066 | } |
| 1067 | |
| 1068 | func freeLoopbackAddr(t *testing.T) string { |
| 1069 | t.Helper() |
| 1070 | ln, err := net.Listen("tcp", "127.0.0.1:0") |
| 1071 | if err != nil { |
| 1072 | t.Fatal(err) |
| 1073 | } |
| 1074 | defer ln.Close() |
| 1075 | return ln.Addr().String() |
| 1076 | } |
| 1077 | |
| 1078 | // writeLargeBoxCerts puts one self-signed certificate for hosts into the |
| 1079 | // edge's autocert directory cache, under each host's name, so the edge in |
| 1080 | // production mode serves them TLS without ordering anything. |
| 1081 | func writeLargeBoxCerts(t *testing.T, dir string, hosts []string) *x509.CertPool { |
| 1082 | t.Helper() |
| 1083 | key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) |
| 1084 | if err != nil { |
| 1085 | t.Fatal(err) |
| 1086 | } |
| 1087 | tmpl := &x509.Certificate{ |
| 1088 | SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "large box"}, DNSNames: hosts, |
| 1089 | NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour), |
| 1090 | KeyUsage: x509.KeyUsageDigitalSignature | x509.KeyUsageCertSign, |
| 1091 | ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, BasicConstraintsValid: true, IsCA: true, |
| 1092 | } |
| 1093 | der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) |
| 1094 | if err != nil { |
| 1095 | t.Fatal(err) |
| 1096 | } |
| 1097 | keyDER, err := x509.MarshalECPrivateKey(key) |
| 1098 | if err != nil { |
| 1099 | t.Fatal(err) |
| 1100 | } |
| 1101 | pemBytes := append(pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER}), pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})...) |
| 1102 | for _, h := range hosts { |
| 1103 | if err := os.WriteFile(filepath.Join(dir, h), pemBytes, 0o600); err != nil { |
| 1104 | t.Fatal(err) |
| 1105 | } |
| 1106 | } |
| 1107 | cert, err := x509.ParseCertificate(der) |
| 1108 | if err != nil { |
| 1109 | t.Fatal(err) |
| 1110 | } |
| 1111 | pool := x509.NewCertPool() |
| 1112 | pool.AddCert(cert) |
| 1113 | return pool |
| 1114 | } |
| 1115 | |
| 1116 | // --- instance control --------------------------------------------------------- |
| 1117 | |
| 1118 | // largeBoxControl is the real instance-control server with a cold systemd |
| 1119 | // behind its Systemctl seam: every call waits for warm, then the box's |
| 1120 | // leftover units are listed active and each stop takes largeBoxUnitStop. |
| 1121 | type largeBoxControl struct { |
| 1122 | warm chan struct{} |
| 1123 | units string |
| 1124 | stops atomic.Int32 |
| 1125 | } |
| 1126 | |
| 1127 | func startLargeBoxControl(t *testing.T, path string, hosts []string) *largeBoxControl { |
| 1128 | t.Helper() |
| 1129 | c := &largeBoxControl{warm: make(chan struct{})} |
| 1130 | var units strings.Builder |
| 1131 | for _, h := range hosts { |
| 1132 | fmt.Fprintf(&units, "%s loaded active running carlos instance\n", instancectl.UnitName(instancectl.RoleRun, h)) |
| 1133 | } |
| 1134 | c.units = units.String() |
| 1135 | srv := &instancectl.Server{ |
| 1136 | Systemctl: c.systemctl, |
| 1137 | PeerOK: func(net.Conn) error { return nil }, |
| 1138 | CgroupRoot: t.TempDir(), |
| 1139 | Build: version.Build, |
| 1140 | Logf: func(string, ...any) {}, |
| 1141 | } |
| 1142 | ln, err := net.Listen("unix", path) |
| 1143 | if err != nil { |
| 1144 | t.Fatal(err) |
| 1145 | } |
| 1146 | ctx, cancel := context.WithCancel(context.Background()) |
| 1147 | done := make(chan struct{}) |
| 1148 | go func() { defer close(done); _ = srv.Serve(ctx, ln) }() |
| 1149 | t.Cleanup(func() { cancel(); <-done }) |
| 1150 | return c |
| 1151 | } |
| 1152 | |
| 1153 | func (c *largeBoxControl) systemctl(ctx context.Context, args ...string) (string, error) { |
| 1154 | select { |
| 1155 | case <-c.warm: |
| 1156 | case <-ctx.Done(): |
| 1157 | return "", ctx.Err() |
| 1158 | } |
| 1159 | switch args[0] { |
| 1160 | case "list-units": |
| 1161 | return c.units, nil |
| 1162 | case "list-jobs": |
| 1163 | return "", nil |
| 1164 | case "stop": |
| 1165 | select { |
| 1166 | case <-time.After(largeBoxUnitStop): |
| 1167 | case <-ctx.Done(): |
| 1168 | return "", ctx.Err() |
| 1169 | } |
| 1170 | c.stops.Add(1) |
| 1171 | return "", nil |
| 1172 | case "show": |
| 1173 | return "inactive\n", nil |
| 1174 | } |
| 1175 | return "", fmt.Errorf("systemctl %v: not part of a sweep", args) |
| 1176 | } |
| 1177 | |
| 1178 | // --- the agent process ---------------------------------------------------------- |
| 1179 | |
| 1180 | type largeBoxAgent struct { |
| 1181 | cmd *exec.Cmd |
| 1182 | began time.Time |
| 1183 | listen time.Duration // process start to the gate probe's first answer |
| 1184 | eventsPath string |
| 1185 | log string |
| 1186 | exited chan struct{} // closed once the process has been waited for |
| 1187 | exitErr error // Wait's answer, read after exited closes |
| 1188 | b *largeBox |
| 1189 | } |
| 1190 | |
| 1191 | // startAgent starts the agent and returns it once the gate's probe is |
| 1192 | // answered and its TLS listener accepts. The edge's -http and -https take |
| 1193 | // an address, and :0 would bind, but the edge never says which port it |
| 1194 | // got, so the test picks free ports and hands them over. A port taken in |
| 1195 | // between makes the edge's listener fail and the agent exit; only that is |
| 1196 | // retried on fresh ports (largeBoxRetryable). A deadline missed is the |
| 1197 | // failure this test exists to catch, and is never retried. |
| 1198 | func (b *largeBox) startAgent(t *testing.T) *largeBoxAgent { |
| 1199 | t.Helper() |
| 1200 | for attempt := 1; ; attempt++ { |
| 1201 | b.httpAddr, b.httpsAddr = freeLoopbackAddr(t), freeLoopbackAddr(t) |
| 1202 | a := b.launchAgent(t) |
| 1203 | outcome, err := a.awaitListening() |
| 1204 | if outcome == largeBoxUp { |
| 1205 | outcome, err = a.awaitTLS() |
| 1206 | } |
| 1207 | if outcome == largeBoxUp { |
| 1208 | return a |
| 1209 | } |
| 1210 | agentLog, _ := os.ReadFile(a.log) |
| 1211 | if largeBoxRetryable(outcome, string(agentLog), b.httpAddr, b.httpsAddr) && attempt < largeBoxBindAttempts { |
| 1212 | t.Logf("attempt %d: %v; a port the test handed the agent was taken before it bound it, starting again on fresh ports", attempt, err) |
| 1213 | continue |
| 1214 | } |
| 1215 | t.Fatal(err) |
| 1216 | } |
| 1217 | } |
| 1218 | |
| 1219 | // largeBoxOutcome is how waiting for the agent to come up ended. |
| 1220 | type largeBoxOutcome int |
| 1221 | |
| 1222 | const ( |
| 1223 | largeBoxUp largeBoxOutcome = iota |
| 1224 | largeBoxLate // a deadline passed: a failure, never retried |
| 1225 | largeBoxExited // the agent exited before it was up |
| 1226 | ) |
| 1227 | |
| 1228 | // largeBoxRetryable reports whether a start may be retried on fresh |
| 1229 | // ports: only when the agent exited, and its log says the edge could not |
| 1230 | // bind one of the addresses the test handed it because it was in use. |
| 1231 | // Edge listener failures are logged as "edge: http listener <addr>: ..." |
| 1232 | // (edge.RunWithActivatorReady). |
| 1233 | func largeBoxRetryable(o largeBoxOutcome, agentLog string, addrs ...string) bool { |
| 1234 | if o != largeBoxExited { |
| 1235 | return false |
| 1236 | } |
| 1237 | for _, line := range strings.Split(agentLog, "\n") { |
| 1238 | if !strings.Contains(line, "address already in use") { |
| 1239 | continue |
| 1240 | } |
| 1241 | for _, addr := range addrs { |
| 1242 | if strings.Contains(line, " listener "+addr+": ") { |
| 1243 | return true |
| 1244 | } |
| 1245 | } |
| 1246 | } |
| 1247 | return false |
| 1248 | } |
| 1249 | |
| 1250 | func TestLargeBoxRetriesOnlyItsOwnBindFailure(t *testing.T) { |
| 1251 | const plain, tls = "127.0.0.1:40001", "127.0.0.1:40002" |
| 1252 | bind := "edge: https listener " + tls + ": listen tcp " + tls + ": bind: address already in use\n" |
| 1253 | for _, c := range []struct { |
| 1254 | name string |
| 1255 | o largeBoxOutcome |
| 1256 | log string |
| 1257 | want bool |
| 1258 | }{ |
| 1259 | {"own bind failure", largeBoxExited, "edge: serving 300 route(s)\n" + bind, true}, |
| 1260 | {"deadline missed, bind failure logged too", largeBoxLate, bind, false}, |
| 1261 | {"exited for another reason", largeBoxExited, "edge: tunnels: invalid\n", false}, |
| 1262 | {"someone else's address", largeBoxExited, "edge: http listener 127.0.0.1:1: listen tcp 127.0.0.1:1: bind: address already in use\n", false}, |
| 1263 | {"in use, but not a listener line", largeBoxExited, "agent: something: address already in use\n", false}, |
| 1264 | } { |
| 1265 | if got := largeBoxRetryable(c.o, c.log, plain, tls); got != c.want { |
| 1266 | t.Errorf("%s: retry = %v, want %v", c.name, got, c.want) |
| 1267 | } |
| 1268 | } |
| 1269 | } |
| 1270 | |
| 1271 | func (b *largeBox) launchAgent(t *testing.T) *largeBoxAgent { |
| 1272 | t.Helper() |
| 1273 | cfgDir := t.TempDir() |
| 1274 | cfg := largeBoxConfig{ |
| 1275 | DataDir: b.dataDir, SocketDir: b.socketDir, StoreDir: b.storeDir, AcmeDir: b.acmeDir, |
| 1276 | Control: filepath.Join(b.socketDir, "control.sock"), Events: filepath.Join(cfgDir, "events"), |
| 1277 | HTTP: b.httpAddr, HTTPS: b.httpsAddr, |
| 1278 | } |
| 1279 | raw, err := json.Marshal(cfg) |
| 1280 | if err != nil { |
| 1281 | t.Fatal(err) |
| 1282 | } |
| 1283 | cfgPath := filepath.Join(cfgDir, "agent.json") |
| 1284 | if err := os.WriteFile(cfgPath, raw, 0o600); err != nil { |
| 1285 | t.Fatal(err) |
| 1286 | } |
| 1287 | exe, err := os.Executable() |
| 1288 | if err != nil { |
| 1289 | t.Fatal(err) |
| 1290 | } |
| 1291 | a := &largeBoxAgent{eventsPath: cfg.Events, log: filepath.Join(cfgDir, "agent.log"), exited: make(chan struct{}), b: b} |
| 1292 | logFile, err := os.Create(a.log) |
| 1293 | if err != nil { |
| 1294 | t.Fatal(err) |
| 1295 | } |
| 1296 | t.Cleanup(func() { logFile.Close() }) |
| 1297 | a.cmd = exec.Command(exe, "-test.run=^TestStartupLargeBoxEndToEnd$", "-test.count=1") |
| 1298 | // A clean environment: nothing of the developer's or the runner's |
| 1299 | // (a deployment bucket, AWS credentials, CARLOS_*) reaches the agent. |
| 1300 | a.cmd.Env = []string{"HOME=" + b.dataDir, "TMPDIR=" + os.TempDir(), "CARLOS_DATA_DIR=" + b.dataDir, largeBoxAgentEnv + "=" + cfgPath} |
| 1301 | a.cmd.Stdout, a.cmd.Stderr = logFile, logFile |
| 1302 | a.began = time.Now() |
| 1303 | if err := a.cmd.Start(); err != nil { |
| 1304 | t.Fatal(err) |
| 1305 | } |
| 1306 | go func() { |
| 1307 | a.exitErr = a.cmd.Wait() |
| 1308 | close(a.exited) |
| 1309 | }() |
| 1310 | t.Cleanup(func() { |
| 1311 | select { |
| 1312 | case <-a.exited: |
| 1313 | default: |
| 1314 | _ = a.cmd.Process.Kill() |
| 1315 | <-a.exited |
| 1316 | } |
| 1317 | if t.Failed() { |
| 1318 | if out, err := os.ReadFile(a.log); err == nil { |
| 1319 | if len(out) > 16<<10 { |
| 1320 | out = out[len(out)-16<<10:] |
| 1321 | } |
| 1322 | t.Logf("agent log (tail):\n%s", out) |
| 1323 | } |
| 1324 | } |
| 1325 | }) |
| 1326 | return a |
| 1327 | } |
| 1328 | |
| 1329 | // awaitListening probes as the update gate does (systemapply.go |
| 1330 | // productionSysOps.probeEdge): GET / on the edge's plain-HTTP listener, |
| 1331 | // three-second client timeout, no redirect followed, and any HTTP answer |
| 1332 | // passes. It records how long after the process started the first answer |
| 1333 | // came; an answer past largeBoxListenBound, or none by then, is late. |
| 1334 | func (a *largeBoxAgent) awaitListening() (largeBoxOutcome, error) { |
| 1335 | probe := &http.Client{ |
| 1336 | Timeout: 3 * time.Second, |
| 1337 | Transport: &http.Transport{Proxy: nil, DisableKeepAlives: true}, |
| 1338 | CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }, |
| 1339 | } |
| 1340 | deadline := a.began.Add(largeBoxListenBound) |
| 1341 | for { |
| 1342 | resp, err := probe.Get("http://" + a.b.httpAddr + "/") |
| 1343 | if err == nil { |
| 1344 | resp.Body.Close() |
| 1345 | a.listen = time.Since(a.began) |
| 1346 | if a.listen > largeBoxListenBound { |
| 1347 | return largeBoxLate, fmt.Errorf("gate probe answered %s after the agent started, over %s", a.listen, largeBoxListenBound) |
| 1348 | } |
| 1349 | return largeBoxUp, nil |
| 1350 | } |
| 1351 | if time.Now().After(deadline) { |
| 1352 | return largeBoxLate, fmt.Errorf("gate probe unanswered %s after the agent started: %v", time.Since(a.began), err) |
| 1353 | } |
| 1354 | select { |
| 1355 | case <-a.exited: |
| 1356 | return largeBoxExited, fmt.Errorf("agent exited before listening: %v", a.exitErr) |
| 1357 | default: |
| 1358 | } |
| 1359 | time.Sleep(20 * time.Millisecond) |
| 1360 | } |
| 1361 | } |
| 1362 | |
| 1363 | // awaitTLS waits, boundedly, for the edge's TLS listener: the edge binds |
| 1364 | // it in its own goroutine, beside the plain-HTTP one the gate probes, so |
| 1365 | // it can come up a moment later. Not part of the gate's measurement. |
| 1366 | func (a *largeBoxAgent) awaitTLS() (largeBoxOutcome, error) { |
| 1367 | deadline := time.Now().Add(largeBoxListenBound) |
| 1368 | for { |
| 1369 | c, err := net.DialTimeout("tcp", a.b.httpsAddr, time.Second) |
| 1370 | if err == nil { |
| 1371 | c.Close() |
| 1372 | } |
| 1373 | select { |
| 1374 | case <-a.exited: |
| 1375 | return largeBoxExited, fmt.Errorf("agent exited before its TLS listener was up: %v", a.exitErr) |
| 1376 | default: |
| 1377 | } |
| 1378 | if err == nil { |
| 1379 | return largeBoxUp, nil |
| 1380 | } |
| 1381 | if time.Now().After(deadline) { |
| 1382 | return largeBoxLate, fmt.Errorf("TLS listener not up %s after the gate probe: %v", largeBoxListenBound, err) |
| 1383 | } |
| 1384 | time.Sleep(20 * time.Millisecond) |
| 1385 | } |
| 1386 | } |
| 1387 | |
| 1388 | func (a *largeBoxAgent) signal(t *testing.T, sig os.Signal) { |
| 1389 | t.Helper() |
| 1390 | if err := a.cmd.Process.Signal(sig); err != nil { |
| 1391 | t.Fatal(err) |
| 1392 | } |
| 1393 | } |
| 1394 | |
| 1395 | func (a *largeBoxAgent) waitExit(t *testing.T, bound time.Duration) { |
| 1396 | t.Helper() |
| 1397 | select { |
| 1398 | case <-a.exited: |
| 1399 | if a.exitErr != nil { |
| 1400 | t.Fatalf("agent exit: %v", a.exitErr) |
| 1401 | } |
| 1402 | case <-time.After(bound): |
| 1403 | t.Fatalf("agent did not exit within %s of the signal", bound) |
| 1404 | } |
| 1405 | } |
| 1406 | |
| 1407 | // assertNeverStarted: startup never finished, so the activator never |
| 1408 | // started and nothing reached the backend or ran. |
| 1409 | func (a *largeBoxAgent) assertNeverStarted(t *testing.T) { |
| 1410 | t.Helper() |
| 1411 | for _, kind := range []string{"ready", "activator-start", "acquire", "start", "worker-start", "served"} { |
| 1412 | if evs := a.events(kind); len(evs) != 0 { |
| 1413 | t.Errorf("%s with startup unfinished: %v", kind, evs) |
| 1414 | } |
| 1415 | } |
| 1416 | } |
| 1417 | |
| 1418 | // assertAfter waits briefly for the event kind with args and fails unless |
| 1419 | // it came after ref. |
| 1420 | func (a *largeBoxAgent) assertAfter(t *testing.T, ref largeBoxEvent, kind string, args ...string) { |
| 1421 | t.Helper() |
| 1422 | e := a.waitEvent(t, 10*time.Second, kind, args...) |
| 1423 | if e.at.Before(ref.at) { |
| 1424 | t.Errorf("%s %v came %s before %s", kind, args, ref.at.Sub(e.at), ref.kind) |
| 1425 | } |
| 1426 | } |
| 1427 | |
| 1428 | // assertCleanShutdown reads the agent's own account of its shutdown: the |
| 1429 | // startup work joined within runAgent's bound, and no goroutine of the |
| 1430 | // startup sequence, the activator or the fake instances left after it. |
| 1431 | func (a *largeBoxAgent) assertCleanShutdown(t *testing.T) { |
| 1432 | t.Helper() |
| 1433 | j, ok := a.event("joined") |
| 1434 | if !ok { |
| 1435 | t.Fatal("agent did not report its startup join") |
| 1436 | } |
| 1437 | took, err := time.ParseDuration(j.arg(1)) |
| 1438 | if j.arg(0) != "true" || err != nil || took > largeBoxJoinBound { |
| 1439 | t.Fatalf("startup join: %v", j) |
| 1440 | } |
| 1441 | t.Logf("startup work joined %s after the edge returned", took) |
| 1442 | if _, ok := a.event("leak-free"); !ok { |
| 1443 | t.Fatal("goroutines of the startup sequence outlived shutdown; see the agent log") |
| 1444 | } |
| 1445 | } |
| 1446 | |
| 1447 | type largeBoxEvent struct { |
| 1448 | at time.Time |
| 1449 | kind string |
| 1450 | args []string |
| 1451 | } |
| 1452 | |
| 1453 | func (e largeBoxEvent) arg(i int) string { |
| 1454 | if i < len(e.args) { |
| 1455 | return e.args[i] |
| 1456 | } |
| 1457 | return "" |
| 1458 | } |
| 1459 | |
| 1460 | func (a *largeBoxAgent) all() []largeBoxEvent { |
| 1461 | f, err := os.Open(a.eventsPath) |
| 1462 | if err != nil { |
| 1463 | return nil |
| 1464 | } |
| 1465 | defer f.Close() |
| 1466 | var out []largeBoxEvent |
| 1467 | sc := bufio.NewScanner(f) |
| 1468 | for sc.Scan() { |
| 1469 | fields := strings.Fields(sc.Text()) |
| 1470 | if len(fields) < 2 { |
| 1471 | continue |
| 1472 | } |
| 1473 | ns, err := strconv.ParseInt(fields[0], 10, 64) |
| 1474 | if err != nil { |
| 1475 | continue |
| 1476 | } |
| 1477 | out = append(out, largeBoxEvent{at: time.Unix(0, ns), kind: fields[1], args: fields[2:]}) |
| 1478 | } |
| 1479 | return out |
| 1480 | } |
| 1481 | |
| 1482 | func (a *largeBoxAgent) events(kind string) []largeBoxEvent { |
| 1483 | var out []largeBoxEvent |
| 1484 | for _, e := range a.all() { |
| 1485 | if e.kind == kind { |
| 1486 | out = append(out, e) |
| 1487 | } |
| 1488 | } |
| 1489 | return out |
| 1490 | } |
| 1491 | |
| 1492 | func (a *largeBoxAgent) event(kind string, args ...string) (largeBoxEvent, bool) { |
| 1493 | for _, e := range a.events(kind) { |
| 1494 | match := true |
| 1495 | for i, want := range args { |
| 1496 | if e.arg(i) != want { |
| 1497 | match = false |
| 1498 | } |
| 1499 | } |
| 1500 | if match { |
| 1501 | return e, true |
| 1502 | } |
| 1503 | } |
| 1504 | return largeBoxEvent{}, false |
| 1505 | } |
| 1506 | |
| 1507 | func (a *largeBoxAgent) waitEvent(t *testing.T, bound time.Duration, kind string, args ...string) largeBoxEvent { |
| 1508 | t.Helper() |
| 1509 | deadline := time.Now().Add(bound) |
| 1510 | for { |
| 1511 | if e, ok := a.event(kind, args...); ok { |
| 1512 | return e |
| 1513 | } |
| 1514 | if time.Now().After(deadline) { |
| 1515 | t.Fatalf("agent did not report %s %v within %s", kind, args, bound) |
| 1516 | } |
| 1517 | time.Sleep(10 * time.Millisecond) |
| 1518 | } |
| 1519 | } |
| 1520 | |
| 1521 | // --- inside the agent process --------------------------------------------------- |
| 1522 | |
| 1523 | type largeBoxEvents struct { |
| 1524 | mu sync.Mutex |
| 1525 | f *os.File |
| 1526 | } |
| 1527 | |
| 1528 | func (e *largeBoxEvents) log(kind string, args ...any) { |
| 1529 | var b strings.Builder |
| 1530 | fmt.Fprintf(&b, "%d %s", time.Now().UnixNano(), kind) |
| 1531 | for _, a := range args { |
| 1532 | fmt.Fprintf(&b, " %v", a) |
| 1533 | } |
| 1534 | b.WriteByte('\n') |
| 1535 | e.mu.Lock() |
| 1536 | defer e.mu.Unlock() |
| 1537 | _, _ = e.f.WriteString(b.String()) |
| 1538 | } |
| 1539 | |
| 1540 | // runLargeBoxAgent is the agent: runAgent's legacy-box startup over the |
| 1541 | // fixture, then the edge, then runAgent's shutdown. It returns the exit |
| 1542 | // code. |
| 1543 | func runLargeBoxAgent(cfgPath string) int { |
| 1544 | raw, err := os.ReadFile(cfgPath) |
| 1545 | if err != nil { |
| 1546 | log.Printf("largebox agent: %v", err) |
| 1547 | return 2 |
| 1548 | } |
| 1549 | var cfg largeBoxConfig |
| 1550 | if err := json.Unmarshal(raw, &cfg); err != nil { |
| 1551 | log.Printf("largebox agent: %v", err) |
| 1552 | return 2 |
| 1553 | } |
| 1554 | f, err := os.OpenFile(cfg.Events, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600) |
| 1555 | if err != nil { |
| 1556 | log.Printf("largebox agent: %v", err) |
| 1557 | return 2 |
| 1558 | } |
| 1559 | defer f.Close() |
| 1560 | ev := &largeBoxEvents{f: f} |
| 1561 | |
| 1562 | agentCtx, cancelAgent := context.WithCancel(context.Background()) |
| 1563 | defer cancelAgent() |
| 1564 | startupCtx, cancelStartup := context.WithCancel(agentCtx) |
| 1565 | defer cancelStartup() |
| 1566 | |
| 1567 | dd := cfg.DataDir |
| 1568 | regPath := filepath.Join(dd, "registry.db") |
| 1569 | reg, err := registry.Open(regPath) |
| 1570 | if err != nil { |
| 1571 | log.Printf("largebox agent: %v", err) |
| 1572 | return 2 |
| 1573 | } |
| 1574 | defer reg.Close() |
| 1575 | |
| 1576 | control := instancectl.NewClient(cfg.Control) |
| 1577 | nestedOK := nestedGate(func() error { return instancefs.SelfCheck(dd) }, instancefs.MountNoSymfollow, control.Ready) |
| 1578 | started := make(chan struct{}) |
| 1579 | backend := &largeBoxBackend{ev: ev, started: started} |
| 1580 | svc := activator.New(activator.Config{Bucket: "unused", Idle: time.Hour}, backend, reg) |
| 1581 | backend.svc = svc |
| 1582 | svc.Routes = reg |
| 1583 | svc.Startup = started |
| 1584 | openStore := func(context.Context) (release.Store, error) { return release.NewDirStore(cfg.StoreDir), nil } |
| 1585 | |
| 1586 | startupDone := activatorStart{ |
| 1587 | Ready: func() { |
| 1588 | ev.log("ready") // before the close, so no acquire can be stamped earlier |
| 1589 | close(started) |
| 1590 | }, |
| 1591 | Sweep: func(ctx context.Context) error { |
| 1592 | sw, err := control.Sweep(ctx) |
| 1593 | if err != nil { |
| 1594 | log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err) |
| 1595 | } else if len(sw.Units) > 0 { |
| 1596 | log.Printf("agent: swept %d instance unit(s) left from a previous agent", len(sw.Units)) |
| 1597 | } |
| 1598 | ev.log("swept", len(sw.Units), err == nil) |
| 1599 | return err |
| 1600 | }, |
| 1601 | Recover: func(ctx context.Context, swept bool) *layout.Pass { |
| 1602 | pass := newLayoutPass(reg, svc, dd, nestedOK, swept) |
| 1603 | pass.SocketDir, pass.SocketsGID = cfg.SocketDir, -1 // the fixture's run dir; no carlos-sockets group |
| 1604 | intent := startupRecoveryIntent(ctx, dd, openStore, "largebox") |
| 1605 | recoverLayoutAtStart(ctx, pass, intent) |
| 1606 | ev.log("recovered", intent) |
| 1607 | return pass |
| 1608 | }, |
| 1609 | Routes: reg.List, |
| 1610 | Route: reg.Get, |
| 1611 | Keys: largeBoxKeys{ev}, |
| 1612 | Slot: svc.PrepareScheduledToken, |
| 1613 | Publish: armedLayoutPass.Store, |
| 1614 | // The checkpoint: when Start runs, every hold marker's host must |
| 1615 | // already be fenced again and every journal complete. |
| 1616 | Start: func() { |
| 1617 | held, _ := layout.HoldMarkers(dd) |
| 1618 | fenced := 0 |
| 1619 | for _, h := range held { |
| 1620 | if svc.FenceGeneration(h) != 0 { |
| 1621 | fenced++ |
| 1622 | } |
| 1623 | } |
| 1624 | journals, _ := layout.JournalHosts(dd) |
| 1625 | ev.log("activator-start", "held="+strconv.Itoa(len(held)), "fenced="+strconv.Itoa(fenced), "journals="+strconv.Itoa(len(journals))) |
| 1626 | svc.Start() |
| 1627 | }, |
| 1628 | }.Run(startupCtx) |
| 1629 | go func() { |
| 1630 | <-startupDone |
| 1631 | ev.log("startup-done") |
| 1632 | }() |
| 1633 | // runAgent's worker boot pass, which waits on the same latch. |
| 1634 | go func() { |
| 1635 | select { |
| 1636 | case <-started: |
| 1637 | case <-startupCtx.Done(): |
| 1638 | return |
| 1639 | } |
| 1640 | routes, err := reg.List() |
| 1641 | if err != nil { |
| 1642 | return |
| 1643 | } |
| 1644 | runWorkerBootPass(routes, func(rt registry.Route) error { |
| 1645 | err := svc.WakeWorker(rt) |
| 1646 | ev.log("worker-boot", rt.Host, err == nil) |
| 1647 | return err |
| 1648 | }) |
| 1649 | }() |
| 1650 | |
| 1651 | // Admissions from inside the agent that do not pass runAgent's own |
| 1652 | // gates, so they meet the activator's latch on their own: a worker |
| 1653 | // wake as an operator command or the sidecar-wake leg makes it, and |
| 1654 | // one scheduler tick through the real runner and deliverer with a |
| 1655 | // delivery already due. |
| 1656 | if rt, err := reg.Get(largeBoxWorker); err == nil && rt != nil { |
| 1657 | go func() { |
| 1658 | ev.log("worker-direct-begin", rt.Host) |
| 1659 | err := svc.WakeWorker(*rt) |
| 1660 | ev.log("worker-direct", rt.Host, err == nil) |
| 1661 | }() |
| 1662 | } |
| 1663 | runner := &schedule.Runner{ |
| 1664 | Reg: reg, Deliver: &activatorDeliverer{svc: svc}, Token: scheduleToken(dd, svc, reg), |
| 1665 | Now: time.Now, Timeout: largeBoxLatchBound + 15*time.Second, Box: "largebox", Build: "test", |
| 1666 | } |
| 1667 | runner.Tick(agentCtx) // delivers in the background, and returns |
| 1668 | ev.log("tick-done") |
| 1669 | |
| 1670 | edge.RunWithActivatorReady([]string{"-registry", regPath, "-acme-cache", cfg.AcmeDir, "-http", cfg.HTTP, "-https", cfg.HTTPS}, |
| 1671 | largeBoxActivator{Service: svc, ev: ev}, edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes), |
| 1672 | func(e *edge.Edge) { |
| 1673 | reload := func() { _ = e.Reload() } |
| 1674 | layoutReload.Store(&reload) |
| 1675 | }) |
| 1676 | |
| 1677 | // runAgent's shutdown, in its order. |
| 1678 | cancelStartup() |
| 1679 | joinCtx, joinCancel := context.WithTimeout(context.Background(), largeBoxJoinBound) |
| 1680 | began := time.Now() |
| 1681 | joined := joinAgentStartup(joinCtx, startupDone) |
| 1682 | joinCancel() |
| 1683 | ev.log("joined", joined, time.Since(began)) |
| 1684 | ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) |
| 1685 | svc.Shutdown(ctx) |
| 1686 | cancel() |
| 1687 | cancelAgent() |
| 1688 | |
| 1689 | deadline := time.Now().Add(2 * time.Second) |
| 1690 | for { |
| 1691 | left := largeBoxStartupGoroutines() |
| 1692 | if len(left) == 0 { |
| 1693 | ev.log("leak-free") |
| 1694 | return 0 |
| 1695 | } |
| 1696 | if time.Now().After(deadline) { |
| 1697 | log.Printf("largebox agent: %d goroutine(s) outlived shutdown:\n\n%s", len(left), strings.Join(left, "\n\n")) |
| 1698 | ev.log("leak", len(left)) |
| 1699 | return 0 |
| 1700 | } |
| 1701 | time.Sleep(20 * time.Millisecond) |
| 1702 | } |
| 1703 | } |
| 1704 | |
| 1705 | // largeBoxStartupGoroutines lists every goroutine but this one that runs, |
| 1706 | // or was started by, the startup sequence, the layout pass, the control |
| 1707 | // client, the activator, the scheduler, or this agent's own fakes, and |
| 1708 | // every server connection still open (a request held at the latch is |
| 1709 | // answered only once the activator shuts down, after the edge has stopped |
| 1710 | // waiting for it). The edge's process-lifetime loops (its registry watch, |
| 1711 | // its SIGHUP reader) are not startup's and are left alone. |
| 1712 | func largeBoxStartupGoroutines() []string { |
| 1713 | buf := make([]byte, 8<<20) |
| 1714 | n := runtime.Stack(buf, true) |
| 1715 | blocks := strings.Split(string(buf[:n]), "\n\n") |
| 1716 | markers := []string{ |
| 1717 | "host.activatorStart", "host.startActivator", "host.backfillInstanceKeys", "host.keyBackfillInSlot", |
| 1718 | "host.recoverLayoutAtStart", "host.startupRecoveryIntent", "host.runWorkerBootPass", |
| 1719 | "host.runLargeBoxAgent", "host.(*largeBoxBackend)", "host.largeBoxActivator", "host.(*activatorDeliverer)", |
| 1720 | "layout.(*Pass)", "instancectl.(*Client)", "activator.(*Service)", "schedule.(*Runner)", |
| 1721 | "net/http.(*conn).serve", |
| 1722 | } |
| 1723 | var left []string |
| 1724 | for _, g := range blocks[1:] { // [0] is the calling goroutine |
| 1725 | for _, m := range markers { |
| 1726 | if strings.Contains(g, m) { |
| 1727 | left = append(left, g) |
| 1728 | break |
| 1729 | } |
| 1730 | } |
| 1731 | } |
| 1732 | return left |
| 1733 | } |
| 1734 | |
| 1735 | // largeBoxActivator is the activator as the edge sees it, reporting each |
| 1736 | // instance request as it reaches it. |
| 1737 | type largeBoxActivator struct { |
| 1738 | *activator.Service |
| 1739 | ev *largeBoxEvents |
| 1740 | } |
| 1741 | |
| 1742 | func (a largeBoxActivator) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool { |
| 1743 | if rt.Kind == "instance" { |
| 1744 | a.ev.log("enter", rt.Host, rt.Socket) |
| 1745 | } |
| 1746 | return a.Service.ServeHTTP(w, r, rt, proxy) |
| 1747 | } |
| 1748 | |
| 1749 | // largeBoxBackend is activator.Backend with no S3 and no processes: a wake |
| 1750 | // serves an app on the row's socket that names the host and the socket it |
| 1751 | // was started at, and a worker reports ready at once, as a child does |
| 1752 | // over the agent socket. It records every acquire, start and request |
| 1753 | // served, and any acquire the startup latch should have prevented. |
| 1754 | type largeBoxBackend struct { |
| 1755 | ev *largeBoxEvents |
| 1756 | started <-chan struct{} |
| 1757 | svc *activator.Service |
| 1758 | } |
| 1759 | |
| 1760 | func (b *largeBoxBackend) Acquire(_ context.Context, rt registry.Route) (activator.Lease, error) { |
| 1761 | select { |
| 1762 | case <-b.started: |
| 1763 | default: |
| 1764 | b.ev.log("violation", "acquire-before-ready", rt.Host) |
| 1765 | } |
| 1766 | b.ev.log("acquire", rt.Host) |
| 1767 | return nopLease{}, nil |
| 1768 | } |
| 1769 | |
| 1770 | func (b *largeBoxBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil } |
| 1771 | func (b *largeBoxBackend) Seed(context.Context, registry.Route) error { return nil } |
| 1772 | |
| 1773 | func (b *largeBoxBackend) Start(_ context.Context, rt registry.Route) (activator.Proc, string, error) { |
| 1774 | b.ev.log("start", rt.Host, rt.Socket) |
| 1775 | if err := os.MkdirAll(filepath.Dir(rt.Socket), 0o750); err != nil { |
| 1776 | return nil, "", err |
| 1777 | } |
| 1778 | ln, err := net.Listen("unix", rt.Socket) |
| 1779 | if err != nil { |
| 1780 | return nil, "", err |
| 1781 | } |
| 1782 | body := "host=" + rt.Host + " socket=" + rt.Socket |
| 1783 | p := &largeBoxProc{done: make(chan struct{})} |
| 1784 | p.srv = &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 1785 | b.ev.log("served", rt.Host, r.URL.Path) |
| 1786 | io.WriteString(w, body) |
| 1787 | })} |
| 1788 | go func() { defer close(p.done); _ = p.srv.Serve(ln) }() |
| 1789 | return p, "", nil |
| 1790 | } |
| 1791 | |
| 1792 | func (b *largeBoxBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil } |
| 1793 | func (b *largeBoxBackend) HasSidecar(registry.Route) bool { return false } |
| 1794 | func (b *largeBoxBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) { |
| 1795 | return nil, errors.New("largeBoxBackend: no co-spawned sidecars on this box") |
| 1796 | } |
| 1797 | |
| 1798 | func (b *largeBoxBackend) StartWorker(_ context.Context, rt registry.Route, _ map[string]string) (activator.Proc, string, error) { |
| 1799 | b.ev.log("worker-start", rt.Host) |
| 1800 | go b.svc.NoteWorkerReady(rt.Host, "") |
| 1801 | return &largeBoxWorkerProc{}, "", nil |
| 1802 | } |
| 1803 | |
| 1804 | type largeBoxProc struct { |
| 1805 | srv *http.Server |
| 1806 | done chan struct{} |
| 1807 | exited atomic.Bool |
| 1808 | } |
| 1809 | |
| 1810 | func (p *largeBoxProc) Stop(context.Context) error { |
| 1811 | _ = p.srv.Close() |
| 1812 | <-p.done |
| 1813 | p.exited.Store(true) |
| 1814 | return nil |
| 1815 | } |
| 1816 | |
| 1817 | func (p *largeBoxProc) Exited() bool { return p.exited.Load() } |
| 1818 | |
| 1819 | type largeBoxWorkerProc struct{ exited atomic.Bool } |
| 1820 | |
| 1821 | func (p *largeBoxWorkerProc) Stop(context.Context) error { p.exited.Store(true); return nil } |
| 1822 | func (p *largeBoxWorkerProc) Exited() bool { return p.exited.Load() } |
| 1823 | |
| 1824 | // largeBoxKeys is the exec backend's key writer as the key pass sees it: |
| 1825 | // every row it is handed already has its key. |
| 1826 | type largeBoxKeys struct{ ev *largeBoxEvents } |
| 1827 | |
| 1828 | func (k largeBoxKeys) BackfillInstanceKey(_ context.Context, rt registry.Route) (activator.InstanceKeyOutcome, error) { |
| 1829 | k.ev.log("key", rt.Host) |
| 1830 | return activator.InstanceKeyPresent, nil |
| 1831 | } |
| 1832 | |
amadan