1package daemon
2
3import (
4 "errors"
5 "io"
6 "strings"
7 "sync"
8 "testing"
9 "time"
10
11 vt "github.com/bnema/vev-vt"
12 renderer "github.com/bnema/vev-vt/ansi"
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/stretchr/testify/require"
16)
17
18type datagramTestTransport struct {
19 sends chan ports.Frame
20 recv chan ports.Frame
21 once sync.Once
22}
23
24type failingOutputTransport struct{}
25
26func (failingOutputTransport) Send(ports.Frame) error { return errors.New("send failed") }
27func (failingOutputTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
28func (failingOutputTransport) Close() error { return nil }
29
30type asyncPaintTransport struct {
31 syncSends chan ports.Frame
32 asyncSends chan ports.Frame
33}
34
35func (t *asyncPaintTransport) Send(f ports.Frame) error { t.syncSends <- f; return nil }
36func (t *asyncPaintTransport) SendAsync(f ports.Frame) error { t.asyncSends <- f; return nil }
37func (t *asyncPaintTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF }
38func (t *asyncPaintTransport) Close() error { return nil }
39
40type timedSideEffectTransport struct {
41 closeTrackingTransport
42}
43
44func (t *timedSideEffectTransport) SendSynchronous(f ports.Frame) error {
45 return t.Send(f)
46}
47
48type noWatchdogClock struct{}
49
50func (noWatchdogClock) Now() time.Time { return time.Time{} }
51func (noWatchdogClock) NewTimer(time.Duration) ports.Timer {
52 panic("owned synchronous send must not install daemon watchdog")
53}
54
55func newDatagramTestTransport() *datagramTestTransport {
56 return &datagramTestTransport{sends: make(chan ports.Frame, 64), recv: make(chan ports.Frame, 64)}
57}
58
59func (t *datagramTestTransport) Send(f ports.Frame) error { t.sends <- f; return nil }
60func (t *datagramTestTransport) Recv() (ports.Frame, error) {
61 f, ok := <-t.recv
62 if !ok {
63 return ports.Frame{}, io.EOF
64 }
65 return f, nil
66}
67func (t *datagramTestTransport) Close() error { t.once.Do(func() { close(t.recv) }); return nil }
68
69func TestDatagramAttachPipelinesRendererBeforeAck(t *testing.T) {
70 p, releasePTY := newBlockingPTY(t)
71 defer releasePTY()
72 d, sess, _, _ := newManualSessionWithPTYs(t, p)
73 tr := newDatagramTestTransport()
74 routed, ac, err := d.route(ports.Hello{
75 Version: ports.ProtocolVersion,
76 Intent: ports.IntentAttach,
77 Name: sess.name,
78 Size: domain.Size{Cols: 80, Rows: 24},
79 MaxOutputInFlight: 8,
80 }, tr)
81 require.NoError(t, err)
82 require.Same(t, sess, routed)
83
84 // route publishes membership only; mark the attachment ready before the
85 // direct pipeline paints so coordinator admission is deterministic.
86 rc := sess.renderCoordinator()
87 require.NotNil(t, rc)
88 require.True(t, rc.markAttachmentReady(rc.attachmentLease(ac)))
89
90 sess.tabs[0].focusedPane().screen.Write([]byte("A"))
91 d.paint(sess, ac, false, nil)
92 first := awaitFrame(t, tr.sends, ports.MsgOutput)
93 out, err := ports.UnmarshalOutput(first.Payload)
94 require.NoError(t, err)
95 require.Equal(t, uint64(1), out.New)
96
97 sess.tabs[0].focusedPane().screen.Write([]byte("\rB"))
98 d.paint(sess, ac, false, nil)
99 updated := awaitFrame(t, tr.sends, ports.MsgOutput)
100 out, err = ports.UnmarshalOutput(updated.Payload)
101 require.NoError(t, err)
102 require.Equal(t, uint64(2), out.New)
103 // Both states were emitted before the MsgAck; the ordered dependency chain
104 // must therefore remain valid without waiting for renderer acknowledgement.
105 tr.recv <- ports.Frame{Type: ports.MsgAck, Payload: mustMarshalAck(ports.Ack{Epoch: out.Epoch, State: out.New})}
106 require.NoError(t, tr.Close())
107 d.runConnLoop(ac)
108
109 // The production MsgAck path retires retained states without moving the
110 // renderer baseline backward.
111 d.paint(sess, ac, false, nil)
112 require.Empty(t, tr.sends)
113}
114
115func TestPaintExplicitlyUsesAsyncTransportCapability(t *testing.T) {
116 p, releasePTY := newBlockingPTY(t)
117 defer releasePTY()
118 d, sess, ac, _ := newManualSessionWithPTYs(t, p)
119 tr := &asyncPaintTransport{syncSends: make(chan ports.Frame, 1), asyncSends: make(chan ports.Frame, 1)}
120 ac.replaceTransport(tr)
121 sess.tabs[0].focusedPane().screen.Write([]byte("A"))
122
123 d.paint(sess, ac, false, nil)
124
125 awaitFrame(t, tr.asyncSends, ports.MsgOutput)
126 select {
127 case frame := <-tr.syncSends:
128 t.Fatalf("paint used synchronous Send: %+v", frame)
129 default:
130 }
131}
132
133func TestDatagramMultipleUnackedScrollPaintsMatchLatestFrame(t *testing.T) {
134 p, releasePTY := newBlockingPTY(t)
135 defer releasePTY()
136 d, sess, _, _ := newManualSessionWithPTYs(t, p)
137 tr := newDatagramTestTransport()
138 _, ac, err := d.route(ports.Hello{
139 Version: ports.ProtocolVersion,
140 Intent: ports.IntentAttach,
141 Name: sess.name,
142 Size: domain.Size{Cols: 80, Rows: 24},
143 MaxOutputInFlight: 8,
144 }, tr)
145 require.NoError(t, err)
146
147 pane := sess.tabs[0].focusedPane()
148 for y := range pane.screen.Frame.Height {
149 cell := renderer.Cell{Rune: rune('A' + y), Style: renderer.DefaultStyle()}
150 for x := range pane.screen.Frame.Width {
151 pane.screen.Frame.Set(x, y, cell)
152 }
153 }
154 pane.screen.ClearDamage()
155 client := vt.NewScreen(80, 25)
156 d.paint(sess, ac, true, nil)
157 first := mustApplyOutput(t, client, awaitFrame(t, tr.sends, ports.MsgOutput))
158 ac.ackOutputState(first.Epoch, first.New)
159
160 // Preserve the frame after one scroll while inducing a real VT scroll
161 // damage event before each production paint. On main, both unacknowledged
162 // paints are generated from the ACKed frame and apply the scroll twice to
163 // the client. The ordered stream instead renders the second paint from the
164 // preceding emitted frame and overwrites the incompatible repeated damage.
165 desired := pane.screen.Frame.Clone()
166 desired.ScrollUp(0, desired.Height-1, 1)
167 for x := range desired.Width {
168 desired.Set(x, desired.Height-1, renderer.Cell{Rune: 'z', Style: renderer.DefaultStyle()})
169 }
170 for range 2 {
171 pane.screen.Frame = desired.Clone()
172 pane.screen.Row = pane.screen.Frame.Height - 1
173 pane.screen.Col = 0
174 pane.screen.Write([]byte("\nq"))
175 pane.screen.Frame = desired.Clone()
176 d.paint(sess, ac, false, nil)
177 mustApplyOutput(t, client, awaitFrame(t, tr.sends, ports.MsgOutput))
178 }
179
180 require.Equal(t, 'B', client.Frame.At(0, 1).Rune,
181 "client-visible content must equal the latest daemon-composed frame after pipelined scroll paints")
182}
183
184func fillOutputStateRows(frame renderer.Frame, rows []string) {
185 for y, row := range rows {
186 for x, r := range row {
187 frame.Set(x, y, renderer.Cell{Rune: r, Style: renderer.DefaultStyle()})
188 }
189 }
190}
191
192func outputStateRow(row []renderer.Cell) string {
193 runes := make([]rune, len(row))
194 for i, cell := range row {
195 runes[i] = cell.Rune
196 }
197 return string(runes)
198}
199
200func mustApplyOutput(t *testing.T, screen *vt.Screen, frame ports.Frame) ports.Output {
201 t.Helper()
202 out, err := ports.UnmarshalOutput(frame.Payload)
203 require.NoError(t, err)
204 screen.Write(out.Data)
205 return out
206}
207
208func TestLocalAttachStillAdvancesRendererOnSend(t *testing.T) {
209 p, releasePTY := newBlockingPTY(t)
210 defer releasePTY()
211 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
212
213 sess.tabs[0].focusedPane().screen.Write([]byte("A"))
214 d.paint(sess, ac, false, nil)
215 awaitFrame(t, sends, ports.MsgOutput)
216
217 // The first paint updates the renderer shadow without waiting for MsgAck.
218 d.paint(sess, ac, false, nil)
219 requireNoOutputFrame(t, sends)
220}
221
222func TestLocalOutputAckDoesNotMoveRendererShadowBackward(t *testing.T) {
223 p, releasePTY := newBlockingPTY(t)
224 defer releasePTY()
225 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
226 pane := sess.tabs[0].focusedPane()
227
228 pane.screen.Write([]byte("A"))
229 d.paint(sess, ac, false, nil)
230 first := awaitFrame(t, sends, ports.MsgOutput)
231 out, err := ports.UnmarshalOutput(first.Payload)
232 require.NoError(t, err)
233
234 pane.screen.Write([]byte("\rB"))
235 d.paint(sess, ac, false, nil)
236 awaitFrame(t, sends, ports.MsgOutput)
237
238 ac.ackOutputState(out.Epoch, out.New)
239 d.paint(sess, ac, false, nil)
240 requireNoOutputFrame(t, sends)
241}
242
243func TestRawTerminalSideEffectDoesNotEnterFullOutputWindow(t *testing.T) {
244 d := newTestDaemon(t, nil, stubClock{})
245 tr := &closeTrackingTransport{}
246 ac := &attachedClient{tr: tr, output: newOutputStateStream()}
247 ac.output.next = maxUnackedOutputStates
248
249 require.NoError(t, d.boundedSendOutputErr(ac, []byte("\x1b]52;c;YQ==\x07")))
250 require.Equal(t, uint64(maxUnackedOutputStates), ac.output.next)
251 require.Equal(t, uint64(maxUnackedOutputStates), ac.output.outstanding())
252 sends := tr.Sends()
253 require.Len(t, sends, 1)
254 out, err := ports.UnmarshalOutput(sends[0].Payload)
255 require.NoError(t, err)
256 require.Zero(t, out.Base)
257 require.Zero(t, out.New)
258}
259
260func TestOwnedSynchronousSideEffectSkipsOuterWatchdog(t *testing.T) {
261 d := newTestDaemon(t, nil, noWatchdogClock{})
262 tr := &timedSideEffectTransport{}
263 ac := &attachedClient{tr: tr, output: newOutputStateStream()}
264
265 require.NoError(t, d.boundedSendOutputErr(ac, make([]byte, 100*1024)))
266 require.Len(t, tr.Sends(), 1)
267}
268
269func TestFailedPaintSendRollsBackOutputState(t *testing.T) {
270 p, releasePTY := newBlockingPTY(t)
271 defer releasePTY()
272 d, sess, ac, _ := newManualSessionWithPTYs(t, p)
273 ac.replaceTransport(failingOutputTransport{})
274 sess.tabs[0].focusedPane().screen.Write([]byte("A"))
275
276 d.paint(sess, ac, false, nil)
277
278 require.Zero(t, ac.output.next)
279 require.Zero(t, ac.output.acked)
280}
281
282func TestResizeGrowthFirstFrameIncludesConcurrentPTYRedraw(t *testing.T) {
283 for _, tc := range []struct {
284 name string
285 window uint64
286 }{
287 {name: "datagram window", window: 1},
288 {name: "stream window", window: maxUnackedOutputStates},
289 } {
290 t.Run(tc.name, func(t *testing.T) {
291 // resizeReaderPTY delivers the child's SIGWINCH redraw through the
292 // actual ptyReader. Its Resize callback does not return until Read has
293 // accepted the bytes, at which point resizeApplying is still true.
294 p := newResizeReaderPTY([]byte("\x1b[1;81H" + strings.Repeat("B", 40)))
295 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
296 ac.output.maxOutstanding = tc.window
297 pane := testAttachmentTab(sess).focusedPane()
298 p.applying = func() bool {
299 pane.mu.Lock()
300 defer pane.mu.Unlock()
301 return pane.resizeApplying
302 }
303 pane.screen.Write([]byte(strings.Repeat("A", 79)))
304 d.paint(sess, ac, true, nil)
305 client := vt.NewScreen(80, 24)
306 initial := mustApplyOutput(t, client, awaitFrame(t, sends, ports.MsgOutput))
307 ac.ackOutputState(initial.Epoch, initial.New)
308
309 d.sessWg.Add(1)
310 pane.onExit = func() {}
311 go d.ptyReader(sess, testAttachmentTab(sess), pane)
312 defer func() { p.close(); d.sessWg.Wait() }()
313
314 // The resize deadline drives coordinator prepare/apply/commit. No test
315 // writes the grown columns directly.
316 clock := &signalClock{timers: make(chan *signalTimer, 2)}
317 d.clock = clock
318 d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24})
319 timer := <-clock.timers
320 done := captureResizeCallbackDone(t, sess.renderCoordinator())
321 timer.ch <- time.Time{}
322 awaitTestCompletion(t, done, "resize callback did not complete")
323 require.True(t, p.deliveredWhileApplying(), "redraw must arrive while PTY.Resize owns the apply gate")
324 require.Equal(t, 'B', pane.screen.Frame.At(100, 0).Rune, "replay must parse the redraw before commit")
325
326 client.Resize(120, 24)
327 mustApplyOutput(t, client, awaitFrame(t, sends, ports.MsgOutput))
328 require.Equal(t, 'B', client.Frame.At(100, 1).Rune,
329 "first grown frame exposed stale pre-SIGWINCH pane content")
330 select {
331 case extra := <-sends:
332 t.Fatalf("resize emitted more than one frame: %#v", extra)
333 default:
334 }
335 })
336 }
337}
338
339func TestResizeWithoutPTYOutputFlushesOneFullFrameAtDeadline(t *testing.T) {
340 p, releasePTY := newBlockingPTY(t)
341 defer releasePTY()
342 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
343 clk := &signalClock{timers: make(chan *signalTimer, 1)}
344 d.clock = clk
345
346 d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24})
347 var timer *signalTimer
348 select {
349 case timer = <-clk.timers:
350 case <-time.After(time.Second):
351 t.Fatal("resize did not schedule a bounded paint")
352 }
353 requireNoOutputFrame(t, sends)
354
355 done := captureResizeCallbackDone(t, sess.renderCoordinator())
356 timer.ch <- time.Now()
357 awaitTestCompletion(t, done, "resize callback did not complete")
358 frame := awaitFrame(t, sends, ports.MsgOutput)
359 client := vt.NewScreen(120, 24)
360 out := mustApplyOutput(t, client, frame)
361 require.Zero(t, out.Base)
362 require.Contains(t, screenLineText(client, 0), "1")
363 require.Contains(t, screenLineText(client, 23), "work")
364 require.Equal(t, domain.Size{Cols: 120, Rows: 24}, ac.size)
365 requireNoOutputFrame(t, sends)
366}
367
368func TestResizeBurstFlushesOnlyLatestGeometry(t *testing.T) {
369 p, releasePTY := newBlockingPTY(t)
370 defer releasePTY()
371 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
372 clk := &signalClock{timers: make(chan *signalTimer, 2)}
373 d.clock = clk
374
375 d.resize(sess, ac, domain.Size{Cols: 100, Rows: 24})
376 var first *signalTimer
377 select {
378 case first = <-clk.timers:
379 case <-time.After(time.Second):
380 t.Fatal("first resize did not schedule a bounded paint")
381 }
382 firstDone := captureResizeCallbackDone(t, sess.renderCoordinator())
383 d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24})
384 var latest *signalTimer
385 select {
386 case latest = <-clk.timers:
387 case <-time.After(time.Second):
388 t.Fatal("latest resize did not replace the bounded paint")
389 }
390
391 latestDone := captureResizeCallbackDone(t, sess.renderCoordinator())
392 first.ch <- time.Now()
393 awaitTestCompletion(t, firstDone, "obsolete resize callback did not complete")
394 requireNoOutputFrame(t, sends)
395 latest.ch <- time.Now()
396 awaitTestCompletion(t, latestDone, "latest resize callback did not complete")
397 awaitFrame(t, sends, ports.MsgOutput)
398 require.Equal(t, domain.Size{Cols: 120, Rows: 24}, ac.size)
399 require.Equal(t, 120, testAttachmentTab(sess).focusedPane().screen.Frame.Width)
400 requireNoOutputFrame(t, sends)
401}