1package daemon
2
3import (
4 "errors"
5 "runtime"
6 "testing"
7
8 vt "github.com/bnema/vev-vt"
9 renderer "github.com/bnema/vev-vt/ansi"
10 "github.com/bnema/vev/internal/domain"
11 "github.com/bnema/vev/internal/ports"
12 "github.com/stretchr/testify/require"
13)
14
15func TestOutputStateStreamBuildsPipelinedDependencyChain(t *testing.T) {
16 stream := newOutputStateStream()
17 firstFrame := renderer.NewFrame(3, 1)
18 fillOutputStateRows(firstFrame, []string{"abc"})
19 first, ok, err := drawOutputState(t, stream, firstFrame, nil, true, 7)
20 require.NoError(t, err)
21 require.True(t, ok)
22 firstOut, err := ports.UnmarshalOutput(first.Payload)
23 require.NoError(t, err)
24 require.Equal(t, uint64(0), firstOut.Base)
25 require.Equal(t, uint64(1), firstOut.New)
26 require.Equal(t, uint64(7), firstOut.Echo)
27
28 secondFrame := firstFrame.Clone()
29 secondFrame.Set(0, 0, renderer.Cell{Rune: 'x', Style: renderer.DefaultStyle()})
30 second, ok, err := drawOutputState(t, stream, secondFrame, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false, 8)
31 require.NoError(t, err)
32 require.True(t, ok)
33 secondOut, err := ports.UnmarshalOutput(second.Payload)
34 require.NoError(t, err)
35 require.Equal(t, firstOut.New, secondOut.Base)
36 require.Equal(t, uint64(2), secondOut.New)
37 require.Equal(t, uint64(2), stream.outstanding())
38}
39
40func TestOutputStateStreamCapacityProbeDoesNotRaceWithSend(t *testing.T) {
41 stream := newOutputStateStream(1)
42 frame := renderer.NewFrame(3, 1)
43 fillOutputStateRows(frame, []string{"abc"})
44 prepared, err := stream.prepare(frame, nil, true)
45 require.NoError(t, err)
46
47 entered := make(chan struct{})
48 release := make(chan struct{})
49 sendDone := make(chan error, 1)
50 go func() {
51 sendDone <- prepared.send(prepared.data, 0, func(ports.Frame) error {
52 close(entered)
53 <-release
54 return nil
55 })
56 }()
57 <-entered
58
59 probed := make(chan struct{})
60 go func() {
61 for range 100_000 {
62 _ = stream.atCapacity()
63 runtime.Gosched()
64 }
65 close(probed)
66 }()
67 close(release)
68 require.NoError(t, <-sendDone)
69 <-probed
70 require.True(t, stream.atCapacity())
71
72 stream.ack(1, 1)
73 require.False(t, stream.atCapacity())
74 stream.rebase()
75 require.False(t, stream.atCapacity())
76}
77
78func TestOutputStateStreamEpochsAreAttachmentLocalAndPreparedFramesAreFenced(t *testing.T) {
79 frame := renderer.NewFrame(3, 1)
80 fillOutputStateRows(frame, []string{"abc"})
81 newAttachment := func() *outputStateStream {
82 stream := newOutputStateStream()
83 ac := &attachedClient{output: stream, size: domain.Size{Cols: 3, Rows: 1}}
84 stream.attachment = ac
85 return stream
86 }
87
88 first := newAttachment()
89 second := newAttachment()
90 firstPrepared, err := first.prepare(frame, nil, true)
91 require.NoError(t, err)
92 var firstFrame ports.Frame
93 require.NoError(t, firstPrepared.send(firstPrepared.data, 0, func(frame ports.Frame) error {
94 firstFrame = frame
95 return nil
96 }))
97 firstOutput, err := ports.UnmarshalOutput(firstFrame.Payload)
98 require.NoError(t, err)
99 require.Equal(t, uint64(1), firstOutput.Epoch)
100 require.Equal(t, uint64(1), firstOutput.New)
101 require.True(t, firstOutput.Full)
102 first.ack(firstOutput.Epoch, firstOutput.New)
103
104 secondPrepared, err := second.prepare(frame, nil, true)
105 require.NoError(t, err)
106 require.NoError(t, secondPrepared.send(secondPrepared.data, 0, func(ports.Frame) error { return nil }))
107 require.Equal(t, uint64(1), second.epoch)
108 require.Equal(t, uint64(1), second.next)
109 require.Equal(t, uint64(1), first.next)
110
111 changed := frame.Clone()
112 changed.Set(0, 0, renderer.Cell{Rune: 'x', Style: renderer.DefaultStyle()})
113 pending, err := first.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
114 require.NoError(t, err)
115 first.rebase()
116 require.Equal(t, uint64(2), first.epoch)
117 require.Zero(t, first.next)
118 require.Zero(t, first.acked)
119 var staleSent bool
120 require.NoError(t, pending.send(pending.data, 0, func(ports.Frame) error {
121 staleSent = true
122 return nil
123 }))
124 require.False(t, staleSent)
125 require.Zero(t, first.next)
126 first.ack(1, 1)
127 require.Zero(t, first.acked, "an ACK from the retired epoch must not advance the new epoch")
128
129 reset, err := first.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
130 require.NoError(t, err)
131 var resetFrame ports.Frame
132 require.NoError(t, reset.send(reset.data, 0, func(frame ports.Frame) error {
133 resetFrame = frame
134 return nil
135 }))
136 resetOutput, err := ports.UnmarshalOutput(resetFrame.Payload)
137 require.NoError(t, err)
138 require.Equal(t, uint64(2), resetOutput.Epoch)
139 require.Zero(t, resetOutput.Base)
140 require.Equal(t, uint64(1), resetOutput.New)
141 require.True(t, resetOutput.Full)
142}
143
144func TestPreparedOutputDropsReplacedConnectionAndView(t *testing.T) {
145 for _, tt := range []struct {
146 name string
147 change func(*attachedClient)
148 }{
149 {name: "connection generation", change: func(ac *attachedClient) { ac.connectionGeneration.Add(1) }},
150 {name: "view revision", change: func(ac *attachedClient) {
151 ac.viewMu.Lock()
152 ac.view.revision++
153 ac.viewMu.Unlock()
154 }},
155 } {
156 t.Run(tt.name, func(t *testing.T) {
157 stream := newOutputStateStream()
158 ac := &attachedClient{output: stream, size: domain.Size{Cols: 3, Rows: 1}}
159 stream.attachment = ac
160 frame := renderer.NewFrame(3, 1)
161 fillOutputStateRows(frame, []string{"abc"})
162 initial, err := stream.prepare(frame, nil, true)
163 require.NoError(t, err)
164 require.NoError(t, initial.send(initial.data, 0, func(ports.Frame) error { return nil }))
165 changed := frame.Clone()
166 changed.Set(0, 0, renderer.Cell{Rune: 'x', Style: renderer.DefaultStyle()})
167 pending, err := stream.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
168 require.NoError(t, err)
169 tt.change(ac)
170 var sent bool
171 require.NoError(t, pending.send(pending.data, 0, func(ports.Frame) error {
172 sent = true
173 return nil
174 }))
175 require.False(t, sent)
176 require.Equal(t, uint64(1), stream.next)
177
178 retry, err := stream.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
179 require.NoError(t, err)
180 require.NoError(t, retry.send(retry.data, 0, func(ports.Frame) error { return nil }))
181 require.Equal(t, uint64(2), stream.next)
182 })
183 }
184}
185
186func TestOutputStateAckRequiresCurrentEpoch(t *testing.T) {
187 stream := newOutputStateStream()
188 stream.next = 2
189 stream.ack(2, 1)
190 require.Zero(t, stream.acked, "a stale epoch must not retire output state")
191 stream.ack(1, 1)
192 require.Equal(t, uint64(1), stream.acked)
193 stream.ack(1, 3)
194 require.Equal(t, uint64(1), stream.acked, "a future state must not advance the ACK window")
195}
196
197func TestOutputStateSideEffectsDoNotAdvanceEpochStateOrACK(t *testing.T) {
198 stream := newOutputStateStream()
199 stream.ack(1, 0)
200 beforeEpoch, beforeNext, beforeAck := stream.epoch, stream.next, stream.acked
201 frame, err := stream.sideEffect([]byte("pty"), 9)
202 require.NoError(t, err)
203 out, err := ports.UnmarshalOutput(frame.Payload)
204 require.NoError(t, err)
205 require.Equal(t, beforeEpoch, out.Epoch)
206 require.Zero(t, out.Base)
207 require.Zero(t, out.New)
208 require.False(t, out.Full)
209 require.Equal(t, uint64(9), out.Echo)
210 require.Equal(t, beforeNext, stream.next)
211 require.Equal(t, beforeAck, stream.acked)
212}
213
214func TestOutputStateStreamFailedSendRetriesSnapshotWithoutAdvancing(t *testing.T) {
215 stream := newOutputStateStream()
216 initial := renderer.NewFrame(3, 1)
217 fillOutputStateRows(initial, []string{"abc"})
218 first, err := stream.prepare(initial, nil, true)
219 require.NoError(t, err)
220 require.NoError(t, first.send(first.data, 0, func(ports.Frame) error { return nil }))
221 require.Equal(t, uint64(1), stream.next)
222
223 changed := initial.Clone()
224 changed.Set(0, 0, renderer.Cell{Rune: 'x', Style: renderer.DefaultStyle()})
225 pending, err := stream.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
226 require.NoError(t, err)
227 probe, err := stream.renderer.Prepare(initial, nil, false)
228 require.NoError(t, err)
229 require.Empty(t, probe.Bytes(), "preparation must not advance the renderer shadow")
230
231 sendErr := errors.New("send failed")
232 require.ErrorIs(t, pending.send(pending.data, 0, func(ports.Frame) error { return sendErr }), sendErr)
233 require.Equal(t, uint64(1), stream.next, "failed send must not advance the state chain")
234 pending.commitNoSend()
235 probe, err = stream.renderer.Prepare(initial, nil, false)
236 require.NoError(t, err)
237 require.Empty(t, probe.Bytes(), "failed send must retain the committed renderer shadow")
238
239 retry, err := stream.prepare(changed, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false)
240 require.NoError(t, err)
241 var sent ports.Frame
242 require.NoError(t, retry.send(retry.data, 0, func(frame ports.Frame) error {
243 sent = frame
244 return nil
245 }))
246 out, err := ports.UnmarshalOutput(sent.Payload)
247 require.NoError(t, err)
248 require.Zero(t, out.Base, "retry after ambiguous send failure must be dependency-free")
249 require.Equal(t, uint64(2), out.New, "state numbers remain monotonic across rebases")
250 require.Equal(t, uint64(2), stream.next, "successful retry advances the chain exactly once")
251 require.NoError(t, retry.send(retry.data, 0, func(ports.Frame) error {
252 t.Fatal("completed output sent twice")
253 return nil
254 }))
255 require.Equal(t, uint64(2), stream.next)
256 probe, err = stream.renderer.Prepare(changed, nil, false)
257 require.NoError(t, err)
258 require.Empty(t, probe.Bytes(), "successful retry commits the candidate shadow")
259}
260
261func TestOutputStateStreamNoByteCommitAdvancesShadowWithoutState(t *testing.T) {
262 stream := newOutputStateStream()
263 frame := renderer.NewFrame(3, 1)
264 fillOutputStateRows(frame, []string{"abc"})
265 initial, err := stream.prepare(frame, nil, true)
266 require.NoError(t, err)
267 require.NoError(t, initial.send(initial.data, 0, func(ports.Frame) error { return nil }))
268
269 noOp, err := stream.prepare(frame, nil, false)
270 require.NoError(t, err)
271 require.Empty(t, noOp.data)
272 noOp.commitNoSend()
273 noOp.commitNoSend()
274 require.Equal(t, uint64(1), stream.next, "no-byte commit must not create a state frame")
275
276 probe, err := stream.renderer.Prepare(frame, nil, false)
277 require.NoError(t, err)
278 require.Empty(t, probe.Bytes(), "no-byte commit retains the prepared visual candidate")
279}
280
281func TestOutputStateStreamRepeatedUnackedScrollRemainsClientCorrect(t *testing.T) {
282 stream := newOutputStateStream()
283 client := vt.NewScreen(4, 3)
284 initial := renderer.NewFrame(4, 3)
285 fillOutputStateRows(initial, []string{"AAAA", "BBBB", "CCCC"})
286 first, ok, err := drawOutputState(t, stream, initial, nil, true, 0)
287 require.NoError(t, err)
288 require.True(t, ok)
289 mustApplyOutput(t, client, first)
290 stream.ack(1, 1)
291
292 scrolled := renderer.NewFrame(4, 3)
293 fillOutputStateRows(scrolled, []string{"BBBB", "CCCC", "N "})
294 damage := []renderer.Damage{
295 {Kind: renderer.DamageScrollUp, X: 0, Y: 0, Width: 4, Height: 3, Count: 1},
296 {Kind: renderer.DamageText, X: 0, Y: 2, Width: 4, Height: 1},
297 }
298 second, ok, err := drawOutputState(t, stream, scrolled, damage, false, 0)
299 require.NoError(t, err)
300 require.True(t, ok)
301 mustApplyOutput(t, client, second)
302
303 // The renderer baseline is the preceding emitted frame, so repeating stale
304 // scroll damage falls back to an overwrite instead of applying a second
305 // incompatible scroll program.
306 third, ok, err := drawOutputState(t, stream, scrolled, damage, false, 0)
307 require.NoError(t, err)
308 require.True(t, ok)
309 mustApplyOutput(t, client, third)
310 for y, want := range []string{"BBBB", "CCCC", "N "} {
311 require.Equal(t, want, outputStateRow(client.Frame.Row(y)))
312 }
313}
314
315func (s *outputStateStream) render(frame renderer.Frame, damage []renderer.Damage, reset bool) ([]byte, error) {
316 prepared, err := s.prepare(frame, damage, reset)
317 if err != nil {
318 return nil, err
319 }
320 if len(prepared.data) == 0 {
321 prepared.commitNoSend()
322 return nil, nil
323 }
324 // This helper tests renderer replay bytes without publishing a wire frame.
325 // Production state advancement happens only through preparedOutput.send.
326 prepared.draw.Commit()
327 s.forceSnapshot = false
328 return prepared.data, nil
329}
330
331func outputStateFrame(stream *outputStateStream, data []byte, reset bool, echoAck uint64) ports.Frame {
332 stream.next++
333 base := stream.next - 1
334 if reset {
335 base = 0
336 }
337 frame, err := frameOutputState(data, base, stream.next, echoAck)
338 if err != nil {
339 panic(err)
340 }
341 return frame
342}
343
344func drawOutputState(t *testing.T, stream *outputStateStream, frame renderer.Frame, damage []renderer.Damage, reset bool, echoAck uint64) (ports.Frame, bool, error) {
345 t.Helper()
346 prepared, err := stream.prepare(frame, damage, reset)
347 if err != nil || len(prepared.data) == 0 {
348 if prepared != nil {
349 prepared.commitNoSend()
350 }
351 return ports.Frame{}, false, err
352 }
353 var output ports.Frame
354 err = prepared.send(prepared.data, echoAck, func(frame ports.Frame) error {
355 output = frame
356 return nil
357 })
358 if err != nil {
359 return ports.Frame{}, false, err
360 }
361 return output, true, nil
362}
363
364// A resize is a state reset, but it must not turn subsequent idle or damaged
365// renders into full frames.
366func TestOutputStateStreamResizeFrameThenNoopAndDamageAreDifferential(t *testing.T) {
367 screen := vt.NewScreen(4, 2)
368 screen.Write([]byte("abcd"))
369 screen.ClearDamage()
370 screen.Resize(6, 2)
371
372 stream := newOutputStateStream()
373 resized, ok, err := drawOutputState(t, stream, screen.Frame, screen.Damage(), true, 0)
374 require.NoError(t, err)
375 require.True(t, ok)
376 resizeOutput, err := ports.UnmarshalOutput(resized.Payload)
377 require.NoError(t, err)
378 require.Zero(t, resizeOutput.Base, "resize must emit the one reset frame")
379 screen.ClearDamage()
380
381 _, ok, err = drawOutputState(t, stream, screen.Frame, screen.Damage(), false, 0)
382 require.NoError(t, err)
383 require.False(t, ok, "an unchanged render after resize must emit nothing")
384
385 screen.Frame.Set(0, 0, renderer.Cell{Rune: 'x', Style: renderer.DefaultStyle()})
386 damaged, ok, err := drawOutputState(t, stream, screen.Frame, []renderer.Damage{{Kind: renderer.DamageText, Width: 1, Height: 1}}, false, 0)
387 require.NoError(t, err)
388 require.True(t, ok)
389 damageOutput, err := ports.UnmarshalOutput(damaged.Payload)
390 require.NoError(t, err)
391 require.Equal(t, resizeOutput.New, damageOutput.Base, "later damage must remain incremental")
392}
393
394func TestOutputStateStreamDefaultsAndNormalizesWindow(t *testing.T) {
395 require.Equal(t, uint64(8), newOutputStateStream().maxOutstanding)
396 require.Equal(t, uint64(8), newOutputStateStream(0).maxOutstanding)
397 require.Equal(t, uint64(8), newOutputStateStream(9).maxOutstanding)
398 require.Equal(t, uint64(1), newOutputStateStream(1).maxOutstanding)
399}
400
401func TestNormalizeOutputWindow(t *testing.T) {
402 for _, tt := range []struct {
403 input, want uint8
404 }{{0, 8}, {1, 1}, {8, 8}, {9, 8}, {255, 8}} {
405 require.Equal(t, tt.want, normalizeOutputWindow(tt.input))
406 }
407}