fix(obs): validate harness terminal evidence
- commit
- 1ffeb5584c39ba44ff282bae80a3ed00ab2a5bec
- author
- brice · · 2026-07-12 23:35 UTC
- parent
- 72845c9
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +177 -24
M cmd/vev-perf-harness/main.go +104 -21
| @@ -210,15 +210,16 @@ type terminalOutput interface { | |||
| 210 | 210 | } | |
| 211 | 211 | ||
| 212 | 212 | type cliProcess struct { | |
| 213 | - | cmd *exec.Cmd | |
| 214 | - | pty io.ReadWriteCloser | |
| 215 | - | output terminalOutput | |
| 216 | - | chunks chan []byte | |
| 217 | - | done chan struct{} | |
| 218 | - | closed sync.Once | |
| 219 | - | waitErr chan error | |
| 220 | - | waitTimeout func() <-chan time.Time | |
| 221 | - | forceCleanup func() | |
| 213 | + | cmd *exec.Cmd | |
| 214 | + | pty io.ReadWriteCloser | |
| 215 | + | output terminalOutput | |
| 216 | + | chunks chan []byte | |
| 217 | + | done chan struct{} | |
| 218 | + | closed sync.Once | |
| 219 | + | waitErr chan error | |
| 220 | + | waitTimeout func() <-chan time.Time | |
| 221 | + | forceCleanup func() | |
| 222 | + | terminalReady bool | |
| 222 | 223 | } | |
| 223 | 224 | ||
| 224 | 225 | func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) { | |
| @@ -413,17 +414,27 @@ func (p *cliProcess) Warmup(input []byte) error { | |||
| 413 | 414 | if p.pty == nil || len(input) == 0 { | |
| 414 | 415 | return nil | |
| 415 | 416 | } | |
| 417 | + | // A client cannot write application output until it has received Welcome and | |
| 418 | + | // entered raw mode. Require that observable output before submitting any | |
| 419 | + | // warmup input, so the PTY's line discipline cannot echo it locally. | |
| 420 | + | if !p.terminalReady { | |
| 421 | + | if err := p.waitForTerminalReady(); err != nil { | |
| 422 | + | return err | |
| 423 | + | } | |
| 424 | + | p.terminalReady = true | |
| 425 | + | } | |
| 416 | 426 | if _, err := p.pty.Write(input); err != nil { | |
| 417 | 427 | return err | |
| 418 | 428 | } | |
| 419 | 429 | // Do not begin measured input until the public client has rendered a real | |
| 420 | 430 | // shell response to warmup input. This is intentionally not a measurement. | |
| 421 | - | return p.waitForTerminalMarker(inputMarker(input)) | |
| 431 | + | return p.waitForTerminalMarker(inputMarker(input), input) | |
| 422 | 432 | } | |
| 423 | 433 | ||
| 424 | 434 | // Measure records a boundary only after the harness-owned terminal output file | |
| 425 | - | // has accepted data and Sync has reported a successful flush. The marker is | |
| 426 | - | // injected through the real client PTY, never fabricated by the harness. | |
| 435 | + | // has accepted application output and Sync has reported a successful flush. | |
| 436 | + | // The marker is injected through the real client PTY, never fabricated by the | |
| 437 | + | // harness. PTY local echo is discarded and cannot satisfy this boundary. | |
| 427 | 438 | func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error { | |
| 428 | 439 | if p.pty == nil { | |
| 429 | 440 | return errors.New("measured process has no terminal PTY") | |
| @@ -435,7 +446,7 @@ func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error | |||
| 435 | 446 | if err := injected(); err != nil { | |
| 436 | 447 | return err | |
| 437 | 448 | } | |
| 438 | - | if err := p.waitForTerminalMarker(inputMarker(input)); err != nil { | |
| 449 | + | if err := p.waitForTerminalMarker(inputMarker(input), input); err != nil { | |
| 439 | 450 | return err | |
| 440 | 451 | } | |
| 441 | 452 | if err := p.output.Sync(); err != nil { | |
| @@ -444,26 +455,98 @@ func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error | |||
| 444 | 455 | return flushed() | |
| 445 | 456 | } | |
| 446 | 457 | ||
| 447 | - | func (p *cliProcess) waitForTerminalMarker(marker []byte) error { | |
| 458 | + | // waitForTerminalReady waits for public client output produced before any | |
| 459 | + | // harness input. The client enters raw mode before its output loop can render | |
| 460 | + | // this initial prompt/state, so this is an observable readiness boundary. | |
| 461 | + | func (p *cliProcess) waitForTerminalReady() error { | |
| 462 | + | timer := time.NewTimer(10 * time.Second) | |
| 463 | + | defer timer.Stop() | |
| 464 | + | for { | |
| 465 | + | select { | |
| 466 | + | case chunk, ok := <-p.chunks: | |
| 467 | + | if !ok { | |
| 468 | + | return errors.New("client terminal closed before readiness output") | |
| 469 | + | } | |
| 470 | + | if len(chunk) == 0 { | |
| 471 | + | continue | |
| 472 | + | } | |
| 473 | + | if err := p.output.Sync(); err != nil { | |
| 474 | + | return fmt.Errorf("flush terminal readiness output: %w", err) | |
| 475 | + | } | |
| 476 | + | return nil | |
| 477 | + | case err := <-p.waitErr: | |
| 478 | + | // Keep the exit result available to Close, which owns reaping. | |
| 479 | + | p.waitErr <- err | |
| 480 | + | if err == nil { | |
| 481 | + | return errors.New("client exited before readiness output") | |
| 482 | + | } | |
| 483 | + | return fmt.Errorf("client exited before readiness output: %w", err) | |
| 484 | + | case <-timer.C: | |
| 485 | + | return errors.New("timed out waiting for client readiness output") | |
| 486 | + | } | |
| 487 | + | } | |
| 488 | + | } | |
| 489 | + | ||
| 490 | + | func (p *cliProcess) waitForTerminalMarker(marker, input []byte) error { | |
| 448 | 491 | var received []byte | |
| 492 | + | echo := newPTYLocalEcho(input) | |
| 449 | 493 | timer := time.NewTimer(10 * time.Second) | |
| 450 | 494 | defer timer.Stop() | |
| 451 | 495 | for !bytes.Contains(received, marker) { | |
| 452 | 496 | select { | |
| 453 | 497 | case chunk, ok := <-p.chunks: | |
| 454 | 498 | if !ok { | |
| 455 | - | return errors.New("client terminal closed before terminal output") | |
| 499 | + | return errors.New("client terminal closed before application output") | |
| 456 | 500 | } | |
| 457 | - | received = append(received, chunk...) | |
| 501 | + | received = append(received, echo.applicationOutput(chunk)...) | |
| 458 | 502 | case err := <-p.waitErr: | |
| 459 | 503 | // Keep the exit result available to Close, which owns reaping. | |
| 460 | 504 | p.waitErr <- err | |
| 461 | 505 | if err == nil { | |
| 462 | - | return errors.New("client exited before terminal output") | |
| 506 | + | return errors.New("client exited before application output") | |
| 463 | 507 | } | |
| 464 | - | return fmt.Errorf("client exited before terminal output: %w", err) | |
| 508 | + | return fmt.Errorf("client exited before application output: %w", err) | |
| 465 | 509 | case <-timer.C: | |
| 466 | - | return errors.New("timed out waiting for terminal output") | |
| 510 | + | return errors.New("timed out waiting for application terminal output") | |
| 511 | + | } | |
| 512 | + | } | |
| 513 | + | return nil | |
| 514 | + | } | |
| 515 | + | ||
| 516 | + | // ptyLocalEcho removes the one ordered input echo that a not-yet-raw slave PTY | |
| 517 | + | // can produce. It normalizes NL to the conventional ONLCR echo spelling; any | |
| 518 | + | // divergence is retained as application output. The line discipline emits the | |
| 519 | + | // echo before the child can read and respond to the same write. | |
| 520 | + | type ptyLocalEcho struct { | |
| 521 | + | expected []byte | |
| 522 | + | matched int | |
| 523 | + | done bool | |
| 524 | + | } | |
| 525 | + | ||
| 526 | + | func newPTYLocalEcho(input []byte) *ptyLocalEcho { | |
| 527 | + | expected := make([]byte, 0, len(input)) | |
| 528 | + | for _, b := range input { | |
| 529 | + | if b == '\n' { | |
| 530 | + | expected = append(expected, '\r') | |
| 531 | + | } | |
| 532 | + | expected = append(expected, b) | |
| 533 | + | } | |
| 534 | + | return &ptyLocalEcho{expected: expected} | |
| 535 | + | } | |
| 536 | + | ||
| 537 | + | func (e *ptyLocalEcho) applicationOutput(chunk []byte) []byte { | |
| 538 | + | if e.done || len(e.expected) == 0 { | |
| 539 | + | return chunk | |
| 540 | + | } | |
| 541 | + | for i, b := range chunk { | |
| 542 | + | if b != e.expected[e.matched] { | |
| 543 | + | e.done = true | |
| 544 | + | return append([]byte(nil), chunk[i:]...) | |
| 545 | + | } | |
| 546 | + | e.matched++ | |
| 547 | + | if e.matched == len(e.expected) { | |
| 548 | + | e.done = true | |
| 549 | + | return append([]byte(nil), chunk[i+1:]...) | |
| 467 | 550 | } | |
| 468 | 551 | } | |
| 469 | 552 | return nil | |
| @@ -1073,9 +1156,9 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) { | |||
| 1073 | 1156 | _ = f.Close() | |
| 1074 | 1157 | return nil, err | |
| 1075 | 1158 | } | |
| 1076 | - | if r.ProcessID != m.ProcessID { | |
| 1159 | + | if r.ProcessID != m.ProcessID || r.Scenario != m.Scenario || r.Run != uint64(m.Run) { | |
| 1077 | 1160 | _ = f.Close() | |
| 1078 | - | return nil, errors.New("trace record process_id does not match manifest") | |
| 1161 | + | return nil, errors.New("trace record identity does not match manifest") | |
| 1079 | 1162 | } | |
| 1080 | 1163 | if r.Scenario == "" || r.Run == 0 || r.Component == "" || r.Sequence == 0 || r.RequestID == 0 || r.Epoch == 0 { | |
| 1081 | 1164 | _ = f.Close() | |
| @@ -210,15 +210,16 @@ type terminalOutput interface { | |||||
| 210 | } | 210 | } | ||
| 211 | 211 | ||||
| 212 | type cliProcess struct { | 212 | type cliProcess struct { | ||
| 213 | - | cmd *exec.Cmd | 213 | + | cmd *exec.Cmd |
| 214 | - | pty io.ReadWriteCloser | 214 | + | pty io.ReadWriteCloser |
| 215 | - | output terminalOutput | 215 | + | output terminalOutput |
| 216 | - | chunks chan []byte | 216 | + | chunks chan []byte |
| 217 | - | done chan struct{} | 217 | + | done chan struct{} |
| 218 | - | closed sync.Once | 218 | + | closed sync.Once |
| 219 | - | waitErr chan error | 219 | + | waitErr chan error |
| 220 | - | waitTimeout func() <-chan time.Time | 220 | + | waitTimeout func() <-chan time.Time |
| 221 | - | forceCleanup func() | 221 | + | forceCleanup func() |
| 222 | + | terminalReady bool | |||
| 222 | } | 223 | } | ||
| 223 | 224 | ||||
| 224 | func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) { | 225 | func (l *cliLauncher) Launch(m processMapping, role roleCommand) (launchedProcess, error) { | ||
| @@ -413,17 +414,27 @@ func (p *cliProcess) Warmup(input []byte) error { | |||||
| 413 | if p.pty == nil || len(input) == 0 { | 414 | if p.pty == nil || len(input) == 0 { | ||
| 414 | return nil | 415 | return nil | ||
| 415 | } | 416 | } | ||
| 417 | + | // A client cannot write application output until it has received Welcome and | |||
| 418 | + | // entered raw mode. Require that observable output before submitting any | |||
| 419 | + | // warmup input, so the PTY's line discipline cannot echo it locally. | |||
| 420 | + | if !p.terminalReady { | |||
| 421 | + | if err := p.waitForTerminalReady(); err != nil { | |||
| 422 | + | return err | |||
| 423 | + | } | |||
| 424 | + | p.terminalReady = true | |||
| 425 | + | } | |||
| 416 | if _, err := p.pty.Write(input); err != nil { | 426 | if _, err := p.pty.Write(input); err != nil { | ||
| 417 | return err | 427 | return err | ||
| 418 | } | 428 | } | ||
| 419 | // Do not begin measured input until the public client has rendered a real | 429 | // Do not begin measured input until the public client has rendered a real | ||
| 420 | // shell response to warmup input. This is intentionally not a measurement. | 430 | // shell response to warmup input. This is intentionally not a measurement. | ||
| 421 | - | return p.waitForTerminalMarker(inputMarker(input)) | 431 | + | return p.waitForTerminalMarker(inputMarker(input), input) |
| 422 | } | 432 | } | ||
| 423 | 433 | ||||
| 424 | // Measure records a boundary only after the harness-owned terminal output file | 434 | // Measure records a boundary only after the harness-owned terminal output file | ||
| 425 | - | // has accepted data and Sync has reported a successful flush. The marker is | 435 | + | // has accepted application output and Sync has reported a successful flush. |
| 426 | - | // injected through the real client PTY, never fabricated by the harness. | 436 | + | // The marker is injected through the real client PTY, never fabricated by the |
| 437 | + | // harness. PTY local echo is discarded and cannot satisfy this boundary. | |||
| 427 | func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error { | 438 | func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error { | ||
| 428 | if p.pty == nil { | 439 | if p.pty == nil { | ||
| 429 | return errors.New("measured process has no terminal PTY") | 440 | return errors.New("measured process has no terminal PTY") | ||
| @@ -435,7 +446,7 @@ func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error | |||||
| 435 | if err := injected(); err != nil { | 446 | if err := injected(); err != nil { | ||
| 436 | return err | 447 | return err | ||
| 437 | } | 448 | } | ||
| 438 | - | if err := p.waitForTerminalMarker(inputMarker(input)); err != nil { | 449 | + | if err := p.waitForTerminalMarker(inputMarker(input), input); err != nil { |
| 439 | return err | 450 | return err | ||
| 440 | } | 451 | } | ||
| 441 | if err := p.output.Sync(); err != nil { | 452 | if err := p.output.Sync(); err != nil { | ||
| @@ -444,26 +455,98 @@ func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error | |||||
| 444 | return flushed() | 455 | return flushed() | ||
| 445 | } | 456 | } | ||
| 446 | 457 | ||||
| 447 | - | func (p *cliProcess) waitForTerminalMarker(marker []byte) error { | 458 | + | // waitForTerminalReady waits for public client output produced before any |
| 459 | + | // harness input. The client enters raw mode before its output loop can render | |||
| 460 | + | // this initial prompt/state, so this is an observable readiness boundary. | |||
| 461 | + | func (p *cliProcess) waitForTerminalReady() error { | |||
| 462 | + | timer := time.NewTimer(10 * time.Second) | |||
| 463 | + | defer timer.Stop() | |||
| 464 | + | for { | |||
| 465 | + | select { | |||
| 466 | + | case chunk, ok := <-p.chunks: | |||
| 467 | + | if !ok { | |||
| 468 | + | return errors.New("client terminal closed before readiness output") | |||
| 469 | + | } | |||
| 470 | + | if len(chunk) == 0 { | |||
| 471 | + | continue | |||
| 472 | + | } | |||
| 473 | + | if err := p.output.Sync(); err != nil { | |||
| 474 | + | return fmt.Errorf("flush terminal readiness output: %w", err) | |||
| 475 | + | } | |||
| 476 | + | return nil | |||
| 477 | + | case err := <-p.waitErr: | |||
| 478 | + | // Keep the exit result available to Close, which owns reaping. | |||
| 479 | + | p.waitErr <- err | |||
| 480 | + | if err == nil { | |||
| 481 | + | return errors.New("client exited before readiness output") | |||
| 482 | + | } | |||
| 483 | + | return fmt.Errorf("client exited before readiness output: %w", err) | |||
| 484 | + | case <-timer.C: | |||
| 485 | + | return errors.New("timed out waiting for client readiness output") | |||
| 486 | + | } | |||
| 487 | + | } | |||
| 488 | + | } | |||
| 489 | + | ||||
| 490 | + | func (p *cliProcess) waitForTerminalMarker(marker, input []byte) error { | |||
| 448 | var received []byte | 491 | var received []byte | ||
| 492 | + | echo := newPTYLocalEcho(input) | |||
| 449 | timer := time.NewTimer(10 * time.Second) | 493 | timer := time.NewTimer(10 * time.Second) | ||
| 450 | defer timer.Stop() | 494 | defer timer.Stop() | ||
| 451 | for !bytes.Contains(received, marker) { | 495 | for !bytes.Contains(received, marker) { | ||
| 452 | select { | 496 | select { | ||
| 453 | case chunk, ok := <-p.chunks: | 497 | case chunk, ok := <-p.chunks: | ||
| 454 | if !ok { | 498 | if !ok { | ||
| 455 | - | return errors.New("client terminal closed before terminal output") | 499 | + | return errors.New("client terminal closed before application output") |
| 456 | } | 500 | } | ||
| 457 | - | received = append(received, chunk...) | 501 | + | received = append(received, echo.applicationOutput(chunk)...) |
| 458 | case err := <-p.waitErr: | 502 | case err := <-p.waitErr: | ||
| 459 | // Keep the exit result available to Close, which owns reaping. | 503 | // Keep the exit result available to Close, which owns reaping. | ||
| 460 | p.waitErr <- err | 504 | p.waitErr <- err | ||
| 461 | if err == nil { | 505 | if err == nil { | ||
| 462 | - | return errors.New("client exited before terminal output") | 506 | + | return errors.New("client exited before application output") |
| 463 | } | 507 | } | ||
| 464 | - | return fmt.Errorf("client exited before terminal output: %w", err) | 508 | + | return fmt.Errorf("client exited before application output: %w", err) |
| 465 | case <-timer.C: | 509 | case <-timer.C: | ||
| 466 | - | return errors.New("timed out waiting for terminal output") | 510 | + | return errors.New("timed out waiting for application terminal output") |
| 511 | + | } | |||
| 512 | + | } | |||
| 513 | + | return nil | |||
| 514 | + | } | |||
| 515 | + | ||||
| 516 | + | // ptyLocalEcho removes the one ordered input echo that a not-yet-raw slave PTY | |||
| 517 | + | // can produce. It normalizes NL to the conventional ONLCR echo spelling; any | |||
| 518 | + | // divergence is retained as application output. The line discipline emits the | |||
| 519 | + | // echo before the child can read and respond to the same write. | |||
| 520 | + | type ptyLocalEcho struct { | |||
| 521 | + | expected []byte | |||
| 522 | + | matched int | |||
| 523 | + | done bool | |||
| 524 | + | } | |||
| 525 | + | ||||
| 526 | + | func newPTYLocalEcho(input []byte) *ptyLocalEcho { | |||
| 527 | + | expected := make([]byte, 0, len(input)) | |||
| 528 | + | for _, b := range input { | |||
| 529 | + | if b == '\n' { | |||
| 530 | + | expected = append(expected, '\r') | |||
| 531 | + | } | |||
| 532 | + | expected = append(expected, b) | |||
| 533 | + | } | |||
| 534 | + | return &ptyLocalEcho{expected: expected} | |||
| 535 | + | } | |||
| 536 | + | ||||
| 537 | + | func (e *ptyLocalEcho) applicationOutput(chunk []byte) []byte { | |||
| 538 | + | if e.done || len(e.expected) == 0 { | |||
| 539 | + | return chunk | |||
| 540 | + | } | |||
| 541 | + | for i, b := range chunk { | |||
| 542 | + | if b != e.expected[e.matched] { | |||
| 543 | + | e.done = true | |||
| 544 | + | return append([]byte(nil), chunk[i:]...) | |||
| 545 | + | } | |||
| 546 | + | e.matched++ | |||
| 547 | + | if e.matched == len(e.expected) { | |||
| 548 | + | e.done = true | |||
| 549 | + | return append([]byte(nil), chunk[i+1:]...) | |||
| 467 | } | 550 | } | ||
| 468 | } | 551 | } | ||
| 469 | return nil | 552 | return nil | ||
| @@ -1073,9 +1156,9 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) { | |||||
| 1073 | _ = f.Close() | 1156 | _ = f.Close() | ||
| 1074 | return nil, err | 1157 | return nil, err | ||
| 1075 | } | 1158 | } | ||
| 1076 | - | if r.ProcessID != m.ProcessID { | 1159 | + | if r.ProcessID != m.ProcessID || r.Scenario != m.Scenario || r.Run != uint64(m.Run) { |
| 1077 | _ = f.Close() | 1160 | _ = f.Close() | ||
| 1078 | - | return nil, errors.New("trace record process_id does not match manifest") | 1161 | + | return nil, errors.New("trace record identity does not match manifest") |
| 1079 | } | 1162 | } | ||
| 1080 | if r.Scenario == "" || r.Run == 0 || r.Component == "" || r.Sequence == 0 || r.RequestID == 0 || r.Epoch == 0 { | 1163 | if r.Scenario == "" || r.Run == 0 || r.Component == "" || r.Sequence == 0 || r.RequestID == 0 || r.Epoch == 0 { | ||
| 1081 | _ = f.Close() | 1164 | _ = f.Close() | ||
M cmd/vev-perf-harness/main_test.go +73 -3
| @@ -157,6 +157,21 @@ func (p *fakePTY) Write(b []byte) (int, error) { | |||
| 157 | 157 | } | |
| 158 | 158 | func (*fakePTY) Close() error { return nil } | |
| 159 | 159 | ||
| 160 | + | type stagedPTY struct { | |
| 161 | + | allow <-chan struct{} | |
| 162 | + | writes chan []byte | |
| 163 | + | } | |
| 164 | + | ||
| 165 | + | func (*stagedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed } | |
| 166 | + | func (p *stagedPTY) Write(b []byte) (int, error) { | |
| 167 | + | if p.allow != nil { | |
| 168 | + | <-p.allow | |
| 169 | + | } | |
| 170 | + | p.writes <- append([]byte(nil), b...) | |
| 171 | + | return len(b), nil | |
| 172 | + | } | |
| 173 | + | func (*stagedPTY) Close() error { return nil } | |
| 174 | + | ||
| 160 | 175 | type fakeOutput struct { | |
| 161 | 176 | bytes.Buffer | |
| 162 | 177 | syncs int | |
| @@ -218,11 +233,59 @@ func TestHarnessDoesNotInheritNestedSession(t *testing.T) { | |||
| 218 | 233 | } | |
| 219 | 234 | } | |
| 220 | 235 | ||
| 221 | - | func TestCLIProcessOwnsPTYInjectionAndSuccessfulFlushBoundary(t *testing.T) { | |
| 236 | + | func TestCLIProcessWarmupWaitsForDelayedTerminalReadiness(t *testing.T) { | |
| 237 | + | allowWrite := make(chan struct{}) | |
| 238 | + | pty, output := &stagedPTY{allow: allowWrite, writes: make(chan []byte, 1)}, &fakeOutput{} | |
| 239 | + | p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte)} | |
| 240 | + | input := workloadInput(scenario{ID: "s"}, 1, "warmup") | |
| 241 | + | done := make(chan error, 1) | |
| 242 | + | go func() { done <- p.Warmup(input) }() | |
| 243 | + | ||
| 244 | + | // This is public client output from the initial application prompt/state, | |
| 245 | + | // emitted only after the client has entered raw mode. A premature PTY write | |
| 246 | + | // blocks on allowWrite and therefore cannot consume this delayed readiness. | |
| 247 | + | go func() { | |
| 248 | + | p.chunks <- []byte("shell prompt$ ") | |
| 249 | + | close(allowWrite) | |
| 250 | + | }() | |
| 251 | + | select { | |
| 252 | + | case got := <-pty.writes: | |
| 253 | + | if !bytes.Equal(got, input) { | |
| 254 | + | t.Fatalf("warmup write=%q want %q", got, input) | |
| 255 | + | } | |
| 256 | + | case <-time.After(time.Second): | |
| 257 | + | t.Fatal("warmup did not wait for and then proceed from client readiness") | |
| 258 | + | } | |
| 259 | + | go func() { p.chunks <- append([]byte("application "), inputMarker(input)...); close(p.chunks) }() | |
| 260 | + | if err := <-done; err != nil { | |
| 261 | + | t.Fatal(err) | |
| 262 | + | } | |
| 263 | + | if output.syncs != 1 { | |
| 264 | + | t.Fatalf("readiness output was not flushed: syncs=%d", output.syncs) | |
| 265 | + | } | |
| 266 | + | } | |
| 267 | + | ||
| 268 | + | func TestCLIProcessRejectsPTYLocalEchoAsFlushEvidence(t *testing.T) { | |
| 269 | + | pty, output := &fakePTY{}, &fakeOutput{} | |
| 270 | + | p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte, 1)} | |
| 271 | + | input := workloadInput(scenario{ID: "s"}, 1, "measured-1") | |
| 272 | + | p.chunks <- newPTYLocalEcho(input).expected | |
| 273 | + | close(p.chunks) | |
| 274 | + | flushed := false | |
| 275 | + | if err := p.Measure(input, func() error { return nil }, func() error { flushed = true; return nil }); err == nil { | |
| 276 | + | t.Fatal("local PTY echo satisfied terminal flush boundary") | |
| 277 | + | } | |
| 278 | + | if flushed || output.syncs != 0 { | |
| 279 | + | t.Fatalf("local echo stamped a flush: flushed=%t syncs=%d", flushed, output.syncs) | |
| 280 | + | } | |
| 281 | + | } | |
| 282 | + | ||
| 283 | + | func TestCLIProcessPairsApplicationOutputWithSuccessfulFlush(t *testing.T) { | |
| 222 | 284 | pty, output := &fakePTY{}, &fakeOutput{} | |
| 223 | 285 | p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte, 1)} | |
| 224 | 286 | input := workloadInput(scenario{ID: "s"}, 1, "measured-1") | |
| 225 | - | p.chunks <- append([]byte("terminal "), inputMarker(input)...) | |
| 287 | + | echo := newPTYLocalEcho(input).expected | |
| 288 | + | p.chunks <- append(append([]byte(nil), echo...), append([]byte("application "), inputMarker(input)...)...) | |
| 226 | 289 | order := []string{} | |
| 227 | 290 | if err := p.Measure(input, func() error { order = append(order, "injected"); return nil }, func() error { order = append(order, "flushed"); return nil }); err != nil { | |
| 228 | 291 | t.Fatal(err) | |
| @@ -630,7 +693,14 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | |||
| 630 | 693 | base := func(kind string, tick int64) traceRecord { | |
| 631 | 694 | return traceRecord{ProcessID: "one", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick} | |
| 632 | 695 | } | |
| 633 | - | for _, records := range [][]traceRecord{{base("diff_end", 1)}, {base("diff_start", 2), base("diff_end", 1)}, {base("diff_start", 1)}, {func() traceRecord { r := base("diff_start", 1); r.ProcessID = "other"; return r }()}} { | |
| 696 | + | for _, records := range [][]traceRecord{ | |
| 697 | + | {base("diff_end", 1)}, | |
| 698 | + | {base("diff_start", 2), base("diff_end", 1)}, | |
| 699 | + | {base("diff_start", 1)}, | |
| 700 | + | {func() traceRecord { r := base("diff_start", 1); r.ProcessID = "other"; return r }()}, | |
| 701 | + | {func() traceRecord { r := base("diff_start", 1); r.Scenario = "other-scenario"; return r }()}, | |
| 702 | + | {func() traceRecord { r := base("diff_start", 1); r.Run = 2; return r }()}, | |
| 703 | + | } { | |
| 634 | 704 | if _, err := mergeProcessTraces([]processMapping{write(t, records...)}); err == nil { | |
| 635 | 705 | t.Fatalf("invalid trace accepted: %+v", records) | |
| 636 | 706 | } | |
| @@ -157,6 +157,21 @@ func (p *fakePTY) Write(b []byte) (int, error) { | |||||
| 157 | } | 157 | } | ||
| 158 | func (*fakePTY) Close() error { return nil } | 158 | func (*fakePTY) Close() error { return nil } | ||
| 159 | 159 | ||||
| 160 | + | type stagedPTY struct { | |||
| 161 | + | allow <-chan struct{} | |||
| 162 | + | writes chan []byte | |||
| 163 | + | } | |||
| 164 | + | ||||
| 165 | + | func (*stagedPTY) Read([]byte) (int, error) { return 0, os.ErrClosed } | |||
| 166 | + | func (p *stagedPTY) Write(b []byte) (int, error) { | |||
| 167 | + | if p.allow != nil { | |||
| 168 | + | <-p.allow | |||
| 169 | + | } | |||
| 170 | + | p.writes <- append([]byte(nil), b...) | |||
| 171 | + | return len(b), nil | |||
| 172 | + | } | |||
| 173 | + | func (*stagedPTY) Close() error { return nil } | |||
| 174 | + | ||||
| 160 | type fakeOutput struct { | 175 | type fakeOutput struct { | ||
| 161 | bytes.Buffer | 176 | bytes.Buffer | ||
| 162 | syncs int | 177 | syncs int | ||
| @@ -218,11 +233,59 @@ func TestHarnessDoesNotInheritNestedSession(t *testing.T) { | |||||
| 218 | } | 233 | } | ||
| 219 | } | 234 | } | ||
| 220 | 235 | ||||
| 221 | - | func TestCLIProcessOwnsPTYInjectionAndSuccessfulFlushBoundary(t *testing.T) { | 236 | + | func TestCLIProcessWarmupWaitsForDelayedTerminalReadiness(t *testing.T) { |
| 237 | + | allowWrite := make(chan struct{}) | |||
| 238 | + | pty, output := &stagedPTY{allow: allowWrite, writes: make(chan []byte, 1)}, &fakeOutput{} | |||
| 239 | + | p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte)} | |||
| 240 | + | input := workloadInput(scenario{ID: "s"}, 1, "warmup") | |||
| 241 | + | done := make(chan error, 1) | |||
| 242 | + | go func() { done <- p.Warmup(input) }() | |||
| 243 | + | ||||
| 244 | + | // This is public client output from the initial application prompt/state, | |||
| 245 | + | // emitted only after the client has entered raw mode. A premature PTY write | |||
| 246 | + | // blocks on allowWrite and therefore cannot consume this delayed readiness. | |||
| 247 | + | go func() { | |||
| 248 | + | p.chunks <- []byte("shell prompt$ ") | |||
| 249 | + | close(allowWrite) | |||
| 250 | + | }() | |||
| 251 | + | select { | |||
| 252 | + | case got := <-pty.writes: | |||
| 253 | + | if !bytes.Equal(got, input) { | |||
| 254 | + | t.Fatalf("warmup write=%q want %q", got, input) | |||
| 255 | + | } | |||
| 256 | + | case <-time.After(time.Second): | |||
| 257 | + | t.Fatal("warmup did not wait for and then proceed from client readiness") | |||
| 258 | + | } | |||
| 259 | + | go func() { p.chunks <- append([]byte("application "), inputMarker(input)...); close(p.chunks) }() | |||
| 260 | + | if err := <-done; err != nil { | |||
| 261 | + | t.Fatal(err) | |||
| 262 | + | } | |||
| 263 | + | if output.syncs != 1 { | |||
| 264 | + | t.Fatalf("readiness output was not flushed: syncs=%d", output.syncs) | |||
| 265 | + | } | |||
| 266 | + | } | |||
| 267 | + | ||||
| 268 | + | func TestCLIProcessRejectsPTYLocalEchoAsFlushEvidence(t *testing.T) { | |||
| 269 | + | pty, output := &fakePTY{}, &fakeOutput{} | |||
| 270 | + | p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte, 1)} | |||
| 271 | + | input := workloadInput(scenario{ID: "s"}, 1, "measured-1") | |||
| 272 | + | p.chunks <- newPTYLocalEcho(input).expected | |||
| 273 | + | close(p.chunks) | |||
| 274 | + | flushed := false | |||
| 275 | + | if err := p.Measure(input, func() error { return nil }, func() error { flushed = true; return nil }); err == nil { | |||
| 276 | + | t.Fatal("local PTY echo satisfied terminal flush boundary") | |||
| 277 | + | } | |||
| 278 | + | if flushed || output.syncs != 0 { | |||
| 279 | + | t.Fatalf("local echo stamped a flush: flushed=%t syncs=%d", flushed, output.syncs) | |||
| 280 | + | } | |||
| 281 | + | } | |||
| 282 | + | ||||
| 283 | + | func TestCLIProcessPairsApplicationOutputWithSuccessfulFlush(t *testing.T) { | |||
| 222 | pty, output := &fakePTY{}, &fakeOutput{} | 284 | pty, output := &fakePTY{}, &fakeOutput{} | ||
| 223 | p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte, 1)} | 285 | p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte, 1)} | ||
| 224 | input := workloadInput(scenario{ID: "s"}, 1, "measured-1") | 286 | input := workloadInput(scenario{ID: "s"}, 1, "measured-1") | ||
| 225 | - | p.chunks <- append([]byte("terminal "), inputMarker(input)...) | 287 | + | echo := newPTYLocalEcho(input).expected |
| 288 | + | p.chunks <- append(append([]byte(nil), echo...), append([]byte("application "), inputMarker(input)...)...) | |||
| 226 | order := []string{} | 289 | order := []string{} | ||
| 227 | if err := p.Measure(input, func() error { order = append(order, "injected"); return nil }, func() error { order = append(order, "flushed"); return nil }); err != nil { | 290 | if err := p.Measure(input, func() error { order = append(order, "injected"); return nil }, func() error { order = append(order, "flushed"); return nil }); err != nil { | ||
| 228 | t.Fatal(err) | 291 | t.Fatal(err) | ||
| @@ -630,7 +693,14 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) { | |||||
| 630 | base := func(kind string, tick int64) traceRecord { | 693 | base := func(kind string, tick int64) traceRecord { | ||
| 631 | return traceRecord{ProcessID: "one", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick} | 694 | return traceRecord{ProcessID: "one", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick} | ||
| 632 | } | 695 | } | ||
| 633 | - | for _, records := range [][]traceRecord{{base("diff_end", 1)}, {base("diff_start", 2), base("diff_end", 1)}, {base("diff_start", 1)}, {func() traceRecord { r := base("diff_start", 1); r.ProcessID = "other"; return r }()}} { | 696 | + | for _, records := range [][]traceRecord{ |
| 697 | + | {base("diff_end", 1)}, | |||
| 698 | + | {base("diff_start", 2), base("diff_end", 1)}, | |||
| 699 | + | {base("diff_start", 1)}, | |||
| 700 | + | {func() traceRecord { r := base("diff_start", 1); r.ProcessID = "other"; return r }()}, | |||
| 701 | + | {func() traceRecord { r := base("diff_start", 1); r.Scenario = "other-scenario"; return r }()}, | |||
| 702 | + | {func() traceRecord { r := base("diff_start", 1); r.Run = 2; return r }()}, | |||
| 703 | + | } { | |||
| 634 | if _, err := mergeProcessTraces([]processMapping{write(t, records...)}); err == nil { | 704 | if _, err := mergeProcessTraces([]processMapping{write(t, records...)}); err == nil { | ||
| 635 | t.Fatalf("invalid trace accepted: %+v", records) | 705 | t.Fatalf("invalid trace accepted: %+v", records) | ||
| 636 | } | 706 | } | ||