fix(obs): close roles before trace merge
- commit
- 72845c92d20c3433a13817838de4c98f92bb9dcb
- author
- brice · · 2026-07-12 23:30 UTC
- parent
- a244e21
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
2 files · +150 -14
M cmd/vev-perf-harness/main.go +38 -8
| @@ -147,6 +147,23 @@ type launcher interface { | |||
| 147 | 147 | Launch(processMapping, roleCommand) (launchedProcess, error) | |
| 148 | 148 | } | |
| 149 | 149 | ||
| 150 | + | type launchedRole struct { | |
| 151 | + | mapping processMapping | |
| 152 | + | process launchedProcess | |
| 153 | + | } | |
| 154 | + | ||
| 155 | + | // closeLaunchedRoles closes every role in reverse launch order. errors.Join | |
| 156 | + | // retains every cleanup failure in that stable order. | |
| 157 | + | func closeLaunchedRoles(processes []launchedRole) error { | |
| 158 | + | var failures []error | |
| 159 | + | for i := len(processes) - 1; i >= 0; i-- { | |
| 160 | + | if err := processes[i].process.Close(); err != nil { | |
| 161 | + | failures = append(failures, err) | |
| 162 | + | } | |
| 163 | + | } | |
| 164 | + | return errors.Join(failures...) | |
| 165 | + | } | |
| 166 | + | ||
| 150 | 167 | // cliLauncher is deliberately limited to executing the supplied public vev | |
| 151 | 168 | // binary. Scenario orchestration never reaches into an in-process daemon. | |
| 152 | 169 | type cliLauncher struct { | |
| @@ -794,17 +811,25 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr | |||
| 794 | 811 | if h.launcher == nil { | |
| 795 | 812 | h.launcher = &cliLauncher{bin: o.vevBin} | |
| 796 | 813 | } | |
| 797 | - | type launchedRole struct { | |
| 798 | - | mapping processMapping | |
| 799 | - | process launchedProcess | |
| 800 | - | } | |
| 801 | 814 | processes := make([]launchedRole, 0, len(maps)) | |
| 802 | - | defer func() { | |
| 815 | + | closed := false | |
| 816 | + | closeRoles := func() error { | |
| 817 | + | if closed { | |
| 818 | + | return nil | |
| 819 | + | } | |
| 820 | + | closed = true | |
| 803 | 821 | // Clients own SSH-stdio descendants; close them before the deferred UDP | |
| 804 | 822 | // peer and daemon so no role survives a failed or completed run. | |
| 805 | - | for i := len(processes) - 1; i >= 0; i-- { | |
| 806 | - | if e := processes[i].process.Close(); err == nil && e != nil { | |
| 807 | - | err = e | |
| 823 | + | return closeLaunchedRoles(processes) | |
| 824 | + | } | |
| 825 | + | defer func() { | |
| 826 | + | if cleanupErr := closeRoles(); cleanupErr != nil { | |
| 827 | + | if err == nil { | |
| 828 | + | err = cleanupErr | |
| 829 | + | } else { | |
| 830 | + | // Keep the primary workload failure first while retaining every | |
| 831 | + | // cleanup failure in reverse launch order. | |
| 832 | + | err = errors.Join(err, cleanupErr) | |
| 808 | 833 | } | |
| 809 | 834 | } | |
| 810 | 835 | }() | |
| @@ -857,6 +882,11 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr | |||
| 857 | 882 | if e != nil { | |
| 858 | 883 | return res, e | |
| 859 | 884 | } | |
| 885 | + | // Process teardown can unblock a receive and cause its failed end mark to | |
| 886 | + | // be serialized. Merge only the complete, post-cleanup per-process traces. | |
| 887 | + | if cleanupErr := closeRoles(); cleanupErr != nil { | |
| 888 | + | return res, cleanupErr | |
| 889 | + | } | |
| 860 | 890 | spans, e := mergeProcessTraces(maps) | |
| 861 | 891 | if e != nil { | |
| 862 | 892 | return res, e | |
| @@ -147,6 +147,23 @@ type launcher interface { | |||||
| 147 | Launch(processMapping, roleCommand) (launchedProcess, error) | 147 | Launch(processMapping, roleCommand) (launchedProcess, error) | ||
| 148 | } | 148 | } | ||
| 149 | 149 | ||||
| 150 | + | type launchedRole struct { | |||
| 151 | + | mapping processMapping | |||
| 152 | + | process launchedProcess | |||
| 153 | + | } | |||
| 154 | + | ||||
| 155 | + | // closeLaunchedRoles closes every role in reverse launch order. errors.Join | |||
| 156 | + | // retains every cleanup failure in that stable order. | |||
| 157 | + | func closeLaunchedRoles(processes []launchedRole) error { | |||
| 158 | + | var failures []error | |||
| 159 | + | for i := len(processes) - 1; i >= 0; i-- { | |||
| 160 | + | if err := processes[i].process.Close(); err != nil { | |||
| 161 | + | failures = append(failures, err) | |||
| 162 | + | } | |||
| 163 | + | } | |||
| 164 | + | return errors.Join(failures...) | |||
| 165 | + | } | |||
| 166 | + | ||||
| 150 | // cliLauncher is deliberately limited to executing the supplied public vev | 167 | // cliLauncher is deliberately limited to executing the supplied public vev | ||
| 151 | // binary. Scenario orchestration never reaches into an in-process daemon. | 168 | // binary. Scenario orchestration never reaches into an in-process daemon. | ||
| 152 | type cliLauncher struct { | 169 | type cliLauncher struct { | ||
| @@ -794,17 +811,25 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr | |||||
| 794 | if h.launcher == nil { | 811 | if h.launcher == nil { | ||
| 795 | h.launcher = &cliLauncher{bin: o.vevBin} | 812 | h.launcher = &cliLauncher{bin: o.vevBin} | ||
| 796 | } | 813 | } | ||
| 797 | - | type launchedRole struct { | |||
| 798 | - | mapping processMapping | |||
| 799 | - | process launchedProcess | |||
| 800 | - | } | |||
| 801 | processes := make([]launchedRole, 0, len(maps)) | 814 | processes := make([]launchedRole, 0, len(maps)) | ||
| 802 | - | defer func() { | 815 | + | closed := false |
| 816 | + | closeRoles := func() error { | |||
| 817 | + | if closed { | |||
| 818 | + | return nil | |||
| 819 | + | } | |||
| 820 | + | closed = true | |||
| 803 | // Clients own SSH-stdio descendants; close them before the deferred UDP | 821 | // Clients own SSH-stdio descendants; close them before the deferred UDP | ||
| 804 | // peer and daemon so no role survives a failed or completed run. | 822 | // peer and daemon so no role survives a failed or completed run. | ||
| 805 | - | for i := len(processes) - 1; i >= 0; i-- { | 823 | + | return closeLaunchedRoles(processes) |
| 806 | - | if e := processes[i].process.Close(); err == nil && e != nil { | 824 | + | } |
| 807 | - | err = e | 825 | + | defer func() { |
| 826 | + | if cleanupErr := closeRoles(); cleanupErr != nil { | |||
| 827 | + | if err == nil { | |||
| 828 | + | err = cleanupErr | |||
| 829 | + | } else { | |||
| 830 | + | // Keep the primary workload failure first while retaining every | |||
| 831 | + | // cleanup failure in reverse launch order. | |||
| 832 | + | err = errors.Join(err, cleanupErr) | |||
| 808 | } | 833 | } | ||
| 809 | } | 834 | } | ||
| 810 | }() | 835 | }() | ||
| @@ -857,6 +882,11 @@ func (h *harness) runOne(o options, mat manifest, s scenario, run int, raw io.Wr | |||||
| 857 | if e != nil { | 882 | if e != nil { | ||
| 858 | return res, e | 883 | return res, e | ||
| 859 | } | 884 | } | ||
| 885 | + | // Process teardown can unblock a receive and cause its failed end mark to | |||
| 886 | + | // be serialized. Merge only the complete, post-cleanup per-process traces. | |||
| 887 | + | if cleanupErr := closeRoles(); cleanupErr != nil { | |||
| 888 | + | return res, cleanupErr | |||
| 889 | + | } | |||
| 860 | spans, e := mergeProcessTraces(maps) | 890 | spans, e := mergeProcessTraces(maps) | ||
| 861 | if e != nil { | 891 | if e != nil { | ||
| 862 | return res, e | 892 | return res, e | ||
M cmd/vev-perf-harness/main_test.go +112 -6
| @@ -17,10 +17,12 @@ func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick } | |||
| 17 | 17 | func (*fakeClock) Sleep(time.Duration) {} | |
| 18 | 18 | ||
| 19 | 19 | type fakeProcess struct { | |
| 20 | - | fail bool | |
| 21 | - | warmups [][]byte | |
| 22 | - | measures [][]byte | |
| 23 | - | closed bool | |
| 20 | + | fail bool | |
| 21 | + | measureErr error | |
| 22 | + | closeErr error | |
| 23 | + | warmups [][]byte | |
| 24 | + | measures [][]byte | |
| 25 | + | closed bool | |
| 24 | 26 | } | |
| 25 | 27 | ||
| 26 | 28 | func (p *fakeProcess) Warmup(input []byte) error { | |
| @@ -29,6 +31,9 @@ func (p *fakeProcess) Warmup(input []byte) error { | |||
| 29 | 31 | } | |
| 30 | 32 | func (p *fakeProcess) Measure(input []byte, injected, flush func() error) error { | |
| 31 | 33 | p.measures = append(p.measures, append([]byte(nil), input...)) | |
| 34 | + | if p.measureErr != nil { | |
| 35 | + | return p.measureErr | |
| 36 | + | } | |
| 32 | 37 | if p.fail { | |
| 33 | 38 | return errors.New("measure failure") | |
| 34 | 39 | } | |
| @@ -37,13 +42,15 @@ func (p *fakeProcess) Measure(input []byte, injected, flush func() error) error | |||
| 37 | 42 | } | |
| 38 | 43 | return flush() | |
| 39 | 44 | } | |
| 40 | - | func (p *fakeProcess) Close() error { p.closed = true; return nil } | |
| 45 | + | func (p *fakeProcess) Close() error { p.closed = true; return p.closeErr } | |
| 41 | 46 | ||
| 42 | 47 | type fakeLauncher struct { | |
| 43 | 48 | mappings []processMapping | |
| 44 | 49 | args [][]string | |
| 45 | 50 | commands []roleCommand | |
| 46 | 51 | process []*fakeProcess | |
| 52 | + | measureErr map[string]error | |
| 53 | + | closeErr map[string]error | |
| 47 | 54 | manifestPresent bool | |
| 48 | 55 | } | |
| 49 | 56 | ||
| @@ -53,7 +60,7 @@ func (l *fakeLauncher) Launch(m processMapping, command roleCommand) (launchedPr | |||
| 53 | 60 | l.commands = append(l.commands, command) | |
| 54 | 61 | _, e := os.Stat(filepath.Join(filepath.Dir(m.TracePath), "manifest.json")) | |
| 55 | 62 | l.manifestPresent = e == nil | |
| 56 | - | p := &fakeProcess{} | |
| 63 | + | p := &fakeProcess{measureErr: l.measureErr[m.Role], closeErr: l.closeErr[m.Role]} | |
| 57 | 64 | l.process = append(l.process, p) | |
| 58 | 65 | return p, nil | |
| 59 | 66 | } | |
| @@ -377,6 +384,35 @@ func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) { | |||
| 377 | 384 | } | |
| 378 | 385 | } | |
| 379 | 386 | ||
| 387 | + | func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) { | |
| 388 | + | dir := t.TempDir() | |
| 389 | + | raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) | |
| 390 | + | if err != nil { | |
| 391 | + | t.Fatal(err) | |
| 392 | + | } | |
| 393 | + | defer raw.Close() | |
| 394 | + | primary := errors.New("primary workload failure") | |
| 395 | + | cleanup := errors.New("daemon cleanup failure") | |
| 396 | + | launcher := &fakeLauncher{ | |
| 397 | + | measureErr: map[string]error{"client": primary}, | |
| 398 | + | closeErr: map[string]error{"daemon": cleanup}, | |
| 399 | + | } | |
| 400 | + | h := defaultHarness() | |
| 401 | + | h.clock, h.launcher = &fakeClock{}, launcher | |
| 402 | + | _, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration, repetitions: minimumRepetitions}, manifest{}, scenario{ID: "errors", Roles: []string{"daemon", "client"}}, 1, raw) | |
| 403 | + | if !errors.Is(err, primary) || !errors.Is(err, cleanup) { | |
| 404 | + | t.Fatalf("workload and cleanup errors were not both preserved: %v", err) | |
| 405 | + | } | |
| 406 | + | if got, want := err.Error(), primary.Error()+"\n"+cleanup.Error(); got != want { | |
| 407 | + | t.Fatalf("error ordering=%q want %q", got, want) | |
| 408 | + | } | |
| 409 | + | for i, p := range launcher.process { | |
| 410 | + | if !p.closed { | |
| 411 | + | t.Fatalf("role %s was not closed", launcher.mappings[i].Role) | |
| 412 | + | } | |
| 413 | + | } | |
| 414 | + | } | |
| 415 | + | ||
| 380 | 416 | func TestHarnessCleansRunDirectoryOnProcessFailure(t *testing.T) { | |
| 381 | 417 | dir := t.TempDir() | |
| 382 | 418 | raw, err := os.OpenFile(filepath.Join(dir, "raw-harness.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) | |
| @@ -460,6 +496,76 @@ func (l *traceLauncher) Launch(m processMapping, command roleCommand) (launchedP | |||
| 460 | 496 | } | |
| 461 | 497 | return p, nil | |
| 462 | 498 | } | |
| 499 | + | ||
| 500 | + | type closeTraceProcess struct { | |
| 501 | + | *fakeProcess | |
| 502 | + | traceEnd func() error | |
| 503 | + | } | |
| 504 | + | ||
| 505 | + | func (p *closeTraceProcess) Close() error { | |
| 506 | + | p.closed = true | |
| 507 | + | return errors.Join(p.closeErr, p.traceEnd()) | |
| 508 | + | } | |
| 509 | + | ||
| 510 | + | type closeTraceLauncher struct{ fakeLauncher } | |
| 511 | + | ||
| 512 | + | func (l *closeTraceLauncher) Launch(m processMapping, command roleCommand) (launchedProcess, error) { | |
| 513 | + | p, err := l.fakeLauncher.Launch(m, command) | |
| 514 | + | if err != nil { | |
| 515 | + | return nil, err | |
| 516 | + | } | |
| 517 | + | start := traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_start", Tick: 10} | |
| 518 | + | if err := appendTraceRecord(m.TracePath, start); err != nil { | |
| 519 | + | return nil, err | |
| 520 | + | } | |
| 521 | + | return &closeTraceProcess{fakeProcess: p.(*fakeProcess), traceEnd: func() error { | |
| 522 | + | 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}) | |
| 523 | + | }}, nil | |
| 524 | + | } | |
| 525 | + | ||
| 526 | + | func appendTraceRecord(path string, record traceRecord) error { | |
| 527 | + | f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0o600) | |
| 528 | + | if err != nil { | |
| 529 | + | return err | |
| 530 | + | } | |
| 531 | + | defer f.Close() | |
| 532 | + | b, err := json.Marshal(record) | |
| 533 | + | if err != nil { | |
| 534 | + | return err | |
| 535 | + | } | |
| 536 | + | _, err = f.Write(append(b, '\n')) | |
| 537 | + | return err | |
| 538 | + | } | |
| 539 | + | ||
| 540 | + | func TestHarnessClosesRolesBeforeMergingReceiveSpans(t *testing.T) { | |
| 541 | + | dir := t.TempDir() | |
| 542 | + | raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) | |
| 543 | + | if err != nil { | |
| 544 | + | t.Fatal(err) | |
| 545 | + | } | |
| 546 | + | defer raw.Close() | |
| 547 | + | launcher := &closeTraceLauncher{} | |
| 548 | + | h := defaultHarness() | |
| 549 | + | h.clock, h.launcher = &fakeClock{}, launcher | |
| 550 | + | 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) | |
| 551 | + | if err != nil { | |
| 552 | + | t.Fatal(err) | |
| 553 | + | } | |
| 554 | + | if len(result.Spans) != 2 { | |
| 555 | + | t.Fatalf("post-cleanup receive spans=%+v", result.Spans) | |
| 556 | + | } | |
| 557 | + | for _, s := range result.Spans { | |
| 558 | + | if s.Name != "adapter_receive_duration" || len(s.Samples) != 1 || s.Samples[0] != 10 { | |
| 559 | + | t.Fatalf("unmatched or wrong receive span: %+v", s) | |
| 560 | + | } | |
| 561 | + | } | |
| 562 | + | for i, p := range launcher.process { | |
| 563 | + | if !p.closed { | |
| 564 | + | t.Fatalf("role %s was not closed before trace merge", launcher.mappings[i].Role) | |
| 565 | + | } | |
| 566 | + | } | |
| 567 | + | } | |
| 568 | + | ||
| 463 | 569 | func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) { | |
| 464 | 570 | base, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json")) | |
| 465 | 571 | if err != nil { | |
| @@ -17,10 +17,12 @@ func (c *fakeClock) Now() int64 { c.tick += 10; return c.tick } | |||||
| 17 | func (*fakeClock) Sleep(time.Duration) {} | 17 | func (*fakeClock) Sleep(time.Duration) {} | ||
| 18 | 18 | ||||
| 19 | type fakeProcess struct { | 19 | type fakeProcess struct { | ||
| 20 | - | fail bool | 20 | + | fail bool |
| 21 | - | warmups [][]byte | 21 | + | measureErr error |
| 22 | - | measures [][]byte | 22 | + | closeErr error |
| 23 | - | closed bool | 23 | + | warmups [][]byte |
| 24 | + | measures [][]byte | |||
| 25 | + | closed bool | |||
| 24 | } | 26 | } | ||
| 25 | 27 | ||||
| 26 | func (p *fakeProcess) Warmup(input []byte) error { | 28 | func (p *fakeProcess) Warmup(input []byte) error { | ||
| @@ -29,6 +31,9 @@ func (p *fakeProcess) Warmup(input []byte) error { | |||||
| 29 | } | 31 | } | ||
| 30 | func (p *fakeProcess) Measure(input []byte, injected, flush func() error) error { | 32 | func (p *fakeProcess) Measure(input []byte, injected, flush func() error) error { | ||
| 31 | p.measures = append(p.measures, append([]byte(nil), input...)) | 33 | p.measures = append(p.measures, append([]byte(nil), input...)) | ||
| 34 | + | if p.measureErr != nil { | |||
| 35 | + | return p.measureErr | |||
| 36 | + | } | |||
| 32 | if p.fail { | 37 | if p.fail { | ||
| 33 | return errors.New("measure failure") | 38 | return errors.New("measure failure") | ||
| 34 | } | 39 | } | ||
| @@ -37,13 +42,15 @@ func (p *fakeProcess) Measure(input []byte, injected, flush func() error) error | |||||
| 37 | } | 42 | } | ||
| 38 | return flush() | 43 | return flush() | ||
| 39 | } | 44 | } | ||
| 40 | - | func (p *fakeProcess) Close() error { p.closed = true; return nil } | 45 | + | func (p *fakeProcess) Close() error { p.closed = true; return p.closeErr } |
| 41 | 46 | ||||
| 42 | type fakeLauncher struct { | 47 | type fakeLauncher struct { | ||
| 43 | mappings []processMapping | 48 | mappings []processMapping | ||
| 44 | args [][]string | 49 | args [][]string | ||
| 45 | commands []roleCommand | 50 | commands []roleCommand | ||
| 46 | process []*fakeProcess | 51 | process []*fakeProcess | ||
| 52 | + | measureErr map[string]error | |||
| 53 | + | closeErr map[string]error | |||
| 47 | manifestPresent bool | 54 | manifestPresent bool | ||
| 48 | } | 55 | } | ||
| 49 | 56 | ||||
| @@ -53,7 +60,7 @@ func (l *fakeLauncher) Launch(m processMapping, command roleCommand) (launchedPr | |||||
| 53 | l.commands = append(l.commands, command) | 60 | l.commands = append(l.commands, command) | ||
| 54 | _, e := os.Stat(filepath.Join(filepath.Dir(m.TracePath), "manifest.json")) | 61 | _, e := os.Stat(filepath.Join(filepath.Dir(m.TracePath), "manifest.json")) | ||
| 55 | l.manifestPresent = e == nil | 62 | l.manifestPresent = e == nil | ||
| 56 | - | p := &fakeProcess{} | 63 | + | p := &fakeProcess{measureErr: l.measureErr[m.Role], closeErr: l.closeErr[m.Role]} |
| 57 | l.process = append(l.process, p) | 64 | l.process = append(l.process, p) | ||
| 58 | return p, nil | 65 | return p, nil | ||
| 59 | } | 66 | } | ||
| @@ -377,6 +384,35 @@ func TestHarnessRejectsInvalidBoundaryPairs(t *testing.T) { | |||||
| 377 | } | 384 | } | ||
| 378 | } | 385 | } | ||
| 379 | 386 | ||||
| 387 | + | func TestHarnessPreservesWorkloadAndCleanupErrors(t *testing.T) { | |||
| 388 | + | dir := t.TempDir() | |||
| 389 | + | raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) | |||
| 390 | + | if err != nil { | |||
| 391 | + | t.Fatal(err) | |||
| 392 | + | } | |||
| 393 | + | defer raw.Close() | |||
| 394 | + | primary := errors.New("primary workload failure") | |||
| 395 | + | cleanup := errors.New("daemon cleanup failure") | |||
| 396 | + | launcher := &fakeLauncher{ | |||
| 397 | + | measureErr: map[string]error{"client": primary}, | |||
| 398 | + | closeErr: map[string]error{"daemon": cleanup}, | |||
| 399 | + | } | |||
| 400 | + | h := defaultHarness() | |||
| 401 | + | h.clock, h.launcher = &fakeClock{}, launcher | |||
| 402 | + | _, err = h.runOne(options{out: dir, warmup: time.Second, duration: minimumDuration, repetitions: minimumRepetitions}, manifest{}, scenario{ID: "errors", Roles: []string{"daemon", "client"}}, 1, raw) | |||
| 403 | + | if !errors.Is(err, primary) || !errors.Is(err, cleanup) { | |||
| 404 | + | t.Fatalf("workload and cleanup errors were not both preserved: %v", err) | |||
| 405 | + | } | |||
| 406 | + | if got, want := err.Error(), primary.Error()+"\n"+cleanup.Error(); got != want { | |||
| 407 | + | t.Fatalf("error ordering=%q want %q", got, want) | |||
| 408 | + | } | |||
| 409 | + | for i, p := range launcher.process { | |||
| 410 | + | if !p.closed { | |||
| 411 | + | t.Fatalf("role %s was not closed", launcher.mappings[i].Role) | |||
| 412 | + | } | |||
| 413 | + | } | |||
| 414 | + | } | |||
| 415 | + | ||||
| 380 | func TestHarnessCleansRunDirectoryOnProcessFailure(t *testing.T) { | 416 | func TestHarnessCleansRunDirectoryOnProcessFailure(t *testing.T) { | ||
| 381 | dir := t.TempDir() | 417 | dir := t.TempDir() | ||
| 382 | raw, err := os.OpenFile(filepath.Join(dir, "raw-harness.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) | 418 | raw, err := os.OpenFile(filepath.Join(dir, "raw-harness.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) | ||
| @@ -460,6 +496,76 @@ func (l *traceLauncher) Launch(m processMapping, command roleCommand) (launchedP | |||||
| 460 | } | 496 | } | ||
| 461 | return p, nil | 497 | return p, nil | ||
| 462 | } | 498 | } | ||
| 499 | + | ||||
| 500 | + | type closeTraceProcess struct { | |||
| 501 | + | *fakeProcess | |||
| 502 | + | traceEnd func() error | |||
| 503 | + | } | |||
| 504 | + | ||||
| 505 | + | func (p *closeTraceProcess) Close() error { | |||
| 506 | + | p.closed = true | |||
| 507 | + | return errors.Join(p.closeErr, p.traceEnd()) | |||
| 508 | + | } | |||
| 509 | + | ||||
| 510 | + | type closeTraceLauncher struct{ fakeLauncher } | |||
| 511 | + | ||||
| 512 | + | func (l *closeTraceLauncher) Launch(m processMapping, command roleCommand) (launchedProcess, error) { | |||
| 513 | + | p, err := l.fakeLauncher.Launch(m, command) | |||
| 514 | + | if err != nil { | |||
| 515 | + | return nil, err | |||
| 516 | + | } | |||
| 517 | + | start := traceRecord{ProcessID: m.ProcessID, Component: m.Role, Scenario: m.Scenario, Run: uint64(m.Run), Sequence: 1, RequestID: 1, Epoch: 1, Kind: "adapter_receive_start", Tick: 10} | |||
| 518 | + | if err := appendTraceRecord(m.TracePath, start); err != nil { | |||
| 519 | + | return nil, err | |||
| 520 | + | } | |||
| 521 | + | return &closeTraceProcess{fakeProcess: p.(*fakeProcess), traceEnd: func() error { | |||
| 522 | + | 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}) | |||
| 523 | + | }}, nil | |||
| 524 | + | } | |||
| 525 | + | ||||
| 526 | + | func appendTraceRecord(path string, record traceRecord) error { | |||
| 527 | + | f, err := os.OpenFile(path, os.O_WRONLY|os.O_APPEND, 0o600) | |||
| 528 | + | if err != nil { | |||
| 529 | + | return err | |||
| 530 | + | } | |||
| 531 | + | defer f.Close() | |||
| 532 | + | b, err := json.Marshal(record) | |||
| 533 | + | if err != nil { | |||
| 534 | + | return err | |||
| 535 | + | } | |||
| 536 | + | _, err = f.Write(append(b, '\n')) | |||
| 537 | + | return err | |||
| 538 | + | } | |||
| 539 | + | ||||
| 540 | + | func TestHarnessClosesRolesBeforeMergingReceiveSpans(t *testing.T) { | |||
| 541 | + | dir := t.TempDir() | |||
| 542 | + | raw, err := os.OpenFile(filepath.Join(dir, "raw.jsonl"), os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o600) | |||
| 543 | + | if err != nil { | |||
| 544 | + | t.Fatal(err) | |||
| 545 | + | } | |||
| 546 | + | defer raw.Close() | |||
| 547 | + | launcher := &closeTraceLauncher{} | |||
| 548 | + | h := defaultHarness() | |||
| 549 | + | h.clock, h.launcher = &fakeClock{}, launcher | |||
| 550 | + | 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) | |||
| 551 | + | if err != nil { | |||
| 552 | + | t.Fatal(err) | |||
| 553 | + | } | |||
| 554 | + | if len(result.Spans) != 2 { | |||
| 555 | + | t.Fatalf("post-cleanup receive spans=%+v", result.Spans) | |||
| 556 | + | } | |||
| 557 | + | for _, s := range result.Spans { | |||
| 558 | + | if s.Name != "adapter_receive_duration" || len(s.Samples) != 1 || s.Samples[0] != 10 { | |||
| 559 | + | t.Fatalf("unmatched or wrong receive span: %+v", s) | |||
| 560 | + | } | |||
| 561 | + | } | |||
| 562 | + | for i, p := range launcher.process { | |||
| 563 | + | if !p.closed { | |||
| 564 | + | t.Fatalf("role %s was not closed before trace merge", launcher.mappings[i].Role) | |||
| 565 | + | } | |||
| 566 | + | } | |||
| 567 | + | } | |||
| 568 | + | ||||
| 463 | func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) { | 569 | func TestHarnessWritesRequiredEvidenceWithSufficientSpans(t *testing.T) { | ||
| 464 | base, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json")) | 570 | base, err := readManifest(filepath.Join("..", "..", "testdata", "perf", "manifest.json")) | ||
| 465 | if err != nil { | 571 | if err != nil { | ||