1//go:build linux
2
3package main
4
5import (
6 "context"
7 "debug/buildinfo"
8 "encoding/json"
9 "errors"
10 "fmt"
11 "io"
12 "os"
13 "os/exec"
14 "path/filepath"
15 "runtime/debug"
16 "sort"
17 "strconv"
18 "strings"
19 "time"
20
21 "github.com/bnema/vev/pkg/safedir"
22)
23
24type harnessMark struct {
25 Scenario string `json:"scenario"`
26 Run int `json:"run"`
27 Sequence uint64 `json:"sequence"`
28 Kind string `json:"kind"`
29 Tick int64 `json:"tick"`
30 Valid bool `json:"valid"`
31}
32
33func (c systemClock) Now() int64 { return time.Since(c.start).Nanoseconds() }
34
35func (systemClock) Sleep(d time.Duration) { time.Sleep(d) }
36
37func newPTYLocalEcho(input []byte) *ptyLocalEcho {
38 expected := make([]byte, 0, len(input))
39 for _, b := range input {
40 if b == '\n' {
41 expected = append(expected, '\r')
42 }
43 expected = append(expected, b)
44 }
45 return &ptyLocalEcho{expected: expected}
46}
47
48type harness struct {
49 clock clock
50 launcher launcher
51 mkdir func(string) error
52 createRunDir func(string) error
53 create func(string) (*os.File, error)
54 removeAll func(string) error
55 gitSHA func() string
56 vevGitSHA func(string) string
57 selectScenarios func(manifest) []scenario // test-only bounded fixture seam
58}
59
60// createExclusiveRunDir prevents stale runtime, socket, state, or trace files
61// from an earlier invocation being mistaken for the next repetition's evidence.
62func createExclusiveRunDir(path string) error {
63 if err := os.Mkdir(path, 0o700); err != nil {
64 return err
65 }
66 if err := safedir.EnsurePrivate(path); err != nil {
67 return errors.Join(err, os.Remove(path))
68 }
69 return nil
70}
71
72// recordedGitSHA prefers build provenance so an installed public harness does
73// not depend on its launch directory being a checkout. The git fallback keeps
74// go run evidence attributable as well.
75const gitSHACommandTimeout = time.Second
76
77func recordedBinaryGitSHA(path string) string {
78 info, err := buildinfo.ReadFile(path)
79 if err != nil {
80 return "unknown"
81 }
82 for _, setting := range info.Settings {
83 if setting.Key == "vcs.revision" && setting.Value != "" {
84 return setting.Value
85 }
86 }
87 return "unknown"
88}
89
90func recordedGitSHA() string {
91 if info, ok := debug.ReadBuildInfo(); ok {
92 for _, setting := range info.Settings {
93 if setting.Key == "vcs.revision" && setting.Value != "" {
94 return setting.Value
95 }
96 }
97 }
98 ctx, cancel := context.WithTimeout(context.Background(), gitSHACommandTimeout)
99 defer cancel()
100 out, err := exec.CommandContext(ctx, "git", "rev-parse", "HEAD").Output()
101 if err == nil && strings.TrimSpace(string(out)) != "" {
102 return strings.TrimSpace(string(out))
103 }
104 return "unknown"
105}
106
107func run(args []string, h *harness) (err error) {
108 opt, err := parseOptions(args)
109 if err != nil {
110 return err
111 }
112 opt, err = resolvePathOptions(opt)
113 if err != nil {
114 return err
115 }
116 m, err := readManifest(opt.manifest)
117 if err != nil {
118 return err
119 }
120 if err := validateManifest(m); err != nil {
121 return err
122 }
123 if err = h.mkdir(opt.out); err != nil {
124 return err
125 }
126 raw, err := os.OpenFile(filepath.Join(opt.out, "raw-harness.jsonl"), os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
127 if err != nil {
128 return err
129 }
130 defer func() {
131 if closeErr := raw.Close(); closeErr != nil {
132 err = errors.Join(err, closeErr)
133 }
134 }()
135 results := make([]runResult, 0, len(m.Scenarios)*opt.repetitions)
136 var localSpans []span
137 scenarios := m.Scenarios
138 if opt.scenario != "" {
139 scenarios, err = selectScenario(m, opt.scenario)
140 if err != nil {
141 return err
142 }
143 }
144 if h.selectScenarios != nil {
145 scenarios = h.selectScenarios(m)
146 }
147 for _, s := range scenarios {
148 if s.InapplicableReason != "" {
149 continue
150 }
151 for r := 1; r <= opt.repetitions; r++ {
152 rr, err := h.runOne(opt, m, s, r, raw)
153 if err != nil {
154 return fmt.Errorf("scenario %s run %d: %w", s.ID, r, err)
155 }
156 results = append(results, rr)
157 localSpans = append(localSpans, rr.Spans...)
158 }
159 }
160 all, cadence, maxGap, runP50s := aggregateEventResults(results)
161 if len(all) == 0 {
162 return errors.New("no measured harness input/flush pairs")
163 }
164 if err := writeJSON(filepath.Join(opt.out, "runs.json"), results); err != nil {
165 return err
166 }
167 return writeJSON(filepath.Join(opt.out, "summary.json"), summary{Schema: 1, GitSHA: h.gitSHA(), VevGitSHA: h.vevGitSHA(opt.vevBin), Warmup: opt.warmup.String(), Duration: opt.duration.String(), Repetitions: opt.repetitions, EndToEnd: percentiles(all), Cadence: percentiles(cadence), MaxGap: maxGap, RunP50Dispersion: percentiles(runP50s), Spans: summarizeSpans(localSpans), Runs: len(results)})
168}
169
170// aggregateEventResults deliberately preserves raw event latencies for
171// end-to-end percentiles. Per-run p50s are a separate dispersion series.
172func aggregateEventResults(results []runResult) (all, cadence []int64, maxGap int64, runP50s []int64) {
173 for _, result := range results {
174 all = append(all, result.EndToEnd...)
175 cadence = append(cadence, result.CadenceSamples...)
176 runP50s = append(runP50s, result.Event.P50)
177 if result.MaxGap > maxGap {
178 maxGap = result.MaxGap
179 }
180 }
181 return all, cadence, maxGap, runP50s
182}
183
184func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Writer) (res runResult, err error) {
185 dir := filepath.Join(o.out, fmt.Sprintf("%s-run-%03d", safeName(s.ID), run))
186 createRunDir := h.createRunDir
187 if createRunDir == nil {
188 createRunDir = createExclusiveRunDir
189 }
190 if err = createRunDir(dir); err != nil {
191 return res, fmt.Errorf("create isolated run directory: %w", err)
192 }
193 success := false
194 defer func() {
195 if !success {
196 if cleanupErr := h.removeAll(dir); cleanupErr != nil {
197 err = errors.Join(err, fmt.Errorf("remove failed run directory: %w", cleanupErr))
198 }
199 }
200 }()
201 maps := make([]processMapping, 0, len(s.Roles))
202 for i, role := range s.Roles {
203 pid := fmt.Sprintf("%s-r%03d-%s-%02d", safeName(s.ID), run, safeName(role), i+1)
204 path := filepath.Join(dir, pid+".jsonl")
205 f, e := h.create(path)
206 if e != nil {
207 return res, fmt.Errorf("allocate exclusive trace %s: %w", role, e)
208 }
209 if e = f.Close(); e != nil {
210 return res, e
211 }
212 maps = append(maps, processMapping{ProcessID: pid, ClockDomain: pid, TracePath: path, Role: role, Scenario: s.ID, Run: run, Identity: "harness-assigned:" + pid})
213 }
214 // Persist the complete mapping before any process can be launched. This is
215 // both crash evidence and the only legal source for later ID correlation.
216 if err = writeJSON(filepath.Join(dir, "manifest.json"), runManifest{Scenario: s.ID, Run: run, Processes: maps}); err != nil {
217 return res, err
218 }
219 if h.launcher == nil {
220 h.launcher = &cliLauncher{bin: o.vevBin}
221 }
222 processes := make([]launchedRole, 0, len(maps))
223 closed := false
224 closeRoles := func() error {
225 if closed {
226 return nil
227 }
228 closed = true
229 // Clients own SSH-stdio descendants; close them before the deferred UDP
230 // peer and daemon so no role survives a failed or completed run.
231 return closeLaunchedRoles(processes)
232 }
233 defer func() {
234 if cleanupErr := closeRoles(); cleanupErr != nil {
235 if err == nil {
236 err = cleanupErr
237 } else {
238 // Keep the primary workload failure first while retaining every
239 // cleanup failure in reverse launch order.
240 err = errors.Join(err, cleanupErr)
241 }
242 }
243 }()
244 selectedTransport, e := manifestTransport(mat, s.Transport)
245 if e != nil {
246 return res, e
247 }
248 for _, pm := range launchOrder(maps) {
249 p, e := h.launcher.Launch(pm, routeRoleArgs(s, pm, selectedTransport))
250 if e != nil {
251 return res, e
252 }
253 processes = append(processes, launchedRole{mapping: pm, process: p})
254 if pm.Role == "daemon" {
255 if ready, ok := p.(interface{ WaitReady() error }); ok {
256 if err = ready.WaitReady(); err != nil {
257 return res, err
258 }
259 }
260 }
261 }
262 marks := []harnessMark{}
263 recordMark := func(mark harnessMark) error {
264 marks = append(marks, mark)
265 if err := json.NewEncoder(raw).Encode(mark); err != nil {
266 return fmt.Errorf("write raw harness mark: %w", err)
267 }
268 return nil
269 }
270 seq := uint64(0)
271 for _, p := range processes {
272 if p.mapping.Role != "client" {
273 continue
274 }
275 seq++
276 sequence := seq
277 if err = recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "input_injected", Tick: h.clock.Now(), Valid: true}); err != nil {
278 return res, err
279 }
280 warmupErr := p.process.Warmup(workloadInput(s, run, "warmup"))
281 if markErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: warmupErr == nil}); markErr != nil {
282 return res, errors.Join(warmupErr, markErr)
283 }
284 if warmupErr != nil {
285 return res, warmupErr
286 }
287 }
288 // The harness clock, rather than a process clock, owns the explicit warmup
289 // and measured intervals. Raw evidence retains every boundary, while only
290 // complete valid pairs within [Start, End) contribute measurement samples.
291 h.clock.Sleep(o.warmup)
292 interval := measuredInterval{Start: h.clock.Now()}
293 interval.End = interval.Start + o.duration.Nanoseconds()
294 nextInjection := interval.Start
295 for {
296 now := h.clock.Now()
297 if now < nextInjection {
298 h.clock.Sleep(time.Duration(nextInjection - now))
299 now = h.clock.Now()
300 }
301 if now >= interval.End {
302 break
303 }
304 for _, p := range processes {
305 if p.mapping.Role != "client" {
306 continue
307 }
308 seq++
309 sequence := seq
310 injectedRecorded := false
311 flushedRecorded := false
312 injected := func() error {
313 if markErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "input_injected", Tick: h.clock.Now(), Valid: true}); markErr != nil {
314 return markErr
315 }
316 injectedRecorded = true
317 return nil
318 }
319 flushed := func() error {
320 if markErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: true}); markErr != nil {
321 return markErr
322 }
323 flushedRecorded = true
324 return nil
325 }
326 measureInjected := injected
327 if cols, rows, resize := workloadResize(s, sequence); resize {
328 resizer, ok := p.process.(processResizer)
329 if !ok {
330 return res, errors.New("client process does not support resize workload")
331 }
332 if err := injected(); err != nil {
333 return res, err
334 }
335 measureInjected = func() error { return nil }
336 if resizeErr := resizer.Resize(cols, rows); resizeErr != nil {
337 diagnosticErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: false})
338 return res, errors.Join(fmt.Errorf("resize client PTY: %w", resizeErr), diagnosticErr)
339 }
340 }
341 if measureErr := p.process.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), measureInjected, flushed); measureErr != nil {
342 if injectedRecorded && !flushedRecorded {
343 diagnosticErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: false})
344 return res, errors.Join(measureErr, diagnosticErr)
345 }
346 return res, measureErr
347 }
348 }
349 // Cap injection rate rather than bursting events after a slow terminal
350 // flush. Any resulting spacing is retained as raw cadence evidence.
351 nextInjection = h.clock.Now() + measuredEventCadence.Nanoseconds()
352 }
353 if now := h.clock.Now(); now < interval.End {
354 h.clock.Sleep(time.Duration(interval.End - now))
355 }
356 events, e := measuredEventSamples(marks, interval)
357 if e != nil {
358 return res, e
359 }
360 if err := requireMinimumEventSamples(events); err != nil {
361 return res, err
362 }
363 samples := eventLatencies(events)
364 cadenceSamples, maxGap := eventCadenceSamples(events)
365 // Process teardown can unblock a receive and cause its failed end mark to
366 // be serialized. Merge only the complete, post-cleanup per-process traces.
367 if cleanupErr := closeRoles(); cleanupErr != nil {
368 return res, cleanupErr
369 }
370 spans, e := mergeProcessTraces(maps)
371 if e != nil {
372 return res, e
373 }
374 success = true
375 return runResult{Spans: spans, Scenario: s.ID, Run: run, Samples: len(samples), EndToEnd: samples, Event: percentiles(samples), Cadence: percentiles(cadenceSamples), CadenceSamples: cadenceSamples, MaxGap: maxGap, Processes: maps}, nil
376}
377
378func roleArgs(s scenario, m processMapping) roleCommand {
379 return routeRoleArgs(s, m, scenarioTransport(s))
380}
381
382func routeRoleArgs(s scenario, m processMapping, selected transport) roleCommand {
383 session := "perf-" + safeName(s.ID) + fmt.Sprintf("-%03d", m.Run)
384 // Remote benchmark sessions are created by the first Hello on the peer
385 // daemon. Omitting the session suffix keeps that Hello ephemeral; a named
386 // attach would fail because the fresh per-run daemon has no session yet.
387 remote := []string{"attach", "harness@127.0.0.1"}
388 switch m.Role {
389 case "daemon":
390 return roleCommand{Args: []string{"--daemon"}}
391 case "client":
392 switch s.Transport {
393 case "local":
394 return roleCommand{Args: []string{"new", session}}
395 case "ssh_stdio", "udp_baseline", "udp_25ms", "udp_100ms", "udp_loss_0pct", "udp_loss_1pct":
396 return roleCommand{Args: remote}
397 }
398 case "ssh_stdio_peer":
399 return roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: s.Transport, Kind: "ssh_stdio"}}
400 case "udp_peer":
401 return roleCommand{Args: []string{"_udp-proxy"}, Transport: selected}
402 }
403 return roleCommand{}
404}
405
406func manifestTransport(m manifest, id string) (transport, error) {
407 for _, t := range m.Transports {
408 if t.ID == id {
409 return t, nil
410 }
411 }
412 // Focused unit tests may call runOne with a minimal manifest; production
413 // already validates every scenario reference before this point.
414 if m.Schema == 0 {
415 return scenarioTransport(scenario{Transport: id}), nil
416 }
417 return transport{}, fmt.Errorf("scenario references missing transport %q", id)
418}
419
420func scenarioTransport(s scenario) transport {
421 switch s.Transport {
422 case "udp_baseline":
423 return transport{ID: s.Transport, Kind: "udp"}
424 case "udp_25ms":
425 return transport{ID: s.Transport, Kind: "udp", RTTMS: 25}
426 case "udp_100ms":
427 return transport{ID: s.Transport, Kind: "udp", RTTMS: 100}
428 case "udp_loss_0pct":
429 return transport{ID: s.Transport, Kind: "udp", LossPercent: 0}
430 case "udp_loss_1pct":
431 return transport{ID: s.Transport, Kind: "udp", LossPercent: 1}
432 default:
433 return transport{ID: s.Transport}
434 }
435}
436
437func launchOrder(mappings []processMapping) []processMapping {
438 out := append([]processMapping(nil), mappings...)
439 priority := func(role string) int {
440 switch role {
441 case "daemon":
442 return 0
443 case "ssh_stdio_peer", "udp_peer":
444 return 1
445 case "client":
446 return 2
447 default:
448 return 3
449 }
450 }
451 sort.SliceStable(out, func(i, j int) bool { return priority(out[i].Role) < priority(out[j].Role) })
452 return out
453}
454
455type processResizer interface {
456 Resize(cols, rows uint16) error
457}
458
459func workloadResize(s scenario, sequence uint64) (uint16, uint16, bool) {
460 if s.Workload != "resize_sweep" && s.Workload != "snapshot_output_resize" {
461 return 0, 0, false
462 }
463 if sequence%2 == 1 {
464 return 119, 40, true
465 }
466 return 120, 40, true
467}
468
469// workloadInput is shell input delivered through the client PTY. It avoids any
470// private daemon API and makes every topology/workload/transport manifest row
471// execute a real terminal workload. The marker is the output boundary awaited
472// by cliProcess; it is not a synthetic timestamp.
473func workloadInput(s scenario, run int, phase string) []byte {
474 marker := fmt.Sprintf("__VEV_HARNESS_%s_r%d_%s__", safeName(s.ID), run, safeName(phase))
475 body := fmt.Sprintf("printf 'vev perf %s\\n'", s.ID)
476 markerSeparator := "; "
477 switch s.Workload {
478 case "active_output":
479 cell := byte('X')
480 if sequence, err := strconv.ParseUint(strings.TrimPrefix(phase, "measured-"), 10, 64); err == nil && strings.HasPrefix(phase, "measured-") {
481 cell = '0' + byte(sequence%2)
482 }
483 body = fmt.Sprintf(`printf '\033[10;20H%c'; : '%s'`, cell, s.ID)
484 case "all_output":
485 body = fmt.Sprintf(`printf '%%-120s' 'vev perf line %s'`, s.ID)
486 case "inactive_output":
487 body = fmt.Sprintf(`printf '\033[38;2;20;120;220mred'; printf '\033[38;2;220;80;40mgreen'; printf '\033[38;2;80;220;120m%s'; printf '\033[0m'`, s.ID)
488 case "interactive_flood":
489 body = strings.TrimSuffix(strings.Repeat("printf 'vev perf output\\n'; ", 128), "; ")
490 case "resize_sweep":
491 body = fmt.Sprintf("printf 'vev perf resize-sweep %s\\n'", s.ID)
492 case "copy_search":
493 body = fmt.Sprintf("printf 'vev perf copy-search %s\\n'", s.ID)
494 case "snapshot_output_resize":
495 body = fmt.Sprintf(`printf '\033[2J\033[Hvev perf snapshot-output-resize %s\n'`, s.ID)
496 case "attach_restore_tab_switch":
497 body = fmt.Sprintf(`"$VEV_PERF_BIN" cmd next-tab && "$VEV_PERF_BIN" cmd previous-tab && printf 'vev perf attach-restore-tab-switch %s\n'`, s.ID)
498 if phase == "warmup" {
499 body = fmt.Sprintf(`"$VEV_PERF_BIN" cmd new-tab && "$VEV_PERF_BIN" cmd previous-tab && printf 'vev perf attach-restore-tab-switch %s\n'`, s.ID)
500 }
501 markerSeparator = " && "
502 }
503 return []byte(body + markerSeparator + fmt.Sprintf("printf '%s\\n'\n", marker))
504}