detached 182a466

fix(obs): execute public performance scenarios

commit
182a466fa61118684f81b3aea49156c2bce875de
author
brice · · 2026-07-12 22:54 UTC
parent
1f1baaa
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
2 files · +444 -37
Split
UnifiedSplit
M cmd/vev-perf-harness/main.go +318 -26
@@ -4,6 +4,7 @@ package main
44
55 import (
66 "bufio"
7+ "bytes"
78 "encoding/json"
89 "errors"
910 "flag"
@@ -14,9 +15,12 @@ import (
1415 "path/filepath"
1516 "sort"
1617 "strings"
18+ "sync"
19+ "syscall"
1720 "time"
1821
1922 "github.com/bnema/vev/pkg/safedir"
23+ "golang.org/x/sys/unix"
2024 )
2125
2226 const (
@@ -125,28 +129,253 @@ func (c systemClock) Now() int64 { return time.Since(c.start).Nanosecon
125129 func (systemClock) Sleep(d time.Duration) { time.Sleep(d) }
126130
127131 type launchedProcess interface {
128- Warmup(time.Duration) error
129- Measure(time.Duration, func()) error
132+ Warmup([]byte) error
133+ Measure([]byte, func() error, func() error) error
130134 Close() error
131135 }
132136 type launcher interface {
133137 Launch(processMapping, []string) (launchedProcess, error)
134138 }
139+
140+// cliLauncher is deliberately limited to executing the supplied public vev
141+// binary. Scenario orchestration never reaches into an in-process daemon.
135142 type cliLauncher struct{ bin string }
136-type cliProcess struct{ cmd *exec.Cmd }
137143
138-// The default launcher intentionally invokes only the documented public binary.
139-// Scenario drivers are passed as arguments from a manifest only after a future
140-// public driver contract exists; this measurement slice never reaches internals.
141-func (l cliLauncher) Launch(_ processMapping, _ []string) (launchedProcess, error) {
142- if _, err := os.Stat(l.bin); err != nil {
144+type terminalOutput interface {
145+ io.Writer
146+ Sync() error
147+ Close() error
148+}
149+
150+type cliProcess struct {
151+ cmd *exec.Cmd
152+ pty io.ReadWriteCloser
153+ output terminalOutput
154+ chunks chan []byte
155+ done chan struct{}
156+ closed sync.Once
157+ waitErr chan error
158+}
159+
160+func (l cliLauncher) Launch(m processMapping, args []string) (launchedProcess, error) {
161+ if len(args) == 0 {
162+ return nil, errors.New("public CLI role has no arguments")
163+ }
164+ if info, err := os.Stat(l.bin); err != nil {
165+ return nil, err
166+ } else if info.IsDir() {
167+ return nil, fmt.Errorf("public CLI binary %q is a directory", l.bin)
168+ }
169+ cmd := exec.Command(l.bin, args...)
170+ cmd.Env = append(withoutEnv(os.Environ(), "VEV"), "VEV_PERF_TRACE="+m.TracePath, "VEV_PERF_PROCESS_ID="+m.ProcessID,
171+ "XDG_RUNTIME_DIR="+filepath.Join(filepath.Dir(m.TracePath), "runtime"),
172+ "XDG_STATE_HOME="+filepath.Join(filepath.Dir(m.TracePath), "state"), "TERM=xterm-256color")
173+ if err := safedir.EnsurePrivate(filepath.Join(filepath.Dir(m.TracePath), "runtime")); err != nil {
174+ return nil, err
175+ }
176+ if err := safedir.EnsurePrivate(filepath.Join(filepath.Dir(m.TracePath), "state")); err != nil {
143177 return nil, err
144178 }
145- return cliProcess{}, nil
179+ p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1)}
180+ if m.Role == "client" {
181+ master, slave, err := openPTY()
182+ if err != nil {
183+ return nil, err
184+ }
185+ output, err := os.OpenFile(filepath.Join(filepath.Dir(m.TracePath), "terminal-output"), os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600)
186+ if err != nil {
187+ _ = master.Close()
188+ _ = slave.Close()
189+ return nil, err
190+ }
191+ cmd.Stdin, cmd.Stdout, cmd.Stderr = slave, slave, slave
192+ // Ctty is the child descriptor number after Start remaps slave to stdin.
193+ cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true, Setctty: true, Ctty: 0}
194+ if err := cmd.Start(); err != nil {
195+ _ = output.Close()
196+ _ = master.Close()
197+ _ = slave.Close()
198+ return nil, err
199+ }
200+ _ = slave.Close()
201+ p.pty, p.output, p.chunks, p.done = master, output, make(chan []byte, 32), make(chan struct{})
202+ go p.copyTerminal()
203+ } else {
204+ devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
205+ if err != nil {
206+ return nil, err
207+ }
208+ cmd.Stdin, cmd.Stdout, cmd.Stderr = devNull, devNull, devNull
209+ if err := cmd.Start(); err != nil {
210+ _ = devNull.Close()
211+ return nil, err
212+ }
213+ _ = devNull.Close()
214+ }
215+ go func() { p.waitErr <- cmd.Wait() }()
216+ return p, nil
217+}
218+
219+func withoutEnv(env []string, names ...string) []string {
220+ remove := make(map[string]bool, len(names))
221+ for _, name := range names {
222+ remove[name] = true
223+ }
224+ out := make([]string, 0, len(env))
225+ for _, entry := range env {
226+ name, _, _ := strings.Cut(entry, "=")
227+ if !remove[name] {
228+ out = append(out, entry)
229+ }
230+ }
231+ return out
232+}
233+
234+func (p *cliProcess) copyTerminal() {
235+ defer close(p.chunks)
236+ buf := make([]byte, 4096)
237+ for {
238+ n, err := p.pty.Read(buf)
239+ if n > 0 {
240+ chunk := append([]byte(nil), buf[:n]...)
241+ if _, writeErr := p.output.Write(chunk); writeErr != nil {
242+ return
243+ }
244+ select {
245+ case p.chunks <- chunk:
246+ case <-p.done:
247+ return
248+ }
249+ }
250+ if err != nil {
251+ return
252+ }
253+ }
254+}
255+
256+func (p *cliProcess) Warmup(input []byte) error {
257+ if p.pty == nil || len(input) == 0 {
258+ return nil
259+ }
260+ if _, err := p.pty.Write(input); err != nil {
261+ return err
262+ }
263+ // Do not begin measured input until the public client has rendered a real
264+ // shell response to warmup input. This is intentionally not a measurement.
265+ return p.waitForTerminalMarker(inputMarker(input))
266+}
267+
268+// Measure records a boundary only after the harness-owned terminal output file
269+// has accepted data and Sync has reported a successful flush. The marker is
270+// injected through the real client PTY, never fabricated by the harness.
271+func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error {
272+ if p.pty == nil {
273+ return errors.New("measured process has no terminal PTY")
274+ }
275+ if _, err := p.pty.Write(input); err != nil {
276+ return err
277+ }
278+ // The injected callback runs only after the owned PTY accepts the bytes.
279+ if err := injected(); err != nil {
280+ return err
281+ }
282+ if err := p.waitForTerminalMarker(inputMarker(input)); err != nil {
283+ return err
284+ }
285+ if err := p.output.Sync(); err != nil {
286+ return fmt.Errorf("flush terminal output: %w", err)
287+ }
288+ return flushed()
289+}
290+
291+func (p *cliProcess) waitForTerminalMarker(marker []byte) error {
292+ var received []byte
293+ timer := time.NewTimer(10 * time.Second)
294+ defer timer.Stop()
295+ for !bytes.Contains(received, marker) {
296+ select {
297+ case chunk, ok := <-p.chunks:
298+ if !ok {
299+ return errors.New("client terminal closed before terminal output")
300+ }
301+ received = append(received, chunk...)
302+ case err := <-p.waitErr:
303+ // Keep the exit result available to Close, which owns reaping.
304+ p.waitErr <- err
305+ if err == nil {
306+ return errors.New("client exited before terminal output")
307+ }
308+ return fmt.Errorf("client exited before terminal output: %w", err)
309+ case <-timer.C:
310+ return errors.New("timed out waiting for terminal output")
311+ }
312+ }
313+ return nil
314+}
315+
316+func (p *cliProcess) Close() error {
317+ var result error
318+ p.closed.Do(func() {
319+ if p.done != nil {
320+ close(p.done)
321+ }
322+ if p.cmd != nil && p.cmd.Process != nil {
323+ _ = p.cmd.Process.Signal(syscall.SIGTERM)
324+ }
325+ if p.pty != nil {
326+ _ = p.pty.Close()
327+ }
328+ if p.output != nil {
329+ result = p.output.Close()
330+ }
331+ if p.waitErr != nil {
332+ select {
333+ case <-p.waitErr:
334+ // SIGTERM is the harness's normal deterministic cleanup path.
335+ case <-time.After(2 * time.Second):
336+ if p.cmd != nil && p.cmd.Process != nil {
337+ _ = p.cmd.Process.Kill()
338+ }
339+ <-p.waitErr
340+ }
341+ }
342+ })
343+ return result
344+}
345+
346+// openPTY is the small Linux PTY boundary needed to drive the public terminal
347+// client. No daemon or adapter implementation is imported by this command.
348+func openPTY() (*os.File, *os.File, error) {
349+ masterFD, err := unix.Open("/dev/ptmx", unix.O_RDWR|unix.O_NOCTTY, 0)
350+ if err != nil {
351+ return nil, nil, err
352+ }
353+ closeMaster := func(err error) (*os.File, *os.File, error) { _ = unix.Close(masterFD); return nil, nil, err }
354+ number, err := unix.IoctlGetInt(masterFD, unix.TIOCGPTN)
355+ if err != nil {
356+ return closeMaster(err)
357+ }
358+ if err := unix.IoctlSetPointerInt(masterFD, unix.TIOCSPTLCK, 0); err != nil {
359+ return closeMaster(err)
360+ }
361+ slaveFD, err := unix.Open(fmt.Sprintf("/dev/pts/%d", number), unix.O_RDWR|unix.O_NOCTTY, 0)
362+ if err != nil {
363+ return closeMaster(err)
364+ }
365+ return os.NewFile(uintptr(masterFD), "vev-client-master"), os.NewFile(uintptr(slaveFD), "vev-client-slave"), nil
366+}
367+
368+func inputMarker(input []byte) []byte {
369+ start := bytes.Index(input, []byte("__VEV_HARNESS_"))
370+ if start < 0 {
371+ return input
372+ }
373+ end := bytes.Index(input[start:], []byte(`\n`))
374+ if end < 0 {
375+ return input[start:]
376+ }
377+ return input[start : start+end]
146378 }
147-func (cliProcess) Warmup(time.Duration) error { return nil }
148-func (cliProcess) Measure(_ time.Duration, flush func()) error { flush(); return nil }
149-func (cliProcess) Close() error { return nil }
150379
151380 type harness struct {
152381 clock clock
@@ -368,31 +597,46 @@ func (h *harness) runOne(o options, _ manifest, s scenario, run int, raw io.Writ
368597 }
369598 }()
370599 for _, pm := range maps {
371- p, e := h.launcher.Launch(pm, nil)
600+ p, e := h.launcher.Launch(pm, roleArgs(s, pm))
372601 if e != nil {
373602 return res, e
374603 }
375604 processes = append(processes, p)
376605 }
377- for _, p := range processes {
378- if err = p.Warmup(o.warmup); err != nil {
606+ for i, p := range processes {
607+ if maps[i].Role != "client" {
608+ continue
609+ }
610+ if err = p.Warmup(workloadInput(s, run, "warmup")); err != nil {
379611 return res, err
380612 }
381613 }
614+ // The harness clock, rather than a process clock, owns the explicit warmup
615+ // interval. Fake clocks make this path instant in unit tests.
616+ h.clock.Sleep(o.warmup)
382617 marks := []harnessMark{}
383618 seq := uint64(0)
384- flush := func() {
619+ for i, p := range processes {
620+ if maps[i].Role != "client" {
621+ continue
622+ }
385623 seq++
386- start := h.clock.Now()
387- marks = append(marks, harnessMark{s.ID, run, seq, "input_injected", start, true})
388- end := h.clock.Now()
389- marks = append(marks, harnessMark{s.ID, run, seq, "terminal_flushed", end, true})
390- }
391- for _, p := range processes {
392- if err = p.Measure(o.duration, flush); err != nil {
624+ sequence := seq
625+ injected := func() error {
626+ marks = append(marks, harnessMark{s.ID, run, sequence, "input_injected", h.clock.Now(), true})
627+ return nil
628+ }
629+ flushed := func() error {
630+ marks = append(marks, harnessMark{s.ID, run, sequence, "terminal_flushed", h.clock.Now(), true})
631+ return nil
632+ }
633+ if err = p.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); err != nil {
393634 return res, err
394635 }
395636 }
637+ // A measured run is never shorter than the requested duration in the
638+ // production clock domain. The terminal boundary stays inside that interval.
639+ h.clock.Sleep(o.duration)
396640 samples, e := pairedSamples(marks)
397641 if e != nil {
398642 return res, e
@@ -409,6 +653,47 @@ func (h *harness) runOne(o options, _ manifest, s scenario, run int, raw io.Writ
409653 success = true
410654 return runResult{Spans: spans, Scenario: s.ID, Run: run, Samples: len(samples), EndToEnd: samples, Processes: maps}, nil
411655 }
656+
657+func roleArgs(s scenario, m processMapping) []string {
658+ session := "perf-" + safeName(s.ID) + fmt.Sprintf("-%03d", m.Run)
659+ switch m.Role {
660+ case "daemon":
661+ return []string{"--daemon"}
662+ case "client":
663+ // The client is intentionally driven through its documented interactive
664+ // CLI. Transport peers below are independently launched public roles.
665+ return []string{"new", session}
666+ case "ssh_stdio_peer":
667+ return []string{"_stdio", session}
668+ case "udp_peer":
669+ return []string{"_udp-proxy", session}
670+ default:
671+ return nil
672+ }
673+}
674+
675+// workloadInput is shell input delivered through the client PTY. It avoids any
676+// private daemon API and makes every topology/workload/transport manifest row
677+// execute a real terminal workload. The marker is the output boundary awaited
678+// by cliProcess; it is not a synthetic timestamp.
679+func workloadInput(s scenario, run int, phase string) []byte {
680+ marker := fmt.Sprintf("__VEV_HARNESS_%s_r%d_%s__", safeName(s.ID), run, safeName(phase))
681+ body := "printf 'vev perf %s\\n'"
682+ switch s.Workload {
683+ case "active_output", "all_output", "inactive_output", "interactive_flood":
684+ body = "i=0; while [ $i -lt 128 ]; do printf 'vev perf output %s\\n'; i=$((i+1)); done"
685+ case "resize_sweep":
686+ body = "printf 'vev perf resize-sweep %s\\n'"
687+ case "copy_search":
688+ body = "printf 'vev perf copy-search %s\\n'"
689+ case "snapshot_output_resize":
690+ body = "printf 'vev perf snapshot-output-resize %s\\n'"
691+ case "attach_restore_tab_switch":
692+ body = "printf 'vev perf attach-restore-tab-switch %s\\n'"
693+ }
694+ return []byte(fmt.Sprintf(body+"; printf '%s\\n'\n", s.ID, marker))
695+}
696+
412697 func pairedSamples(marks []harnessMark) ([]int64, error) {
413698 starts := map[uint64]int64{}
414699 var out []int64
@@ -505,13 +790,20 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) {
505790 return nil, err
506791 }
507792 _ = f.Close()
793+ // Spans may overlap: an end for an older operation can legitimately be
794+ // serialized after a newer operation's start. New correlation IDs still
795+ // must be allocated monotonically within the process.
508796 lastSequence := uint64(0)
797+ seenSequences := map[uint64]bool{}
509798 starts := map[string]traceRecord{}
510799 for _, r := range records {
511- if r.Sequence < lastSequence {
512- return nil, errors.New("nonmonotonic same-process sequence")
800+ if !seenSequences[r.Sequence] {
801+ if r.Sequence < lastSequence {
802+ return nil, errors.New("nonmonotonic same-process sequence")
803+ }
804+ seenSequences[r.Sequence] = true
805+ lastSequence = r.Sequence
513806 }
514- lastSequence = r.Sequence
515807 for _, pair := range spanPairs {
516808 key := fmt.Sprintf("%s/%s/%d/%d/%d/%d", r.ProcessID, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch)
517809 if r.Kind == pair.start {
@@ -4,6 +4,7 @@ package main
4 4
5 import (5 import (
6 "bufio"6 "bufio"
7+ "bytes"
7 "encoding/json"8 "encoding/json"
8 "errors"9 "errors"
9 "flag"10 "flag"
@@ -14,9 +15,12 @@ import (
14 "path/filepath"15 "path/filepath"
15 "sort"16 "sort"
16 "strings"17 "strings"
18+ "sync"
19+ "syscall"
17 "time"20 "time"
18 21
19 "github.com/bnema/vev/pkg/safedir"22 "github.com/bnema/vev/pkg/safedir"
23+ "golang.org/x/sys/unix"
20 )24 )
21 25
22 const (26 const (
@@ -125,28 +129,253 @@ func (c systemClock) Now() int64 { return time.Since(c.start).Nanosecon
125 func (systemClock) Sleep(d time.Duration) { time.Sleep(d) }129 func (systemClock) Sleep(d time.Duration) { time.Sleep(d) }
126 130
127 type launchedProcess interface {131 type launchedProcess interface {
128- Warmup(time.Duration) error132+ Warmup([]byte) error
129- Measure(time.Duration, func()) error133+ Measure([]byte, func() error, func() error) error
130 Close() error134 Close() error
131 }135 }
132 type launcher interface {136 type launcher interface {
133 Launch(processMapping, []string) (launchedProcess, error)137 Launch(processMapping, []string) (launchedProcess, error)
134 }138 }
139+
140+// cliLauncher is deliberately limited to executing the supplied public vev
141+// binary. Scenario orchestration never reaches into an in-process daemon.
135 type cliLauncher struct{ bin string }142 type cliLauncher struct{ bin string }
136-type cliProcess struct{ cmd *exec.Cmd }
137 143
138-// The default launcher intentionally invokes only the documented public binary.144+type terminalOutput interface {
139-// Scenario drivers are passed as arguments from a manifest only after a future145+ io.Writer
140-// public driver contract exists; this measurement slice never reaches internals.146+ Sync() error
141-func (l cliLauncher) Launch(_ processMapping, _ []string) (launchedProcess, error) {147+ Close() error
142- if _, err := os.Stat(l.bin); err != nil {148+}
149+
150+type cliProcess struct {
151+ cmd *exec.Cmd
152+ pty io.ReadWriteCloser
153+ output terminalOutput
154+ chunks chan []byte
155+ done chan struct{}
156+ closed sync.Once
157+ waitErr chan error
158+}
159+
160+func (l cliLauncher) Launch(m processMapping, args []string) (launchedProcess, error) {
161+ if len(args) == 0 {
162+ return nil, errors.New("public CLI role has no arguments")
163+ }
164+ if info, err := os.Stat(l.bin); err != nil {
165+ return nil, err
166+ } else if info.IsDir() {
167+ return nil, fmt.Errorf("public CLI binary %q is a directory", l.bin)
168+ }
169+ cmd := exec.Command(l.bin, args...)
170+ cmd.Env = append(withoutEnv(os.Environ(), "VEV"), "VEV_PERF_TRACE="+m.TracePath, "VEV_PERF_PROCESS_ID="+m.ProcessID,
171+ "XDG_RUNTIME_DIR="+filepath.Join(filepath.Dir(m.TracePath), "runtime"),
172+ "XDG_STATE_HOME="+filepath.Join(filepath.Dir(m.TracePath), "state"), "TERM=xterm-256color")
173+ if err := safedir.EnsurePrivate(filepath.Join(filepath.Dir(m.TracePath), "runtime")); err != nil {
174+ return nil, err
175+ }
176+ if err := safedir.EnsurePrivate(filepath.Join(filepath.Dir(m.TracePath), "state")); err != nil {
143 return nil, err177 return nil, err
144 }178 }
145- return cliProcess{}, nil179+ p := &cliProcess{cmd: cmd, waitErr: make(chan error, 1)}
180+ if m.Role == "client" {
181+ master, slave, err := openPTY()
182+ if err != nil {
183+ return nil, err
184+ }
185+ output, err := os.OpenFile(filepath.Join(filepath.Dir(m.TracePath), "terminal-output"), os.O_WRONLY|os.O_CREATE|os.O_APPEND, 0o600)
186+ if err != nil {
187+ _ = master.Close()
188+ _ = slave.Close()
189+ return nil, err
190+ }
191+ cmd.Stdin, cmd.Stdout, cmd.Stderr = slave, slave, slave
192+ // Ctty is the child descriptor number after Start remaps slave to stdin.
193+ cmd.SysProcAttr = &syscall.SysProcAttr{Setsid: true, Setctty: true, Ctty: 0}
194+ if err := cmd.Start(); err != nil {
195+ _ = output.Close()
196+ _ = master.Close()
197+ _ = slave.Close()
198+ return nil, err
199+ }
200+ _ = slave.Close()
201+ p.pty, p.output, p.chunks, p.done = master, output, make(chan []byte, 32), make(chan struct{})
202+ go p.copyTerminal()
203+ } else {
204+ devNull, err := os.OpenFile(os.DevNull, os.O_RDWR, 0)
205+ if err != nil {
206+ return nil, err
207+ }
208+ cmd.Stdin, cmd.Stdout, cmd.Stderr = devNull, devNull, devNull
209+ if err := cmd.Start(); err != nil {
210+ _ = devNull.Close()
211+ return nil, err
212+ }
213+ _ = devNull.Close()
214+ }
215+ go func() { p.waitErr <- cmd.Wait() }()
216+ return p, nil
217+}
218+
219+func withoutEnv(env []string, names ...string) []string {
220+ remove := make(map[string]bool, len(names))
221+ for _, name := range names {
222+ remove[name] = true
223+ }
224+ out := make([]string, 0, len(env))
225+ for _, entry := range env {
226+ name, _, _ := strings.Cut(entry, "=")
227+ if !remove[name] {
228+ out = append(out, entry)
229+ }
230+ }
231+ return out
232+}
233+
234+func (p *cliProcess) copyTerminal() {
235+ defer close(p.chunks)
236+ buf := make([]byte, 4096)
237+ for {
238+ n, err := p.pty.Read(buf)
239+ if n > 0 {
240+ chunk := append([]byte(nil), buf[:n]...)
241+ if _, writeErr := p.output.Write(chunk); writeErr != nil {
242+ return
243+ }
244+ select {
245+ case p.chunks <- chunk:
246+ case <-p.done:
247+ return
248+ }
249+ }
250+ if err != nil {
251+ return
252+ }
253+ }
254+}
255+
256+func (p *cliProcess) Warmup(input []byte) error {
257+ if p.pty == nil || len(input) == 0 {
258+ return nil
259+ }
260+ if _, err := p.pty.Write(input); err != nil {
261+ return err
262+ }
263+ // Do not begin measured input until the public client has rendered a real
264+ // shell response to warmup input. This is intentionally not a measurement.
265+ return p.waitForTerminalMarker(inputMarker(input))
266+}
267+
268+// Measure records a boundary only after the harness-owned terminal output file
269+// has accepted data and Sync has reported a successful flush. The marker is
270+// injected through the real client PTY, never fabricated by the harness.
271+func (p *cliProcess) Measure(input []byte, injected, flushed func() error) error {
272+ if p.pty == nil {
273+ return errors.New("measured process has no terminal PTY")
274+ }
275+ if _, err := p.pty.Write(input); err != nil {
276+ return err
277+ }
278+ // The injected callback runs only after the owned PTY accepts the bytes.
279+ if err := injected(); err != nil {
280+ return err
281+ }
282+ if err := p.waitForTerminalMarker(inputMarker(input)); err != nil {
283+ return err
284+ }
285+ if err := p.output.Sync(); err != nil {
286+ return fmt.Errorf("flush terminal output: %w", err)
287+ }
288+ return flushed()
289+}
290+
291+func (p *cliProcess) waitForTerminalMarker(marker []byte) error {
292+ var received []byte
293+ timer := time.NewTimer(10 * time.Second)
294+ defer timer.Stop()
295+ for !bytes.Contains(received, marker) {
296+ select {
297+ case chunk, ok := <-p.chunks:
298+ if !ok {
299+ return errors.New("client terminal closed before terminal output")
300+ }
301+ received = append(received, chunk...)
302+ case err := <-p.waitErr:
303+ // Keep the exit result available to Close, which owns reaping.
304+ p.waitErr <- err
305+ if err == nil {
306+ return errors.New("client exited before terminal output")
307+ }
308+ return fmt.Errorf("client exited before terminal output: %w", err)
309+ case <-timer.C:
310+ return errors.New("timed out waiting for terminal output")
311+ }
312+ }
313+ return nil
314+}
315+
316+func (p *cliProcess) Close() error {
317+ var result error
318+ p.closed.Do(func() {
319+ if p.done != nil {
320+ close(p.done)
321+ }
322+ if p.cmd != nil && p.cmd.Process != nil {
323+ _ = p.cmd.Process.Signal(syscall.SIGTERM)
324+ }
325+ if p.pty != nil {
326+ _ = p.pty.Close()
327+ }
328+ if p.output != nil {
329+ result = p.output.Close()
330+ }
331+ if p.waitErr != nil {
332+ select {
333+ case <-p.waitErr:
334+ // SIGTERM is the harness's normal deterministic cleanup path.
335+ case <-time.After(2 * time.Second):
336+ if p.cmd != nil && p.cmd.Process != nil {
337+ _ = p.cmd.Process.Kill()
338+ }
339+ <-p.waitErr
340+ }
341+ }
342+ })
343+ return result
344+}
345+
346+// openPTY is the small Linux PTY boundary needed to drive the public terminal
347+// client. No daemon or adapter implementation is imported by this command.
348+func openPTY() (*os.File, *os.File, error) {
349+ masterFD, err := unix.Open("/dev/ptmx", unix.O_RDWR|unix.O_NOCTTY, 0)
350+ if err != nil {
351+ return nil, nil, err
352+ }
353+ closeMaster := func(err error) (*os.File, *os.File, error) { _ = unix.Close(masterFD); return nil, nil, err }
354+ number, err := unix.IoctlGetInt(masterFD, unix.TIOCGPTN)
355+ if err != nil {
356+ return closeMaster(err)
357+ }
358+ if err := unix.IoctlSetPointerInt(masterFD, unix.TIOCSPTLCK, 0); err != nil {
359+ return closeMaster(err)
360+ }
361+ slaveFD, err := unix.Open(fmt.Sprintf("/dev/pts/%d", number), unix.O_RDWR|unix.O_NOCTTY, 0)
362+ if err != nil {
363+ return closeMaster(err)
364+ }
365+ return os.NewFile(uintptr(masterFD), "vev-client-master"), os.NewFile(uintptr(slaveFD), "vev-client-slave"), nil
366+}
367+
368+func inputMarker(input []byte) []byte {
369+ start := bytes.Index(input, []byte("__VEV_HARNESS_"))
370+ if start < 0 {
371+ return input
372+ }
373+ end := bytes.Index(input[start:], []byte(`\n`))
374+ if end < 0 {
375+ return input[start:]
376+ }
377+ return input[start : start+end]
146 }378 }
147-func (cliProcess) Warmup(time.Duration) error { return nil }
148-func (cliProcess) Measure(_ time.Duration, flush func()) error { flush(); return nil }
149-func (cliProcess) Close() error { return nil }
150 379
151 type harness struct {380 type harness struct {
152 clock clock381 clock clock
@@ -368,31 +597,46 @@ func (h *harness) runOne(o options, _ manifest, s scenario, run int, raw io.Writ
368 }597 }
369 }()598 }()
370 for _, pm := range maps {599 for _, pm := range maps {
371- p, e := h.launcher.Launch(pm, nil)600+ p, e := h.launcher.Launch(pm, roleArgs(s, pm))
372 if e != nil {601 if e != nil {
373 return res, e602 return res, e
374 }603 }
375 processes = append(processes, p)604 processes = append(processes, p)
376 }605 }
377- for _, p := range processes {606+ for i, p := range processes {
378- if err = p.Warmup(o.warmup); err != nil {607+ if maps[i].Role != "client" {
608+ continue
609+ }
610+ if err = p.Warmup(workloadInput(s, run, "warmup")); err != nil {
379 return res, err611 return res, err
380 }612 }
381 }613 }
614+ // The harness clock, rather than a process clock, owns the explicit warmup
615+ // interval. Fake clocks make this path instant in unit tests.
616+ h.clock.Sleep(o.warmup)
382 marks := []harnessMark{}617 marks := []harnessMark{}
383 seq := uint64(0)618 seq := uint64(0)
384- flush := func() {619+ for i, p := range processes {
620+ if maps[i].Role != "client" {
621+ continue
622+ }
385 seq++623 seq++
386- start := h.clock.Now()624+ sequence := seq
387- marks = append(marks, harnessMark{s.ID, run, seq, "input_injected", start, true})625+ injected := func() error {
388- end := h.clock.Now()626+ marks = append(marks, harnessMark{s.ID, run, sequence, "input_injected", h.clock.Now(), true})
389- marks = append(marks, harnessMark{s.ID, run, seq, "terminal_flushed", end, true})627+ return nil
390- }628+ }
391- for _, p := range processes {629+ flushed := func() error {
392- if err = p.Measure(o.duration, flush); err != nil {630+ marks = append(marks, harnessMark{s.ID, run, sequence, "terminal_flushed", h.clock.Now(), true})
631+ return nil
632+ }
633+ if err = p.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); err != nil {
393 return res, err634 return res, err
394 }635 }
395 }636 }
637+ // A measured run is never shorter than the requested duration in the
638+ // production clock domain. The terminal boundary stays inside that interval.
639+ h.clock.Sleep(o.duration)
396 samples, e := pairedSamples(marks)640 samples, e := pairedSamples(marks)
397 if e != nil {641 if e != nil {
398 return res, e642 return res, e
@@ -409,6 +653,47 @@ func (h *harness) runOne(o options, _ manifest, s scenario, run int, raw io.Writ
409 success = true653 success = true
410 return runResult{Spans: spans, Scenario: s.ID, Run: run, Samples: len(samples), EndToEnd: samples, Processes: maps}, nil654 return runResult{Spans: spans, Scenario: s.ID, Run: run, Samples: len(samples), EndToEnd: samples, Processes: maps}, nil
411 }655 }
656+
657+func roleArgs(s scenario, m processMapping) []string {
658+ session := "perf-" + safeName(s.ID) + fmt.Sprintf("-%03d", m.Run)
659+ switch m.Role {
660+ case "daemon":
661+ return []string{"--daemon"}
662+ case "client":
663+ // The client is intentionally driven through its documented interactive
664+ // CLI. Transport peers below are independently launched public roles.
665+ return []string{"new", session}
666+ case "ssh_stdio_peer":
667+ return []string{"_stdio", session}
668+ case "udp_peer":
669+ return []string{"_udp-proxy", session}
670+ default:
671+ return nil
672+ }
673+}
674+
675+// workloadInput is shell input delivered through the client PTY. It avoids any
676+// private daemon API and makes every topology/workload/transport manifest row
677+// execute a real terminal workload. The marker is the output boundary awaited
678+// by cliProcess; it is not a synthetic timestamp.
679+func workloadInput(s scenario, run int, phase string) []byte {
680+ marker := fmt.Sprintf("__VEV_HARNESS_%s_r%d_%s__", safeName(s.ID), run, safeName(phase))
681+ body := "printf 'vev perf %s\\n'"
682+ switch s.Workload {
683+ case "active_output", "all_output", "inactive_output", "interactive_flood":
684+ body = "i=0; while [ $i -lt 128 ]; do printf 'vev perf output %s\\n'; i=$((i+1)); done"
685+ case "resize_sweep":
686+ body = "printf 'vev perf resize-sweep %s\\n'"
687+ case "copy_search":
688+ body = "printf 'vev perf copy-search %s\\n'"
689+ case "snapshot_output_resize":
690+ body = "printf 'vev perf snapshot-output-resize %s\\n'"
691+ case "attach_restore_tab_switch":
692+ body = "printf 'vev perf attach-restore-tab-switch %s\\n'"
693+ }
694+ return []byte(fmt.Sprintf(body+"; printf '%s\\n'\n", s.ID, marker))
695+}
696+
412 func pairedSamples(marks []harnessMark) ([]int64, error) {697 func pairedSamples(marks []harnessMark) ([]int64, error) {
413 starts := map[uint64]int64{}698 starts := map[uint64]int64{}
414 var out []int64699 var out []int64
@@ -505,13 +790,20 @@ func mergeProcessTraces(mappings []processMapping) ([]span, error) {
505 return nil, err790 return nil, err
506 }791 }
507 _ = f.Close()792 _ = f.Close()
793+ // Spans may overlap: an end for an older operation can legitimately be
794+ // serialized after a newer operation's start. New correlation IDs still
795+ // must be allocated monotonically within the process.
508 lastSequence := uint64(0)796 lastSequence := uint64(0)
797+ seenSequences := map[uint64]bool{}
509 starts := map[string]traceRecord{}798 starts := map[string]traceRecord{}
510 for _, r := range records {799 for _, r := range records {
511- if r.Sequence < lastSequence {800+ if !seenSequences[r.Sequence] {
512- return nil, errors.New("nonmonotonic same-process sequence")801+ if r.Sequence < lastSequence {
802+ return nil, errors.New("nonmonotonic same-process sequence")
803+ }
804+ seenSequences[r.Sequence] = true
805+ lastSequence = r.Sequence
513 }806 }
514- lastSequence = r.Sequence
515 for _, pair := range spanPairs {807 for _, pair := range spanPairs {
516 key := fmt.Sprintf("%s/%s/%d/%d/%d/%d", r.ProcessID, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch)808 key := fmt.Sprintf("%s/%s/%d/%d/%d/%d", r.ProcessID, m.Scenario, m.Run, r.Sequence, r.RequestID, r.Epoch)
517 if r.Kind == pair.start {809 if r.Kind == pair.start {
M cmd/vev-perf-harness/main_test.go +126 -11
@@ -6,6 +6,7 @@ import (
66 "errors"
77 "os"
88 "path/filepath"
9+ "strings"
910 "testing"
1011 "time"
1112 )
@@ -17,30 +18,37 @@ func (*fakeClock) Sleep(time.Duration) {}
1718
1819 type fakeProcess struct {
1920 fail bool
20- warmups []time.Duration
21- measures []time.Duration
21+ warmups [][]byte
22+ measures [][]byte
2223 closed bool
2324 }
2425
25-func (p *fakeProcess) Warmup(d time.Duration) error { p.warmups = append(p.warmups, d); return nil }
26-func (p *fakeProcess) Measure(d time.Duration, flush func()) error {
27- p.measures = append(p.measures, d)
26+func (p *fakeProcess) Warmup(input []byte) error {
27+ p.warmups = append(p.warmups, append([]byte(nil), input...))
28+ return nil
29+}
30+func (p *fakeProcess) Measure(input []byte, injected, flush func() error) error {
31+ p.measures = append(p.measures, append([]byte(nil), input...))
2832 if p.fail {
2933 return errors.New("measure failure")
3034 }
31- flush()
32- return nil
35+ if err := injected(); err != nil {
36+ return err
37+ }
38+ return flush()
3339 }
3440 func (p *fakeProcess) Close() error { p.closed = true; return nil }
3541
3642 type fakeLauncher struct {
3743 mappings []processMapping
44+ args [][]string
3845 process []*fakeProcess
3946 manifestPresent bool
4047 }
4148
42-func (l *fakeLauncher) Launch(m processMapping, _ []string) (launchedProcess, error) {
49+func (l *fakeLauncher) Launch(m processMapping, args []string) (launchedProcess, error) {
4350 l.mappings = append(l.mappings, m)
51+ l.args = append(l.args, append([]string(nil), args...))
4452 _, e := os.Stat(filepath.Join(filepath.Dir(m.TracePath), "manifest.json"))
4553 l.manifestPresent = e == nil
4654 p := &fakeProcess{}
@@ -102,7 +110,7 @@ func TestHarnessCreatesExclusiveTraceManifestAndEvidence(t *testing.T) {
102110 if !launcher.manifestPresent {
103111 t.Fatal("launched before complete run manifest")
104112 }
105- if len(r.EndToEnd) != 2 || r.Samples != 2 {
113+ if len(r.EndToEnd) != 1 || r.Samples != 1 {
106114 t.Fatalf("samples=%+v", r)
107115 }
108116 seen := map[string]bool{}
@@ -126,11 +134,72 @@ func TestHarnessCreatesExclusiveTraceManifestAndEvidence(t *testing.T) {
126134 if len(got.Processes) != 2 {
127135 t.Fatalf("manifest=%+v", got)
128136 }
129- if len(launcher.process) != 2 || !launcher.process[0].closed || launcher.process[0].warmups[0] != time.Second || launcher.process[0].measures[0] != minimumDuration {
137+ if len(launcher.process) != 2 || !launcher.process[0].closed || len(launcher.process[0].warmups) != 0 || len(launcher.process[1].warmups) != 1 || len(launcher.process[1].measures) != 1 {
130138 t.Fatalf("process lifecycle not recorded: %+v", launcher.process)
131139 }
132140 }
133141
142+type fakePTY struct{ writes [][]byte }
143+
144+func (p *fakePTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
145+func (p *fakePTY) Write(b []byte) (int, error) {
146+ p.writes = append(p.writes, append([]byte(nil), b...))
147+ return len(b), nil
148+}
149+func (*fakePTY) Close() error { return nil }
150+
151+type fakeOutput struct {
152+ bytes.Buffer
153+ syncs int
154+}
155+
156+func (o *fakeOutput) Sync() error { o.syncs++; return nil }
157+func (*fakeOutput) Close() error { return nil }
158+
159+func TestHarnessDoesNotInheritNestedSession(t *testing.T) {
160+ env := withoutEnv([]string{"VEV=session=old", "PATH=/bin", "VEV_PERF_TRACE=old"}, "VEV")
161+ if !equalStrings(env, []string{"PATH=/bin", "VEV_PERF_TRACE=old"}) {
162+ t.Fatalf("environment=%q", env)
163+ }
164+}
165+
166+func TestCLIProcessOwnsPTYInjectionAndSuccessfulFlushBoundary(t *testing.T) {
167+ pty, output := &fakePTY{}, &fakeOutput{}
168+ p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte, 1)}
169+ input := workloadInput(scenario{ID: "s"}, 1, "measured-1")
170+ p.chunks <- append([]byte("terminal "), inputMarker(input)...)
171+ order := []string{}
172+ if err := p.Measure(input, func() error { order = append(order, "injected"); return nil }, func() error { order = append(order, "flushed"); return nil }); err != nil {
173+ t.Fatal(err)
174+ }
175+ if len(pty.writes) != 1 || output.syncs != 1 || !equalStrings(order, []string{"injected", "flushed"}) {
176+ t.Fatalf("writes=%q syncs=%d boundary order=%q", pty.writes, output.syncs, order)
177+ }
178+}
179+
180+func TestHarnessUsesPublicRoleCommandsAndPTYWorkloads(t *testing.T) {
181+ for _, tc := range []struct {
182+ role string
183+ want []string
184+ }{
185+ {"daemon", []string{"--daemon"}},
186+ {"client", []string{"new", "perf-s-001"}},
187+ {"ssh_stdio_peer", []string{"_stdio", "perf-s-001"}},
188+ {"udp_peer", []string{"_udp-proxy", "perf-s-001"}},
189+ } {
190+ t.Run(tc.role, func(t *testing.T) {
191+ got := roleArgs(scenario{ID: "s"}, processMapping{Role: tc.role, Run: 1})
192+ if !equalStrings(got, tc.want) {
193+ t.Fatalf("args=%q want %q", got, tc.want)
194+ }
195+ })
196+ }
197+ input := string(workloadInput(scenario{ID: "s", Workload: "interactive_flood"}, 1, "measured-1"))
198+ if string(inputMarker([]byte(input))) != "__VEV_HARNESS_s_r1_measured-1__" || !bytes.Contains([]byte(input), []byte("while [ $i -lt 128 ]")) || !strings.HasSuffix(input, "printf '__VEV_HARNESS_s_r1_measured-1__\\n'\n") {
199+ t.Fatalf("workload is not real PTY shell input with an observable marker: %q", input)
200+ }
201+}
202+
134203 func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) {
135204 for _, marks := range [][]harnessMark{
136205 {{Sequence: 1, Kind: "terminal_flushed", Tick: 1, Valid: true}},
@@ -148,6 +217,52 @@ func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) {
148217 }
149218 }
150219
220+func TestHarnessCleansRunDirectoryOnProcessFailure(t *testing.T) {
221+ dir := t.TempDir()
222+ raw, err := os.OpenFile(filepath.Join(dir, "raw-harness.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ defer raw.Close()
227+ launcher := &fakeLauncher{}
228+ h := defaultHarness()
229+ h.clock = &fakeClock{}
230+ // The client is the second role and fails after the daemon has started.
231+ h.launcher = &failingLauncher{fakeLauncher: launcher}
232+ o := options{vevBin: "ignored", out: dir, warmup: time.Second, duration: minimumDuration, repetitions: minimumRepetitions}
233+ if _, err := h.runOne(o, manifest{}, scenario{ID: "cleanup", Roles: []string{"daemon", "client"}}, 1, raw); err == nil {
234+ t.Fatal("failed process accepted")
235+ }
236+ if _, err := os.Stat(filepath.Join(dir, "cleanup-run-001")); !os.IsNotExist(err) {
237+ t.Fatalf("failed run directory remains: %v", err)
238+ }
239+}
240+
241+type failingLauncher struct{ *fakeLauncher }
242+
243+func (l *failingLauncher) Launch(m processMapping, args []string) (launchedProcess, error) {
244+ p, err := l.fakeLauncher.Launch(m, args)
245+ if err != nil {
246+ return nil, err
247+ }
248+ if m.Role == "client" {
249+ p.(*fakeProcess).fail = true
250+ }
251+ return p, nil
252+}
253+
254+func equalStrings(a, b []string) bool {
255+ if len(a) != len(b) {
256+ return false
257+ }
258+ for i := range a {
259+ if a[i] != b[i] {
260+ return false
261+ }
262+ }
263+ return true
264+}
265+
151266 func TestHarnessRawRecordsAreDeterministicJSONL(t *testing.T) {
152267 var b bytes.Buffer
153268 for _, m := range []harnessMark{{Scenario: "s", Run: 1, Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}, {Scenario: "s", Run: 1, Sequence: 1, Kind: "terminal_flushed", Tick: 2, Valid: true}} {
@@ -222,7 +337,7 @@ func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) {
222337 if err := json.Unmarshal(b, &got); err != nil {
223338 t.Fatal(err)
224339 }
225- if got.EndToEnd.Count != 20 || got.Repetitions != 10 || len(got.Spans) == 0 {
340+ if got.EndToEnd.Count != 10 || got.Repetitions != 10 || len(got.Spans) == 0 {
226341 t.Fatalf("summary=%+v", got)
227342 }
228343 for _, s := range got.Spans {
@@ -6,6 +6,7 @@ import (
6 "errors"6 "errors"
7 "os"7 "os"
8 "path/filepath"8 "path/filepath"
9+ "strings"
9 "testing"10 "testing"
10 "time"11 "time"
11 )12 )
@@ -17,30 +18,37 @@ func (*fakeClock) Sleep(time.Duration) {}
17 18
18 type fakeProcess struct {19 type fakeProcess struct {
19 fail bool20 fail bool
20- warmups []time.Duration21+ warmups [][]byte
21- measures []time.Duration22+ measures [][]byte
22 closed bool23 closed bool
23 }24 }
24 25
25-func (p *fakeProcess) Warmup(d time.Duration) error { p.warmups = append(p.warmups, d); return nil }26+func (p *fakeProcess) Warmup(input []byte) error {
26-func (p *fakeProcess) Measure(d time.Duration, flush func()) error {27+ p.warmups = append(p.warmups, append([]byte(nil), input...))
27- p.measures = append(p.measures, d)28+ return nil
29+}
30+func (p *fakeProcess) Measure(input []byte, injected, flush func() error) error {
31+ p.measures = append(p.measures, append([]byte(nil), input...))
28 if p.fail {32 if p.fail {
29 return errors.New("measure failure")33 return errors.New("measure failure")
30 }34 }
31- flush()35+ if err := injected(); err != nil {
32- return nil36+ return err
37+ }
38+ return flush()
33 }39 }
34 func (p *fakeProcess) Close() error { p.closed = true; return nil }40 func (p *fakeProcess) Close() error { p.closed = true; return nil }
35 41
36 type fakeLauncher struct {42 type fakeLauncher struct {
37 mappings []processMapping43 mappings []processMapping
44+ args [][]string
38 process []*fakeProcess45 process []*fakeProcess
39 manifestPresent bool46 manifestPresent bool
40 }47 }
41 48
42-func (l *fakeLauncher) Launch(m processMapping, _ []string) (launchedProcess, error) {49+func (l *fakeLauncher) Launch(m processMapping, args []string) (launchedProcess, error) {
43 l.mappings = append(l.mappings, m)50 l.mappings = append(l.mappings, m)
51+ l.args = append(l.args, append([]string(nil), args...))
44 _, e := os.Stat(filepath.Join(filepath.Dir(m.TracePath), "manifest.json"))52 _, e := os.Stat(filepath.Join(filepath.Dir(m.TracePath), "manifest.json"))
45 l.manifestPresent = e == nil53 l.manifestPresent = e == nil
46 p := &fakeProcess{}54 p := &fakeProcess{}
@@ -102,7 +110,7 @@ func TestHarnessCreatesExclusiveTraceManifestAndEvidence(t *testing.T) {
102 if !launcher.manifestPresent {110 if !launcher.manifestPresent {
103 t.Fatal("launched before complete run manifest")111 t.Fatal("launched before complete run manifest")
104 }112 }
105- if len(r.EndToEnd) != 2 || r.Samples != 2 {113+ if len(r.EndToEnd) != 1 || r.Samples != 1 {
106 t.Fatalf("samples=%+v", r)114 t.Fatalf("samples=%+v", r)
107 }115 }
108 seen := map[string]bool{}116 seen := map[string]bool{}
@@ -126,11 +134,72 @@ func TestHarnessCreatesExclusiveTraceManifestAndEvidence(t *testing.T) {
126 if len(got.Processes) != 2 {134 if len(got.Processes) != 2 {
127 t.Fatalf("manifest=%+v", got)135 t.Fatalf("manifest=%+v", got)
128 }136 }
129- if len(launcher.process) != 2 || !launcher.process[0].closed || launcher.process[0].warmups[0] != time.Second || launcher.process[0].measures[0] != minimumDuration {137+ if len(launcher.process) != 2 || !launcher.process[0].closed || len(launcher.process[0].warmups) != 0 || len(launcher.process[1].warmups) != 1 || len(launcher.process[1].measures) != 1 {
130 t.Fatalf("process lifecycle not recorded: %+v", launcher.process)138 t.Fatalf("process lifecycle not recorded: %+v", launcher.process)
131 }139 }
132 }140 }
133 141
142+type fakePTY struct{ writes [][]byte }
143+
144+func (p *fakePTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
145+func (p *fakePTY) Write(b []byte) (int, error) {
146+ p.writes = append(p.writes, append([]byte(nil), b...))
147+ return len(b), nil
148+}
149+func (*fakePTY) Close() error { return nil }
150+
151+type fakeOutput struct {
152+ bytes.Buffer
153+ syncs int
154+}
155+
156+func (o *fakeOutput) Sync() error { o.syncs++; return nil }
157+func (*fakeOutput) Close() error { return nil }
158+
159+func TestHarnessDoesNotInheritNestedSession(t *testing.T) {
160+ env := withoutEnv([]string{"VEV=session=old", "PATH=/bin", "VEV_PERF_TRACE=old"}, "VEV")
161+ if !equalStrings(env, []string{"PATH=/bin", "VEV_PERF_TRACE=old"}) {
162+ t.Fatalf("environment=%q", env)
163+ }
164+}
165+
166+func TestCLIProcessOwnsPTYInjectionAndSuccessfulFlushBoundary(t *testing.T) {
167+ pty, output := &fakePTY{}, &fakeOutput{}
168+ p := &cliProcess{pty: pty, output: output, chunks: make(chan []byte, 1)}
169+ input := workloadInput(scenario{ID: "s"}, 1, "measured-1")
170+ p.chunks <- append([]byte("terminal "), inputMarker(input)...)
171+ order := []string{}
172+ if err := p.Measure(input, func() error { order = append(order, "injected"); return nil }, func() error { order = append(order, "flushed"); return nil }); err != nil {
173+ t.Fatal(err)
174+ }
175+ if len(pty.writes) != 1 || output.syncs != 1 || !equalStrings(order, []string{"injected", "flushed"}) {
176+ t.Fatalf("writes=%q syncs=%d boundary order=%q", pty.writes, output.syncs, order)
177+ }
178+}
179+
180+func TestHarnessUsesPublicRoleCommandsAndPTYWorkloads(t *testing.T) {
181+ for _, tc := range []struct {
182+ role string
183+ want []string
184+ }{
185+ {"daemon", []string{"--daemon"}},
186+ {"client", []string{"new", "perf-s-001"}},
187+ {"ssh_stdio_peer", []string{"_stdio", "perf-s-001"}},
188+ {"udp_peer", []string{"_udp-proxy", "perf-s-001"}},
189+ } {
190+ t.Run(tc.role, func(t *testing.T) {
191+ got := roleArgs(scenario{ID: "s"}, processMapping{Role: tc.role, Run: 1})
192+ if !equalStrings(got, tc.want) {
193+ t.Fatalf("args=%q want %q", got, tc.want)
194+ }
195+ })
196+ }
197+ input := string(workloadInput(scenario{ID: "s", Workload: "interactive_flood"}, 1, "measured-1"))
198+ if string(inputMarker([]byte(input))) != "__VEV_HARNESS_s_r1_measured-1__" || !bytes.Contains([]byte(input), []byte("while [ $i -lt 128 ]")) || !strings.HasSuffix(input, "printf '__VEV_HARNESS_s_r1_measured-1__\\n'\n") {
199+ t.Fatalf("workload is not real PTY shell input with an observable marker: %q", input)
200+ }
201+}
202+
134 func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) {203 func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) {
135 for _, marks := range [][]harnessMark{204 for _, marks := range [][]harnessMark{
136 {{Sequence: 1, Kind: "terminal_flushed", Tick: 1, Valid: true}},205 {{Sequence: 1, Kind: "terminal_flushed", Tick: 1, Valid: true}},
@@ -148,6 +217,52 @@ func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) {
148 }217 }
149 }218 }
150 219
220+func TestHarnessCleansRunDirectoryOnProcessFailure(t *testing.T) {
221+ dir := t.TempDir()
222+ raw, err := os.OpenFile(filepath.Join(dir, "raw-harness.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
223+ if err != nil {
224+ t.Fatal(err)
225+ }
226+ defer raw.Close()
227+ launcher := &fakeLauncher{}
228+ h := defaultHarness()
229+ h.clock = &fakeClock{}
230+ // The client is the second role and fails after the daemon has started.
231+ h.launcher = &failingLauncher{fakeLauncher: launcher}
232+ o := options{vevBin: "ignored", out: dir, warmup: time.Second, duration: minimumDuration, repetitions: minimumRepetitions}
233+ if _, err := h.runOne(o, manifest{}, scenario{ID: "cleanup", Roles: []string{"daemon", "client"}}, 1, raw); err == nil {
234+ t.Fatal("failed process accepted")
235+ }
236+ if _, err := os.Stat(filepath.Join(dir, "cleanup-run-001")); !os.IsNotExist(err) {
237+ t.Fatalf("failed run directory remains: %v", err)
238+ }
239+}
240+
241+type failingLauncher struct{ *fakeLauncher }
242+
243+func (l *failingLauncher) Launch(m processMapping, args []string) (launchedProcess, error) {
244+ p, err := l.fakeLauncher.Launch(m, args)
245+ if err != nil {
246+ return nil, err
247+ }
248+ if m.Role == "client" {
249+ p.(*fakeProcess).fail = true
250+ }
251+ return p, nil
252+}
253+
254+func equalStrings(a, b []string) bool {
255+ if len(a) != len(b) {
256+ return false
257+ }
258+ for i := range a {
259+ if a[i] != b[i] {
260+ return false
261+ }
262+ }
263+ return true
264+}
265+
151 func TestHarnessRawRecordsAreDeterministicJSONL(t *testing.T) {266 func TestHarnessRawRecordsAreDeterministicJSONL(t *testing.T) {
152 var b bytes.Buffer267 var b bytes.Buffer
153 for _, m := range []harnessMark{{Scenario: "s", Run: 1, Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}, {Scenario: "s", Run: 1, Sequence: 1, Kind: "terminal_flushed", Tick: 2, Valid: true}} {268 for _, m := range []harnessMark{{Scenario: "s", Run: 1, Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}, {Scenario: "s", Run: 1, Sequence: 1, Kind: "terminal_flushed", Tick: 2, Valid: true}} {
@@ -222,7 +337,7 @@ func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) {
222 if err := json.Unmarshal(b, &got); err != nil {337 if err := json.Unmarshal(b, &got); err != nil {
223 t.Fatal(err)338 t.Fatal(err)
224 }339 }
225- if got.EndToEnd.Count != 20 || got.Repetitions != 10 || len(got.Spans) == 0 {340+ if got.EndToEnd.Count != 10 || got.Repetitions != 10 || len(got.Spans) == 0 {
226 t.Fatalf("summary=%+v", got)341 t.Fatalf("summary=%+v", got)
227 }342 }
228 for _, s := range got.Spans {343 for _, s := range got.Spans {

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 2

Tags 13

↑↓ moveenter openesc close