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