fix(perf): start isolated daemon before probing broker catalogue
- commit
- 7f208e1361feb2fe7725fdc4919592199adca93e
- author
- brice · · 2026-09-23 14:04 UTC
- parent
- cf66883
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +35 -8
M cmd/vev-perf-harness/launcher.go +8 -5
| @@ -221,7 +221,9 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces | |||
| 221 | 221 | } | |
| 222 | 222 | p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1), serverOutput: &synchronizedBuffer{}} | |
| 223 | 223 | if m.Role == "daemon" { | |
| 224 | - | p.startCommand = exec.Command(bin, "ls") | |
| 224 | + | // ls is read-only and cannot start an absent daemon. Start the daemon | |
| 225 | + | // explicitly before requiring its observed local catalogue. | |
| 226 | + | p.startCommand = exec.Command(bin, "--daemon-launcher") | |
| 225 | 227 | p.startCommand.Env = cmd.Env | |
| 226 | 228 | p.readyCommand = exec.Command(bin, "_broker-ready", "--require", "local-catalogue", "--timeout", "10s") | |
| 227 | 229 | p.readyCommand.Env = cmd.Env | |
| @@ -516,10 +518,11 @@ func (p *cliProcess) copyTerminal() { | |||
| 516 | 518 | // stdout, stderr and owner-process death are retained in any failure. | |
| 517 | 519 | func (p *cliProcess) WaitReady() error { | |
| 518 | 520 | if p.startCommand != nil { | |
| 519 | - | // The first control stream is only a start authorization. The daemon may | |
| 520 | - | // close that stream while publishing its first catalogue, so readiness is | |
| 521 | - | // decided exclusively by the semantic probe below. | |
| 522 | - | _, _ = p.startCommand.CombinedOutput() | |
| 521 | + | // Starting the daemon is separate from readiness: only the semantic | |
| 522 | + | // catalogue probe below can establish that it is serving. | |
| 523 | + | if output, err := p.startCommand.CombinedOutput(); err != nil { | |
| 524 | + | return fmt.Errorf("start daemon for broker readiness: %w (output=%q)", err, output) | |
| 525 | + | } | |
| 523 | 526 | } | |
| 524 | 527 | if p.readyCommand == nil { | |
| 525 | 528 | return errors.New("broker readiness command is empty") | |
| @@ -221,7 +221,9 @@ func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProces | |||||
| 221 | } | 221 | } | ||
| 222 | p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1), serverOutput: &synchronizedBuffer{}} | 222 | p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1), serverOutput: &synchronizedBuffer{}} | ||
| 223 | if m.Role == "daemon" { | 223 | if m.Role == "daemon" { | ||
| 224 | - | p.startCommand = exec.Command(bin, "ls") | 224 | + | // ls is read-only and cannot start an absent daemon. Start the daemon |
| 225 | + | // explicitly before requiring its observed local catalogue. | |||
| 226 | + | p.startCommand = exec.Command(bin, "--daemon-launcher") | |||
| 225 | p.startCommand.Env = cmd.Env | 227 | p.startCommand.Env = cmd.Env | ||
| 226 | p.readyCommand = exec.Command(bin, "_broker-ready", "--require", "local-catalogue", "--timeout", "10s") | 228 | p.readyCommand = exec.Command(bin, "_broker-ready", "--require", "local-catalogue", "--timeout", "10s") | ||
| 227 | p.readyCommand.Env = cmd.Env | 229 | p.readyCommand.Env = cmd.Env | ||
| @@ -516,10 +518,11 @@ func (p *cliProcess) copyTerminal() { | |||||
| 516 | // stdout, stderr and owner-process death are retained in any failure. | 518 | // stdout, stderr and owner-process death are retained in any failure. | ||
| 517 | func (p *cliProcess) WaitReady() error { | 519 | func (p *cliProcess) WaitReady() error { | ||
| 518 | if p.startCommand != nil { | 520 | if p.startCommand != nil { | ||
| 519 | - | // The first control stream is only a start authorization. The daemon may | 521 | + | // Starting the daemon is separate from readiness: only the semantic |
| 520 | - | // close that stream while publishing its first catalogue, so readiness is | 522 | + | // catalogue probe below can establish that it is serving. |
| 521 | - | // decided exclusively by the semantic probe below. | 523 | + | if output, err := p.startCommand.CombinedOutput(); err != nil { |
| 522 | - | _, _ = p.startCommand.CombinedOutput() | 524 | + | return fmt.Errorf("start daemon for broker readiness: %w (output=%q)", err, output) |
| 525 | + | } | |||
| 523 | } | 526 | } | ||
| 524 | if p.readyCommand == nil { | 527 | if p.readyCommand == nil { | ||
| 525 | return errors.New("broker readiness command is empty") | 528 | return errors.New("broker readiness command is empty") | ||
M cmd/vev-perf-harness/launcher_test.go +27 -3
| @@ -337,9 +337,33 @@ func TestCLIProcessCloseEscalationOutcome(t *testing.T) { | |||
| 337 | 337 | } | |
| 338 | 338 | ||
| 339 | 339 | func TestCLIProcessWaitReadyRunsJSONProbe(t *testing.T) { | |
| 340 | - | command := exec.Command("sh", "-c", `printf '%s\n' '{"schema":"vev.broker-ready/v1","status":"ready"}'`) | |
| 341 | - | if err := (&cliProcess{readyCommand: command, waitErr: make(chan error, 1)}).WaitReady(); err != nil { | |
| 342 | - | t.Fatal(err) | |
| 340 | + | tests := []struct { | |
| 341 | + | name string | |
| 342 | + | start string | |
| 343 | + | wantErr string | |
| 344 | + | wantProbe bool | |
| 345 | + | }{ | |
| 346 | + | {name: "daemon started", start: "exit 0", wantProbe: true}, | |
| 347 | + | {name: "daemon start failed", start: "exit 7", wantErr: "start daemon for broker readiness: exit status 7"}, | |
| 348 | + | } | |
| 349 | + | for _, tt := range tests { | |
| 350 | + | t.Run(tt.name, func(t *testing.T) { | |
| 351 | + | marker := filepath.Join(t.TempDir(), "probe") | |
| 352 | + | probe := exec.Command("sh", "-c", `printf '%s\n' '{"schema":"vev.broker-ready/v1","status":"ready"}'; touch "$1"`, "sh", marker) | |
| 353 | + | p := &cliProcess{startCommand: exec.Command("sh", "-c", tt.start), readyCommand: probe, waitErr: make(chan error, 1)} | |
| 354 | + | err := p.WaitReady() | |
| 355 | + | if tt.wantErr == "" { | |
| 356 | + | if err != nil { | |
| 357 | + | t.Fatal(err) | |
| 358 | + | } | |
| 359 | + | } else if err == nil || !strings.Contains(err.Error(), tt.wantErr) { | |
| 360 | + | t.Fatalf("WaitReady() = %v, want %q", err, tt.wantErr) | |
| 361 | + | } | |
| 362 | + | _, statErr := os.Stat(marker) | |
| 363 | + | if (statErr == nil) != tt.wantProbe { | |
| 364 | + | t.Fatalf("readiness probe ran = %t, want %t (stat: %v)", statErr == nil, tt.wantProbe, statErr) | |
| 365 | + | } | |
| 366 | + | }) | |
| 343 | 367 | } | |
| 344 | 368 | } | |
| 345 | 369 | ||
| @@ -337,9 +337,33 @@ func TestCLIProcessCloseEscalationOutcome(t *testing.T) { | |||||
| 337 | } | 337 | } | ||
| 338 | 338 | ||||
| 339 | func TestCLIProcessWaitReadyRunsJSONProbe(t *testing.T) { | 339 | func TestCLIProcessWaitReadyRunsJSONProbe(t *testing.T) { | ||
| 340 | - | command := exec.Command("sh", "-c", `printf '%s\n' '{"schema":"vev.broker-ready/v1","status":"ready"}'`) | 340 | + | tests := []struct { |
| 341 | - | if err := (&cliProcess{readyCommand: command, waitErr: make(chan error, 1)}).WaitReady(); err != nil { | 341 | + | name string |
| 342 | - | t.Fatal(err) | 342 | + | start string |
| 343 | + | wantErr string | |||
| 344 | + | wantProbe bool | |||
| 345 | + | }{ | |||
| 346 | + | {name: "daemon started", start: "exit 0", wantProbe: true}, | |||
| 347 | + | {name: "daemon start failed", start: "exit 7", wantErr: "start daemon for broker readiness: exit status 7"}, | |||
| 348 | + | } | |||
| 349 | + | for _, tt := range tests { | |||
| 350 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 351 | + | marker := filepath.Join(t.TempDir(), "probe") | |||
| 352 | + | probe := exec.Command("sh", "-c", `printf '%s\n' '{"schema":"vev.broker-ready/v1","status":"ready"}'; touch "$1"`, "sh", marker) | |||
| 353 | + | p := &cliProcess{startCommand: exec.Command("sh", "-c", tt.start), readyCommand: probe, waitErr: make(chan error, 1)} | |||
| 354 | + | err := p.WaitReady() | |||
| 355 | + | if tt.wantErr == "" { | |||
| 356 | + | if err != nil { | |||
| 357 | + | t.Fatal(err) | |||
| 358 | + | } | |||
| 359 | + | } else if err == nil || !strings.Contains(err.Error(), tt.wantErr) { | |||
| 360 | + | t.Fatalf("WaitReady() = %v, want %q", err, tt.wantErr) | |||
| 361 | + | } | |||
| 362 | + | _, statErr := os.Stat(marker) | |||
| 363 | + | if (statErr == nil) != tt.wantProbe { | |||
| 364 | + | t.Fatalf("readiness probe ran = %t, want %t (stat: %v)", statErr == nil, tt.wantProbe, statErr) | |||
| 365 | + | } | |||
| 366 | + | }) | |||
| 343 | } | 367 | } | ||
| 344 | } | 368 | } | ||
| 345 | 369 | ||||