1package ports
2
3import (
4 "errors"
5 "sync"
6 "sync/atomic"
7)
8
9// RuntimeMarkSchema is the version of the process-local performance trace.
10const RuntimeMarkSchema uint16 = 1
11
12// RuntimeMarkKind identifies a process-local runtime boundary. It deliberately
13// contains no protocol payload, cells, or rendering policy.
14type RuntimeMarkKind string
15
16const (
17 RuntimeInputInjected RuntimeMarkKind = "input_injected"
18 RuntimeResizeRequested RuntimeMarkKind = "resize_requested"
19 RuntimeResizeCommitted RuntimeMarkKind = "resize_committed"
20 RuntimeCaptureStart RuntimeMarkKind = "capture_start"
21 RuntimeCaptureEnd RuntimeMarkKind = "capture_end"
22 RuntimeComposeStart RuntimeMarkKind = "compose_start"
23 RuntimeComposeEnd RuntimeMarkKind = "compose_end"
24 RuntimeDiffStart RuntimeMarkKind = "diff_start"
25 RuntimeDiffEnd RuntimeMarkKind = "diff_end"
26 RuntimeQueueEnqueued RuntimeMarkKind = "queue_enqueued"
27 RuntimeQueueDequeued RuntimeMarkKind = "queue_dequeued"
28 RuntimeACKBlockedStart RuntimeMarkKind = "ack_blocked_start"
29 RuntimeACKBlockedEnd RuntimeMarkKind = "ack_blocked_end"
30 RuntimeEmitStart RuntimeMarkKind = "emit_start"
31 RuntimeEmitEnd RuntimeMarkKind = "emit_end"
32 RuntimeAdapterSendStart RuntimeMarkKind = "adapter_send_start"
33 RuntimeAdapterSendEnd RuntimeMarkKind = "adapter_send_end"
34 RuntimeAdapterReceiveStart RuntimeMarkKind = "adapter_receive_start"
35 RuntimeAdapterReceiveEnd RuntimeMarkKind = "adapter_receive_end"
36 RuntimeTerminalFlushed RuntimeMarkKind = "terminal_flushed"
37 RuntimeTransportDiagnostic RuntimeMarkKind = "transport_diagnostic"
38)
39
40// RuntimeMark is a process-local observation record. Tick is intentionally
41// left zero by producers: the concrete process observer owns timestamping.
42type RuntimeMark struct {
43 Schema uint16
44 ProcessID string
45 Component string
46 Scenario string
47 Run uint64
48 Sequence uint64
49 RequestID uint64
50 Epoch uint64
51 Kind RuntimeMarkKind
52 Tick int64
53 Bytes uint64
54 Fragments uint64
55 Retransmits uint64
56 Pending uint64
57 AckRTTNanos int64
58 Valid bool
59}
60
61// RuntimeObserver receives process-local marks. Implementations must not let
62// a producer provide a timestamp in another clock domain.
63type RuntimeObserver interface{ ObserveRuntime(RuntimeMark) }
64
65// SerializedRuntimeObserver is the only observer contract accepted by runtime
66// transport and use-case hot paths. ObserveRuntime must return without waiting
67// for sink I/O. Flush and Close wait until every accepted mark has reached the
68// concrete observer; Close is idempotent and rejects later marks.
69type SerializedRuntimeObserver interface {
70 RuntimeObserver
71 Flush()
72 Close()
73}
74
75// NewSerializedRuntimeObserver isolates producers from a potentially slow
76// observer with one ordered worker. capacity bounds queued marks. If the queue
77// fills, dropped marks are represented by one ordered diagnostic mark before
78// reporting resumes, so a trace can never silently claim completeness.
79func NewSerializedRuntimeObserver(observer RuntimeObserver, capacity int) SerializedRuntimeObserver {
80 if observer == nil {
81 return nil
82 }
83 if capacity < 1 {
84 capacity = 1
85 }
86 o := &serializedRuntimeObserver{observer: observer, capacity: capacity, done: make(chan struct{})}
87 o.cond = sync.NewCond(&o.mu)
88 go o.run()
89 return o
90}
91
92// EnsureSerializedRuntimeObserver returns observer's existing serialized
93// owner when it has one. Otherwise it creates an owned worker. Callers close
94// the result only when owned is true, preventing nested workers and duplicate
95// lifecycle ownership when application wiring already serialized the process
96// observer.
97func EnsureSerializedRuntimeObserver(observer RuntimeObserver, capacity int) (serialized SerializedRuntimeObserver, owned bool) {
98 if observer == nil {
99 return nil, false
100 }
101 if serialized, ok := observer.(SerializedRuntimeObserver); ok {
102 return serialized, false
103 }
104 return NewSerializedRuntimeObserver(observer, capacity), true
105}
106
107type serializedRuntimeObserver struct {
108 observer RuntimeObserver
109 capacity int
110
111 mu sync.Mutex
112 cond *sync.Cond
113 marks []RuntimeMark
114 dropped uint64
115 closed bool
116 active bool
117 done chan struct{}
118}
119
120func (o *serializedRuntimeObserver) ObserveRuntime(mark RuntimeMark) {
121 if o == nil {
122 return
123 }
124 o.mu.Lock()
125 defer o.mu.Unlock()
126 if o.closed {
127 return
128 }
129 if o.dropped != 0 {
130 // Reserve space for both the gap marker and the first post-gap mark.
131 // This preserves the producer order around a bounded-queue loss.
132 if len(o.marks)+2 > o.capacity {
133 o.dropped++
134 return
135 }
136 o.marks = append(o.marks, runtimeObserverGapMark(o.dropped), mark)
137 o.dropped = 0
138 o.cond.Signal()
139 return
140 }
141 if len(o.marks) == o.capacity {
142 o.dropped = 1
143 return
144 }
145 o.marks = append(o.marks, mark)
146 o.cond.Signal()
147}
148
149func runtimeObserverGapMark(dropped uint64) RuntimeMark {
150 return NewRuntimeMark("observer", RuntimeTransportDiagnostic, dropped, false)
151}
152
153func (o *serializedRuntimeObserver) Flush() {
154 if o == nil {
155 return
156 }
157 o.mu.Lock()
158 for len(o.marks) != 0 || o.active || o.dropped != 0 {
159 o.cond.Signal()
160 o.cond.Wait()
161 }
162 closed := o.closed
163 o.mu.Unlock()
164 if closed {
165 <-o.done
166 }
167}
168
169func (o *serializedRuntimeObserver) Close() {
170 if o == nil {
171 return
172 }
173 o.mu.Lock()
174 if !o.closed {
175 o.closed = true
176 o.cond.Broadcast()
177 }
178 o.mu.Unlock()
179 <-o.done
180}
181
182func (o *serializedRuntimeObserver) run() {
183 defer close(o.done)
184 for {
185 o.mu.Lock()
186 for len(o.marks) == 0 && o.dropped == 0 && !o.closed {
187 o.cond.Wait()
188 }
189 if len(o.marks) == 0 && o.dropped != 0 {
190 o.marks = append(o.marks, runtimeObserverGapMark(o.dropped))
191 o.dropped = 0
192 }
193 if len(o.marks) == 0 && o.closed {
194 o.cond.Broadcast()
195 o.mu.Unlock()
196 return
197 }
198 mark := o.marks[0]
199 o.marks = o.marks[1:]
200 o.active = true
201 o.mu.Unlock()
202
203 o.observer.ObserveRuntime(mark)
204
205 o.mu.Lock()
206 o.active = false
207 o.cond.Broadcast()
208 o.mu.Unlock()
209 }
210}
211
212// RuntimeMarkValid reports whether a mark has all correlation fields required
213// by the JSONL schema. Tick is checked separately by the timestamp owner.
214func RuntimeMarkValid(mark RuntimeMark) bool {
215 if mark.Schema != RuntimeMarkSchema || mark.Component == "" || mark.Scenario == "" || mark.Run == 0 || mark.Sequence == 0 || mark.RequestID == 0 || mark.Epoch == 0 {
216 return false
217 }
218 switch mark.Kind {
219 case RuntimeInputInjected, RuntimeResizeRequested, RuntimeResizeCommitted,
220 RuntimeCaptureStart, RuntimeCaptureEnd, RuntimeComposeStart, RuntimeComposeEnd,
221 RuntimeDiffStart, RuntimeDiffEnd, RuntimeQueueEnqueued, RuntimeQueueDequeued,
222 RuntimeACKBlockedStart, RuntimeACKBlockedEnd, RuntimeEmitStart, RuntimeEmitEnd,
223 RuntimeAdapterSendStart, RuntimeAdapterSendEnd, RuntimeAdapterReceiveStart,
224 RuntimeAdapterReceiveEnd, RuntimeTerminalFlushed, RuntimeTransportDiagnostic:
225 return true
226 default:
227 return false
228 }
229}
230
231// RuntimeCorrelation identifies one attempted operation. It is deliberately
232// copied to both endpoints of a span; process ID remains owned by the sink.
233type RuntimeCorrelation struct {
234 Scenario string
235 Run, Sequence, RequestID, Epoch uint64
236}
237
238// RuntimeCorrelationInputs are supplied once by process orchestration. They
239// are deliberately data-only: consumer APIs continue to receive only their
240// one RuntimeObserver argument and never a clock or tick.
241type RuntimeCorrelationInputs struct {
242 Scenario string
243 Run uint64
244}
245
246// NewRuntimeCorrelationObserver binds every mark from one OS process to its
247// harness manifest scenario/run. Operation IDs remain producer-owned when
248// supplied; missing IDs are assigned from the process-wide allocator.
249func NewRuntimeCorrelationObserver(observer RuntimeObserver, inputs RuntimeCorrelationInputs) (RuntimeObserver, error) {
250 if observer == nil || inputs.Scenario == "" || inputs.Run == 0 {
251 return nil, errors.New("invalid runtime correlation inputs")
252 }
253 return &runtimeCorrelationObserver{observer: observer, inputs: inputs}, nil
254}
255
256type runtimeCorrelationObserver struct {
257 observer RuntimeObserver
258 inputs RuntimeCorrelationInputs
259}
260
261func (o *runtimeCorrelationObserver) ObserveRuntime(mark RuntimeMark) {
262 if o == nil || o.observer == nil {
263 return
264 }
265 mark.Scenario, mark.Run = o.inputs.Scenario, o.inputs.Run
266 if mark.Sequence == 0 {
267 mark.Sequence = runtimeMarkSequence.Add(1)
268 }
269 if mark.RequestID == 0 {
270 mark.RequestID = mark.Sequence
271 }
272 if mark.Epoch == 0 {
273 mark.Epoch = mark.RequestID
274 }
275 o.observer.ObserveRuntime(mark)
276}
277
278var runtimeMarkSequence atomic.Uint64
279
280// NewRuntimeCorrelation makes a unique, process-local operation identity.
281// Components with a domain request/epoch should use those values directly via
282// NewRuntimeMarkWithCorrelation instead of inventing a second identity.
283func NewRuntimeCorrelation() RuntimeCorrelation {
284 sequence := runtimeMarkSequence.Add(1)
285 return RuntimeCorrelation{Scenario: "runtime", Run: 1, Sequence: sequence, RequestID: sequence, Epoch: sequence}
286}
287
288// NewRuntimeMarkWithCorrelation returns a timestamp-free mark carrying the
289// supplied operation identity. Start/end callers must reuse correlation.
290func NewRuntimeMarkWithCorrelation(component string, correlation RuntimeCorrelation, kind RuntimeMarkKind, bytes uint64, valid bool) RuntimeMark {
291 return RuntimeMark{Schema: RuntimeMarkSchema, Component: component, Scenario: correlation.Scenario, Run: correlation.Run, Sequence: correlation.Sequence, RequestID: correlation.RequestID, Epoch: correlation.Epoch, Kind: kind, Bytes: bytes, Valid: valid}
292}
293
294// NewRuntimeMark is retained for one-shot marks. Unlike the old helper it
295// never returns constant correlation IDs.
296func NewRuntimeMark(component string, kind RuntimeMarkKind, bytes uint64, valid bool) RuntimeMark {
297 return NewRuntimeMarkWithCorrelation(component, NewRuntimeCorrelation(), kind, bytes, valid)
298}