fix(perf-harness): drain terminal output on shutdown
- commit
- 1d11e86430e010f0a07490b461a173c0f7795e6d
- author
- brice · · 2026-07-18 15:33 UTC
- parent
- 19fd124
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +67 -0
M cmd/vev-perf-harness/launcher.go +17 -0
| @@ -628,6 +628,23 @@ func (e *ptyLocalEcho) applicationOutput(chunk []byte) []byte { | |||
| 628 | 628 | func (p *cliProcess) Close() error { | |
| 629 | 629 | var result error | |
| 630 | 630 | p.closed.Do(func() { | |
| 631 | + | // No measurement waits for terminal chunks during teardown. Continue | |
| 632 | + | // consuming them so a full channel cannot block the client before it | |
| 633 | + | // receives the shell exit request and closes its transport receive span. | |
| 634 | + | if p.chunks != nil && p.done != nil { | |
| 635 | + | go func() { | |
| 636 | + | for { | |
| 637 | + | select { | |
| 638 | + | case _, ok := <-p.chunks: | |
| 639 | + | if !ok { | |
| 640 | + | return | |
| 641 | + | } | |
| 642 | + | case <-p.done: | |
| 643 | + | return | |
| 644 | + | } | |
| 645 | + | } | |
| 646 | + | }() | |
| 647 | + | } | |
| 631 | 648 | // A public terminal client owns a shell session. Ask that shell to exit | |
| 632 | 649 | // before closing the PTY so its transport can finish its receive span; | |
| 633 | 650 | // this avoids turning ordinary harness teardown into an aborted trace. | |
| @@ -628,6 +628,23 @@ func (e *ptyLocalEcho) applicationOutput(chunk []byte) []byte { | |||||
| 628 | func (p *cliProcess) Close() error { | 628 | func (p *cliProcess) Close() error { | ||
| 629 | var result error | 629 | var result error | ||
| 630 | p.closed.Do(func() { | 630 | p.closed.Do(func() { | ||
| 631 | + | // No measurement waits for terminal chunks during teardown. Continue | |||
| 632 | + | // consuming them so a full channel cannot block the client before it | |||
| 633 | + | // receives the shell exit request and closes its transport receive span. | |||
| 634 | + | if p.chunks != nil && p.done != nil { | |||
| 635 | + | go func() { | |||
| 636 | + | for { | |||
| 637 | + | select { | |||
| 638 | + | case _, ok := <-p.chunks: | |||
| 639 | + | if !ok { | |||
| 640 | + | return | |||
| 641 | + | } | |||
| 642 | + | case <-p.done: | |||
| 643 | + | return | |||
| 644 | + | } | |||
| 645 | + | } | |||
| 646 | + | }() | |||
| 647 | + | } | |||
| 631 | // A public terminal client owns a shell session. Ask that shell to exit | 648 | // A public terminal client owns a shell session. Ask that shell to exit | ||
| 632 | // before closing the PTY so its transport can finish its receive span; | 649 | // before closing the PTY so its transport can finish its receive span; | ||
| 633 | // this avoids turning ordinary harness teardown into an aborted trace. | 650 | // this avoids turning ordinary harness teardown into an aborted trace. | ||
M cmd/vev-perf-harness/launcher_test.go +50 -0
| @@ -18,6 +18,56 @@ import ( | |||
| 18 | 18 | "github.com/bnema/vev/pkg/safedir" | |
| 19 | 19 | ) | |
| 20 | 20 | ||
| 21 | + | type discardTerminalOutput struct{} | |
| 22 | + | ||
| 23 | + | func (discardTerminalOutput) Write(p []byte) (int, error) { return len(p), nil } | |
| 24 | + | func (discardTerminalOutput) Sync() error { return nil } | |
| 25 | + | func (discardTerminalOutput) Close() error { return nil } | |
| 26 | + | ||
| 27 | + | func TestCLIProcessCloseDrainsQueuedTerminalOutputBeforeClientExit(t *testing.T) { | |
| 28 | + | harnessPTY, clientPTY := net.Pipe() | |
| 29 | + | t.Cleanup(func() { _ = clientPTY.Close() }) | |
| 30 | + | waitErr := make(chan error) | |
| 31 | + | secondRead := make(chan struct{}) | |
| 32 | + | p := &cliProcess{ | |
| 33 | + | pty: harnessPTY, | |
| 34 | + | output: discardTerminalOutput{}, | |
| 35 | + | chunks: make(chan []byte, 1), | |
| 36 | + | done: make(chan struct{}), | |
| 37 | + | waitErr: waitErr, | |
| 38 | + | waitTimeout: func() <-chan time.Time { return time.After(time.Second) }, | |
| 39 | + | } | |
| 40 | + | go p.copyTerminal() | |
| 41 | + | go func() { | |
| 42 | + | _, _ = clientPTY.Write([]byte("first")) | |
| 43 | + | _, _ = clientPTY.Write([]byte("second")) | |
| 44 | + | close(secondRead) | |
| 45 | + | buf := make([]byte, len("exit\n")) | |
| 46 | + | if _, err := clientPTY.Read(buf); err != nil { | |
| 47 | + | return | |
| 48 | + | } | |
| 49 | + | if _, err := clientPTY.Write([]byte("third")); err != nil { | |
| 50 | + | return | |
| 51 | + | } | |
| 52 | + | waitErr <- nil | |
| 53 | + | }() | |
| 54 | + | select { | |
| 55 | + | case <-secondRead: | |
| 56 | + | case <-time.After(time.Second): | |
| 57 | + | t.Fatal("terminal reader did not queue output") | |
| 58 | + | } | |
| 59 | + | closed := make(chan error, 1) | |
| 60 | + | go func() { closed <- p.Close() }() | |
| 61 | + | select { | |
| 62 | + | case err := <-closed: | |
| 63 | + | if err != nil { | |
| 64 | + | t.Fatal(err) | |
| 65 | + | } | |
| 66 | + | case <-time.After(time.Second): | |
| 67 | + | t.Fatal("Close did not drain queued terminal output before waiting for client exit") | |
| 68 | + | } | |
| 69 | + | } | |
| 70 | + | ||
| 21 | 71 | func TestCLIProcessCloseKeepsTerminalDrainActiveUntilClientExit(t *testing.T) { | |
| 22 | 72 | pty, err := os.CreateTemp(t.TempDir(), "pty") | |
| 23 | 73 | if err != nil { | |
| @@ -18,6 +18,56 @@ import ( | |||||
| 18 | "github.com/bnema/vev/pkg/safedir" | 18 | "github.com/bnema/vev/pkg/safedir" | ||
| 19 | ) | 19 | ) | ||
| 20 | 20 | ||||
| 21 | + | type discardTerminalOutput struct{} | |||
| 22 | + | ||||
| 23 | + | func (discardTerminalOutput) Write(p []byte) (int, error) { return len(p), nil } | |||
| 24 | + | func (discardTerminalOutput) Sync() error { return nil } | |||
| 25 | + | func (discardTerminalOutput) Close() error { return nil } | |||
| 26 | + | ||||
| 27 | + | func TestCLIProcessCloseDrainsQueuedTerminalOutputBeforeClientExit(t *testing.T) { | |||
| 28 | + | harnessPTY, clientPTY := net.Pipe() | |||
| 29 | + | t.Cleanup(func() { _ = clientPTY.Close() }) | |||
| 30 | + | waitErr := make(chan error) | |||
| 31 | + | secondRead := make(chan struct{}) | |||
| 32 | + | p := &cliProcess{ | |||
| 33 | + | pty: harnessPTY, | |||
| 34 | + | output: discardTerminalOutput{}, | |||
| 35 | + | chunks: make(chan []byte, 1), | |||
| 36 | + | done: make(chan struct{}), | |||
| 37 | + | waitErr: waitErr, | |||
| 38 | + | waitTimeout: func() <-chan time.Time { return time.After(time.Second) }, | |||
| 39 | + | } | |||
| 40 | + | go p.copyTerminal() | |||
| 41 | + | go func() { | |||
| 42 | + | _, _ = clientPTY.Write([]byte("first")) | |||
| 43 | + | _, _ = clientPTY.Write([]byte("second")) | |||
| 44 | + | close(secondRead) | |||
| 45 | + | buf := make([]byte, len("exit\n")) | |||
| 46 | + | if _, err := clientPTY.Read(buf); err != nil { | |||
| 47 | + | return | |||
| 48 | + | } | |||
| 49 | + | if _, err := clientPTY.Write([]byte("third")); err != nil { | |||
| 50 | + | return | |||
| 51 | + | } | |||
| 52 | + | waitErr <- nil | |||
| 53 | + | }() | |||
| 54 | + | select { | |||
| 55 | + | case <-secondRead: | |||
| 56 | + | case <-time.After(time.Second): | |||
| 57 | + | t.Fatal("terminal reader did not queue output") | |||
| 58 | + | } | |||
| 59 | + | closed := make(chan error, 1) | |||
| 60 | + | go func() { closed <- p.Close() }() | |||
| 61 | + | select { | |||
| 62 | + | case err := <-closed: | |||
| 63 | + | if err != nil { | |||
| 64 | + | t.Fatal(err) | |||
| 65 | + | } | |||
| 66 | + | case <-time.After(time.Second): | |||
| 67 | + | t.Fatal("Close did not drain queued terminal output before waiting for client exit") | |||
| 68 | + | } | |||
| 69 | + | } | |||
| 70 | + | ||||
| 21 | func TestCLIProcessCloseKeepsTerminalDrainActiveUntilClientExit(t *testing.T) { | 71 | func TestCLIProcessCloseKeepsTerminalDrainActiveUntilClientExit(t *testing.T) { | ||
| 22 | pty, err := os.CreateTemp(t.TempDir(), "pty") | 72 | pty, err := os.CreateTemp(t.TempDir(), "pty") | ||
| 23 | if err != nil { | 73 | if err != nil { | ||