test(app): run broker election tests against the production layout
- commit
- b5b719eeeab7bccb26d50db1f1557ff03bb31fc0
- author
- brice · · 2026-09-23 14:56 UTC
- parent
- 644ac29
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +77 -101
M internal/app/broker_offline_status_subprocess_test.go +76 -100
| @@ -77,35 +77,38 @@ func shortTempDir(t *testing.T, prefix string) string { | |||
| 77 | 77 | // brokerSubprocessSandbox isolates one real offline root and records every | |
| 78 | 78 | // spawned helper process. | |
| 79 | 79 | type brokerSubprocessSandbox struct { | |
| 80 | - | t *testing.T | |
| 81 | - | root string | |
| 82 | - | layout brokerconfig.Layout | |
| 83 | - | recordDir string | |
| 84 | - | prodRuntime string | |
| 85 | - | prodState string | |
| 80 | + | t *testing.T | |
| 81 | + | layout brokerconfig.Layout | |
| 82 | + | recordDir string | |
| 86 | 83 | } | |
| 87 | 84 | ||
| 88 | 85 | // newBrokerSubprocessSandbox isolates XDG roots, writes one marker-valid empty | |
| 89 | 86 | // sandbox, and arranges cleanup of every recorded broker and launcher process. | |
| 90 | 87 | func newBrokerSubprocessSandbox(t *testing.T, idleGrace string) *brokerSubprocessSandbox { | |
| 91 | 88 | t.Helper() | |
| 92 | - | prodRuntime, prodState := t.TempDir(), t.TempDir() | |
| 93 | - | t.Setenv("XDG_RUNTIME_DIR", prodRuntime) | |
| 94 | - | t.Setenv("XDG_STATE_HOME", prodState) | |
| 95 | - | root := filepath.Join(shortTempDir(t, "vevb"), "s") | |
| 96 | - | require.NoError(t, os.MkdirAll(root, 0o700)) | |
| 97 | - | writeSandboxConfig(t, root, sandboxEmptyDocument(idleGrace)) | |
| 89 | + | layout := emptyProductionBrokerLayout(t, idleGrace) | |
| 98 | 90 | recordDir := filepath.Join(shortTempDir(t, "vevr"), "r") | |
| 99 | 91 | require.NoError(t, os.MkdirAll(recordDir, 0o700)) | |
| 100 | 92 | t.Setenv(brokerHelperEnv, "1") | |
| 101 | 93 | t.Setenv(brokerHelperRecordDirEnv, recordDir) | |
| 102 | - | layout, err := offlineLayout(root) | |
| 103 | - | require.NoError(t, err) | |
| 104 | - | sandbox := &brokerSubprocessSandbox{t: t, root: root, layout: layout, recordDir: recordDir, prodRuntime: prodRuntime, prodState: prodState} | |
| 94 | + | sandbox := &brokerSubprocessSandbox{t: t, layout: layout, recordDir: recordDir} | |
| 105 | 95 | t.Cleanup(sandbox.cleanup) | |
| 106 | 96 | return sandbox | |
| 107 | 97 | } | |
| 108 | 98 | ||
| 99 | + | // ensure uses the production connect-or-spawn entry point and observes the | |
| 100 | + | // registered epoch after the broker publishes its initial snapshot. | |
| 101 | + | func (s *brokerSubprocessSandbox) ensure(timeout time.Duration) (brokerStatusReport, error) { | |
| 102 | + | ctx, cancel := context.WithTimeout(context.Background(), timeout) | |
| 103 | + | defer cancel() | |
| 104 | + | service, err := connectProductionBroker(ctx) | |
| 105 | + | if err != nil { | |
| 106 | + | return brokerStatusReport{}, err | |
| 107 | + | } | |
| 108 | + | defer service.Close() | |
| 109 | + | return probeBrokerStatus(ctx, s.socketPath()) | |
| 110 | + | } | |
| 111 | + | ||
| 109 | 112 | // runBrokerCommand executes one helper through the test binary. | |
| 110 | 113 | func runBrokerCommand(t *testing.T, args ...string) (stdout, stderr string, code int) { | |
| 111 | 114 | t.Helper() | |
| @@ -125,13 +128,6 @@ func runBrokerCommand(t *testing.T, args ...string) (stdout, stderr string, code | |||
| 125 | 128 | return out.String(), errb.String(), code | |
| 126 | 129 | } | |
| 127 | 130 | ||
| 128 | - | // status runs `_broker-status` for this sandbox with the supplied extra flags. | |
| 129 | - | func (s *brokerSubprocessSandbox) status(extra ...string) (stdout, stderr string, code int) { | |
| 130 | - | s.t.Helper() | |
| 131 | - | args := append([]string{brokerStatusCommand, "--offline-root", s.root}, extra...) | |
| 132 | - | return runBrokerCommand(s.t, args...) | |
| 133 | - | } | |
| 134 | - | ||
| 135 | 131 | // socketPath is the broker endpoint inside the sandbox runtime directory. | |
| 136 | 132 | func (s *brokerSubprocessSandbox) socketPath() string { | |
| 137 | 133 | return brokeripc.SocketPath(s.layout.Runtime) | |
| @@ -193,7 +189,7 @@ func (s *brokerSubprocessSandbox) waitForSocketGone(timeout time.Duration) bool | |||
| 193 | 189 | // cleanup terminates every recorded broker and launcher so no test leaks an | |
| 194 | 190 | // orphan process. | |
| 195 | 191 | func (s *brokerSubprocessSandbox) cleanup() { | |
| 196 | - | for _, role := range []string{brokerServeCommand, brokerLauncherCommand} { | |
| 192 | + | for _, role := range []string{productionBrokerServeCommand, productionBrokerLauncherCommand} { | |
| 197 | 193 | for _, pid := range s.recorded(role) { | |
| 198 | 194 | if err := syscall.Kill(pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) { | |
| 199 | 195 | continue | |
| @@ -209,10 +205,9 @@ func (s *brokerSubprocessSandbox) cleanup() { | |||
| 209 | 205 | func TestBrokerStatusSubprocessElectsExactlyOneBroker(t *testing.T) { | |
| 210 | 206 | sandbox := newBrokerSubprocessSandbox(t, "5s") | |
| 211 | 207 | const racers = 4 | |
| 212 | - | ||
| 213 | 208 | type outcome struct { | |
| 214 | - | stdout, stderr string | |
| 215 | - | code int | |
| 209 | + | report brokerStatusReport | |
| 210 | + | err error | |
| 216 | 211 | } | |
| 217 | 212 | results := make([]outcome, racers) | |
| 218 | 213 | var wg sync.WaitGroup | |
| @@ -220,24 +215,20 @@ func TestBrokerStatusSubprocessElectsExactlyOneBroker(t *testing.T) { | |||
| 220 | 215 | wg.Add(1) | |
| 221 | 216 | go func(index int) { | |
| 222 | 217 | defer wg.Done() | |
| 223 | - | stdout, stderr, code := runBrokerCommand(t, brokerStatusCommand, "--offline-root", sandbox.root, "--ensure", "--timeout", "20s") | |
| 224 | - | results[index] = outcome{stdout: stdout, stderr: stderr, code: code} | |
| 218 | + | results[index].report, results[index].err = sandbox.ensure(20 * time.Second) | |
| 225 | 219 | }(i) | |
| 226 | 220 | } | |
| 227 | 221 | wg.Wait() | |
| 228 | - | ||
| 229 | 222 | epochs := make(map[uint64]struct{}, racers) | |
| 230 | - | for i, result := range results { | |
| 231 | - | require.Equalf(t, 0, result.code, "racer %d exited %d: %s", i, result.code, result.stderr) | |
| 232 | - | report := decodeOneReport(t, result.stdout) | |
| 233 | - | require.Equal(t, "ready", report.Status) | |
| 234 | - | require.Equal(t, sandbox.socketPath(), report.Endpoint) | |
| 235 | - | epochs[uint64(report.Epoch)] = struct{}{} | |
| 223 | + | for _, result := range results { | |
| 224 | + | require.NoError(t, result.err) | |
| 225 | + | require.Equal(t, "ready", result.report.Status) | |
| 226 | + | require.Equal(t, sandbox.socketPath(), result.report.Endpoint) | |
| 227 | + | epochs[uint64(result.report.Epoch)] = struct{}{} | |
| 236 | 228 | } | |
| 237 | - | require.Len(t, epochs, 1, "every racer must observe the same broker epoch") | |
| 238 | - | require.Len(t, sandbox.recorded(brokerServeCommand), 1, "exactly one broker process must ever be spawned") | |
| 239 | - | require.Len(t, sandbox.live(brokerServeCommand), 1, "exactly one broker process must be live") | |
| 240 | - | requireProductionUntouched(t, sandbox.prodRuntime, sandbox.prodState) | |
| 229 | + | require.Len(t, epochs, 1) | |
| 230 | + | require.Len(t, sandbox.recorded(productionBrokerServeCommand), 1) | |
| 231 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1) | |
| 241 | 232 | } | |
| 242 | 233 | ||
| 243 | 234 | // TestBrokerStatusSubprocessIdleExitRestartsWithNewEpoch lets the broker idle | |
| @@ -246,25 +237,23 @@ func TestBrokerStatusSubprocessElectsExactlyOneBroker(t *testing.T) { | |||
| 246 | 237 | func TestBrokerStatusSubprocessIdleExitRestartsWithNewEpoch(t *testing.T) { | |
| 247 | 238 | sandbox := newBrokerSubprocessSandbox(t, "2s") | |
| 248 | 239 | ||
| 249 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "20s") | |
| 250 | - | require.Equal(t, 0, code, stderr) | |
| 251 | - | first := decodeOneReport(t, stdout) | |
| 240 | + | first, err := sandbox.ensure(20 * time.Second) | |
| 241 | + | require.NoError(t, err) | |
| 252 | 242 | require.Equal(t, "ready", first.Status) | |
| 253 | 243 | require.Positive(t, first.Epoch) | |
| 254 | 244 | ||
| 255 | 245 | require.True(t, sandbox.waitForSocketGone(15*time.Second), "the idle broker must shut itself down") | |
| 256 | - | for _, pid := range sandbox.recorded(brokerServeCommand) { | |
| 246 | + | for _, pid := range sandbox.recorded(productionBrokerServeCommand) { | |
| 257 | 247 | require.NoError(t, waitForProcessExit(pid, 10*time.Second), "the idle broker must exit") | |
| 258 | 248 | } | |
| 259 | - | require.Empty(t, sandbox.live(brokerServeCommand), "the idle broker must exit") | |
| 249 | + | require.Empty(t, sandbox.live(productionBrokerServeCommand), "the idle broker must exit") | |
| 260 | 250 | ||
| 261 | - | stdout, stderr, code = sandbox.status("--ensure", "--timeout", "20s") | |
| 262 | - | require.Equal(t, 0, code, stderr) | |
| 263 | - | second := decodeOneReport(t, stdout) | |
| 251 | + | second, err := sandbox.ensure(20 * time.Second) | |
| 252 | + | require.NoError(t, err) | |
| 264 | 253 | require.Equal(t, "ready", second.Status) | |
| 265 | 254 | require.NotEqual(t, first.Epoch, second.Epoch, "an on-demand restart must sample a new epoch") | |
| 266 | - | require.Len(t, sandbox.recorded(brokerServeCommand), 2, "one broker per ensure is expected") | |
| 267 | - | require.Len(t, sandbox.live(brokerServeCommand), 1, "only the restarted broker may remain") | |
| 255 | + | require.Len(t, sandbox.recorded(productionBrokerServeCommand), 2, "one broker per ensure is expected") | |
| 256 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1, "only the restarted broker may remain") | |
| 268 | 257 | } | |
| 269 | 258 | ||
| 270 | 259 | // TestBrokerStatusSubprocessRecoversStaleSocket proves a stale socket left by a | |
| @@ -274,11 +263,10 @@ func TestBrokerStatusSubprocessRecoversStaleSocket(t *testing.T) { | |||
| 274 | 263 | require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700)) | |
| 275 | 264 | bindStaleUnixSocket(t, sandbox.socketPath()) | |
| 276 | 265 | ||
| 277 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "20s") | |
| 278 | - | require.Equal(t, 0, code, stderr) | |
| 279 | - | report := decodeOneReport(t, stdout) | |
| 266 | + | report, err := sandbox.ensure(20 * time.Second) | |
| 267 | + | require.NoError(t, err) | |
| 280 | 268 | require.Equal(t, "ready", report.Status) | |
| 281 | - | require.Len(t, sandbox.live(brokerServeCommand), 1) | |
| 269 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1) | |
| 282 | 270 | } | |
| 283 | 271 | ||
| 284 | 272 | // TestBrokerStatusSubprocessWaitsForLiveSpawnerThenRecovers holds the | |
| @@ -290,18 +278,17 @@ func TestBrokerStatusSubprocessWaitsForLiveSpawnerThenRecovers(t *testing.T) { | |||
| 290 | 278 | spawner, err := lifecycle.TryAcquire(sandbox.layout.Spawn) | |
| 291 | 279 | require.NoError(t, err) | |
| 292 | 280 | ||
| 293 | - | stdout, _, code := sandbox.status("--ensure", "--timeout", "500ms") | |
| 294 | - | require.Equal(t, 3, code) | |
| 295 | - | require.Equal(t, "offline", decodeOneReport(t, stdout).Status) | |
| 296 | - | require.Empty(t, sandbox.recorded(brokerServeCommand), "a waiter must not spawn while another spawner is live") | |
| 281 | + | _, err = sandbox.ensure(500 * time.Millisecond) | |
| 282 | + | require.Error(t, err) | |
| 283 | + | require.Empty(t, sandbox.recorded(productionBrokerServeCommand), "a waiter must not spawn while another spawner is live") | |
| 297 | 284 | ||
| 298 | 285 | // The kernel releases the spawner's descriptor when it dies; releasing here | |
| 299 | 286 | // is exactly that event. | |
| 300 | 287 | require.NoError(t, spawner.Release()) | |
| 301 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "20s") | |
| 302 | - | require.Equal(t, 0, code, stderr) | |
| 303 | - | require.Equal(t, "ready", decodeOneReport(t, stdout).Status) | |
| 304 | - | require.Len(t, sandbox.live(brokerServeCommand), 1) | |
| 288 | + | report, err := sandbox.ensure(20 * time.Second) | |
| 289 | + | require.NoError(t, err) | |
| 290 | + | require.Equal(t, "ready", report.Status) | |
| 291 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1) | |
| 305 | 292 | } | |
| 306 | 293 | ||
| 307 | 294 | // TestBrokerStatusSubprocessSilentEndpointFailsWithoutSpawning proves a live | |
| @@ -316,12 +303,10 @@ func TestBrokerStatusSubprocessSilentEndpointFailsWithoutSpawning(t *testing.T) | |||
| 316 | 303 | // The timeout must exceed the probe budget so the stall is classified as a | |
| 317 | 304 | // live-but-incompatible endpoint instead of exhausting the overall deadline. | |
| 318 | 305 | start := time.Now() | |
| 319 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "5s") | |
| 320 | - | require.Equal(t, 3, code) | |
| 321 | - | require.Equal(t, "offline", decodeOneReport(t, stdout).Status) | |
| 322 | - | require.Contains(t, stderr, "live but incompatible", "a silent live endpoint must be classified incompatible") | |
| 306 | + | _, err := sandbox.ensure(5 * time.Second) | |
| 307 | + | require.ErrorContains(t, err, "live but incompatible") | |
| 323 | 308 | require.Less(t, time.Since(start), 5*time.Second, "the probe budget must classify a silent endpoint before the overall deadline") | |
| 324 | - | require.Empty(t, sandbox.recorded(brokerServeCommand), "a live-but-silent endpoint must not be respawned over") | |
| 309 | + | require.Empty(t, sandbox.recorded(productionBrokerServeCommand), "a live-but-silent endpoint must not be respawned over") | |
| 325 | 310 | } | |
| 326 | 311 | ||
| 327 | 312 | // TestBrokerLauncherSpawnWaitIsBoundedByContext proves a controlled blocking | |
| @@ -340,7 +325,7 @@ func TestBrokerLauncherSpawnWaitIsBoundedByContext(t *testing.T) { | |||
| 340 | 325 | ctx, cancel := context.WithCancel(context.Background()) | |
| 341 | 326 | defer cancel() | |
| 342 | 327 | result := make(chan error, 1) | |
| 343 | - | go func() { result <- spawnBrokerLauncher(ctx, t.TempDir(), 0, false) }() | |
| 328 | + | go func() { result <- spawnProductionBrokerLauncher(ctx) }() | |
| 344 | 329 | ||
| 345 | 330 | pid := waitForProcessRecord(t, record) | |
| 346 | 331 | select { | |
| @@ -362,31 +347,23 @@ func TestBrokerLauncherSpawnWaitIsBoundedByContext(t *testing.T) { | |||
| 362 | 347 | // TestBrokerStatusSubprocessLauncherDiagnosesConflicts proves an explicit | |
| 363 | 348 | // launcher grace that conflicts with the provisioned one is refused before any | |
| 364 | 349 | // process is spawned, and that a missing configuration fails the launcher. | |
| 350 | + | // A production launcher accepts no idle override; malformed broker.json is | |
| 351 | + | // diagnosed by the detached serve before it can claim the lifetime lock. | |
| 365 | 352 | func TestBrokerStatusSubprocessLauncherDiagnosesConflicts(t *testing.T) { | |
| 366 | - | t.Run("conflicting idle grace", func(t *testing.T) { | |
| 353 | + | t.Run("unsupported idle override", func(t *testing.T) { | |
| 367 | 354 | sandbox := newBrokerSubprocessSandbox(t, "90s") | |
| 368 | - | stdout, stderr, code := runBrokerCommand(t, brokerLauncherCommand, "--offline-root", sandbox.root, "--idle-grace", "1m") | |
| 369 | - | require.NotEqual(t, 0, code) | |
| 370 | - | require.Contains(t, stderr, "conflicts with the provisioned idle grace") | |
| 371 | - | require.Empty(t, strings.TrimSpace(stdout)) | |
| 372 | - | require.Empty(t, sandbox.recorded(brokerServeCommand)) | |
| 355 | + | _, stderr, code := runBrokerCommand(t, productionBrokerLauncherCommand, "--idle-grace", "1m") | |
| 356 | + | require.NotZero(t, code) | |
| 357 | + | require.Contains(t, stderr, "does not accept arguments") | |
| 358 | + | require.Empty(t, sandbox.recorded(productionBrokerServeCommand)) | |
| 373 | 359 | }) | |
| 374 | - | ||
| 375 | - | t.Run("missing configuration", func(t *testing.T) { | |
| 376 | - | prodRuntime, prodState := t.TempDir(), t.TempDir() | |
| 377 | - | t.Setenv("XDG_RUNTIME_DIR", prodRuntime) | |
| 378 | - | t.Setenv("XDG_STATE_HOME", prodState) | |
| 379 | - | recordDir := filepath.Join(t.TempDir(), "records") | |
| 380 | - | require.NoError(t, os.MkdirAll(recordDir, 0o700)) | |
| 381 | - | t.Setenv(brokerHelperEnv, "1") | |
| 382 | - | t.Setenv(brokerHelperRecordDirEnv, recordDir) | |
| 383 | - | root := filepath.Join(shortTempDir(t, "vevx"), "s") | |
| 384 | - | require.NoError(t, os.MkdirAll(root, 0o700)) | |
| 385 | - | ||
| 386 | - | _, stderr, code := runBrokerCommand(t, brokerLauncherCommand, "--offline-root", root) | |
| 387 | - | require.NotEqual(t, 0, code) | |
| 388 | - | require.NotEmpty(t, strings.TrimSpace(stderr)) | |
| 389 | - | require.Empty(t, recordedPIDsIn(t, recordDir, brokerServeCommand)) | |
| 360 | + | t.Run("invalid production configuration", func(t *testing.T) { | |
| 361 | + | sandbox := newBrokerSubprocessSandbox(t, "") | |
| 362 | + | require.NoError(t, os.WriteFile(productionBrokerConfigPath(), []byte("not json"), 0o600)) | |
| 363 | + | _, stderr, code := runBrokerCommand(t, productionBrokerServeCommand) | |
| 364 | + | require.NotZero(t, code) | |
| 365 | + | require.Contains(t, stderr, brokerconfig.ConfigFileName) | |
| 366 | + | require.NoFileExists(t, filepath.Join(sandbox.layout.Runtime, "lifecycle.lock")) | |
| 390 | 367 | }) | |
| 391 | 368 | } | |
| 392 | 369 | ||
| @@ -395,27 +372,26 @@ func TestBrokerStatusSubprocessLauncherDiagnosesConflicts(t *testing.T) { | |||
| 395 | 372 | func TestBrokerStatusSubprocessNoOrphanProcess(t *testing.T) { | |
| 396 | 373 | sandbox := newBrokerSubprocessSandbox(t, "5s") | |
| 397 | 374 | ||
| 398 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "20s") | |
| 399 | - | require.Equal(t, 0, code, stderr) | |
| 400 | - | require.Equal(t, "ready", decodeOneReport(t, stdout).Status) | |
| 375 | + | report, err := sandbox.ensure(20 * time.Second) | |
| 376 | + | require.NoError(t, err) | |
| 377 | + | require.Equal(t, "ready", report.Status) | |
| 401 | 378 | ||
| 402 | - | require.NotEmpty(t, sandbox.recorded(brokerLauncherCommand)) | |
| 403 | - | for _, pid := range sandbox.recorded(brokerLauncherCommand) { | |
| 379 | + | require.NotEmpty(t, sandbox.recorded(productionBrokerLauncherCommand)) | |
| 380 | + | for _, pid := range sandbox.recorded(productionBrokerLauncherCommand) { | |
| 404 | 381 | require.NoError(t, waitForProcessExit(pid, 5*time.Second), "the short-lived launcher must exit") | |
| 405 | 382 | } | |
| 406 | - | require.Len(t, sandbox.live(brokerServeCommand), 1) | |
| 383 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1) | |
| 407 | 384 | ||
| 408 | - | _, err := lifecycle.TryAcquire(sandbox.layout.Runtime) | |
| 385 | + | _, err = lifecycle.TryAcquire(sandbox.layout.Runtime) | |
| 409 | 386 | require.ErrorIs(t, err, lifecycle.ErrBusy, "the live broker must own the lifetime lock") | |
| 410 | 387 | ||
| 411 | - | sandbox.terminate(brokerServeCommand, syscall.SIGTERM) | |
| 412 | - | require.Empty(t, sandbox.live(brokerLauncherCommand), "no launcher may be left behind") | |
| 413 | - | require.Empty(t, sandbox.live(brokerServeCommand), "no broker may be left behind") | |
| 388 | + | sandbox.terminate(productionBrokerServeCommand, syscall.SIGTERM) | |
| 389 | + | require.Empty(t, sandbox.live(productionBrokerLauncherCommand), "no launcher may be left behind") | |
| 390 | + | require.Empty(t, sandbox.live(productionBrokerServeCommand), "no broker may be left behind") | |
| 414 | 391 | require.True(t, sandbox.waitForSocketGone(5*time.Second), "an orderly broker shutdown must unlink its socket") | |
| 415 | 392 | owner, err := lifecycle.TryAcquire(sandbox.layout.Runtime) | |
| 416 | 393 | require.NoError(t, err, "the terminated broker must release the lifetime lock") | |
| 417 | 394 | require.NoError(t, owner.Release()) | |
| 418 | - | requireProductionUntouched(t, sandbox.prodRuntime, sandbox.prodState) | |
| 419 | 395 | } | |
| 420 | 396 | ||
| 421 | 397 | // recordedPIDsIn reads one role's PID record from an arbitrary directory. | |
| @@ -77,35 +77,38 @@ func shortTempDir(t *testing.T, prefix string) string { | |||||
| 77 | // brokerSubprocessSandbox isolates one real offline root and records every | 77 | // brokerSubprocessSandbox isolates one real offline root and records every | ||
| 78 | // spawned helper process. | 78 | // spawned helper process. | ||
| 79 | type brokerSubprocessSandbox struct { | 79 | type brokerSubprocessSandbox struct { | ||
| 80 | - | t *testing.T | 80 | + | t *testing.T |
| 81 | - | root string | 81 | + | layout brokerconfig.Layout |
| 82 | - | layout brokerconfig.Layout | 82 | + | recordDir string |
| 83 | - | recordDir string | |||
| 84 | - | prodRuntime string | |||
| 85 | - | prodState string | |||
| 86 | } | 83 | } | ||
| 87 | 84 | ||||
| 88 | // newBrokerSubprocessSandbox isolates XDG roots, writes one marker-valid empty | 85 | // newBrokerSubprocessSandbox isolates XDG roots, writes one marker-valid empty | ||
| 89 | // sandbox, and arranges cleanup of every recorded broker and launcher process. | 86 | // sandbox, and arranges cleanup of every recorded broker and launcher process. | ||
| 90 | func newBrokerSubprocessSandbox(t *testing.T, idleGrace string) *brokerSubprocessSandbox { | 87 | func newBrokerSubprocessSandbox(t *testing.T, idleGrace string) *brokerSubprocessSandbox { | ||
| 91 | t.Helper() | 88 | t.Helper() | ||
| 92 | - | prodRuntime, prodState := t.TempDir(), t.TempDir() | 89 | + | layout := emptyProductionBrokerLayout(t, idleGrace) |
| 93 | - | t.Setenv("XDG_RUNTIME_DIR", prodRuntime) | |||
| 94 | - | t.Setenv("XDG_STATE_HOME", prodState) | |||
| 95 | - | root := filepath.Join(shortTempDir(t, "vevb"), "s") | |||
| 96 | - | require.NoError(t, os.MkdirAll(root, 0o700)) | |||
| 97 | - | writeSandboxConfig(t, root, sandboxEmptyDocument(idleGrace)) | |||
| 98 | recordDir := filepath.Join(shortTempDir(t, "vevr"), "r") | 90 | recordDir := filepath.Join(shortTempDir(t, "vevr"), "r") | ||
| 99 | require.NoError(t, os.MkdirAll(recordDir, 0o700)) | 91 | require.NoError(t, os.MkdirAll(recordDir, 0o700)) | ||
| 100 | t.Setenv(brokerHelperEnv, "1") | 92 | t.Setenv(brokerHelperEnv, "1") | ||
| 101 | t.Setenv(brokerHelperRecordDirEnv, recordDir) | 93 | t.Setenv(brokerHelperRecordDirEnv, recordDir) | ||
| 102 | - | layout, err := offlineLayout(root) | 94 | + | sandbox := &brokerSubprocessSandbox{t: t, layout: layout, recordDir: recordDir} |
| 103 | - | require.NoError(t, err) | |||
| 104 | - | sandbox := &brokerSubprocessSandbox{t: t, root: root, layout: layout, recordDir: recordDir, prodRuntime: prodRuntime, prodState: prodState} | |||
| 105 | t.Cleanup(sandbox.cleanup) | 95 | t.Cleanup(sandbox.cleanup) | ||
| 106 | return sandbox | 96 | return sandbox | ||
| 107 | } | 97 | } | ||
| 108 | 98 | ||||
| 99 | + | // ensure uses the production connect-or-spawn entry point and observes the | |||
| 100 | + | // registered epoch after the broker publishes its initial snapshot. | |||
| 101 | + | func (s *brokerSubprocessSandbox) ensure(timeout time.Duration) (brokerStatusReport, error) { | |||
| 102 | + | ctx, cancel := context.WithTimeout(context.Background(), timeout) | |||
| 103 | + | defer cancel() | |||
| 104 | + | service, err := connectProductionBroker(ctx) | |||
| 105 | + | if err != nil { | |||
| 106 | + | return brokerStatusReport{}, err | |||
| 107 | + | } | |||
| 108 | + | defer service.Close() | |||
| 109 | + | return probeBrokerStatus(ctx, s.socketPath()) | |||
| 110 | + | } | |||
| 111 | + | ||||
| 109 | // runBrokerCommand executes one helper through the test binary. | 112 | // runBrokerCommand executes one helper through the test binary. | ||
| 110 | func runBrokerCommand(t *testing.T, args ...string) (stdout, stderr string, code int) { | 113 | func runBrokerCommand(t *testing.T, args ...string) (stdout, stderr string, code int) { | ||
| 111 | t.Helper() | 114 | t.Helper() | ||
| @@ -125,13 +128,6 @@ func runBrokerCommand(t *testing.T, args ...string) (stdout, stderr string, code | |||||
| 125 | return out.String(), errb.String(), code | 128 | return out.String(), errb.String(), code | ||
| 126 | } | 129 | } | ||
| 127 | 130 | ||||
| 128 | - | // status runs `_broker-status` for this sandbox with the supplied extra flags. | |||
| 129 | - | func (s *brokerSubprocessSandbox) status(extra ...string) (stdout, stderr string, code int) { | |||
| 130 | - | s.t.Helper() | |||
| 131 | - | args := append([]string{brokerStatusCommand, "--offline-root", s.root}, extra...) | |||
| 132 | - | return runBrokerCommand(s.t, args...) | |||
| 133 | - | } | |||
| 134 | - | ||||
| 135 | // socketPath is the broker endpoint inside the sandbox runtime directory. | 131 | // socketPath is the broker endpoint inside the sandbox runtime directory. | ||
| 136 | func (s *brokerSubprocessSandbox) socketPath() string { | 132 | func (s *brokerSubprocessSandbox) socketPath() string { | ||
| 137 | return brokeripc.SocketPath(s.layout.Runtime) | 133 | return brokeripc.SocketPath(s.layout.Runtime) | ||
| @@ -193,7 +189,7 @@ func (s *brokerSubprocessSandbox) waitForSocketGone(timeout time.Duration) bool | |||||
| 193 | // cleanup terminates every recorded broker and launcher so no test leaks an | 189 | // cleanup terminates every recorded broker and launcher so no test leaks an | ||
| 194 | // orphan process. | 190 | // orphan process. | ||
| 195 | func (s *brokerSubprocessSandbox) cleanup() { | 191 | func (s *brokerSubprocessSandbox) cleanup() { | ||
| 196 | - | for _, role := range []string{brokerServeCommand, brokerLauncherCommand} { | 192 | + | for _, role := range []string{productionBrokerServeCommand, productionBrokerLauncherCommand} { |
| 197 | for _, pid := range s.recorded(role) { | 193 | for _, pid := range s.recorded(role) { | ||
| 198 | if err := syscall.Kill(pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) { | 194 | if err := syscall.Kill(pid, syscall.SIGKILL); err != nil && !errors.Is(err, syscall.ESRCH) { | ||
| 199 | continue | 195 | continue | ||
| @@ -209,10 +205,9 @@ func (s *brokerSubprocessSandbox) cleanup() { | |||||
| 209 | func TestBrokerStatusSubprocessElectsExactlyOneBroker(t *testing.T) { | 205 | func TestBrokerStatusSubprocessElectsExactlyOneBroker(t *testing.T) { | ||
| 210 | sandbox := newBrokerSubprocessSandbox(t, "5s") | 206 | sandbox := newBrokerSubprocessSandbox(t, "5s") | ||
| 211 | const racers = 4 | 207 | const racers = 4 | ||
| 212 | - | ||||
| 213 | type outcome struct { | 208 | type outcome struct { | ||
| 214 | - | stdout, stderr string | 209 | + | report brokerStatusReport |
| 215 | - | code int | 210 | + | err error |
| 216 | } | 211 | } | ||
| 217 | results := make([]outcome, racers) | 212 | results := make([]outcome, racers) | ||
| 218 | var wg sync.WaitGroup | 213 | var wg sync.WaitGroup | ||
| @@ -220,24 +215,20 @@ func TestBrokerStatusSubprocessElectsExactlyOneBroker(t *testing.T) { | |||||
| 220 | wg.Add(1) | 215 | wg.Add(1) | ||
| 221 | go func(index int) { | 216 | go func(index int) { | ||
| 222 | defer wg.Done() | 217 | defer wg.Done() | ||
| 223 | - | stdout, stderr, code := runBrokerCommand(t, brokerStatusCommand, "--offline-root", sandbox.root, "--ensure", "--timeout", "20s") | 218 | + | results[index].report, results[index].err = sandbox.ensure(20 * time.Second) |
| 224 | - | results[index] = outcome{stdout: stdout, stderr: stderr, code: code} | |||
| 225 | }(i) | 219 | }(i) | ||
| 226 | } | 220 | } | ||
| 227 | wg.Wait() | 221 | wg.Wait() | ||
| 228 | - | ||||
| 229 | epochs := make(map[uint64]struct{}, racers) | 222 | epochs := make(map[uint64]struct{}, racers) | ||
| 230 | - | for i, result := range results { | 223 | + | for _, result := range results { |
| 231 | - | require.Equalf(t, 0, result.code, "racer %d exited %d: %s", i, result.code, result.stderr) | 224 | + | require.NoError(t, result.err) |
| 232 | - | report := decodeOneReport(t, result.stdout) | 225 | + | require.Equal(t, "ready", result.report.Status) |
| 233 | - | require.Equal(t, "ready", report.Status) | 226 | + | require.Equal(t, sandbox.socketPath(), result.report.Endpoint) |
| 234 | - | require.Equal(t, sandbox.socketPath(), report.Endpoint) | 227 | + | epochs[uint64(result.report.Epoch)] = struct{}{} |
| 235 | - | epochs[uint64(report.Epoch)] = struct{}{} | |||
| 236 | } | 228 | } | ||
| 237 | - | require.Len(t, epochs, 1, "every racer must observe the same broker epoch") | 229 | + | require.Len(t, epochs, 1) |
| 238 | - | require.Len(t, sandbox.recorded(brokerServeCommand), 1, "exactly one broker process must ever be spawned") | 230 | + | require.Len(t, sandbox.recorded(productionBrokerServeCommand), 1) |
| 239 | - | require.Len(t, sandbox.live(brokerServeCommand), 1, "exactly one broker process must be live") | 231 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1) |
| 240 | - | requireProductionUntouched(t, sandbox.prodRuntime, sandbox.prodState) | |||
| 241 | } | 232 | } | ||
| 242 | 233 | ||||
| 243 | // TestBrokerStatusSubprocessIdleExitRestartsWithNewEpoch lets the broker idle | 234 | // TestBrokerStatusSubprocessIdleExitRestartsWithNewEpoch lets the broker idle | ||
| @@ -246,25 +237,23 @@ func TestBrokerStatusSubprocessElectsExactlyOneBroker(t *testing.T) { | |||||
| 246 | func TestBrokerStatusSubprocessIdleExitRestartsWithNewEpoch(t *testing.T) { | 237 | func TestBrokerStatusSubprocessIdleExitRestartsWithNewEpoch(t *testing.T) { | ||
| 247 | sandbox := newBrokerSubprocessSandbox(t, "2s") | 238 | sandbox := newBrokerSubprocessSandbox(t, "2s") | ||
| 248 | 239 | ||||
| 249 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "20s") | 240 | + | first, err := sandbox.ensure(20 * time.Second) |
| 250 | - | require.Equal(t, 0, code, stderr) | 241 | + | require.NoError(t, err) |
| 251 | - | first := decodeOneReport(t, stdout) | |||
| 252 | require.Equal(t, "ready", first.Status) | 242 | require.Equal(t, "ready", first.Status) | ||
| 253 | require.Positive(t, first.Epoch) | 243 | require.Positive(t, first.Epoch) | ||
| 254 | 244 | ||||
| 255 | require.True(t, sandbox.waitForSocketGone(15*time.Second), "the idle broker must shut itself down") | 245 | require.True(t, sandbox.waitForSocketGone(15*time.Second), "the idle broker must shut itself down") | ||
| 256 | - | for _, pid := range sandbox.recorded(brokerServeCommand) { | 246 | + | for _, pid := range sandbox.recorded(productionBrokerServeCommand) { |
| 257 | require.NoError(t, waitForProcessExit(pid, 10*time.Second), "the idle broker must exit") | 247 | require.NoError(t, waitForProcessExit(pid, 10*time.Second), "the idle broker must exit") | ||
| 258 | } | 248 | } | ||
| 259 | - | require.Empty(t, sandbox.live(brokerServeCommand), "the idle broker must exit") | 249 | + | require.Empty(t, sandbox.live(productionBrokerServeCommand), "the idle broker must exit") |
| 260 | 250 | ||||
| 261 | - | stdout, stderr, code = sandbox.status("--ensure", "--timeout", "20s") | 251 | + | second, err := sandbox.ensure(20 * time.Second) |
| 262 | - | require.Equal(t, 0, code, stderr) | 252 | + | require.NoError(t, err) |
| 263 | - | second := decodeOneReport(t, stdout) | |||
| 264 | require.Equal(t, "ready", second.Status) | 253 | require.Equal(t, "ready", second.Status) | ||
| 265 | require.NotEqual(t, first.Epoch, second.Epoch, "an on-demand restart must sample a new epoch") | 254 | require.NotEqual(t, first.Epoch, second.Epoch, "an on-demand restart must sample a new epoch") | ||
| 266 | - | require.Len(t, sandbox.recorded(brokerServeCommand), 2, "one broker per ensure is expected") | 255 | + | require.Len(t, sandbox.recorded(productionBrokerServeCommand), 2, "one broker per ensure is expected") |
| 267 | - | require.Len(t, sandbox.live(brokerServeCommand), 1, "only the restarted broker may remain") | 256 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1, "only the restarted broker may remain") |
| 268 | } | 257 | } | ||
| 269 | 258 | ||||
| 270 | // TestBrokerStatusSubprocessRecoversStaleSocket proves a stale socket left by a | 259 | // TestBrokerStatusSubprocessRecoversStaleSocket proves a stale socket left by a | ||
| @@ -274,11 +263,10 @@ func TestBrokerStatusSubprocessRecoversStaleSocket(t *testing.T) { | |||||
| 274 | require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700)) | 263 | require.NoError(t, os.MkdirAll(sandbox.layout.Runtime, 0o700)) | ||
| 275 | bindStaleUnixSocket(t, sandbox.socketPath()) | 264 | bindStaleUnixSocket(t, sandbox.socketPath()) | ||
| 276 | 265 | ||||
| 277 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "20s") | 266 | + | report, err := sandbox.ensure(20 * time.Second) |
| 278 | - | require.Equal(t, 0, code, stderr) | 267 | + | require.NoError(t, err) |
| 279 | - | report := decodeOneReport(t, stdout) | |||
| 280 | require.Equal(t, "ready", report.Status) | 268 | require.Equal(t, "ready", report.Status) | ||
| 281 | - | require.Len(t, sandbox.live(brokerServeCommand), 1) | 269 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1) |
| 282 | } | 270 | } | ||
| 283 | 271 | ||||
| 284 | // TestBrokerStatusSubprocessWaitsForLiveSpawnerThenRecovers holds the | 272 | // TestBrokerStatusSubprocessWaitsForLiveSpawnerThenRecovers holds the | ||
| @@ -290,18 +278,17 @@ func TestBrokerStatusSubprocessWaitsForLiveSpawnerThenRecovers(t *testing.T) { | |||||
| 290 | spawner, err := lifecycle.TryAcquire(sandbox.layout.Spawn) | 278 | spawner, err := lifecycle.TryAcquire(sandbox.layout.Spawn) | ||
| 291 | require.NoError(t, err) | 279 | require.NoError(t, err) | ||
| 292 | 280 | ||||
| 293 | - | stdout, _, code := sandbox.status("--ensure", "--timeout", "500ms") | 281 | + | _, err = sandbox.ensure(500 * time.Millisecond) |
| 294 | - | require.Equal(t, 3, code) | 282 | + | require.Error(t, err) |
| 295 | - | require.Equal(t, "offline", decodeOneReport(t, stdout).Status) | 283 | + | require.Empty(t, sandbox.recorded(productionBrokerServeCommand), "a waiter must not spawn while another spawner is live") |
| 296 | - | require.Empty(t, sandbox.recorded(brokerServeCommand), "a waiter must not spawn while another spawner is live") | |||
| 297 | 284 | ||||
| 298 | // The kernel releases the spawner's descriptor when it dies; releasing here | 285 | // The kernel releases the spawner's descriptor when it dies; releasing here | ||
| 299 | // is exactly that event. | 286 | // is exactly that event. | ||
| 300 | require.NoError(t, spawner.Release()) | 287 | require.NoError(t, spawner.Release()) | ||
| 301 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "20s") | 288 | + | report, err := sandbox.ensure(20 * time.Second) |
| 302 | - | require.Equal(t, 0, code, stderr) | 289 | + | require.NoError(t, err) |
| 303 | - | require.Equal(t, "ready", decodeOneReport(t, stdout).Status) | 290 | + | require.Equal(t, "ready", report.Status) |
| 304 | - | require.Len(t, sandbox.live(brokerServeCommand), 1) | 291 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1) |
| 305 | } | 292 | } | ||
| 306 | 293 | ||||
| 307 | // TestBrokerStatusSubprocessSilentEndpointFailsWithoutSpawning proves a live | 294 | // TestBrokerStatusSubprocessSilentEndpointFailsWithoutSpawning proves a live | ||
| @@ -316,12 +303,10 @@ func TestBrokerStatusSubprocessSilentEndpointFailsWithoutSpawning(t *testing.T) | |||||
| 316 | // The timeout must exceed the probe budget so the stall is classified as a | 303 | // The timeout must exceed the probe budget so the stall is classified as a | ||
| 317 | // live-but-incompatible endpoint instead of exhausting the overall deadline. | 304 | // live-but-incompatible endpoint instead of exhausting the overall deadline. | ||
| 318 | start := time.Now() | 305 | start := time.Now() | ||
| 319 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "5s") | 306 | + | _, err := sandbox.ensure(5 * time.Second) |
| 320 | - | require.Equal(t, 3, code) | 307 | + | require.ErrorContains(t, err, "live but incompatible") |
| 321 | - | require.Equal(t, "offline", decodeOneReport(t, stdout).Status) | |||
| 322 | - | require.Contains(t, stderr, "live but incompatible", "a silent live endpoint must be classified incompatible") | |||
| 323 | require.Less(t, time.Since(start), 5*time.Second, "the probe budget must classify a silent endpoint before the overall deadline") | 308 | require.Less(t, time.Since(start), 5*time.Second, "the probe budget must classify a silent endpoint before the overall deadline") | ||
| 324 | - | require.Empty(t, sandbox.recorded(brokerServeCommand), "a live-but-silent endpoint must not be respawned over") | 309 | + | require.Empty(t, sandbox.recorded(productionBrokerServeCommand), "a live-but-silent endpoint must not be respawned over") |
| 325 | } | 310 | } | ||
| 326 | 311 | ||||
| 327 | // TestBrokerLauncherSpawnWaitIsBoundedByContext proves a controlled blocking | 312 | // TestBrokerLauncherSpawnWaitIsBoundedByContext proves a controlled blocking | ||
| @@ -340,7 +325,7 @@ func TestBrokerLauncherSpawnWaitIsBoundedByContext(t *testing.T) { | |||||
| 340 | ctx, cancel := context.WithCancel(context.Background()) | 325 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 341 | defer cancel() | 326 | defer cancel() | ||
| 342 | result := make(chan error, 1) | 327 | result := make(chan error, 1) | ||
| 343 | - | go func() { result <- spawnBrokerLauncher(ctx, t.TempDir(), 0, false) }() | 328 | + | go func() { result <- spawnProductionBrokerLauncher(ctx) }() |
| 344 | 329 | ||||
| 345 | pid := waitForProcessRecord(t, record) | 330 | pid := waitForProcessRecord(t, record) | ||
| 346 | select { | 331 | select { | ||
| @@ -362,31 +347,23 @@ func TestBrokerLauncherSpawnWaitIsBoundedByContext(t *testing.T) { | |||||
| 362 | // TestBrokerStatusSubprocessLauncherDiagnosesConflicts proves an explicit | 347 | // TestBrokerStatusSubprocessLauncherDiagnosesConflicts proves an explicit | ||
| 363 | // launcher grace that conflicts with the provisioned one is refused before any | 348 | // launcher grace that conflicts with the provisioned one is refused before any | ||
| 364 | // process is spawned, and that a missing configuration fails the launcher. | 349 | // process is spawned, and that a missing configuration fails the launcher. | ||
| 350 | + | // A production launcher accepts no idle override; malformed broker.json is | |||
| 351 | + | // diagnosed by the detached serve before it can claim the lifetime lock. | |||
| 365 | func TestBrokerStatusSubprocessLauncherDiagnosesConflicts(t *testing.T) { | 352 | func TestBrokerStatusSubprocessLauncherDiagnosesConflicts(t *testing.T) { | ||
| 366 | - | t.Run("conflicting idle grace", func(t *testing.T) { | 353 | + | t.Run("unsupported idle override", func(t *testing.T) { |
| 367 | sandbox := newBrokerSubprocessSandbox(t, "90s") | 354 | sandbox := newBrokerSubprocessSandbox(t, "90s") | ||
| 368 | - | stdout, stderr, code := runBrokerCommand(t, brokerLauncherCommand, "--offline-root", sandbox.root, "--idle-grace", "1m") | 355 | + | _, stderr, code := runBrokerCommand(t, productionBrokerLauncherCommand, "--idle-grace", "1m") |
| 369 | - | require.NotEqual(t, 0, code) | 356 | + | require.NotZero(t, code) |
| 370 | - | require.Contains(t, stderr, "conflicts with the provisioned idle grace") | 357 | + | require.Contains(t, stderr, "does not accept arguments") |
| 371 | - | require.Empty(t, strings.TrimSpace(stdout)) | 358 | + | require.Empty(t, sandbox.recorded(productionBrokerServeCommand)) |
| 372 | - | require.Empty(t, sandbox.recorded(brokerServeCommand)) | |||
| 373 | }) | 359 | }) | ||
| 374 | - | 360 | + | t.Run("invalid production configuration", func(t *testing.T) { | |
| 375 | - | t.Run("missing configuration", func(t *testing.T) { | 361 | + | sandbox := newBrokerSubprocessSandbox(t, "") |
| 376 | - | prodRuntime, prodState := t.TempDir(), t.TempDir() | 362 | + | require.NoError(t, os.WriteFile(productionBrokerConfigPath(), []byte("not json"), 0o600)) |
| 377 | - | t.Setenv("XDG_RUNTIME_DIR", prodRuntime) | 363 | + | _, stderr, code := runBrokerCommand(t, productionBrokerServeCommand) |
| 378 | - | t.Setenv("XDG_STATE_HOME", prodState) | 364 | + | require.NotZero(t, code) |
| 379 | - | recordDir := filepath.Join(t.TempDir(), "records") | 365 | + | require.Contains(t, stderr, brokerconfig.ConfigFileName) |
| 380 | - | require.NoError(t, os.MkdirAll(recordDir, 0o700)) | 366 | + | require.NoFileExists(t, filepath.Join(sandbox.layout.Runtime, "lifecycle.lock")) |
| 381 | - | t.Setenv(brokerHelperEnv, "1") | |||
| 382 | - | t.Setenv(brokerHelperRecordDirEnv, recordDir) | |||
| 383 | - | root := filepath.Join(shortTempDir(t, "vevx"), "s") | |||
| 384 | - | require.NoError(t, os.MkdirAll(root, 0o700)) | |||
| 385 | - | ||||
| 386 | - | _, stderr, code := runBrokerCommand(t, brokerLauncherCommand, "--offline-root", root) | |||
| 387 | - | require.NotEqual(t, 0, code) | |||
| 388 | - | require.NotEmpty(t, strings.TrimSpace(stderr)) | |||
| 389 | - | require.Empty(t, recordedPIDsIn(t, recordDir, brokerServeCommand)) | |||
| 390 | }) | 367 | }) | ||
| 391 | } | 368 | } | ||
| 392 | 369 | ||||
| @@ -395,27 +372,26 @@ func TestBrokerStatusSubprocessLauncherDiagnosesConflicts(t *testing.T) { | |||||
| 395 | func TestBrokerStatusSubprocessNoOrphanProcess(t *testing.T) { | 372 | func TestBrokerStatusSubprocessNoOrphanProcess(t *testing.T) { | ||
| 396 | sandbox := newBrokerSubprocessSandbox(t, "5s") | 373 | sandbox := newBrokerSubprocessSandbox(t, "5s") | ||
| 397 | 374 | ||||
| 398 | - | stdout, stderr, code := sandbox.status("--ensure", "--timeout", "20s") | 375 | + | report, err := sandbox.ensure(20 * time.Second) |
| 399 | - | require.Equal(t, 0, code, stderr) | 376 | + | require.NoError(t, err) |
| 400 | - | require.Equal(t, "ready", decodeOneReport(t, stdout).Status) | 377 | + | require.Equal(t, "ready", report.Status) |
| 401 | 378 | ||||
| 402 | - | require.NotEmpty(t, sandbox.recorded(brokerLauncherCommand)) | 379 | + | require.NotEmpty(t, sandbox.recorded(productionBrokerLauncherCommand)) |
| 403 | - | for _, pid := range sandbox.recorded(brokerLauncherCommand) { | 380 | + | for _, pid := range sandbox.recorded(productionBrokerLauncherCommand) { |
| 404 | require.NoError(t, waitForProcessExit(pid, 5*time.Second), "the short-lived launcher must exit") | 381 | require.NoError(t, waitForProcessExit(pid, 5*time.Second), "the short-lived launcher must exit") | ||
| 405 | } | 382 | } | ||
| 406 | - | require.Len(t, sandbox.live(brokerServeCommand), 1) | 383 | + | require.Len(t, sandbox.live(productionBrokerServeCommand), 1) |
| 407 | 384 | ||||
| 408 | - | _, err := lifecycle.TryAcquire(sandbox.layout.Runtime) | 385 | + | _, err = lifecycle.TryAcquire(sandbox.layout.Runtime) |
| 409 | require.ErrorIs(t, err, lifecycle.ErrBusy, "the live broker must own the lifetime lock") | 386 | require.ErrorIs(t, err, lifecycle.ErrBusy, "the live broker must own the lifetime lock") | ||
| 410 | 387 | ||||
| 411 | - | sandbox.terminate(brokerServeCommand, syscall.SIGTERM) | 388 | + | sandbox.terminate(productionBrokerServeCommand, syscall.SIGTERM) |
| 412 | - | require.Empty(t, sandbox.live(brokerLauncherCommand), "no launcher may be left behind") | 389 | + | require.Empty(t, sandbox.live(productionBrokerLauncherCommand), "no launcher may be left behind") |
| 413 | - | require.Empty(t, sandbox.live(brokerServeCommand), "no broker may be left behind") | 390 | + | require.Empty(t, sandbox.live(productionBrokerServeCommand), "no broker may be left behind") |
| 414 | require.True(t, sandbox.waitForSocketGone(5*time.Second), "an orderly broker shutdown must unlink its socket") | 391 | require.True(t, sandbox.waitForSocketGone(5*time.Second), "an orderly broker shutdown must unlink its socket") | ||
| 415 | owner, err := lifecycle.TryAcquire(sandbox.layout.Runtime) | 392 | owner, err := lifecycle.TryAcquire(sandbox.layout.Runtime) | ||
| 416 | require.NoError(t, err, "the terminated broker must release the lifetime lock") | 393 | require.NoError(t, err, "the terminated broker must release the lifetime lock") | ||
| 417 | require.NoError(t, owner.Release()) | 394 | require.NoError(t, owner.Release()) | ||
| 418 | - | requireProductionUntouched(t, sandbox.prodRuntime, sandbox.prodState) | |||
| 419 | } | 395 | } | ||
| 420 | 396 | ||||
| 421 | // recordedPIDsIn reads one role's PID record from an arbitrary directory. | 397 | // recordedPIDsIn reads one role's PID record from an arbitrary directory. | ||
M internal/app/spawn_test.go +1 -1
| @@ -35,7 +35,7 @@ func TestMain(m *testing.M) { | |||
| 35 | 35 | brokerReadyCommand, | |
| 36 | 36 | productionBrokerServeCommand, productionBrokerLauncherCommand, "--daemon": | |
| 37 | 37 | recordBrokerHelperProcess(os.Args[1]) | |
| 38 | - | if os.Args[1] == brokerLauncherCommand { | |
| 38 | + | if os.Args[1] == brokerLauncherCommand || os.Args[1] == productionBrokerLauncherCommand { | |
| 39 | 39 | if record := os.Getenv(brokerLauncherBlockRecordEnv); record != "" { | |
| 40 | 40 | blockBrokerLauncherHelper(record, os.Getenv(brokerLauncherBlockReleaseEnv)) | |
| 41 | 41 | } | |
| @@ -35,7 +35,7 @@ func TestMain(m *testing.M) { | |||||
| 35 | brokerReadyCommand, | 35 | brokerReadyCommand, | ||
| 36 | productionBrokerServeCommand, productionBrokerLauncherCommand, "--daemon": | 36 | productionBrokerServeCommand, productionBrokerLauncherCommand, "--daemon": | ||
| 37 | recordBrokerHelperProcess(os.Args[1]) | 37 | recordBrokerHelperProcess(os.Args[1]) | ||
| 38 | - | if os.Args[1] == brokerLauncherCommand { | 38 | + | if os.Args[1] == brokerLauncherCommand || os.Args[1] == productionBrokerLauncherCommand { |
| 39 | if record := os.Getenv(brokerLauncherBlockRecordEnv); record != "" { | 39 | if record := os.Getenv(brokerLauncherBlockRecordEnv); record != "" { | ||
| 40 | blockBrokerLauncherHelper(record, os.Getenv(brokerLauncherBlockReleaseEnv)) | 40 | blockBrokerLauncherHelper(record, os.Getenv(brokerLauncherBlockReleaseEnv)) | ||
| 41 | } | 41 | } | ||