1package daemon
2
3import (
4 "runtime"
5 "sync"
6 "sync/atomic"
7 "testing"
8 "time"
9
10 "github.com/bnema/vev/internal/adapters/observability"
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13)
14
15// This acceptance boundary exercises the attached resize, render, and ACK
16// path rather than merely listing the marks the implementation should emit.
17func TestTransportObservabilityDaemonBoundaries(t *testing.T) {
18 observer := &daemonRuntimeObserver{}
19 d, sess, ac, sends := newManualSessionWithPTYs(t, &quietPTY{})
20 d.runtimeObserver = observer
21 // One in-flight state makes the second public resize take the ACK-blocked
22 // path. The real coordinator owns both renders and the ACK wake.
23 ac.output = newOutputStateStream(1)
24 rc := d.attachCoordinator(sess, nil, ac, true)
25
26 if !d.resizeForFirstPaint(sess, ac, domain.Size{Cols: 90, Rows: 25}) {
27 t.Fatal("first resize was not accepted")
28 }
29 first := awaitFrame(t, sends, "Output")
30 if envelopeMessageName(t, first.Payload) != "Output" {
31 t.Fatalf("first frame = %v, want output", envelopeMessageName(t, first.Payload))
32 }
33 output := unmarshalTestOutput(t, first.Payload)
34 if !d.resizeForFirstPaint(sess, ac, domain.Size{Cols: 100, Rows: 26}) {
35 t.Fatal("second resize was not accepted")
36 }
37 ac.ackOutputState(output.Epoch, output.New)
38 rc.notifyAck()
39 second := awaitFrame(t, sends, "Output")
40 if envelopeMessageName(t, second.Payload) != "Output" {
41 t.Fatalf("second frame = %v, want output", envelopeMessageName(t, second.Payload))
42 }
43
44 want := []ports.RuntimeMarkKind{
45 ports.RuntimeResizeRequested, ports.RuntimeResizeCommitted,
46 ports.RuntimeQueueEnqueued, ports.RuntimeQueueDequeued,
47 ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd,
48 ports.RuntimeComposeStart, ports.RuntimeComposeEnd,
49 ports.RuntimeDiffStart, ports.RuntimeDiffEnd,
50 ports.RuntimeEmitStart, ports.RuntimeEmitEnd,
51 ports.RuntimeResizeRequested, ports.RuntimeResizeCommitted,
52 ports.RuntimeQueueEnqueued, ports.RuntimeACKBlockedStart,
53 ports.RuntimeACKBlockedEnd, ports.RuntimeQueueDequeued,
54 ports.RuntimeCaptureStart, ports.RuntimeCaptureEnd,
55 ports.RuntimeComposeStart, ports.RuntimeComposeEnd,
56 ports.RuntimeDiffStart, ports.RuntimeDiffEnd,
57 ports.RuntimeEmitStart, ports.RuntimeEmitEnd,
58 }
59 if len(observer.marks) != len(want) {
60 t.Fatalf("runtime marks = %#v, want kinds %#v", observer.marks, want)
61 }
62 for i, kind := range want {
63 if observer.marks[i].Kind != kind {
64 t.Fatalf("runtime mark %d = %q, want %q; all=%#v", i, observer.marks[i].Kind, kind, observer.marks)
65 }
66 }
67}
68
69type daemonRuntimeObserver struct{ marks []ports.RuntimeMark }
70
71func (o *daemonRuntimeObserver) ObserveRuntime(mark ports.RuntimeMark) {
72 o.marks = append(o.marks, mark)
73}
74
75type blockingDaemonRuntimeObserver struct {
76 mu sync.Mutex
77 entered chan struct{}
78 release chan struct{}
79 blocked bool
80}
81
82func newBlockingDaemonRuntimeObserver() *blockingDaemonRuntimeObserver {
83 return &blockingDaemonRuntimeObserver{entered: make(chan struct{}), release: make(chan struct{})}
84}
85
86func (o *blockingDaemonRuntimeObserver) ObserveRuntime(ports.RuntimeMark) {
87 o.mu.Lock()
88 defer o.mu.Unlock()
89 if !o.blocked {
90 o.blocked = true
91 close(o.entered)
92 <-o.release
93 }
94}
95
96func awaitDaemonObserver(t *testing.T, ch <-chan struct{}, what string) {
97 t.Helper()
98 select {
99 case <-ch:
100 case <-time.After(time.Second):
101 t.Fatalf("timed out waiting for %s", what)
102 }
103}
104
105// A runtime observer may synchronously write JSONL. Its I/O must run only
106// after the render transaction has released attachment, session, and pane
107// ownership, so a blocked trace cannot stall live output processing.
108func TestTransportObservabilityBlockedRenderObserverReleasesArchitectureLocks(t *testing.T) {
109 d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
110 observer := newBlockingDaemonRuntimeObserver()
111 d.runtimeObserver = observer
112
113 paintDone := make(chan struct{})
114 go func() {
115 d.paint(sess, ac, true, nil)
116 close(paintDone)
117 }()
118 awaitDaemonObserver(t, observer.entered, "blocked render observer")
119 frame := awaitFrame(t, sends, "Output") // The frame was emitted before the observer began its I/O.
120 output := unmarshalTestOutput(t, frame.Payload)
121 ac.ackOutputState(output.Epoch, output.New)
122
123 pane := testAttachmentTab(sess).focusedPane()
124 for _, lock := range []struct {
125 name string
126 fn func()
127 }{
128 {"attachment", func() { ac.sendMu.Lock(); runtime.Gosched(); ac.sendMu.Unlock() }},
129 {"session", func() { sess.mu.Lock(); runtime.Gosched(); sess.mu.Unlock() }},
130 {"pane", func() { pane.mu.Lock(); runtime.Gosched(); pane.mu.Unlock() }},
131 } {
132 done := make(chan struct{})
133 go func(fn func()) { fn(); close(done) }(lock.fn)
134 awaitDaemonObserver(t, done, lock.name+" lock")
135 }
136
137 livePTY := make(chan struct{})
138 go func() {
139 d.processPTYData(sess, testAttachmentTab(sess), pane, []byte("live"), false)
140 close(livePTY)
141 }()
142 awaitDaemonObserver(t, livePTY, "live PTY processing")
143
144 // A second render can complete its actual pipeline while the first trace
145 // write is blocked; it is not serialized behind the first attachment lock.
146 secondPaint := make(chan struct{})
147 go func() {
148 d.paint(sess, ac, false, nil)
149 close(secondPaint)
150 }()
151 _ = awaitFrame(t, sends, "Output") // The second frame is live even though serialized observer I/O waits.
152
153 close(observer.release)
154 awaitDaemonObserver(t, paintDone, "first render completion")
155 awaitDaemonObserver(t, secondPaint, "unrelated render completion")
156}
157
158// ACK-blocked start is captured while coordinator state changes, then emitted
159// after c.mu is unlocked. This closes the remaining coordinator-observer path.
160func TestTransportObservabilityACKBlockedSpansEndExactlyOnce(t *testing.T) {
161 cases := []struct {
162 name string
163 finish func(*renderCoordinator, *bool)
164 }{
165 {
166 name: "notify ACK",
167 finish: func(rc *renderCoordinator, ready *bool) {
168 *ready = true
169 rc.notifyAck()
170 },
171 },
172 {
173 name: "successful consume",
174 finish: func(rc *renderCoordinator, ready *bool) {
175 *ready = true
176 rc.fire(rc.normalLane.generation, false, false)
177 },
178 },
179 {
180 name: "teardown",
181 finish: func(rc *renderCoordinator, ready *bool) {
182 rc.beginSessionTeardown().finish()
183 },
184 },
185 }
186 for _, tt := range cases {
187 t.Run(tt.name, func(t *testing.T) {
188 observer := &daemonRuntimeObserver{}
189 ready := false
190 rc := newRenderCoordinator(renderCoordinatorOptions{
191 observer: observer,
192 ackReadyFor: func(*attachedClient) bool { return ready },
193 })
194 rc.attach(&attachedClient{})
195 rc.invalidate(renderInvalidation{class: invalidateOutput})
196 rc.fire(rc.normalLane.generation, false, true)
197 tt.finish(rc, &ready)
198 // A late notification after consuming or tearing down must not emit a
199 // second end for the already-cleared blocked interval.
200 rc.notifyAck()
201
202 var starts, ends []ports.RuntimeMark
203 for _, mark := range observer.marks {
204 switch mark.Kind {
205 case ports.RuntimeACKBlockedStart:
206 starts = append(starts, mark)
207 case ports.RuntimeACKBlockedEnd:
208 ends = append(ends, mark)
209 }
210 }
211 if len(starts) != 1 || len(ends) != 1 {
212 t.Fatalf("ACK-blocked span marks: starts=%+v ends=%+v all=%+v", starts, ends, observer.marks)
213 }
214 if starts[0].Sequence != ends[0].Sequence || starts[0].RequestID != ends[0].RequestID || starts[0].Epoch != ends[0].Epoch {
215 t.Fatalf("ACK-blocked span correlation changed: start=%+v end=%+v", starts[0], ends[0])
216 }
217 })
218 }
219}
220
221// A raw observer supplied through daemon configuration is serialized by the
222// daemon. Once Start has been published, a blocked End must not delay the ACK
223// that consumes deferred work and delivers its wake.
224func TestTransportObservabilityBlockedACKEndDoesNotDelayNotifyAck(t *testing.T) {
225 observer := &blockingACKEndObserver{
226 startObserved: make(chan struct{}),
227 endEntered: make(chan struct{}),
228 releaseEnd: make(chan struct{}),
229 }
230 reporter := observability.NewSerialized(observer, 64)
231 defer reporter.Close()
232 d := New(nil, nil, nil, WithRuntimeObserver(reporter))
233 wakes := make(chan renderWake, 1)
234 var ready atomic.Bool
235 rc := newRenderCoordinator(renderCoordinatorOptions{
236 observer: d.runtimeObserver,
237 ackReadyFor: func(*attachedClient) bool { return ready.Load() },
238 wake: func(w renderWake) { wakes <- w },
239 })
240 rc.attach(&attachedClient{})
241 rc.invalidate(renderInvalidation{class: invalidateOutput})
242
243 // Enter the ACK-blocked state and wait until Start has reached the injected
244 // observer. End is the only deliberately blocked call in this test.
245 rc.fire(rc.normalLane.generation, false, true)
246 awaitDaemonObserver(t, observer.startObserved, "ACK-blocked start observer")
247
248 ready.Store(true)
249 ackDone := make(chan struct{})
250 go func() {
251 rc.notifyAck()
252 close(ackDone)
253 }()
254 awaitDaemonObserver(t, observer.endEntered, "blocked ACK-blocked end observer")
255 awaitDaemonObserver(t, ackDone, "nonblocking ACK progress")
256 awaitWake(t, wakes)
257 rc.mu.Lock()
258 pending := rc.pending
259 rc.mu.Unlock()
260 if pending {
261 t.Fatal("notifyAck left consumed ACK-deferred work pending")
262 }
263
264 close(observer.releaseEnd)
265 reporter.Close()
266}
267
268type blockingACKEndObserver struct {
269 startObserved chan struct{}
270 endEntered chan struct{}
271 releaseEnd chan struct{}
272}
273
274func (o *blockingACKEndObserver) ObserveRuntime(mark ports.RuntimeMark) {
275 switch mark.Kind {
276 case ports.RuntimeACKBlockedStart:
277 close(o.startObserved)
278 case ports.RuntimeACKBlockedEnd:
279 close(o.endEntered)
280 <-o.releaseEnd
281 }
282}
283
284func TestTransportObservabilityBlockedCoordinatorObserverReleasesLock(t *testing.T) {
285 observer := newBlockingDaemonRuntimeObserver()
286 rc := newRenderCoordinator(renderCoordinatorOptions{
287 observer: observer,
288 ackReadyFor: func(*attachedClient) bool { return false },
289 })
290 rc.attach(&attachedClient{})
291 rc.mu.Lock()
292 rc.pending = true
293 gen := rc.normalLane.generation
294 rc.mu.Unlock()
295
296 fired := make(chan struct{})
297 go func() {
298 rc.fire(gen, false, true)
299 close(fired)
300 }()
301 awaitDaemonObserver(t, observer.entered, "blocked ACK observer")
302
303 acquired := make(chan struct{})
304 go func() {
305 rc.mu.Lock()
306 runtime.Gosched()
307 rc.mu.Unlock()
308 close(acquired)
309 }()
310 awaitDaemonObserver(t, acquired, "coordinator lock")
311
312 close(observer.release)
313 awaitDaemonObserver(t, fired, "ACK fire completion")
314}
315
316type ackSpanInterleavingObserver struct {
317 mu sync.Mutex
318 marks []ports.RuntimeMark
319 startEntered chan struct{}
320 releaseStart chan struct{}
321}
322
323func newACKSpanInterleavingObserver() *ackSpanInterleavingObserver {
324 return &ackSpanInterleavingObserver{startEntered: make(chan struct{}), releaseStart: make(chan struct{})}
325}
326
327func (o *ackSpanInterleavingObserver) ObserveRuntime(mark ports.RuntimeMark) {
328 if mark.Kind == ports.RuntimeACKBlockedStart {
329 close(o.startEntered)
330 <-o.releaseStart
331 }
332 o.mu.Lock()
333 o.marks = append(o.marks, mark)
334 o.mu.Unlock()
335}
336
337func (o *ackSpanInterleavingObserver) snapshot() []ports.RuntimeMark {
338 o.mu.Lock()
339 defer o.mu.Unlock()
340 return append([]ports.RuntimeMark(nil), o.marks...)
341}
342
343// An ACK can consume deferred work while the start mark is still waiting on
344// observer I/O. The End must wait for that Start publication, but the ACK
345// itself must not wait for either observer call.
346func TestTransportObservabilityACKBlockedStartPublishesBeforeEnd(t *testing.T) {
347 observer := newACKSpanInterleavingObserver()
348 ready := false
349 rc := newRenderCoordinator(renderCoordinatorOptions{
350 observer: observer,
351 ackReadyFor: func(*attachedClient) bool { return ready },
352 })
353 rc.attach(&attachedClient{})
354 rc.mu.Lock()
355 rc.pending = true
356 gen := rc.normalLane.generation
357 rc.mu.Unlock()
358
359 fireDone := make(chan struct{})
360 go func() {
361 rc.fire(gen, false, true)
362 close(fireDone)
363 }()
364 awaitDaemonObserver(t, observer.startEntered, "ACK-blocked start observer")
365
366 ready = true
367 ackDone := make(chan struct{})
368 go func() {
369 rc.notifyAck()
370 close(ackDone)
371 }()
372 awaitDaemonObserver(t, ackDone, "nonblocking ACK progress")
373 if marks := observer.snapshot(); len(marks) != 0 {
374 t.Fatalf("ACK-blocked end published before its blocked start: %#v", marks)
375 }
376
377 close(observer.releaseStart)
378 awaitDaemonObserver(t, fireDone, "ACK-blocked start completion")
379 marks := observer.snapshot()
380 if len(marks) != 2 || marks[0].Kind != ports.RuntimeACKBlockedStart || marks[1].Kind != ports.RuntimeACKBlockedEnd {
381 t.Fatalf("ACK-blocked marks = %#v, want one ordered start/end pair", marks)
382 }
383 if marks[0].Sequence != marks[1].Sequence || marks[0].RequestID != marks[1].RequestID || marks[0].Epoch != marks[1].Epoch {
384 t.Fatalf("ACK-blocked span correlation changed: start=%+v end=%+v", marks[0], marks[1])
385 }
386}