1package daemon
2
3import (
4 "errors"
5 "sync/atomic"
6 "testing"
7
8 "github.com/stretchr/testify/mock"
9
10 "github.com/bnema/vev/internal/protocol"
11 "github.com/bnema/vev/internal/protocol/wire"
12 "github.com/stretchr/testify/require"
13)
14
15func TestUIFenceDispatchReceivesACKAtFullOutputWindow(t *testing.T) {
16 writes := make(chan []byte, 1)
17 pty, release := newBlockingPTYWithWrites(t, writes)
18 t.Cleanup(release)
19 d, sess, ac, sends := newManualSessionWithPTYs(t, pty)
20 rc := d.attachCoordinator(sess, nil, ac, true)
21 lease := rc.attachmentLease(ac)
22 invalidated := make(chan struct{}, 1)
23 rc.opts.onInvalidate = func(inv renderInvalidation) {
24 if inv.producer == "ui-fence" {
25 invalidated <- struct{}{}
26 }
27 }
28 t.Cleanup(func() { rc.beginSessionTeardown().finish(); rc.waitForTimerWorkers() })
29 ac.output.setWindow(1)
30 require.Equal(t, paintEmitted, d.paint(sess, ac, true, lease))
31 initial := unmarshalTestOutput(t, awaitFrame(t, sends, "Output").Payload)
32 require.True(t, ac.output.atCapacity())
33 require.Empty(t, drainAllFrames(sends))
34
35 // A single dispatch owner receives Input, UIFence, and then the ACK that
36 // makes room. Neither registration nor its synchronous test-clock render
37 // fallback may wait for that same owner to dispatch the ACK.
38 messages := make(chan protocol.ClientMessage)
39 dispatched := make(chan bool, 1)
40 done := make(chan struct{})
41 go func() {
42 defer close(done)
43 for message := range messages {
44 capability := captureAttachmentCapability(sess, ac, ac.transport())
45 dispatched <- d.handleAttachmentClientMessage(capability, message)
46 }
47 }()
48 t.Cleanup(func() { close(messages); awaitTestCompletion(t, done, "dispatch owner leaked") })
49 messages <- protocol.Input{InputSeq: 1, ActionID: 9, Data: []byte("x")}
50 require.False(t, awaitTestValue(t, dispatched, "input dispatch blocked"))
51 require.Equal(t, []byte("x"), awaitTestValue(t, writes, "normal PTY input was bypassed"))
52 messages <- protocol.UIFence{ActionID: 9}
53 require.False(t, awaitTestValue(t, dispatched, "fence dispatch waited for an output ACK"))
54 awaitTestValue(t, invalidated, "fence did not reach ACK-deferred render scheduling")
55 require.Empty(t, drainAllFrames(sends), "full output window cannot publish or confirm the fence")
56 messages <- protocol.Ack{Epoch: initial.Epoch, State: initial.New}
57 require.False(t, awaitTestValue(t, dispatched, "releasing ACK did not complete"))
58 update := decodeServerMessage(t, awaitFrame(t, sends, "UIViewUpdate")).(protocol.UIViewUpdate)
59 receipt := decodeServerMessage(t, awaitFrame(t, sends, "UIReceipt")).(protocol.UIReceipt)
60 require.Equal(t, protocol.UIReceipt{ActionID: 9, Epoch: update.Epoch, State: update.State, ViewPublication: update.Context.Publication, Outcome: protocol.UIReceiptProcessed}, receipt)
61 require.Equal(t, initial.New, update.State, "no-op confirmation must not advance the ACK chain")
62 require.Equal(t, initial.Context.Publication+1, update.Context.Publication)
63 require.Zero(t, ac.echoAck.Load(), "processing is not child echo")
64 require.Empty(t, drainAllFrames(sends))
65}
66
67func TestUIFencePreRegistrationCaptureCannotConfirmAction(t *testing.T) {
68 d, sess, ac, sends := newManualSessionWithPTYs(t, nil)
69 // A coordinator without a clock leaves scheduling under the test's exact
70 // capture boundary. The registration still uses the real dispatch owner.
71 rc := newRenderCoordinator(renderCoordinatorOptions{})
72 installAttachmentRenderCoordinator(sess, rc)
73 lease := rc.attachWithReadiness(ac, true)
74 t.Cleanup(func() { rc.beginSessionTeardown().finish(); rc.waitForTimerWorkers() })
75 capture := func(reset bool) *capturedRenderState {
76 t.Helper()
77 ac.sendMu.Lock()
78 state, ok := captureRenderState(sess, ac, renderCaptureRequest{reset: reset, lease: lease, uiFence: rc.captureUIFence(lease)})
79 if !ok {
80 ac.sendMu.Unlock()
81 t.Fatal("capture rejected")
82 }
83 return state
84 }
85 emit := func(state *capturedRenderState) {
86 t.Helper()
87 composed := composeFrame(*state, ac.render.cache)
88 require.True(t, d.publishCapturedFrameForTest(sess, ac, state, composed))
89 }
90 state := capture(true)
91 dispatched := make(chan bool, 1)
92 go func() {
93 dispatched <- d.handleAttachmentClientMessage(captureAttachmentCapability(sess, ac, ac.transport()), protocol.UIFence{ActionID: 5})
94 }()
95 require.False(t, awaitTestValue(t, dispatched, "fence dispatch acquired render sendMu"))
96 emit(state)
97 initial := unmarshalTestOutput(t, awaitFrame(t, sends, "Output").Payload)
98 require.Empty(t, drainAllFrames(sends), "an earlier capture cannot retire a later fence")
99 require.True(t, rc.needsUIFence(lease, rc.captureUIFence(lease)))
100 emit(capture(false))
101 update := decodeServerMessage(t, awaitFrame(t, sends, "UIViewUpdate")).(protocol.UIViewUpdate)
102 receipt := decodeServerMessage(t, awaitFrame(t, sends, "UIReceipt")).(protocol.UIReceipt)
103 require.Equal(t, uint64(5), receipt.ActionID)
104 require.Equal(t, initial.New, receipt.State)
105 require.Equal(t, update.Context.Publication, receipt.ViewPublication)
106 require.False(t, rc.needsUIFence(lease, rc.captureUIFence(lease)))
107}
108
109func TestUIFenceFailedPublicationOrReceiptDoesNotConfirm(t *testing.T) {
110 for _, failedType := range []string{"UIViewUpdate", "UIReceipt"} {
111 t.Run(map[string]string{"UIViewUpdate": "metadata", "UIReceipt": "receipt"}[failedType], func(t *testing.T) {
112 d, sess, ac, _ := newManualSessionWithPTYs(t, nil)
113 tr := newMockServerConnection(t)
114 sends := make(chan wire.Envelope, 16)
115 var fail atomic.Bool
116 tr.EXPECT().Send(mock.Anything).RunAndReturn(func(frame wire.Envelope) error {
117 if fail.Load() && envelopeMessageName(nil, frame.Payload) == failedType {
118 return errors.New("injected send failure")
119 }
120 sends <- frame
121 return nil
122 }).Maybe()
123 tr.EXPECT().Close().Return(nil).Maybe()
124 ac.tr = tr
125 rc := newRenderCoordinator(renderCoordinatorOptions{})
126 installAttachmentRenderCoordinator(sess, rc)
127 lease := rc.attachWithReadiness(ac, true)
128 cleanupEntered, releaseCleanup := make(chan struct{}), make(chan struct{})
129 d.beforeAttachmentSendErrorCleanup = func(attachmentCapability) {
130 close(cleanupEntered)
131 <-releaseCleanup
132 }
133 t.Cleanup(func() {
134 close(releaseCleanup)
135 d.attachmentCleanupWg.Wait()
136 rc.beginSessionTeardown().finish()
137 rc.waitForTimerWorkers()
138 })
139 require.Equal(t, paintEmitted, d.paint(sess, ac, true, lease))
140 initial := unmarshalTestOutput(t, awaitFrame(t, sends, "Output").Payload)
141 require.Empty(t, drainAllFrames(sends))
142 require.False(t, d.handleAttachmentClientMessage(captureAttachmentCapability(sess, ac, tr), protocol.UIFence{ActionID: 13}))
143 fail.Store(true)
144 require.Equal(t, paintEmitted, d.paint(sess, ac, false, lease))
145 awaitTestValue(t, cleanupEntered, "failed send did not reserve exact-attachment cleanup")
146 if failedType == "UIViewUpdate" {
147 require.Equal(t, initial.Context.Publication, ac.output.viewPublication)
148 require.True(t, rc.needsUIFence(lease, rc.captureUIFence(lease)), "failed metadata cannot select a receipt")
149 } else {
150 update, ok := decodeServerMessage(t, awaitFrame(t, sends, "UIViewUpdate")).(protocol.UIViewUpdate)
151 require.True(t, ok)
152 require.Equal(t, initial.Context.Publication+1, update.Context.Publication)
153 rc.mu.Lock()
154 pending := lease.pendingUIFence
155 rc.mu.Unlock()
156 require.Nil(t, pending, "failed receipt send must release its slot")
157 require.False(t, rc.needsUIFence(lease, rc.captureUIFence(lease)))
158 }
159 require.Empty(t, drainAllFrames(sends), "no successful receipt after a failed publication/send")
160 })
161 }
162}