1package app
2
3import (
4 "errors"
5 "fmt"
6 "io"
7 "os"
8 "strconv"
9 "sync"
10
11 "github.com/bnema/vev/internal/adapters/observability"
12 "github.com/bnema/vev/internal/ports"
13)
14
15// newPerformanceTrace is the trace factory seam; tests replace it to observe
16// propagation without opening real trace files.
17var newPerformanceTrace = performanceTrace
18
19// performanceTrace creates one serialized timestamp owner for this process.
20// An empty trace environment leaves all production behavior and wire bytes
21// unchanged.
22func performanceTrace(clk ports.Clock) (ports.SerializedRuntimeObserver, io.Closer, error) {
23 return performanceTraceWithFactories(clk, observability.NewJSONL, ports.NewRuntimeCorrelationObserver)
24}
25
26// performanceTraceWithFactories keeps setup rollback behavior directly
27// testable without requiring an operating-system file close failure.
28func performanceTraceWithFactories(
29 clk ports.Clock,
30 newSink func(string, ports.Clock, string) (ports.RuntimeObserver, io.Closer, error),
31 newCorrelation func(ports.RuntimeObserver, ports.RuntimeCorrelationInputs) (ports.RuntimeObserver, error),
32) (ports.SerializedRuntimeObserver, io.Closer, error) {
33 path, processID := os.Getenv("VEV_PERF_TRACE"), os.Getenv("VEV_PERF_PROCESS_ID")
34 if path == "" {
35 return nil, nil, nil
36 }
37 observer, closer, err := newSink(path, clk, processID)
38 if err != nil {
39 return nil, nil, err
40 }
41 // The harness supplies these manifest fields to every launched role. Keep a
42 // valid standalone trace for operators that set only the original trace
43 // variables, while ensuring harness traces match their process mapping.
44 inputs := ports.RuntimeCorrelationInputs{Scenario: os.Getenv("VEV_PERF_SCENARIO"), Run: 1}
45 if inputs.Scenario == "" {
46 inputs.Scenario = "runtime"
47 }
48 if rawRun := os.Getenv("VEV_PERF_RUN"); rawRun != "" {
49 inputs.Run, err = strconv.ParseUint(rawRun, 10, 64)
50 if err != nil {
51 setupErr := fmt.Errorf("invalid VEV_PERF_RUN %q: %w", rawRun, err)
52 return nil, nil, closeTraceAfterSetupFailure(closer, setupErr)
53 }
54 if inputs.Run == 0 {
55 setupErr := fmt.Errorf("invalid VEV_PERF_RUN %q", rawRun)
56 return nil, nil, closeTraceAfterSetupFailure(closer, setupErr)
57 }
58 }
59 observer, err = newCorrelation(observer, inputs)
60 if err != nil {
61 setupErr := fmt.Errorf("configure runtime trace correlation: %w", err)
62 return nil, nil, closeTraceAfterSetupFailure(closer, setupErr)
63 }
64 reporter := observability.NewSerialized(observer, runtimeTraceQueueDepth)
65 return reporter, &runtimeTraceCloser{reporter: reporter, closer: closer}, nil
66}
67
68func closeTraceAfterSetupFailure(closer io.Closer, setupErr error) error {
69 if closer == nil {
70 return setupErr
71 }
72 if closeErr := closer.Close(); closeErr != nil {
73 return errors.Join(setupErr, fmt.Errorf("close performance trace after setup failure: %w", closeErr))
74 }
75 return setupErr
76}
77
78// runtimeTraceQueueDepth bounds all process-local trace producer handoffs.
79// A full queue emits the serialized diagnostic gap rather than delaying a
80// terminal, transport, or ACK progress path.
81const runtimeTraceQueueDepth = 256
82
83// runtimeTraceCloser owns the two-stage trace shutdown: drain the reporter
84// before closing the concrete timestamp/file owner. sync.Once makes every
85// deferred and explicit close path safe without closing shared workers twice.
86type runtimeTraceCloser struct {
87 reporter ports.SerializedRuntimeObserver
88 closer io.Closer
89 once sync.Once
90 err error
91}
92
93func (c *runtimeTraceCloser) Close() error {
94 if c == nil {
95 return nil
96 }
97 c.once.Do(func() {
98 if c.reporter != nil {
99 c.reporter.Flush()
100 c.reporter.Close()
101 }
102 if c.closer != nil {
103 c.err = c.closer.Close()
104 }
105 })
106 return c.err
107}