detached 1f305e9

fix(harness): preserve complete diagnostic evidence

commit
1f305e95de96bf19eb4453042da1f8e0de4d0ee7
author
brice · · 2026-07-13 09:27 UTC
parent
ba3714c
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
7 files · +304 -66
Split
UnifiedSplit
M cmd/vev-perf-harness/evidence.go +8 -2
@@ -35,7 +35,7 @@ func measuredEventSamples(marks []harnessMark, interval measuredInterval) ([]eve
3535 return nil, errors.New("negative harness duration")
3636 }
3737 delete(starts, m.Sequence)
38- if start >= interval.Start && m.Tick <= interval.End {
38+ if start >= interval.Start && m.Tick < interval.End {
3939 events = append(events, eventSample{Sequence: m.Sequence, Injected: start, Flushed: m.Tick, Latency: m.Tick - start})
4040 }
4141 default:
@@ -283,7 +283,13 @@ func writeJSON(path string, v any) error {
283283 return f.Close()
284284 }
285285
286-func mustJSON(v any) string { b, _ := json.Marshal(v); return string(b) }
286+func mustJSON(v any) string {
287+ b, err := json.Marshal(v)
288+ if err != nil {
289+ panic(fmt.Sprintf("marshal JSON: %v", err))
290+ }
291+ return string(b)
292+}
287293
288294 func safeName(s string) string {
289295 return strings.Map(func(r rune) rune {
@@ -35,7 +35,7 @@ func measuredEventSamples(marks []harnessMark, interval measuredInterval) ([]eve
35 return nil, errors.New("negative harness duration")35 return nil, errors.New("negative harness duration")
36 }36 }
37 delete(starts, m.Sequence)37 delete(starts, m.Sequence)
38- if start >= interval.Start && m.Tick <= interval.End {38+ if start >= interval.Start && m.Tick < interval.End {
39 events = append(events, eventSample{Sequence: m.Sequence, Injected: start, Flushed: m.Tick, Latency: m.Tick - start})39 events = append(events, eventSample{Sequence: m.Sequence, Injected: start, Flushed: m.Tick, Latency: m.Tick - start})
40 }40 }
41 default:41 default:
@@ -283,7 +283,13 @@ func writeJSON(path string, v any) error {
283 return f.Close()283 return f.Close()
284 }284 }
285 285
286-func mustJSON(v any) string { b, _ := json.Marshal(v); return string(b) }286+func mustJSON(v any) string {
287+ b, err := json.Marshal(v)
288+ if err != nil {
289+ panic(fmt.Sprintf("marshal JSON: %v", err))
290+ }
291+ return string(b)
292+}
287 293
288 func safeName(s string) string {294 func safeName(s string) string {
289 return strings.Map(func(r rune) rune {295 return strings.Map(func(r rune) rune {
M cmd/vev-perf-harness/evidence_test.go +33 -5
@@ -3,6 +3,7 @@ package main
33 import (
44 "bytes"
55 "encoding/json"
6+ "errors"
67 "os"
78 "path/filepath"
89 "strings"
@@ -99,6 +100,8 @@ func TestHarnessCollectsOnlyCompleteMeasuredIntervalEvents(t *testing.T) {
99100 {Sequence: 3, Kind: "terminal_flushed", Tick: 201, Valid: true}, // exits interval
100101 {Sequence: 4, Kind: "input_injected", Tick: 99, Valid: true},
101102 {Sequence: 4, Kind: "terminal_flushed", Tick: 110, Valid: true}, // enters interval
103+ {Sequence: 5, Kind: "input_injected", Tick: 180, Valid: true},
104+ {Sequence: 5, Kind: "terminal_flushed", Tick: 200, Valid: true}, // end is excluded
102105 }
103106 events, err := measuredEventSamples(marks, interval)
104107 if err != nil {
@@ -132,7 +135,11 @@ func TestHarnessRunRejectsStraddlingEventsAsInsufficient(t *testing.T) {
132135 if err != nil {
133136 t.Fatal(err)
134137 }
135- defer raw.Close()
138+ t.Cleanup(func() {
139+ if err := raw.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
140+ t.Error(err)
141+ }
142+ })
136143 clock := &fakeClock{}
137144 h := defaultHarness()
138145 h.clock, h.launcher = clock, insufficientEventLauncher{clock: clock}
@@ -140,6 +147,16 @@ func TestHarnessRunRejectsStraddlingEventsAsInsufficient(t *testing.T) {
140147 if err == nil || !strings.Contains(err.Error(), "insufficient in-interval event samples") {
141148 t.Fatalf("straddling run error=%v, want insufficient in-interval sample rejection", err)
142149 }
150+ if err := raw.Close(); err != nil {
151+ t.Fatal(err)
152+ }
153+ marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
154+ if got, want := len(marks), 4; got != want {
155+ t.Fatalf("raw marks=%d, want warmup and straddling pair=%d: %+v", got, want, marks)
156+ }
157+ if straddling := marks[len(marks)-1]; straddling.Sequence != 2 || straddling.Kind != "terminal_flushed" || !straddling.Valid {
158+ t.Fatalf("straddling boundary=%+v", straddling)
159+ }
143160 }
144161
145162 func TestHarnessCalculatesEventCadenceAndMaxGapFromRawPairs(t *testing.T) {
@@ -186,7 +203,11 @@ func TestHarnessClosesRolesBeforeMergingReceiveSpans(t *testing.T) {
186203 if err != nil {
187204 t.Fatal(err)
188205 }
189- defer raw.Close()
206+ t.Cleanup(func() {
207+ if err := raw.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
208+ t.Error(err)
209+ }
210+ })
190211 launcher := &closeTraceLauncher{}
191212 h := defaultHarness()
192213 h.clock, h.launcher = &fakeClock{}, launcher
@@ -282,10 +303,17 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) {
282303 t.Fatal(err)
283304 }
284305 for _, r := range records {
285- b, _ := json.Marshal(r)
286- _, _ = f.Write(append(b, '\n'))
306+ b, err := json.Marshal(r)
307+ if err != nil {
308+ t.Fatal(err)
309+ }
310+ if _, err := f.Write(append(b, '\n')); err != nil {
311+ t.Fatal(err)
312+ }
313+ }
314+ if err := f.Close(); err != nil {
315+ t.Fatal(err)
287316 }
288- _ = f.Close()
289317 return processMapping{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}
290318 }
291319 base := func(kind string, tick int64) traceRecord {
@@ -3,6 +3,7 @@ package main
3 import (3 import (
4 "bytes"4 "bytes"
5 "encoding/json"5 "encoding/json"
6+ "errors"
6 "os"7 "os"
7 "path/filepath"8 "path/filepath"
8 "strings"9 "strings"
@@ -99,6 +100,8 @@ func TestHarnessCollectsOnlyCompleteMeasuredIntervalEvents(t *testing.T) {
99 {Sequence: 3, Kind: "terminal_flushed", Tick: 201, Valid: true}, // exits interval100 {Sequence: 3, Kind: "terminal_flushed", Tick: 201, Valid: true}, // exits interval
100 {Sequence: 4, Kind: "input_injected", Tick: 99, Valid: true},101 {Sequence: 4, Kind: "input_injected", Tick: 99, Valid: true},
101 {Sequence: 4, Kind: "terminal_flushed", Tick: 110, Valid: true}, // enters interval102 {Sequence: 4, Kind: "terminal_flushed", Tick: 110, Valid: true}, // enters interval
103+ {Sequence: 5, Kind: "input_injected", Tick: 180, Valid: true},
104+ {Sequence: 5, Kind: "terminal_flushed", Tick: 200, Valid: true}, // end is excluded
102 }105 }
103 events, err := measuredEventSamples(marks, interval)106 events, err := measuredEventSamples(marks, interval)
104 if err != nil {107 if err != nil {
@@ -132,7 +135,11 @@ func TestHarnessRunRejectsStraddlingEventsAsInsufficient(t *testing.T) {
132 if err != nil {135 if err != nil {
133 t.Fatal(err)136 t.Fatal(err)
134 }137 }
135- defer raw.Close()138+ t.Cleanup(func() {
139+ if err := raw.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
140+ t.Error(err)
141+ }
142+ })
136 clock := &fakeClock{}143 clock := &fakeClock{}
137 h := defaultHarness()144 h := defaultHarness()
138 h.clock, h.launcher = clock, insufficientEventLauncher{clock: clock}145 h.clock, h.launcher = clock, insufficientEventLauncher{clock: clock}
@@ -140,6 +147,16 @@ func TestHarnessRunRejectsStraddlingEventsAsInsufficient(t *testing.T) {
140 if err == nil || !strings.Contains(err.Error(), "insufficient in-interval event samples") {147 if err == nil || !strings.Contains(err.Error(), "insufficient in-interval event samples") {
141 t.Fatalf("straddling run error=%v, want insufficient in-interval sample rejection", err)148 t.Fatalf("straddling run error=%v, want insufficient in-interval sample rejection", err)
142 }149 }
150+ if err := raw.Close(); err != nil {
151+ t.Fatal(err)
152+ }
153+ marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
154+ if got, want := len(marks), 4; got != want {
155+ t.Fatalf("raw marks=%d, want warmup and straddling pair=%d: %+v", got, want, marks)
156+ }
157+ if straddling := marks[len(marks)-1]; straddling.Sequence != 2 || straddling.Kind != "terminal_flushed" || !straddling.Valid {
158+ t.Fatalf("straddling boundary=%+v", straddling)
159+ }
143 }160 }
144 161
145 func TestHarnessCalculatesEventCadenceAndMaxGapFromRawPairs(t *testing.T) {162 func TestHarnessCalculatesEventCadenceAndMaxGapFromRawPairs(t *testing.T) {
@@ -186,7 +203,11 @@ func TestHarnessClosesRolesBeforeMergingReceiveSpans(t *testing.T) {
186 if err != nil {203 if err != nil {
187 t.Fatal(err)204 t.Fatal(err)
188 }205 }
189- defer raw.Close()206+ t.Cleanup(func() {
207+ if err := raw.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
208+ t.Error(err)
209+ }
210+ })
190 launcher := &closeTraceLauncher{}211 launcher := &closeTraceLauncher{}
191 h := defaultHarness()212 h := defaultHarness()
192 h.clock, h.launcher = &fakeClock{}, launcher213 h.clock, h.launcher = &fakeClock{}, launcher
@@ -282,10 +303,17 @@ func TestHarnessRejectsCrossProcessAndBadTraceSpans(t *testing.T) {
282 t.Fatal(err)303 t.Fatal(err)
283 }304 }
284 for _, r := range records {305 for _, r := range records {
285- b, _ := json.Marshal(r)306+ b, err := json.Marshal(r)
286- _, _ = f.Write(append(b, '\n'))307+ if err != nil {
308+ t.Fatal(err)
309+ }
310+ if _, err := f.Write(append(b, '\n')); err != nil {
311+ t.Fatal(err)
312+ }
313+ }
314+ if err := f.Close(); err != nil {
315+ t.Fatal(err)
287 }316 }
288- _ = f.Close()
289 return processMapping{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}317 return processMapping{ProcessID: "one", ClockDomain: "one", TracePath: path, Scenario: "s", Run: 1}
290 }318 }
291 base := func(kind string, tick int64) traceRecord {319 base := func(kind string, tick int64) traceRecord {
M cmd/vev-perf-harness/runner.go +42 -24
@@ -2,6 +2,7 @@ package main
22
33 import (
44 "context"
5+ "encoding/json"
56 "errors"
67 "fmt"
78 "io"
@@ -58,8 +59,7 @@ func createExclusiveRunDir(path string) error {
5859 return err
5960 }
6061 if err := safedir.EnsurePrivate(path); err != nil {
61- _ = os.Remove(path)
62- return err
62+ return errors.Join(err, os.Remove(path))
6363 }
6464 return nil
6565 }
@@ -175,7 +175,9 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
175175 success := false
176176 defer func() {
177177 if !success {
178- _ = h.removeAll(dir)
178+ if cleanupErr := h.removeAll(dir); cleanupErr != nil {
179+ err = errors.Join(err, fmt.Errorf("remove failed run directory: %w", cleanupErr))
180+ }
179181 }
180182 }()
181183 maps := make([]processMapping, 0, len(s.Roles))
@@ -239,22 +241,38 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
239241 }
240242 }
241243 }
244+ marks := []harnessMark{}
245+ recordMark := func(mark harnessMark) error {
246+ marks = append(marks, mark)
247+ if err := json.NewEncoder(raw).Encode(mark); err != nil {
248+ return fmt.Errorf("write raw harness mark: %w", err)
249+ }
250+ return nil
251+ }
252+ seq := uint64(0)
242253 for _, p := range processes {
243254 if p.mapping.Role != "client" {
244255 continue
245256 }
246- if err = p.process.Warmup(workloadInput(s, run, "warmup")); err != nil {
257+ seq++
258+ sequence := seq
259+ if err = recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "input_injected", Tick: h.clock.Now(), Valid: true}); err != nil {
247260 return res, err
248261 }
262+ warmupErr := p.process.Warmup(workloadInput(s, run, "warmup"))
263+ if markErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: warmupErr == nil}); markErr != nil {
264+ return res, errors.Join(warmupErr, markErr)
265+ }
266+ if warmupErr != nil {
267+ return res, warmupErr
268+ }
249269 }
250270 // The harness clock, rather than a process clock, owns the explicit warmup
251- // and measured intervals. Warmup boundaries are intentionally never added to
252- // marks, and each measured event must complete before interval.End to count.
271+ // and measured intervals. Raw evidence retains every boundary, while only
272+ // complete valid pairs within [Start, End) contribute measurement samples.
253273 h.clock.Sleep(o.warmup)
254274 interval := measuredInterval{Start: h.clock.Now()}
255275 interval.End = interval.Start + o.duration.Nanoseconds()
256- marks := []harnessMark{}
257- seq := uint64(0)
258276 nextInjection := interval.Start
259277 for {
260278 now := h.clock.Now()
@@ -271,16 +289,28 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
271289 }
272290 seq++
273291 sequence := seq
292+ injectedRecorded := false
293+ flushedRecorded := false
274294 injected := func() error {
275- marks = append(marks, harnessMark{s.ID, run, sequence, "input_injected", h.clock.Now(), true})
295+ if markErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "input_injected", Tick: h.clock.Now(), Valid: true}); markErr != nil {
296+ return markErr
297+ }
298+ injectedRecorded = true
276299 return nil
277300 }
278301 flushed := func() error {
279- marks = append(marks, harnessMark{s.ID, run, sequence, "terminal_flushed", h.clock.Now(), true})
302+ if markErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: true}); markErr != nil {
303+ return markErr
304+ }
305+ flushedRecorded = true
280306 return nil
281307 }
282- if err = p.process.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); err != nil {
283- return res, err
308+ if measureErr := p.process.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); measureErr != nil {
309+ if injectedRecorded && !flushedRecorded {
310+ diagnosticErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: false})
311+ return res, errors.Join(measureErr, diagnosticErr)
312+ }
313+ return res, measureErr
284314 }
285315 }
286316 // Cap injection rate rather than bursting events after a slow terminal
@@ -308,18 +338,6 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
308338 if e != nil {
309339 return res, e
310340 }
311- inInterval := make(map[uint64]bool, len(events))
312- for _, event := range events {
313- inInterval[event.Sequence] = true
314- }
315- for _, m := range marks {
316- if !inInterval[m.Sequence] {
317- continue
318- }
319- if _, e := fmt.Fprintf(raw, "%s\n", mustJSON(m)); e != nil {
320- return res, e
321- }
322- }
323341 success = true
324342 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
325343 }
@@ -2,6 +2,7 @@ package main
2 2
3 import (3 import (
4 "context"4 "context"
5+ "encoding/json"
5 "errors"6 "errors"
6 "fmt"7 "fmt"
7 "io"8 "io"
@@ -58,8 +59,7 @@ func createExclusiveRunDir(path string) error {
58 return err59 return err
59 }60 }
60 if err := safedir.EnsurePrivate(path); err != nil {61 if err := safedir.EnsurePrivate(path); err != nil {
61- _ = os.Remove(path)62+ return errors.Join(err, os.Remove(path))
62- return err
63 }63 }
64 return nil64 return nil
65 }65 }
@@ -175,7 +175,9 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
175 success := false175 success := false
176 defer func() {176 defer func() {
177 if !success {177 if !success {
178- _ = h.removeAll(dir)178+ if cleanupErr := h.removeAll(dir); cleanupErr != nil {
179+ err = errors.Join(err, fmt.Errorf("remove failed run directory: %w", cleanupErr))
180+ }
179 }181 }
180 }()182 }()
181 maps := make([]processMapping, 0, len(s.Roles))183 maps := make([]processMapping, 0, len(s.Roles))
@@ -239,22 +241,38 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
239 }241 }
240 }242 }
241 }243 }
244+ marks := []harnessMark{}
245+ recordMark := func(mark harnessMark) error {
246+ marks = append(marks, mark)
247+ if err := json.NewEncoder(raw).Encode(mark); err != nil {
248+ return fmt.Errorf("write raw harness mark: %w", err)
249+ }
250+ return nil
251+ }
252+ seq := uint64(0)
242 for _, p := range processes {253 for _, p := range processes {
243 if p.mapping.Role != "client" {254 if p.mapping.Role != "client" {
244 continue255 continue
245 }256 }
246- if err = p.process.Warmup(workloadInput(s, run, "warmup")); err != nil {257+ seq++
258+ sequence := seq
259+ if err = recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "input_injected", Tick: h.clock.Now(), Valid: true}); err != nil {
247 return res, err260 return res, err
248 }261 }
262+ warmupErr := p.process.Warmup(workloadInput(s, run, "warmup"))
263+ if markErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: warmupErr == nil}); markErr != nil {
264+ return res, errors.Join(warmupErr, markErr)
265+ }
266+ if warmupErr != nil {
267+ return res, warmupErr
268+ }
249 }269 }
250 // The harness clock, rather than a process clock, owns the explicit warmup270 // The harness clock, rather than a process clock, owns the explicit warmup
251- // and measured intervals. Warmup boundaries are intentionally never added to271+ // and measured intervals. Raw evidence retains every boundary, while only
252- // marks, and each measured event must complete before interval.End to count.272+ // complete valid pairs within [Start, End) contribute measurement samples.
253 h.clock.Sleep(o.warmup)273 h.clock.Sleep(o.warmup)
254 interval := measuredInterval{Start: h.clock.Now()}274 interval := measuredInterval{Start: h.clock.Now()}
255 interval.End = interval.Start + o.duration.Nanoseconds()275 interval.End = interval.Start + o.duration.Nanoseconds()
256- marks := []harnessMark{}
257- seq := uint64(0)
258 nextInjection := interval.Start276 nextInjection := interval.Start
259 for {277 for {
260 now := h.clock.Now()278 now := h.clock.Now()
@@ -271,16 +289,28 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
271 }289 }
272 seq++290 seq++
273 sequence := seq291 sequence := seq
292+ injectedRecorded := false
293+ flushedRecorded := false
274 injected := func() error {294 injected := func() error {
275- marks = append(marks, harnessMark{s.ID, run, sequence, "input_injected", h.clock.Now(), true})295+ if markErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "input_injected", Tick: h.clock.Now(), Valid: true}); markErr != nil {
296+ return markErr
297+ }
298+ injectedRecorded = true
276 return nil299 return nil
277 }300 }
278 flushed := func() error {301 flushed := func() error {
279- marks = append(marks, harnessMark{s.ID, run, sequence, "terminal_flushed", h.clock.Now(), true})302+ if markErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: true}); markErr != nil {
303+ return markErr
304+ }
305+ flushedRecorded = true
280 return nil306 return nil
281 }307 }
282- if err = p.process.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); err != nil {308+ if measureErr := p.process.Measure(workloadInput(s, run, fmt.Sprintf("measured-%d", sequence)), injected, flushed); measureErr != nil {
283- return res, err309+ if injectedRecorded && !flushedRecorded {
310+ diagnosticErr := recordMark(harnessMark{Scenario: s.ID, Run: run, Sequence: sequence, Kind: "terminal_flushed", Tick: h.clock.Now(), Valid: false})
311+ return res, errors.Join(measureErr, diagnosticErr)
312+ }
313+ return res, measureErr
284 }314 }
285 }315 }
286 // Cap injection rate rather than bursting events after a slow terminal316 // Cap injection rate rather than bursting events after a slow terminal
@@ -308,18 +338,6 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr
308 if e != nil {338 if e != nil {
309 return res, e339 return res, e
310 }340 }
311- inInterval := make(map[uint64]bool, len(events))
312- for _, event := range events {
313- inInterval[event.Sequence] = true
314- }
315- for _, m := range marks {
316- if !inInterval[m.Sequence] {
317- continue
318- }
319- if _, e := fmt.Fprintf(raw, "%s\n", mustJSON(m)); e != nil {
320- return res, e
321- }
322- }
323 success = true341 success = true
324 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}, nil342 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
325 }343 }
M cmd/vev-perf-harness/runner_test.go +67 -10
@@ -45,12 +45,12 @@ func TestHarnessCanonicalLocalRolesAreIsolatedAcrossRepetitions(t *testing.T) {
4545 if err != nil {
4646 t.Fatal(err)
4747 }
48- defer os.RemoveAll(out)
48+ removeTestTree(t, out)
4949 raw, err := os.OpenFile(filepath.Join(out, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
5050 if err != nil {
5151 t.Fatal(err)
5252 }
53- defer raw.Close()
53+ closeTestFile(t, raw)
5454 h := defaultHarness()
5555 h.clock = &fakeClock{}
5656 h.launcher = &cliLauncher{bin: bin}
@@ -118,7 +118,7 @@ func TestHarnessResolvesCanonicalRelativePathsBeforeRoleCWDChanges(t *testing.T)
118118 if err != nil {
119119 t.Fatal(err)
120120 }
121- t.Cleanup(func() { _ = os.RemoveAll(workspace) })
121+ removeTestTree(t, workspace)
122122 bin := filepath.Join(workspace, "vev")
123123 build := exec.Command("go", "build", "-o", bin, "./")
124124 build.Dir = root
@@ -198,7 +198,7 @@ func TestHarnessRejectsExistingRunDirectoryBeforeLaunchingRoles(t *testing.T) {
198198 if err != nil {
199199 t.Fatal(err)
200200 }
201- defer raw.Close()
201+ closeTestFile(t, raw)
202202 s := scenario{ID: "isolated", Roles: []string{"daemon", "client"}}
203203 stale := filepath.Join(out, "isolated-run-001")
204204 if err := os.Mkdir(stale, 0o700); err != nil {
@@ -282,7 +282,7 @@ func TestHarnessFakeRunnerRoutesClientToPeerAndCleansEveryRole(t *testing.T) {
282282 if err != nil {
283283 t.Fatal(err)
284284 }
285- defer raw.Close()
285+ closeTestFile(t, raw)
286286 l := &fakeLauncher{}
287287 h := defaultHarness()
288288 h.clock, h.launcher = &fakeClock{}, l
@@ -320,13 +320,70 @@ func TestHarnessAggregatesRawEventsSeparatelyFromRunDispersion(t *testing.T) {
320320 }
321321 }
322322
323+func TestHarnessRawRecordsContainEveryMarkWhileSamplesRemainFiltered(t *testing.T) {
324+ dir := t.TempDir()
325+ raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
326+ if err != nil {
327+ t.Fatal(err)
328+ }
329+ h := defaultHarness()
330+ h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
331+ result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "raw-complete", Roles: []string{"daemon", "client"}}, 1, raw)
332+ if err != nil {
333+ t.Fatal(err)
334+ }
335+ if err := raw.Close(); err != nil {
336+ t.Fatal(err)
337+ }
338+ marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
339+ if got, want := len(marks), 2*(result.Samples+1); got != want {
340+ t.Fatalf("raw marks=%d, want warmup plus every measured boundary=%d", got, want)
341+ }
342+ for i, mark := range marks[:2] {
343+ if mark.Sequence != 1 || !mark.Valid || []string{"input_injected", "terminal_flushed"}[i] != mark.Kind {
344+ t.Fatalf("warmup mark %d=%+v", i, mark)
345+ }
346+ }
347+ for sequence := uint64(2); sequence <= uint64(result.Samples+1); sequence++ {
348+ if marks[2*(sequence-1)].Sequence != sequence || marks[2*(sequence-1)+1].Sequence != sequence {
349+ t.Fatalf("raw marks did not retain both boundaries for sequence %d: %+v", sequence, marks)
350+ }
351+ }
352+}
353+
354+func TestHarnessPersistsFailedFlushDiagnostics(t *testing.T) {
355+ dir := t.TempDir()
356+ raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
357+ if err != nil {
358+ t.Fatal(err)
359+ }
360+ failure := errors.New("flush failed")
361+ h := defaultHarness()
362+ h.clock, h.launcher = &fakeClock{}, failedFlushLauncher{err: failure}
363+ _, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "failed-flush", Roles: []string{"daemon", "client"}}, 1, raw)
364+ if !errors.Is(err, failure) {
365+ t.Fatalf("run error=%v, want failed flush", err)
366+ }
367+ if err := raw.Close(); err != nil {
368+ t.Fatal(err)
369+ }
370+ marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
371+ if got, want := len(marks), 4; got != want {
372+ t.Fatalf("raw marks=%d, want warmup and failed measured pair=%d: %+v", got, want, marks)
373+ }
374+ failed := marks[len(marks)-1]
375+ if failed.Sequence != 2 || failed.Kind != "terminal_flushed" || failed.Valid {
376+ t.Fatalf("failed flush diagnostic=%+v", failed)
377+ }
378+}
379+
323380 func TestHarnessSchedulesMeasuredEventsAcrossFullInterval(t *testing.T) {
324381 dir := t.TempDir()
325382 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
326383 if err != nil {
327384 t.Fatal(err)
328385 }
329- defer raw.Close()
386+ closeTestFile(t, raw)
330387 h := defaultHarness()
331388 h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
332389 result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "schedule", Roles: []string{"daemon", "client"}}, 1, raw)
@@ -343,8 +400,8 @@ func TestHarnessSchedulesMeasuredEventsAcrossFullInterval(t *testing.T) {
343400 if err != nil {
344401 t.Fatal(err)
345402 }
346- if len(records) != result.Samples*2 {
347- t.Fatalf("raw records=%d, want %d complete measured boundaries", len(records), result.Samples*2)
403+ if want := result.Samples*2 + 2; len(records) != want {
404+ t.Fatalf("raw records=%d, want %d warmup and measured boundaries", len(records), want)
348405 }
349406 }
350407
@@ -354,7 +411,7 @@ func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) {
354411 if err != nil {
355412 t.Fatal(err)
356413 }
357- defer raw.Close()
414+ closeTestFile(t, raw)
358415 primary := errors.New("primary workload failure")
359416 cleanup := errors.New("daemon cleanup failure")
360417 launcher := &fakeLauncher{
@@ -383,7 +440,7 @@ func TestHarnessCleansRunDirectoryOnProcessFailure(t *testing.T) {
383440 if err != nil {
384441 t.Fatal(err)
385442 }
386- defer raw.Close()
443+ closeTestFile(t, raw)
387444 launcher := &fakeLauncher{}
388445 h := defaultHarness()
389446 h.clock = &fakeClock{}
@@ -45,12 +45,12 @@ func TestHarnessCanonicalLocalRolesAreIsolatedAcrossRepetitions(t *testing.T) {
45 if err != nil {45 if err != nil {
46 t.Fatal(err)46 t.Fatal(err)
47 }47 }
48- defer os.RemoveAll(out)48+ removeTestTree(t, out)
49 raw, err := os.OpenFile(filepath.Join(out, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)49 raw, err := os.OpenFile(filepath.Join(out, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
50 if err != nil {50 if err != nil {
51 t.Fatal(err)51 t.Fatal(err)
52 }52 }
53- defer raw.Close()53+ closeTestFile(t, raw)
54 h := defaultHarness()54 h := defaultHarness()
55 h.clock = &fakeClock{}55 h.clock = &fakeClock{}
56 h.launcher = &cliLauncher{bin: bin}56 h.launcher = &cliLauncher{bin: bin}
@@ -118,7 +118,7 @@ func TestHarnessResolvesCanonicalRelativePathsBeforeRoleCWDChanges(t *testing.T)
118 if err != nil {118 if err != nil {
119 t.Fatal(err)119 t.Fatal(err)
120 }120 }
121- t.Cleanup(func() { _ = os.RemoveAll(workspace) })121+ removeTestTree(t, workspace)
122 bin := filepath.Join(workspace, "vev")122 bin := filepath.Join(workspace, "vev")
123 build := exec.Command("go", "build", "-o", bin, "./")123 build := exec.Command("go", "build", "-o", bin, "./")
124 build.Dir = root124 build.Dir = root
@@ -198,7 +198,7 @@ func TestHarnessRejectsExistingRunDirectoryBeforeLaunchingRoles(t *testing.T) {
198 if err != nil {198 if err != nil {
199 t.Fatal(err)199 t.Fatal(err)
200 }200 }
201- defer raw.Close()201+ closeTestFile(t, raw)
202 s := scenario{ID: "isolated", Roles: []string{"daemon", "client"}}202 s := scenario{ID: "isolated", Roles: []string{"daemon", "client"}}
203 stale := filepath.Join(out, "isolated-run-001")203 stale := filepath.Join(out, "isolated-run-001")
204 if err := os.Mkdir(stale, 0o700); err != nil {204 if err := os.Mkdir(stale, 0o700); err != nil {
@@ -282,7 +282,7 @@ func TestHarnessFakeRunnerRoutesClientToPeerAndCleansEveryRole(t *testing.T) {
282 if err != nil {282 if err != nil {
283 t.Fatal(err)283 t.Fatal(err)
284 }284 }
285- defer raw.Close()285+ closeTestFile(t, raw)
286 l := &fakeLauncher{}286 l := &fakeLauncher{}
287 h := defaultHarness()287 h := defaultHarness()
288 h.clock, h.launcher = &fakeClock{}, l288 h.clock, h.launcher = &fakeClock{}, l
@@ -320,13 +320,70 @@ func TestHarnessAggregatesRawEventsSeparatelyFromRunDispersion(t *testing.T) {
320 }320 }
321 }321 }
322 322
323+func TestHarnessRawRecordsContainEveryMarkWhileSamplesRemainFiltered(t *testing.T) {
324+ dir := t.TempDir()
325+ raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
326+ if err != nil {
327+ t.Fatal(err)
328+ }
329+ h := defaultHarness()
330+ h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
331+ result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "raw-complete", Roles: []string{"daemon", "client"}}, 1, raw)
332+ if err != nil {
333+ t.Fatal(err)
334+ }
335+ if err := raw.Close(); err != nil {
336+ t.Fatal(err)
337+ }
338+ marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
339+ if got, want := len(marks), 2*(result.Samples+1); got != want {
340+ t.Fatalf("raw marks=%d, want warmup plus every measured boundary=%d", got, want)
341+ }
342+ for i, mark := range marks[:2] {
343+ if mark.Sequence != 1 || !mark.Valid || []string{"input_injected", "terminal_flushed"}[i] != mark.Kind {
344+ t.Fatalf("warmup mark %d=%+v", i, mark)
345+ }
346+ }
347+ for sequence := uint64(2); sequence <= uint64(result.Samples+1); sequence++ {
348+ if marks[2*(sequence-1)].Sequence != sequence || marks[2*(sequence-1)+1].Sequence != sequence {
349+ t.Fatalf("raw marks did not retain both boundaries for sequence %d: %+v", sequence, marks)
350+ }
351+ }
352+}
353+
354+func TestHarnessPersistsFailedFlushDiagnostics(t *testing.T) {
355+ dir := t.TempDir()
356+ raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
357+ if err != nil {
358+ t.Fatal(err)
359+ }
360+ failure := errors.New("flush failed")
361+ h := defaultHarness()
362+ h.clock, h.launcher = &fakeClock{}, failedFlushLauncher{err: failure}
363+ _, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "failed-flush", Roles: []string{"daemon", "client"}}, 1, raw)
364+ if !errors.Is(err, failure) {
365+ t.Fatalf("run error=%v, want failed flush", err)
366+ }
367+ if err := raw.Close(); err != nil {
368+ t.Fatal(err)
369+ }
370+ marks := readHarnessMarks(t, filepath.Join(dir, "raw.jsonl"))
371+ if got, want := len(marks), 4; got != want {
372+ t.Fatalf("raw marks=%d, want warmup and failed measured pair=%d: %+v", got, want, marks)
373+ }
374+ failed := marks[len(marks)-1]
375+ if failed.Sequence != 2 || failed.Kind != "terminal_flushed" || failed.Valid {
376+ t.Fatalf("failed flush diagnostic=%+v", failed)
377+ }
378+}
379+
323 func TestHarnessSchedulesMeasuredEventsAcrossFullInterval(t *testing.T) {380 func TestHarnessSchedulesMeasuredEventsAcrossFullInterval(t *testing.T) {
324 dir := t.TempDir()381 dir := t.TempDir()
325 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)382 raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600)
326 if err != nil {383 if err != nil {
327 t.Fatal(err)384 t.Fatal(err)
328 }385 }
329- defer raw.Close()386+ closeTestFile(t, raw)
330 h := defaultHarness()387 h := defaultHarness()
331 h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}388 h.clock, h.launcher = &fakeClock{}, &fakeLauncher{}
332 result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "schedule", Roles: []string{"daemon", "client"}}, 1, raw)389 result, err := h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration}, manifest{}, scenario{ID: "schedule", Roles: []string{"daemon", "client"}}, 1, raw)
@@ -343,8 +400,8 @@ func TestHarnessSchedulesMeasuredEventsAcrossFullInterval(t *testing.T) {
343 if err != nil {400 if err != nil {
344 t.Fatal(err)401 t.Fatal(err)
345 }402 }
346- if len(records) != result.Samples*2 {403+ if want := result.Samples*2 + 2; len(records) != want {
347- t.Fatalf("raw records=%d, want %d complete measured boundaries", len(records), result.Samples*2)404+ t.Fatalf("raw records=%d, want %d warmup and measured boundaries", len(records), want)
348 }405 }
349 }406 }
350 407
@@ -354,7 +411,7 @@ func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) {
354 if err != nil {411 if err != nil {
355 t.Fatal(err)412 t.Fatal(err)
356 }413 }
357- defer raw.Close()414+ closeTestFile(t, raw)
358 primary := errors.New("primary workload failure")415 primary := errors.New("primary workload failure")
359 cleanup := errors.New("daemon cleanup failure")416 cleanup := errors.New("daemon cleanup failure")
360 launcher := &fakeLauncher{417 launcher := &fakeLauncher{
@@ -383,7 +440,7 @@ func TestHarnessCleansRunDirectoryOnProcessFailure(t *testing.T) {
383 if err != nil {440 if err != nil {
384 t.Fatal(err)441 t.Fatal(err)
385 }442 }
386- defer raw.Close()443+ closeTestFile(t, raw)
387 launcher := &fakeLauncher{}444 launcher := &fakeLauncher{}
388 h := defaultHarness()445 h := defaultHarness()
389 h.clock = &fakeClock{}446 h.clock = &fakeClock{}
M cmd/vev-perf-harness/test_helpers_test.go +84 -9
@@ -4,8 +4,10 @@ import (
44 "bytes"
55 "encoding/json"
66 "errors"
7+ "fmt"
78 "os"
89 "path/filepath"
10+ "testing"
911 "time"
1012 )
1113
@@ -100,6 +102,65 @@ func (l insufficientEventLauncher) Launch(m processMapping, _ roleCommand) (laun
100102 return &fakeProcess{}, nil
101103 }
102104
105+type failedFlushProcess struct{ err error }
106+
107+func (*failedFlushProcess) Warmup([]byte) error { return nil }
108+
109+func (p *failedFlushProcess) Measure(_ []byte, injected, _ func() error) error {
110+ if err := injected(); err != nil {
111+ return err
112+ }
113+ return p.err
114+}
115+
116+func (*failedFlushProcess) Close() error { return nil }
117+
118+type failedFlushLauncher struct{ err error }
119+
120+func (l failedFlushLauncher) Launch(m processMapping, _ roleCommand) (launchedProcess, error) {
121+ if m.Role == "client" {
122+ return &failedFlushProcess{err: l.err}, nil
123+ }
124+ return &fakeProcess{}, nil
125+}
126+
127+func closeTestFile(t *testing.T, f *os.File) {
128+ t.Helper()
129+ t.Cleanup(func() {
130+ if err := f.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
131+ t.Error(err)
132+ }
133+ })
134+}
135+
136+func removeTestTree(t *testing.T, path string) {
137+ t.Helper()
138+ t.Cleanup(func() {
139+ if err := os.RemoveAll(path); err != nil {
140+ t.Error(err)
141+ }
142+ })
143+}
144+
145+func readHarnessMarks(t *testing.T, path string) []harnessMark {
146+ t.Helper()
147+ records, err := readJSONL(path)
148+ if err != nil {
149+ t.Fatal(err)
150+ }
151+ marks := make([]harnessMark, len(records))
152+ for i, record := range records {
153+ encoded, err := json.Marshal(record)
154+ if err != nil {
155+ t.Fatal(err)
156+ }
157+ if err := json.Unmarshal(encoded, &marks[i]); err != nil {
158+ t.Fatal(err)
159+ }
160+ }
161+ return marks
162+}
163+
103164 type fakePTY struct{ writes [][]byte }
104165
105166 func (p *fakePTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
@@ -175,7 +236,11 @@ func (l *failingLauncher) Launch(m processMapping, command roleCommand) (launche
175236 return nil, err
176237 }
177238 if m.Role == "client" {
178- p.(*fakeProcess).fail = true
239+ fake, ok := p.(*fakeProcess)
240+ if !ok {
241+ return nil, fmt.Errorf("fake launcher returned %T, want *fakeProcess", p)
242+ }
243+ fake.fail = true
179244 }
180245 return p, nil
181246 }
@@ -203,15 +268,20 @@ func (l *traceLauncher) Launch(m processMapping, command roleCommand) (launchedP
203268 if err != nil {
204269 return nil, err
205270 }
206- defer f.Close()
207271 for i, pair := range spanPairs {
208272 for n, kind := range []string{pair.start, pair.end} {
209- b, _ := json.Marshal(traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: uint64(i + 1), RequestID: 1, Epoch: 1, Kind: kind, Tick: int64(i*20 + n*5), Valid: true})
273+ b, err := json.Marshal(traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: uint64(i + 1), RequestID: 1, Epoch: 1, Kind: kind, Tick: int64(i*20 + n*5), Valid: true})
274+ if err != nil {
275+ return nil, errors.Join(err, f.Close())
276+ }
210277 if _, err := f.Write(append(b, '\n')); err != nil {
211- return nil, err
278+ return nil, errors.Join(err, f.Close())
212279 }
213280 }
214281 }
282+ if err := f.Close(); err != nil {
283+ return nil, err
284+ }
215285 return p, nil
216286 }
217287
@@ -236,7 +306,11 @@ func (l *closeTraceLauncher) Launch(m processMapping, command roleCommand) (laun
236306 if err := appendTraceRecord(m.TracePath, start); err != nil {
237307 return nil, err
238308 }
239- return &closeTraceProcess{fakeProcess: p.(*fakeProcess), traceEnd: func() error {
309+ fake, ok := p.(*fakeProcess)
310+ if !ok {
311+ return nil, fmt.Errorf("fake launcher returned %T, want *fakeProcess", p)
312+ }
313+ return &closeTraceProcess{fakeProcess: fake, traceEnd: func() error {
240314 return appendTraceRecord(m.TracePath, traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_end", Tick: 20, Valid: true})
241315 }}, nil
242316 }
@@ -246,13 +320,14 @@ func appendTraceRecord(path string, record traceRecord) error {
246320 if err != nil {
247321 return err
248322 }
249- defer f.Close()
250323 b, err := json.Marshal(record)
251324 if err != nil {
252- return err
325+ return errors.Join(err, f.Close())
326+ }
327+ if _, err := f.Write(append(b, '\n')); err != nil {
328+ return errors.Join(err, f.Close())
253329 }
254- _, err = f.Write(append(b, '\n'))
255- return err
330+ return f.Close()
256331 }
257332
258333 // boundedLocalConcurrentTrace is an excerpt from the second run of the
@@ -4,8 +4,10 @@ import (
4 "bytes"4 "bytes"
5 "encoding/json"5 "encoding/json"
6 "errors"6 "errors"
7+ "fmt"
7 "os"8 "os"
8 "path/filepath"9 "path/filepath"
10+ "testing"
9 "time"11 "time"
10 )12 )
11 13
@@ -100,6 +102,65 @@ func (l insufficientEventLauncher) Launch(m processMapping, _ roleCommand) (laun
100 return &fakeProcess{}, nil102 return &fakeProcess{}, nil
101 }103 }
102 104
105+type failedFlushProcess struct{ err error }
106+
107+func (*failedFlushProcess) Warmup([]byte) error { return nil }
108+
109+func (p *failedFlushProcess) Measure(_ []byte, injected, _ func() error) error {
110+ if err := injected(); err != nil {
111+ return err
112+ }
113+ return p.err
114+}
115+
116+func (*failedFlushProcess) Close() error { return nil }
117+
118+type failedFlushLauncher struct{ err error }
119+
120+func (l failedFlushLauncher) Launch(m processMapping, _ roleCommand) (launchedProcess, error) {
121+ if m.Role == "client" {
122+ return &failedFlushProcess{err: l.err}, nil
123+ }
124+ return &fakeProcess{}, nil
125+}
126+
127+func closeTestFile(t *testing.T, f *os.File) {
128+ t.Helper()
129+ t.Cleanup(func() {
130+ if err := f.Close(); err != nil && !errors.Is(err, os.ErrClosed) {
131+ t.Error(err)
132+ }
133+ })
134+}
135+
136+func removeTestTree(t *testing.T, path string) {
137+ t.Helper()
138+ t.Cleanup(func() {
139+ if err := os.RemoveAll(path); err != nil {
140+ t.Error(err)
141+ }
142+ })
143+}
144+
145+func readHarnessMarks(t *testing.T, path string) []harnessMark {
146+ t.Helper()
147+ records, err := readJSONL(path)
148+ if err != nil {
149+ t.Fatal(err)
150+ }
151+ marks := make([]harnessMark, len(records))
152+ for i, record := range records {
153+ encoded, err := json.Marshal(record)
154+ if err != nil {
155+ t.Fatal(err)
156+ }
157+ if err := json.Unmarshal(encoded, &marks[i]); err != nil {
158+ t.Fatal(err)
159+ }
160+ }
161+ return marks
162+}
163+
103 type fakePTY struct{ writes [][]byte }164 type fakePTY struct{ writes [][]byte }
104 165
105 func (p *fakePTY) Read([]byte) (int, error) { return 0, os.ErrClosed }166 func (p *fakePTY) Read([]byte) (int, error) { return 0, os.ErrClosed }
@@ -175,7 +236,11 @@ func (l *failingLauncher) Launch(m processMapping, command roleCommand) (launche
175 return nil, err236 return nil, err
176 }237 }
177 if m.Role == "client" {238 if m.Role == "client" {
178- p.(*fakeProcess).fail = true239+ fake, ok := p.(*fakeProcess)
240+ if !ok {
241+ return nil, fmt.Errorf("fake launcher returned %T, want *fakeProcess", p)
242+ }
243+ fake.fail = true
179 }244 }
180 return p, nil245 return p, nil
181 }246 }
@@ -203,15 +268,20 @@ func (l *traceLauncher) Launch(m processMapping, command roleCommand) (launchedP
203 if err != nil {268 if err != nil {
204 return nil, err269 return nil, err
205 }270 }
206- defer f.Close()
207 for i, pair := range spanPairs {271 for i, pair := range spanPairs {
208 for n, kind := range []string{pair.start, pair.end} {272 for n, kind := range []string{pair.start, pair.end} {
209- b, _ := json.Marshal(traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: uint64(i + 1), RequestID: 1, Epoch: 1, Kind: kind, Tick: int64(i*20 + n*5), Valid: true})273+ b, err := json.Marshal(traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: uint64(i + 1), RequestID: 1, Epoch: 1, Kind: kind, Tick: int64(i*20 + n*5), Valid: true})
274+ if err != nil {
275+ return nil, errors.Join(err, f.Close())
276+ }
210 if _, err := f.Write(append(b, '\n')); err != nil {277 if _, err := f.Write(append(b, '\n')); err != nil {
211- return nil, err278+ return nil, errors.Join(err, f.Close())
212 }279 }
213 }280 }
214 }281 }
282+ if err := f.Close(); err != nil {
283+ return nil, err
284+ }
215 return p, nil285 return p, nil
216 }286 }
217 287
@@ -236,7 +306,11 @@ func (l *closeTraceLauncher) Launch(m processMapping, command roleCommand) (laun
236 if err := appendTraceRecord(m.TracePath, start); err != nil {306 if err := appendTraceRecord(m.TracePath, start); err != nil {
237 return nil, err307 return nil, err
238 }308 }
239- return &closeTraceProcess{fakeProcess: p.(*fakeProcess), traceEnd: func() error {309+ fake, ok := p.(*fakeProcess)
310+ if !ok {
311+ return nil, fmt.Errorf("fake launcher returned %T, want *fakeProcess", p)
312+ }
313+ return &closeTraceProcess{fakeProcess: fake, traceEnd: func() error {
240 return appendTraceRecord(m.TracePath, traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_end", Tick: 20, Valid: true})314 return appendTraceRecord(m.TracePath, traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_end", Tick: 20, Valid: true})
241 }}, nil315 }}, nil
242 }316 }
@@ -246,13 +320,14 @@ func appendTraceRecord(path string, record traceRecord) error {
246 if err != nil {320 if err != nil {
247 return err321 return err
248 }322 }
249- defer f.Close()
250 b, err := json.Marshal(record)323 b, err := json.Marshal(record)
251 if err != nil {324 if err != nil {
252- return err325+ return errors.Join(err, f.Close())
326+ }
327+ if _, err := f.Write(append(b, '\n')); err != nil {
328+ return errors.Join(err, f.Close())
253 }329 }
254- _, err = f.Write(append(b, '\n'))330+ return f.Close()
255- return err
256 }331 }
257 332
258 // boundedLocalConcurrentTrace is an excerpt from the second run of the333 // boundedLocalConcurrentTrace is an excerpt from the second run of the
M cmd/vev-perf-harness/udp_netem.go +38 -12
@@ -46,7 +46,13 @@ type scheduledPackets []scheduledPacket
4646 func (p scheduledPackets) Len() int { return len(p) }
4747 func (p scheduledPackets) Less(i, j int) bool { return p[i].due.Before(p[j].due) }
4848 func (p scheduledPackets) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
49-func (p *scheduledPackets) Push(x any) { *p = append(*p, x.(scheduledPacket)) }
49+func (p *scheduledPackets) Push(x any) {
50+ packet, ok := x.(scheduledPacket)
51+ if !ok {
52+ panic(fmt.Sprintf("scheduled packet type %T", x))
53+ }
54+ *p = append(*p, packet)
55+}
5056 func (p *scheduledPackets) Pop() any {
5157 old := *p
5258 n := len(old)
@@ -56,17 +62,19 @@ func (p *scheduledPackets) Pop() any {
5662 }
5763
5864 type udpNetemRelay struct {
59- config udpNetemConfig
60- conn net.PacketConn
61- done chan struct{}
62- wg sync.WaitGroup
63- once sync.Once
65+ config udpNetemConfig
66+ conn net.PacketConn
67+ done chan struct{}
68+ wg sync.WaitGroup
69+ once sync.Once
70+ closeErr error
6471
6572 // slots bounds queued and deadline-pending packets together. The reader and
6673 // the sole scheduler are the relay's only goroutines.
6774 slots chan struct{}
6875 queue chan scheduledPacket
6976 queueOverflowDrops atomic.Uint64
77+ deliveryErrors atomic.Uint64
7078
7179 mu sync.Mutex
7280 client, dst net.Addr
@@ -95,17 +103,26 @@ func newUDPNetem(config udpNetemConfig) (udpNetem, error) {
95103 }
96104
97105 func (r *udpNetemRelay) Port() int {
98- return r.conn.LocalAddr().(*net.UDPAddr).Port
106+ addr, ok := r.conn.LocalAddr().(*net.UDPAddr)
107+ if !ok {
108+ return 0
109+ }
110+ return addr.Port
99111 }
100112
101113 func (r *udpNetemRelay) Close() error {
102- var closeErr error
103114 r.once.Do(func() {
104115 close(r.done)
105- closeErr = r.conn.Close()
116+ r.closeErr = r.conn.Close()
106117 r.wg.Wait()
118+ if drops := r.queueOverflowDrops.Load(); drops != 0 {
119+ r.closeErr = errors.Join(r.closeErr, fmt.Errorf("udp netem queue overflow drops: %d", drops))
120+ }
121+ if writes := r.deliveryErrors.Load(); writes != 0 {
122+ r.closeErr = errors.Join(r.closeErr, fmt.Errorf("udp netem delivery errors: %d", writes))
123+ }
107124 })
108- return closeErr
125+ return r.closeErr
109126 }
110127
111128 func (r *udpNetemRelay) run() {
@@ -183,12 +200,21 @@ func (r *udpNetemRelay) schedule() {
183200 case <-timerC:
184201 now := time.Now()
185202 for pending.Len() > 0 && !pending[0].due.After(now) {
186- packet := heap.Pop(&pending).(scheduledPacket)
203+ packet, ok := heap.Pop(&pending).(scheduledPacket)
204+ if !ok {
205+ panic("scheduled packet heap contained unexpected type")
206+ }
187207 select {
188208 case <-r.done:
189209 return
190210 default:
191- _, _ = r.conn.WriteTo(packet.packet, packet.to)
211+ if _, err := r.conn.WriteTo(packet.packet, packet.to); err != nil {
212+ select {
213+ case <-r.done:
214+ default:
215+ r.deliveryErrors.Add(1)
216+ }
217+ }
192218 <-r.slots
193219 }
194220 }
@@ -46,7 +46,13 @@ type scheduledPackets []scheduledPacket
46 func (p scheduledPackets) Len() int { return len(p) }46 func (p scheduledPackets) Len() int { return len(p) }
47 func (p scheduledPackets) Less(i, j int) bool { return p[i].due.Before(p[j].due) }47 func (p scheduledPackets) Less(i, j int) bool { return p[i].due.Before(p[j].due) }
48 func (p scheduledPackets) Swap(i, j int) { p[i], p[j] = p[j], p[i] }48 func (p scheduledPackets) Swap(i, j int) { p[i], p[j] = p[j], p[i] }
49-func (p *scheduledPackets) Push(x any) { *p = append(*p, x.(scheduledPacket)) }49+func (p *scheduledPackets) Push(x any) {
50+ packet, ok := x.(scheduledPacket)
51+ if !ok {
52+ panic(fmt.Sprintf("scheduled packet type %T", x))
53+ }
54+ *p = append(*p, packet)
55+}
50 func (p *scheduledPackets) Pop() any {56 func (p *scheduledPackets) Pop() any {
51 old := *p57 old := *p
52 n := len(old)58 n := len(old)
@@ -56,17 +62,19 @@ func (p *scheduledPackets) Pop() any {
56 }62 }
57 63
58 type udpNetemRelay struct {64 type udpNetemRelay struct {
59- config udpNetemConfig65+ config udpNetemConfig
60- conn net.PacketConn66+ conn net.PacketConn
61- done chan struct{}67+ done chan struct{}
62- wg sync.WaitGroup68+ wg sync.WaitGroup
63- once sync.Once69+ once sync.Once
70+ closeErr error
64 71
65 // slots bounds queued and deadline-pending packets together. The reader and72 // slots bounds queued and deadline-pending packets together. The reader and
66 // the sole scheduler are the relay's only goroutines.73 // the sole scheduler are the relay's only goroutines.
67 slots chan struct{}74 slots chan struct{}
68 queue chan scheduledPacket75 queue chan scheduledPacket
69 queueOverflowDrops atomic.Uint6476 queueOverflowDrops atomic.Uint64
77+ deliveryErrors atomic.Uint64
70 78
71 mu sync.Mutex79 mu sync.Mutex
72 client, dst net.Addr80 client, dst net.Addr
@@ -95,17 +103,26 @@ func newUDPNetem(config udpNetemConfig) (udpNetem, error) {
95 }103 }
96 104
97 func (r *udpNetemRelay) Port() int {105 func (r *udpNetemRelay) Port() int {
98- return r.conn.LocalAddr().(*net.UDPAddr).Port106+ addr, ok := r.conn.LocalAddr().(*net.UDPAddr)
107+ if !ok {
108+ return 0
109+ }
110+ return addr.Port
99 }111 }
100 112
101 func (r *udpNetemRelay) Close() error {113 func (r *udpNetemRelay) Close() error {
102- var closeErr error
103 r.once.Do(func() {114 r.once.Do(func() {
104 close(r.done)115 close(r.done)
105- closeErr = r.conn.Close()116+ r.closeErr = r.conn.Close()
106 r.wg.Wait()117 r.wg.Wait()
118+ if drops := r.queueOverflowDrops.Load(); drops != 0 {
119+ r.closeErr = errors.Join(r.closeErr, fmt.Errorf("udp netem queue overflow drops: %d", drops))
120+ }
121+ if writes := r.deliveryErrors.Load(); writes != 0 {
122+ r.closeErr = errors.Join(r.closeErr, fmt.Errorf("udp netem delivery errors: %d", writes))
123+ }
107 })124 })
108- return closeErr125+ return r.closeErr
109 }126 }
110 127
111 func (r *udpNetemRelay) run() {128 func (r *udpNetemRelay) run() {
@@ -183,12 +200,21 @@ func (r *udpNetemRelay) schedule() {
183 case <-timerC:200 case <-timerC:
184 now := time.Now()201 now := time.Now()
185 for pending.Len() > 0 && !pending[0].due.After(now) {202 for pending.Len() > 0 && !pending[0].due.After(now) {
186- packet := heap.Pop(&pending).(scheduledPacket)203+ packet, ok := heap.Pop(&pending).(scheduledPacket)
204+ if !ok {
205+ panic("scheduled packet heap contained unexpected type")
206+ }
187 select {207 select {
188 case <-r.done:208 case <-r.done:
189 return209 return
190 default:210 default:
191- _, _ = r.conn.WriteTo(packet.packet, packet.to)211+ if _, err := r.conn.WriteTo(packet.packet, packet.to); err != nil {
212+ select {
213+ case <-r.done:
214+ default:
215+ r.deliveryErrors.Add(1)
216+ }
217+ }
192 <-r.slots218 <-r.slots
193 }219 }
194 }220 }
M cmd/vev-perf-harness/udp_netem_test.go +32 -4
@@ -9,6 +9,15 @@ import (
99 "time"
1010 )
1111
12+func closeUDPFixture(t *testing.T, c interface{ Close() error }) {
13+ t.Helper()
14+ t.Cleanup(func() {
15+ if err := c.Close(); err != nil {
16+ t.Error(err)
17+ }
18+ })
19+}
20+
1221 func TestUDPNetemDropsWhenBoundedSchedulerIsFull(t *testing.T) {
1322 r := &udpNetemRelay{
1423 done: make(chan struct{}),
@@ -29,6 +38,21 @@ func TestUDPNetemDropsWhenBoundedSchedulerIsFull(t *testing.T) {
2938 }
3039 }
3140
41+func TestUDPNetemCloseRejectsQueueOverflow(t *testing.T) {
42+ netem, err := newUDPNetem(udpNetemConfig{TargetPath: filepath.Join(t.TempDir(), "target")})
43+ if err != nil {
44+ t.Fatal(err)
45+ }
46+ relay, ok := netem.(*udpNetemRelay)
47+ if !ok {
48+ t.Fatalf("netem type=%T, want *udpNetemRelay", netem)
49+ }
50+ relay.queueOverflowDrops.Store(3)
51+ if err := netem.Close(); err == nil || err.Error() != "udp netem queue overflow drops: 3" {
52+ t.Fatalf("Close error=%v, want overflow rejection", err)
53+ }
54+}
55+
3256 func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
3357 fixtures := []struct {
3458 name string
@@ -48,21 +72,25 @@ func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
4872 if err != nil {
4973 t.Fatal(err)
5074 }
51- defer target.Close()
75+ closeUDPFixture(t, target)
76+ targetAddr, ok := target.LocalAddr().(*net.UDPAddr)
77+ if !ok {
78+ t.Fatalf("target address type=%T, want *net.UDPAddr", target.LocalAddr())
79+ }
5280 path := filepath.Join(t.TempDir(), "target")
53- if err := os.WriteFile(path, []byte("VEV-UDP "+strconv.Itoa(target.LocalAddr().(*net.UDPAddr).Port)+" key\\n"), 0o600); err != nil {
81+ if err := os.WriteFile(path, []byte("VEV-UDP "+strconv.Itoa(targetAddr.Port)+" key\\n"), 0o600); err != nil {
5482 t.Fatal(err)
5583 }
5684 netem, err := newUDPNetem(udpNetemConfig{RTT: tc.rtt, LossPercent: tc.loss, TargetPath: path})
5785 if err != nil {
5886 t.Fatal(err)
5987 }
60- defer netem.Close()
88+ closeUDPFixture(t, netem)
6189 client, err := net.ListenPacket("udp", "127.0.0.1:0")
6290 if err != nil {
6391 t.Fatal(err)
6492 }
65- defer client.Close()
93+ closeUDPFixture(t, client)
6694 started := time.Now()
6795 for i := 0; i < tc.sent; i++ {
6896 if _, err := client.WriteTo([]byte{byte(i)}, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: netem.Port()}); err != nil {
@@ -9,6 +9,15 @@ import (
9 "time"9 "time"
10 )10 )
11 11
12+func closeUDPFixture(t *testing.T, c interface{ Close() error }) {
13+ t.Helper()
14+ t.Cleanup(func() {
15+ if err := c.Close(); err != nil {
16+ t.Error(err)
17+ }
18+ })
19+}
20+
12 func TestUDPNetemDropsWhenBoundedSchedulerIsFull(t *testing.T) {21 func TestUDPNetemDropsWhenBoundedSchedulerIsFull(t *testing.T) {
13 r := &udpNetemRelay{22 r := &udpNetemRelay{
14 done: make(chan struct{}),23 done: make(chan struct{}),
@@ -29,6 +38,21 @@ func TestUDPNetemDropsWhenBoundedSchedulerIsFull(t *testing.T) {
29 }38 }
30 }39 }
31 40
41+func TestUDPNetemCloseRejectsQueueOverflow(t *testing.T) {
42+ netem, err := newUDPNetem(udpNetemConfig{TargetPath: filepath.Join(t.TempDir(), "target")})
43+ if err != nil {
44+ t.Fatal(err)
45+ }
46+ relay, ok := netem.(*udpNetemRelay)
47+ if !ok {
48+ t.Fatalf("netem type=%T, want *udpNetemRelay", netem)
49+ }
50+ relay.queueOverflowDrops.Store(3)
51+ if err := netem.Close(); err == nil || err.Error() != "udp netem queue overflow drops: 3" {
52+ t.Fatalf("Close error=%v, want overflow rejection", err)
53+ }
54+}
55+
32 func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {56 func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
33 fixtures := []struct {57 fixtures := []struct {
34 name string58 name string
@@ -48,21 +72,25 @@ func TestUDPNetemExecutesRTTAndLossFixtures(t *testing.T) {
48 if err != nil {72 if err != nil {
49 t.Fatal(err)73 t.Fatal(err)
50 }74 }
51- defer target.Close()75+ closeUDPFixture(t, target)
76+ targetAddr, ok := target.LocalAddr().(*net.UDPAddr)
77+ if !ok {
78+ t.Fatalf("target address type=%T, want *net.UDPAddr", target.LocalAddr())
79+ }
52 path := filepath.Join(t.TempDir(), "target")80 path := filepath.Join(t.TempDir(), "target")
53- if err := os.WriteFile(path, []byte("VEV-UDP "+strconv.Itoa(target.LocalAddr().(*net.UDPAddr).Port)+" key\\n"), 0o600); err != nil {81+ if err := os.WriteFile(path, []byte("VEV-UDP "+strconv.Itoa(targetAddr.Port)+" key\\n"), 0o600); err != nil {
54 t.Fatal(err)82 t.Fatal(err)
55 }83 }
56 netem, err := newUDPNetem(udpNetemConfig{RTT: tc.rtt, LossPercent: tc.loss, TargetPath: path})84 netem, err := newUDPNetem(udpNetemConfig{RTT: tc.rtt, LossPercent: tc.loss, TargetPath: path})
57 if err != nil {85 if err != nil {
58 t.Fatal(err)86 t.Fatal(err)
59 }87 }
60- defer netem.Close()88+ closeUDPFixture(t, netem)
61 client, err := net.ListenPacket("udp", "127.0.0.1:0")89 client, err := net.ListenPacket("udp", "127.0.0.1:0")
62 if err != nil {90 if err != nil {
63 t.Fatal(err)91 t.Fatal(err)
64 }92 }
65- defer client.Close()93+ closeUDPFixture(t, client)
66 started := time.Now()94 started := time.Now()
67 for i := 0; i < tc.sent; i++ {95 for i := 0; i < tc.sent; i++ {
68 if _, err := client.WriteTo([]byte{byte(i)}, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: netem.Port()}); err != nil {96 if _, err := client.WriteTo([]byte{byte(i)}, &net.UDPAddr{IP: net.ParseIP("127.0.0.1"), Port: netem.Port()}); err != nil {

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 12

Tags 11

↑↓ moveenter openesc close