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/bnema/vev/internal/protocol"
16 "github.com/bnema/vev/internal/protocol/wire"
17 "github.com/stretchr/testify/require"
18)
19
20type datagramTestTransport struct {
21 sends chan wire.Envelope
22 recv chan wire.Envelope
23 once sync.Once
24}
25
26type failingOutputTransport struct{}
27
28func (failingOutputTransport) Send(wire.Envelope) error { return errors.New("send failed") }
29func (failingOutputTransport) Recv() (wire.Envelope, error) { return wire.Envelope{}, io.EOF }
30func (failingOutputTransport) Close() error { return nil }
31
32type asyncPaintTransport struct {
33 syncSends chan wire.Envelope
34 asyncSends chan wire.Envelope
35}
36
37func (t *asyncPaintTransport) Send(f wire.Envelope) error { t.syncSends <- f; return nil }
38func (t *asyncPaintTransport) SendAsync(f wire.Envelope) error { t.asyncSends <- f; return nil }
39func (t *asyncPaintTransport) Recv() (wire.Envelope, error) { return wire.Envelope{}, io.EOF }
40func (t *asyncPaintTransport) Close() error { return nil }
41
42type timedSideEffectTransport struct {
43 closeTrackingTransport
44}
45
46func (t *timedSideEffectTransport) SendSynchronous(f wire.Envelope) error {
47 return t.Send(f)
48}
49
50type noWatchdogClock struct{}
51
52func (noWatchdogClock) Now() time.Time { return time.Time{} }
53func (noWatchdogClock) NewTimer(time.Duration) ports.Timer {
54 panic("owned synchronous send must not install daemon watchdog")
55}
56
57func newDatagramTestTransport() *datagramTestTransport {
58 return &datagramTestTransport{sends: make(chan wire.Envelope, 64), recv: make(chan wire.Envelope, 64)}
59}
60
61func (t *datagramTestTransport) Send(f wire.Envelope) error { t.sends <- f; return nil }
62func (t *datagramTestTransport) Recv() (wire.Envelope, error) {
63 f, ok := <-t.recv
64 if !ok {
65 return wire.Envelope{}, io.EOF
66 }
67 return f, nil
68}
69func (t *datagramTestTransport) Close() error { t.once.Do(func() { close(t.recv) }); return nil }
70
71func TestDatagramAttachPipelinesRendererBeforeAck(t *testing.T) {
72 p, releasePTY := newBlockingPTY(t)
73 defer releasePTY()
74 d, sess, _, _ := newManualSessionWithPTYs(t, p)
75 tr := newDatagramTestTransport()
76 routed, ac, err := d.route(protocol.Hello{
77 Version: protocol.Version,
78 Intent: protocol.IntentAttach,
79 Name: sess.name,
80 Size: domain.Size{Cols: 80, Rows: 24},
81 MaxOutputInFlight: 8,
82 }, tr)
83 require.NoError(t, err)
84 require.Same(t, sess, routed)
85
86 // route publishes membership only; mark the attachment ready before the
87 // direct pipeline paints so coordinator admission is deterministic.
88 rc := sess.renderCoordinator()
89 require.NotNil(t, rc)
90 require.True(t, rc.markAttachmentReady(rc.attachmentLease(ac)))
91
92 sess.tabs[0].focusedPane().screen.Write([]byte("A"))
93 d.paint(sess, ac, false, nil)
94 first := awaitFrame(t, tr.sends, "Output")
95 out := unmarshalTestOutput(t, first.Payload)
96 require.Equal(t, uint64(1), out.New)
97
98 sess.tabs[0].focusedPane().screen.Write([]byte("\rB"))
99 d.paint(sess, ac, false, nil)
100 updated := awaitFrame(t, tr.sends, "Output")
101 out = unmarshalTestOutput(t, updated.Payload)
102 require.Equal(t, uint64(2), out.New)
103 // Both states were emitted before the Ack; the ordered dependency chain
104 // must therefore remain valid without waiting for renderer acknowledgement.
105 tr.recv <- mustClientEnvelope(protocol.Ack{Epoch: out.Epoch, State: out.New})
106 require.NoError(t, tr.Close())
107 d.runConnLoop(ac)
108
109 // The production Ack 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 wire.Envelope, 1), asyncSends: make(chan wire.Envelope, 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, "Output")
126 select {
127 case frame := <-tr.syncSends:
128 t.Fatalf("paint used synchronous Send: %+v", envelopeMessageName(t, frame.Payload))
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(protocol.Hello{
139 Version: protocol.Version,
140 Intent: protocol.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 initial := renderer.NewFrame(pane.screen.Columns(), pane.screen.Rows())
149 for y := range initial.Height {
150 cell := renderer.Cell{Rune: rune('A' + y), Style: renderer.DefaultStyle()}
151 initial.FillRow(y, 0, initial.Width, cell)
152 }
153 writeTestFrame(t, pane.screen, initial)
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, "Output"))
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 := captureTestFrame(pane.screen)
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 writeTestFrame(t, pane.screen, desired)
172 pane.screen.Row = pane.screen.Rows() - 1
173 pane.screen.Col = 0
174 pane.screen.Write([]byte("\nq"))
175 writeTestFrame(t, pane.screen, desired)
176 d.paint(sess, ac, false, nil)
177 mustApplyOutput(t, client, awaitFrame(t, tr.sends, "Output"))
178 }
179
180 require.Equal(t, 'B', client.Cell(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 wire.Envelope) protocol.Output {
201 t.Helper()
202 out := unmarshalTestOutput(t, frame.Payload)
203 screen.Write(out.Data)
204 return out
205}
206
207func TestLocalAttachStillAdvancesRendererOnSend(t *testing.T) {
208 p, releasePTY := newBlockingPTY(t)
209 defer releasePTY()
210 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
211
212 sess.tabs[0].focusedPane().screen.Write([]byte("A"))
213 d.paint(sess, ac, false, nil)
214 awaitFrame(t, sends, "Output")
215
216 // The first paint updates the renderer shadow without waiting for Ack.
217 d.paint(sess, ac, false, nil)
218 requireNoOutputFrame(t, sends)
219}
220
221func TestLocalOutputAckDoesNotMoveRendererShadowBackward(t *testing.T) {
222 p, releasePTY := newBlockingPTY(t)
223 defer releasePTY()
224 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
225 pane := sess.tabs[0].focusedPane()
226
227 pane.screen.Write([]byte("A"))
228 d.paint(sess, ac, false, nil)
229 first := awaitFrame(t, sends, "Output")
230 out := unmarshalTestOutput(t, first.Payload)
231
232 pane.screen.Write([]byte("\rB"))
233 d.paint(sess, ac, false, nil)
234 awaitFrame(t, sends, "Output")
235
236 ac.ackOutputState(out.Epoch, out.New)
237 d.paint(sess, ac, false, nil)
238 requireNoOutputFrame(t, sends)
239}
240
241func TestRawTerminalSideEffectDoesNotEnterFullOutputWindow(t *testing.T) {
242 d := newTestDaemon(t, nil, stubClock{})
243 tr := &closeTrackingTransport{}
244 ac := &attachedClient{tr: tr, output: newOutputStateStream()}
245 ac.output.next = maxUnackedOutputStates
246
247 require.NoError(t, d.boundedSendOutputErr(ac, []byte("\x1b]52;c;YQ==\x07")))
248 require.Equal(t, uint64(maxUnackedOutputStates), ac.output.next)
249 require.Equal(t, uint64(maxUnackedOutputStates), ac.output.outstanding())
250 sends := tr.Sends()
251 require.Len(t, sends, 1)
252 out := unmarshalTestOutput(t, sends[0].Payload)
253 require.Zero(t, out.Base)
254 require.Zero(t, out.New)
255}
256
257func TestOwnedSynchronousSideEffectSkipsOuterWatchdog(t *testing.T) {
258 d := newTestDaemon(t, nil, noWatchdogClock{})
259 tr := &timedSideEffectTransport{}
260 ac := &attachedClient{tr: tr, output: newOutputStateStream()}
261
262 require.NoError(t, d.boundedSendOutputErr(ac, make([]byte, 100*1024)))
263 require.Len(t, tr.Sends(), 1)
264}
265
266func TestFailedPaintSendRollsBackOutputState(t *testing.T) {
267 p, releasePTY := newBlockingPTY(t)
268 defer releasePTY()
269 d, sess, ac, _ := newManualSessionWithPTYs(t, p)
270 ac.replaceTransport(failingOutputTransport{})
271 sess.tabs[0].focusedPane().screen.Write([]byte("A"))
272
273 d.paint(sess, ac, false, nil)
274
275 require.Zero(t, ac.output.next)
276 require.Zero(t, ac.output.acked)
277}
278
279func TestResizeGrowthFirstFrameIncludesConcurrentPTYRedraw(t *testing.T) {
280 for _, tc := range []struct {
281 name string
282 window uint64
283 }{
284 {name: "datagram window", window: 1},
285 {name: "stream window", window: maxUnackedOutputStates},
286 } {
287 t.Run(tc.name, func(t *testing.T) {
288 // resizeReaderPTY delivers the child's SIGWINCH redraw through the
289 // actual ptyReader. Its Resize callback does not return until Read has
290 // accepted the bytes, at which point resizeApplying is still true.
291 p := newResizeReaderPTY([]byte("\x1b[1;81H" + strings.Repeat("B", 40)))
292 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
293 ac.output.maxOutstanding = tc.window
294 pane := testAttachmentTab(sess).focusedPane()
295 p.applying = func() bool {
296 pane.mu.Lock()
297 defer pane.mu.Unlock()
298 return pane.resizeApplying
299 }
300 pane.screen.Write([]byte(strings.Repeat("A", 79)))
301 d.paint(sess, ac, true, nil)
302 client := vt.NewScreen(80, 24)
303 initial := mustApplyOutput(t, client, awaitFrame(t, sends, "Output"))
304 ac.ackOutputState(initial.Epoch, initial.New)
305
306 d.sessWg.Add(1)
307 pane.onExit = func() {}
308 go d.ptyReader(sess, testAttachmentTab(sess), pane)
309 defer func() { p.close(); d.sessWg.Wait() }()
310
311 // The resize deadline drives coordinator prepare/apply/commit. No test
312 // writes the grown columns directly.
313 clock := &signalClock{timers: make(chan *signalTimer, 2)}
314 d.clock = clock
315 d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24})
316 timer := <-clock.timers
317 done := captureResizeCallbackDone(t, sess.renderCoordinator())
318 timer.ch <- time.Time{}
319 awaitTestCompletion(t, done, "resize callback did not complete")
320 require.True(t, p.deliveredWhileApplying(), "redraw must arrive while PTY.Resize owns the apply gate")
321 require.Equal(t, 'B', pane.screen.Cell(100, 0).Rune, "replay must parse the redraw before commit")
322
323 client.Resize(120, 24)
324 mustApplyOutput(t, client, awaitFrame(t, sends, "Output"))
325 require.Equal(t, 'B', client.Cell(100, 1).Rune,
326 "first grown frame exposed stale pre-SIGWINCH pane content")
327 select {
328 case extra := <-sends:
329 t.Fatalf("resize emitted more than one frame: %#v", extra)
330 default:
331 }
332 })
333 }
334}
335
336func TestResizeWithoutPTYOutputFlushesOneFullFrameAtDeadline(t *testing.T) {
337 p, releasePTY := newBlockingPTY(t)
338 defer releasePTY()
339 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
340 clk := &signalClock{timers: make(chan *signalTimer, 1)}
341 d.clock = clk
342
343 d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24})
344 var timer *signalTimer
345 select {
346 case timer = <-clk.timers:
347 case <-time.After(time.Second):
348 t.Fatal("resize did not schedule a bounded paint")
349 }
350 requireNoOutputFrame(t, sends)
351
352 done := captureResizeCallbackDone(t, sess.renderCoordinator())
353 timer.ch <- time.Now()
354 awaitTestCompletion(t, done, "resize callback did not complete")
355 frame := awaitFrame(t, sends, "Output")
356 client := vt.NewScreen(120, 24)
357 out := mustApplyOutput(t, client, frame)
358 require.Zero(t, out.Base)
359 require.Contains(t, screenLineText(client, 0), "1")
360 require.Contains(t, screenLineText(client, 23), "work")
361 require.Equal(t, domain.Size{Cols: 120, Rows: 24}, ac.geometry.Size)
362 requireNoOutputFrame(t, sends)
363}
364
365func TestResizeBurstFlushesOnlyLatestGeometry(t *testing.T) {
366 p, releasePTY := newBlockingPTY(t)
367 defer releasePTY()
368 d, sess, ac, sends := newManualSessionWithPTYs(t, p)
369 clk := &signalClock{timers: make(chan *signalTimer, 2)}
370 d.clock = clk
371
372 d.resize(sess, ac, domain.Size{Cols: 100, Rows: 24})
373 var first *signalTimer
374 select {
375 case first = <-clk.timers:
376 case <-time.After(time.Second):
377 t.Fatal("first resize did not schedule a bounded paint")
378 }
379 firstDone := captureResizeCallbackDone(t, sess.renderCoordinator())
380 d.resize(sess, ac, domain.Size{Cols: 120, Rows: 24})
381 var latest *signalTimer
382 select {
383 case latest = <-clk.timers:
384 case <-time.After(time.Second):
385 t.Fatal("latest resize did not replace the bounded paint")
386 }
387
388 latestDone := captureResizeCallbackDone(t, sess.renderCoordinator())
389 first.ch <- time.Now()
390 awaitTestCompletion(t, firstDone, "obsolete resize callback did not complete")
391 requireNoOutputFrame(t, sends)
392 latest.ch <- time.Now()
393 awaitTestCompletion(t, latestDone, "latest resize callback did not complete")
394 awaitFrame(t, sends, "Output")
395 require.Equal(t, domain.Size{Cols: 120, Rows: 24}, ac.geometry.Size)
396 require.Equal(t, 120, testAttachmentTab(sess).focusedPane().screen.Columns())
397 requireNoOutputFrame(t, sends)
398}