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