1//go:build linux
2
3package main
4
5import (
6 "bytes"
7 "encoding/json"
8 "errors"
9 "os"
10 "path/filepath"
11 "strings"
12 "testing"
13 "time"
14
15 "github.com/stretchr/testify/require"
16)
17
18func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) {
19 dir := t.TempDir()
20 if err := os.Chmod(dir, 0o700); err != nil {
21 t.Fatal(err)
22 }
23 l := &cliLauncher{bin: "/bin/true"}
24 m := processMapping{ProcessID: "ssh-stdio-peer", TracePath: filepath.Join(dir, "peer.jsonl"), Role: "ssh_stdio_peer"}
25 if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil {
26 t.Fatal(err)
27 }
28 p, err := l.preparePeer(m, roleCommand{Args: []string{"_broker-mux-stdio", "--production"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}})
29 if err != nil {
30 t.Fatal(err)
31 }
32 shim, err := os.ReadFile(filepath.Join(dir, "ssh"))
33 if err != nil {
34 t.Fatal(err)
35 }
36 text := string(shim)
37 for _, want := range []string{"_broker-mux-stdio", m.TracePath, m.ProcessID} {
38 if !strings.Contains(text, want) {
39 t.Errorf("seam does not retain %q:\n%s", want, text)
40 }
41 }
42 if strings.Contains(text, "VEV_PERF_UDP_") {
43 t.Fatalf("ignored vev UDP environment was used instead of the broker mux stdio seam: %s", text)
44 }
45 // Close must be a no-op for a peer with no owned background process.
46 if err := p.Close(); err != nil {
47 t.Fatal(err)
48 }
49}
50
51func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) {
52 for _, marks := range [][]harnessMark{
53 {{Sequence: 1, Kind: "terminal_flushed", Tick: 1, Valid: true}},
54 {{Sequence: 1, Kind: "input_injected", Tick: 2, Valid: true}, {Sequence: 1, Kind: "terminal_flushed", Tick: 1, Valid: true}},
55 {{Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}},
56 {{Sequence: 1, Kind: "input_injected", Tick: 1, Valid: true}, {Sequence: 1, Kind: "input_injected", Tick: 2, Valid: true}},
57 } {
58 if _, err := pairedSamples(marks); err == nil {
59 t.Fatalf("invalid marks accepted: %+v", marks)
60 }
61 }
62 got, err := pairedSamples([]harnessMark{{Sequence: 9, Kind: "input_injected", Tick: 1, Valid: true}, {Sequence: 9, Kind: "terminal_flushed", Tick: 5, Valid: true}})
63 if err != nil || len(got) != 1 || got[0] != 4 {
64 t.Fatalf("got %v %v", got, err)
65 }
66}
67
68func TestHarnessCollectsOnlyCompleteMeasuredIntervalEvents(t *testing.T) {
69 interval := measuredInterval{Start: 100, End: 200}
70 marks := []harnessMark{
71 {Sequence: 1, Kind: "input_injected", Tick: 10, Valid: true},
72 {Sequence: 1, Kind: "terminal_flushed", Tick: 20, Valid: true}, // warmup
73 {Sequence: 2, Kind: "input_injected", Tick: 100, Valid: true},
74 {Sequence: 2, Kind: "terminal_flushed", Tick: 120, Valid: true},
75 {Sequence: 3, Kind: "input_injected", Tick: 150, Valid: true},
76 {Sequence: 3, Kind: "terminal_flushed", Tick: 201, Valid: true}, // exits interval
77 {Sequence: 4, Kind: "input_injected", Tick: 99, Valid: true},
78 {Sequence: 4, Kind: "terminal_flushed", Tick: 110, Valid: true}, // enters interval
79 {Sequence: 5, Kind: "input_injected", Tick: 180, Valid: true},
80 {Sequence: 5, Kind: "terminal_flushed", Tick: 200, Valid: true}, // end is excluded
81 }
82 events, err := measuredEventSamples(marks, interval)
83 if err != nil {
84 t.Fatal(err)
85 }
86 if len(events) != 1 || events[0].Sequence != 2 || events[0].Latency != 20 {
87 t.Fatalf("measured events=%+v, want only complete sequence 2", events)
88 }
89}
90
91func TestHarnessRejectsInsufficientMeasuredEventSamples(t *testing.T) {
92 marks := make([]harnessMark, 0, 2*(minimumInIntervalEventSamples-1))
93 for i := 0; i < minimumInIntervalEventSamples-1; i++ {
94 marks = append(marks,
95 harnessMark{Sequence: uint64(i + 1), Kind: "input_injected", Tick: int64(i * 100), Valid: true},
96 harnessMark{Sequence: uint64(i + 1), Kind: "terminal_flushed", Tick: int64(i*100 + 10), Valid: true},
97 )
98 }
99 events, err := measuredEventSamples(marks, measuredInterval{Start: 0, End: int64(minimumInIntervalEventSamples * 100)})
100 if err != nil {
101 t.Fatal(err)
102 }
103 if err := requireMinimumEventSamples(events); err == nil {
104 t.Fatal("insufficient measured event samples accepted")
105 }
106}
107
108func TestHarnessRunRejectsStraddlingEventsAsInsufficient(t *testing.T) {
109 dir := t.TempDir()
110 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
111 if err != nil {
112 t.Fatal(err)
113 }
114 t.Cleanup(func() {
115 if err := raw.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
116 t.Error(err)
117 }
118 })
119 clock := &fakeClock{}
120 h := defaultHarness()
121 h.clock, h.launcher = clock, insufficientEventLauncher{clock: clock}
122 _, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "late", Roles: []string{"daemon", "client"}}, 1, raw)
123 if err == nil || !strings.Contains(err.Error(), "insufficient in-interval event samples") {
124 t.Fatalf("straddling run error=%v, want insufficient in-interval sample rejection", err)
125 }
126 if err := raw.Close(); err != nil {
127 t.Fatal(err)
128 }
129 marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
130 if got, want := len(marks), 4; got != want {
131 t.Fatalf("raw marks=%d, want warmup and straddling pair=%d: %+v", got, want, marks)
132 }
133 if straddling := marks[len(marks)-1]; straddling.Sequence != 2 || straddling.Kind != "terminal_flushed" || !straddling.Valid {
134 t.Fatalf("straddling boundary=%+v", straddling)
135 }
136}
137
138func TestHarnessCalculatesEventCadenceAndMaxGapFromRawPairs(t *testing.T) {
139 events := []eventSample{
140 {Sequence: 1, Injected: 100, Flushed: 110, Latency: 10},
141 {Sequence: 2, Injected: 130, Flushed: 140, Latency: 10},
142 {Sequence: 3, Injected: 180, Flushed: 190, Latency: 10},
143 }
144 cadence, maxGap := eventCadence(events)
145 if cadence.Count != 2 || cadence.P50 != 30 || cadence.P95 != 30 || cadence.P99 != 30 || maxGap != 50 {
146 t.Fatalf("cadence=%+v maxGap=%d", cadence, maxGap)
147 }
148}
149
150func TestHarnessIncludesAllRequiredProcessLocalSpans(t *testing.T) {
151 got := make([]string, 0, len(spanPairs))
152 for _, pair := range spanPairs {
153 got = append(got, pair.name)
154 }
155 want := []string{"capture_duration", "compose_duration", "diff_duration", "queue_wait", "ack_blocked_interval", "emit_duration", "adapter_send_duration", "adapter_receive_duration"}
156 if !equalStrings(got, want) {
157 t.Fatalf("span pairs = %v, want %v", got, want)
158 }
159}
160
161func TestHarnessRawRecordsAreDeterministicJSONL(t *testing.T) {
162 var b bytes.Buffer
163 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}} {
164 b.WriteString(mustJSON(m))
165 b.WriteByte('\n')
166 }
167 path := filepath.Join(t.TempDir(), "raw.jsonl")
168 if err := os.WriteFile(path, b.Bytes(), 0o600); err != nil {
169 t.Fatal(err)
170 }
171 if got, err := readJSONL(path); err != nil || len(got) != 2 {
172 t.Fatalf("records=%d err=%v", len(got), err)
173 }
174}
175
176func TestHarnessClosesRolesBeforeMergingReceiveSpans(t *testing.T) {
177 dir := t.TempDir()
178 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
179 if err != nil {
180 t.Fatal(err)
181 }
182 t.Cleanup(func() {
183 if err := raw.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
184 t.Error(err)
185 }
186 })
187 launcher := &closeTraceLauncher{}
188 h := defaultHarness()
189 h.clock, h.launcher = &fakeClock{}, launcher
190 result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration, repetitions: minimumRepetitions}, manifest{}, scenario{ID: "receive-cleanup", Roles: []string{"daemon", "client"}}, 1, raw)
191 if err != nil {
192 t.Fatal(err)
193 }
194 if len(result.Spans) != 2 {
195 t.Fatalf("post-cleanup receive spans=%+v", result.Spans)
196 }
197 for _, s := range result.Spans {
198 if s.Name != "adapter_receive_duration" || len(s.Samples) != 1 || s.Samples[0] != 10 {
199 t.Fatalf("unmatched or wrong receive span: %+v", s)
200 }
201 }
202 for i, p := range launcher.process {
203 if !p.closed {
204 t.Fatalf("role %s was not closed before trace merge", launcher.mappings[i].Role)
205 }
206 }
207}
208
209func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) {
210 base, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json"))
211 if err != nil {
212 t.Fatal(err)
213 }
214 for i := range base.Scenarios {
215 if i > 0 {
216 base.Scenarios[i].InapplicableReason = "fixture limits this test to one public topology"
217 base.Scenarios[i].Roles = nil
218 }
219 }
220 dir := t.TempDir()
221 manifestPath := filepath.Join(dir, "manifest.json")
222 if err := os.WriteFile(manifestPath, []byte(mustJSON(base)), 0o600); err != nil {
223 t.Fatal(err)
224 }
225 l := &traceLauncher{}
226 h := defaultHarness()
227 h.clock = &fakeClock{}
228 h.launcher = l
229 h.gitSHA = func() string { return "harness-sha" }
230 h.vevGitSHA = func(path string) string {
231 if !filepath.IsAbs(path) {
232 t.Fatalf("measured binary path is not resolved: %q", path)
233 }
234 return "measured-vev-sha"
235 }
236 if err := run([]string{"--vev-bin", "ignored", "--manifest", manifestPath, "--out", filepath.Join(dir, "out"), "--warmup", "1s", "--duration", "30s", "--repetitions", "10"}, h); err != nil {
237 t.Fatal(err)
238 }
239 for _, name := range []string{"raw-harness.jsonl", "runs.json", "summary.json"} {
240 if _, err := os.Stat(filepath.Join(dir, "out", name)); err != nil {
241 t.Errorf("%s: %v", name, err)
242 }
243 }
244 var got summary
245 b, err := os.ReadFile(filepath.Join(dir, "out", "summary.json"))
246 if err != nil {
247 t.Fatal(err)
248 }
249 if err := json.Unmarshal(b, &got); err != nil {
250 t.Fatal(err)
251 }
252 if got.GitSHA != "harness-sha" || got.VevGitSHA != "measured-vev-sha" {
253 t.Fatalf("summary provenance = harness %q measured vev %q", got.GitSHA, got.VevGitSHA)
254 }
255 if got.EndToEnd.Count < 10*minimumInIntervalEventSamples || got.Repetitions != 10 || got.RunP50Dispersion.Count != 10 || got.Cadence.Count == 0 || got.MaxGap == 0 || len(got.Spans) == 0 {
256 t.Fatalf("summary=%+v", got)
257 }
258 for _, s := range got.Spans {
259 if s.Distribution.Count < 10 {
260 t.Fatalf("insufficient span summary: %+v", s)
261 }
262 }
263}
264
265func TestTraceRecordRetainsTransportDiagnostics(t *testing.T) {
266 line := []byte(`{"process_id":"p","component":"udp","scenario":"s","run":1,"sequence":1,"request_id":1,"epoch":1,"kind":"transport_diagnostic","tick":2,"bytes":3,"fragments":4,"retransmits":5,"pending":6,"ack_rtt_nanos":7,"valid":true}`)
267 var got traceRecord
268 require.NoError(t, json.Unmarshal(line, &got))
269 require.Equal(t, uint64(3), got.Bytes)
270 require.Equal(t, uint64(4), got.Fragments)
271 require.Equal(t, uint64(5), got.Retransmits)
272 require.Equal(t, uint64(6), got.Pending)
273 require.Equal(t, int64(7), got.AckRTTNanos)
274}
275
276func TestHarnessAcceptsConcurrentProductionTraceInterleaving(t *testing.T) {
277 path := filepath.Join(t.TempDir(), "local-client.jsonl")
278 if err := os.WriteFile(path, []byte(boundedLocalConcurrentTrace), 0o600); err != nil {
279 t.Fatal(err)
280 }
281 spans, err := mergeProcessTraces([]processMapping{{
282 ProcessID: "1x4-idle-local-r002-client-02", ClockDomain: "1x4-idle-local-r002-client-02", TracePath: path,
283 Scenario: "1x4-idle-local", Run: 2,
284 }})
285 if err != nil {
286 t.Fatalf("valid concurrent production trace rejected: %v", err)
287 }
288 if len(spans) != 2 || spans[0].Component != "ipc" || spans[0].Name != "adapter_send_duration" || spans[0].Samples[0] != 21900 || spans[1].Component != "ipc" || spans[1].Name != "adapter_receive_duration" || spans[1].Samples[0] != 178550 {
289 t.Fatalf("spans=%+v", spans)
290 }
291}
292
293func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) {
294 write := func(t *testing.T, records ...traceRecord) processMapping {
295 t.Helper()
296 path := filepath.Join(t.TempDir(), "trace.jsonl")
297 f, err := os.Create(path)
298 if err != nil {
299 t.Fatal(err)
300 }
301 for _, r := range records {
302 b, err := json.Marshal(r)
303 if err != nil {
304 t.Fatal(err)
305 }
306 if _, err := f.Write(append(b, '\n')); err != nil {
307 t.Fatal(err)
308 }
309 }
310 if err := f.Close(); err != nil {
311 t.Fatal(err)
312 }
313 return processMapping{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}
314 }
315 base := func(kind string, tick int64) traceRecord {
316 return traceRecord{ProcessID: "one", Component: "daemon", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick, Valid: true}
317 }
318 for _, records := range [][]traceRecord{
319 {base("diff_end", 1)},
320 {base("diff_start", 2), base("diff_end", 1)},
321 {base("diff_start", 1)},
322 {base("diff_start", 1), base("diff_start", 2)},
323 {base("diff_start", 1), func() traceRecord { r := base("diff_end", 2); r.Component = "ipc"; return r }()},
324 {func() traceRecord { r := base("diff_start", 1); r.ProcessID = "other"; return r }()},
325 {func() traceRecord { r := base("diff_start", 1); r.Scenario = "other-scenario"; return r }()},
326 {func() traceRecord { r := base("diff_start", 1); r.Run = 2; return r }()},
327 } {
328 if _, err := mergeProcessTraces([]processMapping{write(t, records...)}); err == nil {
329 t.Fatalf("invalid trace accepted: %+v", records)
330 }
331 }
332 m := write(t, base("diff_start", 1), base("diff_end", 3))
333 spans, err := mergeProcessTraces([]processMapping{m})
334 if err != nil || len(spans) != 1 || spans[0].Samples[0] != 2 {
335 t.Fatalf("spans=%+v err=%v", spans, err)
336 }
337}
338
339func TestHarnessObserverGapSuppressesPotentiallyDroppedSpanEnds(t *testing.T) {
340 base := func(kind string, sequence, requestID, epoch uint64, tick int64) traceRecord {
341 return traceRecord{ProcessID: "one", Component: "ipc", Scenario: "s", Run: 1, Sequence: sequence, RequestID: requestID, Epoch: epoch, Kind: kind, Tick: tick, Valid: true}
342 }
343 gap := traceRecord{ProcessID: "one", Component: "observer", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: "transport_diagnostic", Valid: false}
344 for _, tc := range []struct {
345 name string
346 records []traceRecord
347 }{
348 {
349 name: "span began before observed gap",
350 records: []traceRecord{
351 base("adapter_receive_start", 2, 2, 2, 10),
352 gap,
353 base("adapter_receive_end", 2, 2, 2, 20),
354 },
355 },
356 {
357 name: "span start was dropped before observed gap",
358 records: []traceRecord{
359 gap,
360 base("adapter_receive_end", 3, 3, 3, 30),
361 },
362 },
363 } {
364 t.Run(tc.name, func(t *testing.T) {
365 path := filepath.Join(t.TempDir(), "trace.jsonl")
366 var lines []string
367 for _, record := range tc.records {
368 lines = append(lines, mustJSON(record))
369 }
370 if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil {
371 t.Fatal(err)
372 }
373 spans, err := mergeProcessTraces([]processMapping{{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}})
374 if err != nil {
375 t.Fatalf("potentially dropped span end was rejected: %v", err)
376 }
377 if len(spans) != 0 {
378 t.Fatalf("spans=%+v, want no sample after observer gap", spans)
379 }
380 })
381 }
382}
383
384func TestHarnessExcludesFailedSpanDurationsWhileValidatingPairing(t *testing.T) {
385 base := func(kind string, tick int64, valid bool) traceRecord {
386 return traceRecord{ProcessID: "one", Component: "adapter", Scenario: "s", Run: 1, Sequence: 1, RequestID: 1, Epoch: 1, Kind: kind, Tick: tick, Valid: valid}
387 }
388 for _, tc := range []struct {
389 name string
390 records []traceRecord
391 wantSpans int
392 wantErr bool
393 }{
394 {"successful adapter send is sampled", []traceRecord{base("adapter_send_start", 10, true), base("adapter_send_end", 20, true)}, 1, false},
395 {"failed adapter send end is excluded", []traceRecord{base("adapter_send_start", 10, true), base("adapter_send_end", 20, false)}, 0, false},
396 {"failed adapter receive end is excluded", []traceRecord{base("adapter_receive_start", 10, true), base("adapter_receive_end", 20, false)}, 0, false},
397 {"failed adapter send start is excluded", []traceRecord{base("adapter_send_start", 10, false), base("adapter_send_end", 20, true)}, 0, false},
398 {"failed adapter receive boundaries are excluded", []traceRecord{base("adapter_receive_start", 10, false), base("adapter_receive_end", 20, false)}, 0, false},
399 {"failed end without start is rejected", []traceRecord{base("adapter_send_end", 20, false)}, 0, true},
400 {"duplicate failed start is rejected", []traceRecord{base("adapter_send_start", 10, false), base("adapter_send_start", 15, false)}, 0, true},
401 {"negative failed span is rejected", []traceRecord{base("adapter_receive_start", 20, false), base("adapter_receive_end", 10, false)}, 0, true},
402 {"cross-component failed end is rejected", []traceRecord{base("adapter_send_start", 10, true), func() traceRecord { r := base("adapter_send_end", 20, false); r.Component = "other"; return r }()}, 0, true},
403 } {
404 t.Run(tc.name, func(t *testing.T) {
405 path := filepath.Join(t.TempDir(), "trace.jsonl")
406 var lines []string
407 for _, record := range tc.records {
408 lines = append(lines, mustJSON(record))
409 }
410 if err := os.WriteFile(path, []byte(strings.Join(lines, "\n")+"\n"), 0o600); err != nil {
411 t.Fatal(err)
412 }
413 spans, err := mergeProcessTraces([]processMapping{{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}})
414 if tc.wantErr {
415 if err == nil {
416 t.Fatalf("invalid trace accepted: %+v", tc.records)
417 }
418 return
419 }
420 if err != nil || len(spans) != tc.wantSpans {
421 t.Fatalf("spans=%+v err=%v, want %d spans", spans, err, tc.wantSpans)
422 }
423 if tc.wantSpans == 1 && spans[0].Samples[0] != 10 {
424 t.Fatalf("successful span=%+v, want 10ns", spans[0])
425 }
426 })
427 }
428}