1package daemon
2
3import (
4 "io"
5 "testing"
6
7 "github.com/bnema/vev/internal/domain"
8 "github.com/bnema/vev/internal/protocol"
9 "github.com/bnema/vev/internal/protocol/wire"
10 "github.com/stretchr/testify/mock"
11 "github.com/stretchr/testify/require"
12)
13
14func TestAttachmentSuspensionRevokesEffectsAndGeometry(t *testing.T) {
15 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
16 rc := d.attachCoordinator(sess, nil, ac, true)
17 token := sess.captureAttachmentCapability(ac, ac.transport())
18 ac.installTestAttachmentCapability(token)
19 _, ok := sess.geometry.claimAttachment(sess, ac)
20 require.True(t, ok)
21 require.NoError(t, d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 1}))
22 require.Nil(t, sess.geometry.latest.Load())
23 require.Nil(t, rc.attachmentLease(ac))
24 current := sess.captureAttachmentCapability(ac, ac.transport())
25 _, admitted := ac.beginAttachmentEffect(current)
26 require.False(t, admitted)
27 require.Equal(t, paintRejected, d.paint(sess, ac, true, nil))
28 require.True(t, ac.currentTransportIs(token.transport.transport))
29}
30
31func TestAttachmentActivationExactLifecycle(t *testing.T) {
32 for _, stale := range []bool{false, true} {
33 t.Run(map[bool]string{false: "retained", true: "stale"}[stale], func(t *testing.T) {
34 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
35 d.attachCoordinator(sess, nil, ac, true)
36 token := sess.captureAttachmentCapability(ac, ac.transport())
37 ac.installTestAttachmentCapability(token)
38 require.NoError(t, d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 1}))
39 target := protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}
40 if stale {
41 target.LifecycleID[0] ^= 1
42 }
43 err := d.activateAttachment(ac, ac.transportSnapshot(), protocol.ActivateAttachment{RequestID: 2, Target: target, Size: domain.Size{Cols: 100, Rows: 30}})
44 if stale {
45 require.Error(t, err)
46 require.Nil(t, sess.geometry.latest.Load())
47 return
48 }
49 require.NoError(t, err)
50 require.Equal(t, ac, sess.geometry.latest.Load().attachment)
51 require.Equal(t, 100, ac.geometrySnapshot().Cols)
52 current := sess.captureAttachmentCapability(ac, ac.transport())
53 effect, admitted := ac.beginAttachmentEffect(current)
54 require.True(t, admitted)
55 effect.End()
56 })
57 }
58}
59
60func TestAttachmentSuspensionBarrierDrainsAdmittedEffects(t *testing.T) {
61 d, sess, ac, sends := newManualSessionWithPTYs(t, newQuietPTY())
62 d.attachCoordinator(sess, nil, ac, true)
63 token := sess.captureAttachmentCapability(ac, ac.transport())
64 ac.installTestAttachmentCapability(token)
65 effect, admitted := ac.beginAttachmentEffect(token)
66 require.True(t, admitted)
67 frozen := make(chan struct{})
68 d.afterAttachmentEffectGateFrozen = func(action string, _ *attachedClient) {
69 if action == "suspend" {
70 close(frozen)
71 }
72 }
73 result := make(chan error, 1)
74 go func() { result <- d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 7}) }()
75 <-frozen
76 select {
77 case <-sends:
78 t.Fatal("ACK before effect drained")
79 default:
80 }
81 _, admitted = ac.beginAttachmentEffect(token)
82 require.False(t, admitted)
83 require.NoError(t, effect.sendControl(protocol.Pong{}))
84 effect.End()
85 require.NoError(t, <-result)
86 require.IsType(t, protocol.Pong{}, decodeServerMessage(t, <-sends))
87 ack := decodeServerMessage(t, <-sends).(protocol.AttachmentSuspended)
88 require.Equal(t, uint64(7), ack.RequestID)
89}
90
91func TestAttachmentActivationPublicationPrecedesSuccess(t *testing.T) {
92 d, sess, ac, sends := newManualSessionWithPTYs(t, newQuietPTY())
93 d.attachCoordinator(sess, nil, ac, true)
94 token := sess.captureAttachmentCapability(ac, ac.transport())
95 ac.installTestAttachmentCapability(token)
96 require.NoError(t, d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 1}))
97 suspended := decodeServerMessage(t, <-sends).(protocol.AttachmentSuspended)
98 require.NoError(t, d.activateAttachment(ac, ac.transportSnapshot(), protocol.ActivateAttachment{RequestID: 2, Target: suspended.Target, Size: ac.sizeSnapshot()}))
99 var full *protocol.Output
100 var position *protocol.RoutePosition
101 for len(sends) != 0 {
102 switch m := decodeServerMessage(t, <-sends).(type) {
103 case protocol.Output:
104 require.Nil(t, full)
105 require.True(t, m.Full)
106 require.Zero(t, m.Base)
107 require.NotNil(t, m.Context)
108 full = &m
109 case protocol.RoutePosition:
110 require.NotNil(t, full, "route position must follow full publication")
111 require.Nil(t, position)
112 position = &m
113 case protocol.AttachmentActivated:
114 require.NotNil(t, full, "activation success must follow full publication")
115 require.NotNil(t, position, "activation success must follow route position")
116 require.Equal(t, full.Epoch, m.Epoch)
117 require.Equal(t, full.New, m.State)
118 require.Equal(t, full.Context.Publication, m.ViewPublication)
119 require.Equal(t, full.Context.Route, m.Identity)
120 return
121 }
122 }
123 t.Fatal("missing activation success")
124}
125
126func TestAttachmentSuspensionCloseDoesNotPark(t *testing.T) {
127 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
128 ac.resumeCapable, ac.resumeToken = true, 123
129 d.attachCoordinator(sess, nil, ac, true)
130 token := sess.captureAttachmentCapability(ac, ac.transport())
131 ac.installTestAttachmentCapability(token)
132 require.NoError(t, d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 1}))
133 d.clientGone(sess, ac, token.transport.transport, false)
134 require.Nil(t, ac.currentAttachmentSession())
135 require.Empty(t, d.resume.parked)
136 require.False(t, ac.parked)
137}
138
139func TestAttachmentSuspensionPeerGeometryFallbackAndReclaim(t *testing.T) {
140 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
141 peerTransport, _ := newCapturingTransport(t)
142 peer, err := d.attachClient(sess, peerTransport, domain.Size{Cols: 60, Rows: 20}, attachClientOptions{})
143 require.NoError(t, err)
144 d.attachCoordinator(sess, nil, ac, true)
145 token := sess.captureAttachmentCapability(ac, ac.transport())
146 ac.installTestAttachmentCapability(token)
147 _, ok := sess.geometry.claimAttachment(sess, ac)
148 require.True(t, ok)
149 require.NoError(t, d.suspendAttachment(token, protocol.SuspendAttachment{RequestID: 1}))
150 require.Equal(t, peer, sess.geometry.latest.Load().attachment)
151 peerToken := sess.captureAttachmentCapability(peer, peer.transport())
152 effect, admitted := peer.beginAttachmentEffect(peerToken)
153 require.True(t, admitted)
154 effect.End()
155 require.NoError(t, d.activateAttachment(ac, ac.transportSnapshot(), protocol.ActivateAttachment{RequestID: 2, Target: protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}, Size: ac.sizeSnapshot()}))
156 require.Equal(t, ac, sess.geometry.latest.Load().attachment)
157}
158
159func TestAttachmentSuspensionReaderKeepsControlAlive(t *testing.T) {
160 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
161 tr := newMockServerConnection(t)
162 incoming := make(chan protocol.ClientMessage, 4)
163 incoming <- protocol.SuspendAttachment{RequestID: 1}
164 incoming <- protocol.Input{Data: []byte("ignored"), InputSeq: 1}
165 incoming <- protocol.Ping{}
166 close(incoming)
167 tr.EXPECT().Recv().RunAndReturn(func() (wire.Envelope, error) {
168 m, ok := <-incoming
169 if !ok {
170 return wire.Envelope{}, io.EOF
171 }
172 return mustClientEnvelope(m), nil
173 })
174 var messages []protocol.ServerMessage
175 tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f wire.Envelope) error {
176 messages = append(messages, decodeServerMessage(t, f))
177 return nil
178 }).Maybe()
179 tr.EXPECT().Close().Return(nil).Maybe()
180 ac.replaceTransport(tr)
181 ac.resumeCapable, ac.resumeToken = true, 123
182 d.attachCoordinator(sess, nil, ac, true)
183 token := sess.captureAttachmentCapability(ac, tr)
184 ac.installTestAttachmentCapability(token)
185 d.runConnLoop(ac)
186 require.IsType(t, protocol.AttachmentSuspended{}, messages[0])
187 require.IsType(t, protocol.Pong{}, messages[1])
188 require.Zero(t, ac.echoAck.Load())
189 require.Nil(t, ac.currentAttachmentSession())
190 require.Empty(t, d.resume.parked)
191}
192
193func TestAttachmentActivationCloseDuringFullPublication(t *testing.T) {
194 d, sess, ac, _ := newManualSessionWithPTYs(t, newQuietPTY())
195 tr := newMockServerConnection(t)
196 var activated bool
197 tr.EXPECT().Send(mock.Anything).RunAndReturn(func(f wire.Envelope) error {
198 switch decodeServerMessage(t, f).(type) {
199 case protocol.Output:
200 return io.ErrClosedPipe
201 case protocol.AttachmentActivated:
202 activated = true
203 }
204 return nil
205 }).Maybe()
206 tr.EXPECT().Close().Return(nil).Maybe()
207 ac.replaceTransport(tr)
208 ac.resumeCapable, ac.resumeToken = true, 123
209 d.attachCoordinator(sess, nil, ac, true)
210 token := sess.captureAttachmentCapability(ac, tr)
211 ac.installTestAttachmentCapability(token)
212 require.False(t, d.handleAttachmentClientMessage(token, protocol.SuspendAttachment{RequestID: 1}))
213 token = sess.captureAttachmentCapability(ac, tr)
214 require.True(t, d.handleAttachmentClientMessage(token, protocol.ActivateAttachment{RequestID: 2, Target: protocol.ExactSessionTarget{LifecycleID: sess.incarnation, SessionName: sess.name}, Size: ac.sizeSnapshot()}))
215 d.attachmentCleanupWg.Wait()
216 require.False(t, activated)
217 require.Nil(t, ac.currentAttachmentSession())
218 require.Empty(t, d.resume.parked)
219}