1package broker
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "testing"
8 "time"
9
10 renderer "github.com/bnema/vev-vt"
11 "github.com/stretchr/testify/require"
12
13 "github.com/bnema/vev/internal/domain"
14 "github.com/bnema/vev/internal/ports"
15 "github.com/bnema/vev/internal/protocol"
16)
17
18// previewWatchConn is one scripted daemon watch stream. The test pushes each
19// server step; ReceiveServer blocks until a step arrives or the stream closes,
20// exactly like a live daemon connection.
21type previewWatchConn struct {
22 stream ports.BrokerStreamID
23 sent chan protocol.ClientMessage
24 steps chan previewStep
25 done chan struct{}
26 once sync.Once
27}
28
29type previewStep struct {
30 message protocol.ServerMessage
31 err error
32}
33
34func newPreviewWatchConn(stream ports.BrokerStreamID) *previewWatchConn {
35 return &previewWatchConn{stream: stream, sent: make(chan protocol.ClientMessage, 4), steps: make(chan previewStep, 8), done: make(chan struct{})}
36}
37
38func (c *previewWatchConn) SendClient(message protocol.ClientMessage) error {
39 select {
40 case <-c.done:
41 return errors.New("closed")
42 case c.sent <- message:
43 return nil
44 }
45}
46
47func (c *previewWatchConn) ReceiveServer() (protocol.ServerMessage, error) {
48 select {
49 case <-c.done:
50 return nil, errors.New("closed")
51 case step := <-c.steps:
52 return step.message, step.err
53 }
54}
55
56func (c *previewWatchConn) Capabilities() protocol.ConnectionCapabilities {
57 return protocol.ConnectionCapabilities{OutputDataLimit: protocol.MaxOutputDataLen}
58}
59func (c *previewWatchConn) LinkState() ports.LinkState { return ports.LinkStateConnected }
60func (c *previewWatchConn) LinkEvents() <-chan ports.LinkEvent { return nil }
61func (c *previewWatchConn) SendEnvelope([]byte) error { return nil }
62func (c *previewWatchConn) RecvEnvelope() ([]byte, error) { <-c.done; return nil, errors.New("closed") }
63func (c *previewWatchConn) Done() <-chan struct{} { return c.done }
64func (c *previewWatchConn) Err() error { return nil }
65func (c *previewWatchConn) Close() error {
66 c.once.Do(func() { close(c.done) })
67 return nil
68}
69
70func (c *previewWatchConn) push(message protocol.ServerMessage) {
71 c.steps <- previewStep{message: message}
72}
73
74func (c *previewWatchConn) fail(err error) { c.steps <- previewStep{err: err} }
75
76// previewTestConnector dials one scripted physical whose every logical open
77// yields a fresh watch stream, published on opened for the test to drive.
78type previewTestConnector struct {
79 opened chan *previewWatchConn
80 scripted *fakePhysical
81}
82
83func (c *previewTestConnector) Connect(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
84 if c.scripted == nil {
85 c.scripted = &fakePhysical{endpoint: e, done: make(chan struct{}), open: func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
86 conn := newPreviewWatchConn(r.Stream)
87 c.opened <- conn
88 return conn, nil
89 }}
90 }
91 return c.scripted, nil
92}
93
94func previewTarget(lifecycle byte, tab string) domain.RemoteSessionTarget {
95 return domain.RemoteSessionTarget{
96 Endpoint: ports.BrokerPreviewLocalEndpoint, DisplayOrigin: "local",
97 LifecycleID: domain.SessionLifecycleID{lifecycle}, SessionName: "work", LiveTabID: domain.TabStableID(tab),
98 }
99}
100
101func previewRequestFor(service ports.BrokerCoreService, generation ports.BrokerPreviewGeneration, target domain.RemoteSessionTarget) ports.BrokerPreviewRequest {
102 return ports.BrokerPreviewRequest{
103 Epoch: 1, Connection: service.ConnectionID(), Generation: generation,
104 Route: ports.BrokerPreviewRoute{Local: true, Policy: poolPolicy()},
105 Preview: protocol.RemotePreviewRequest{Version: protocol.RemotePreviewSchemaVersion, Target: target, Width: 80, Height: 24},
106 }
107}
108
109func previewAnswer(lifecycle byte, tab string, width, height int) protocol.RemotePreview {
110 return protocol.RemotePreview{Version: protocol.RemotePreviewSchemaVersion, Status: protocol.RemotePreviewOK, LifecycleID: domain.SessionLifecycleID{lifecycle}, TabID: domain.TabStableID(tab), Revision: 7, Width: uint16(width), Height: uint16(height), Cells: make([]renderer.Cell, width*height)}
111}
112
113type previewFixture struct {
114 t *testing.T
115 service ports.BrokerCoreService
116 sub ports.BrokerPreviewSubscription
117 connector *previewTestConnector
118 clock *manualClock
119}
120
121// newPreviewFixture subscribes one generation-1 local preview for target on a
122// fresh authority driven by a manual clock.
123func newPreviewFixture(t *testing.T, target domain.RemoteSessionTarget) *previewFixture {
124 t.Helper()
125 connector := &previewTestConnector{opened: make(chan *previewWatchConn, 8)}
126 clock := newManualClock(time.Unix(0, 0))
127 registry, err := NewRegistryWithConfig(1, newTestStore(), nil, clock, nil, RegistryConfig{ObservationDisabled: true})
128 require.NoError(t, err)
129 authority, _, _, _ := composeTestAuthority(t, 1, registry, connector.Connect, clock)
130 authority.codec = testSessionCodec{}
131 service, err := authority.AdmitClient(context.Background())
132 require.NoError(t, err)
133 t.Cleanup(func() { require.NoError(t, service.Close()) })
134 sub, err := service.SubscribePreview(previewRequestFor(service, 1, target))
135 require.NoError(t, err)
136 t.Cleanup(sub.Close)
137 return &previewFixture{t: t, service: service, sub: sub, connector: connector, clock: clock}
138}
139
140// awaitOpen returns the next opened watch stream and the watch it received.
141func (f *previewFixture) awaitOpen() (*previewWatchConn, protocol.RemotePreviewWatch) {
142 f.t.Helper()
143 var conn *previewWatchConn
144 select {
145 case conn = <-f.connector.opened:
146 case <-time.After(5 * time.Second):
147 f.t.Fatal("the preview did not open a watch stream")
148 }
149 select {
150 case message := <-conn.sent:
151 watch, ok := message.(protocol.RemotePreviewWatch)
152 require.True(f.t, ok, "the first message must be a watch, got %T", message)
153 return conn, watch
154 case <-time.After(5 * time.Second):
155 f.t.Fatal("the preview did not send its watch")
156 return nil, protocol.RemotePreviewWatch{}
157 }
158}
159
160func (f *previewFixture) awaitPublication() ports.BrokerPreviewPublication {
161 f.t.Helper()
162 select {
163 case <-f.sub.Changed():
164 return f.sub.Latest()
165 case <-time.After(5 * time.Second):
166 f.t.Fatal("the preview subscription did not publish within the bound")
167 return ports.BrokerPreviewPublication{}
168 }
169}
170
171// advanceUntilOpen drives the manual clock until the reopen backoff fires.
172func (f *previewFixture) advanceUntilOpen(step time.Duration) {
173 f.t.Helper()
174 require.Eventually(f.t, func() bool {
175 f.clock.Advance(step)
176 return len(f.connector.opened) > 0
177 }, 5*time.Second, time.Millisecond)
178}
179
180func awaitClosed(t *testing.T, conn *previewWatchConn) {
181 t.Helper()
182 select {
183 case <-conn.done:
184 case <-time.After(5 * time.Second):
185 t.Fatal("the watch stream was not closed")
186 }
187}
188
189func TestServicePreviewStreamIDsDoNotConsumeClientIDs(t *testing.T) {
190 for _, previewFirst := range []bool{false, true} {
191 name := "client first"
192 if previewFirst {
193 name = "preview first"
194 }
195 t.Run(name, func(t *testing.T) {
196 f := newPreviewFixture(t, previewTarget(0x11, "tab-alpha"))
197 service := f.service.(*Service)
198 openClient := func(want ports.BrokerStreamID) {
199 t.Helper()
200 id, err := service.NextStreamID()
201 require.NoError(t, err)
202 require.Equal(t, want, id)
203 stream, err := service.OpenEnvelopeStream(context.Background(), poolRequest(service.ConnectionID(), uint64(id)))
204 require.NoError(t, err)
205 t.Cleanup(func() { require.NoError(t, stream.Close()) })
206 }
207 if !previewFirst {
208 openClient(1)
209 }
210 for range 3 {
211 req, err := service.previewStreamRequest(ports.BrokerPreviewRoute{Local: true, Policy: poolPolicy()})
212 require.NoError(t, err)
213 stream, err := service.pool.OpenStream(context.Background(), req)
214 require.NoError(t, err)
215 require.NoError(t, stream.Close())
216 }
217 if previewFirst {
218 openClient(1)
219 } else {
220 openClient(2)
221 }
222 })
223 }
224}
225
226func TestServicePreviewKeepsOneStreamAcrossFrames(t *testing.T) {
227 target := previewTarget(0x11, "tab-alpha")
228 f := newPreviewFixture(t, target)
229 conn, watch := f.awaitOpen()
230 require.Equal(t, previewLocalInterval, watch.MinInterval, "the broker picks the local rate")
231 require.Equal(t, target, watch.Request.Target)
232 require.NoError(t, protocol.ValidateRemotePreviewWatch(watch))
233
234 for revision := range uint64(3) {
235 frame := previewAnswer(0x11, "tab-alpha", 2, 1)
236 frame.Revision = revision + 1
237 conn.push(frame)
238 publication := f.awaitPublication()
239 require.NoError(t, publication.Err)
240 require.Equal(t, frame.Revision, publication.Preview.Revision)
241 }
242 require.Empty(t, f.connector.opened, "every frame arrives on the one watch stream")
243}
244
245func TestServicePreviewFirstFrameDeadlinePublishesUnavailableAndKeepsWaiting(t *testing.T) {
246 f := newPreviewFixture(t, previewTarget(0x11, "tab-alpha"))
247 conn, _ := f.awaitOpen()
248 require.Eventually(t, func() bool {
249 f.clock.Advance(previewFirstFrameLimit)
250 return len(f.sub.Changed()) > 0
251 }, 5*time.Second, time.Millisecond)
252 publication := f.awaitPublication()
253 var brokerErr ports.BrokerError
254 require.ErrorAs(t, publication.Err, &brokerErr)
255 require.Equal(t, ports.BrokerErrorUnavailable, brokerErr.Code)
256
257 conn.push(previewAnswer(0x11, "tab-alpha", 2, 1))
258 require.NoError(t, f.awaitPublication().Err, "a slow stream still delivers once it answers")
259 require.Empty(t, f.connector.opened)
260}
261
262func TestServicePreviewFailureReopensWithBackoff(t *testing.T) {
263 cases := []struct {
264 name string
265 fail func(*previewWatchConn)
266 }{
267 {name: "stream error", fail: func(c *previewWatchConn) { c.fail(errors.New("stream lost")) }},
268 {name: "foreign target", fail: func(c *previewWatchConn) { c.push(previewAnswer(0x22, "tab-alpha", 2, 1)) }},
269 {name: "foreign tab", fail: func(c *previewWatchConn) { c.push(previewAnswer(0x11, "tab-beta", 2, 1)) }},
270 {name: "oversized crop", fail: func(c *previewWatchConn) { c.push(previewAnswer(0x11, "tab-alpha", 81, 1)) }},
271 {name: "wrong message", fail: func(c *previewWatchConn) { c.push(protocol.Pong{}) }},
272 }
273 for _, tc := range cases {
274 t.Run(tc.name, func(t *testing.T) {
275 f := newPreviewFixture(t, previewTarget(0x11, "tab-alpha"))
276 first, _ := f.awaitOpen()
277 tc.fail(first)
278 publication := f.awaitPublication()
279 require.Error(t, publication.Err)
280 require.Equal(t, protocol.RemotePreview{}, publication.Preview, "a failure never publishes foreign content")
281 awaitClosed(t, first)
282
283 f.advanceUntilOpen(previewReopenMin)
284 second, _ := f.awaitOpen()
285 require.NotEqual(t, first.stream, second.stream, "a reopen allocates a new stream identity")
286 second.push(previewAnswer(0x11, "tab-alpha", 2, 1))
287 require.NoError(t, f.awaitPublication().Err)
288 })
289 }
290}
291
292func TestServicePreviewBackoffDoublesUntilCap(t *testing.T) {
293 f := newPreviewFixture(t, previewTarget(0x11, "tab-alpha"))
294 conn, _ := f.awaitOpen()
295 want := []time.Duration{previewReopenMin, 2 * previewReopenMin, 4 * previewReopenMin}
296 for _, delay := range want {
297 conn.fail(errors.New("stream lost"))
298 f.awaitPublication()
299 awaitClosed(t, conn)
300 start := f.clock.Now()
301 f.advanceUntilOpen(previewReopenMin / 5)
302 require.GreaterOrEqual(t, f.clock.Now().Sub(start), delay)
303 conn, _ = f.awaitOpen()
304 }
305}
306
307func TestServicePreviewNoSuchTargetStopsRecovery(t *testing.T) {
308 f := newPreviewFixture(t, previewTarget(0x11, "tab-alpha"))
309 conn, _ := f.awaitOpen()
310 conn.push(protocol.RemotePreview{Version: protocol.RemotePreviewSchemaVersion, Status: protocol.RemotePreviewNoSuchTarget})
311 publication := f.awaitPublication()
312 require.NoError(t, publication.Err)
313 require.Equal(t, protocol.RemotePreviewNoSuchTarget, publication.Preview.Status)
314 awaitClosed(t, conn)
315 require.Never(t, func() bool {
316 f.clock.Advance(previewReopenMax)
317 return len(f.connector.opened) > 0
318 }, 100*time.Millisecond, 5*time.Millisecond, "a dead target is never reopened")
319}
320
321func TestServicePreviewCloseClosesTheStream(t *testing.T) {
322 cases := []struct {
323 name string
324 close func(*previewFixture)
325 }{
326 {name: "subscription close", close: func(f *previewFixture) { f.sub.Close() }},
327 {name: "replacement", close: func(f *previewFixture) {
328 replacement, err := f.service.SubscribePreview(previewRequestFor(f.service, 2, previewTarget(0x11, "tab-alpha")))
329 require.NoError(f.t, err)
330 f.t.Cleanup(replacement.Close)
331 }},
332 {name: "service close", close: func(f *previewFixture) { require.NoError(f.t, f.service.Close()) }},
333 }
334 for _, tc := range cases {
335 t.Run(tc.name, func(t *testing.T) {
336 f := newPreviewFixture(t, previewTarget(0x11, "tab-alpha"))
337 conn, _ := f.awaitOpen()
338 conn.push(previewAnswer(0x11, "tab-alpha", 2, 1))
339 f.awaitPublication()
340 tc.close(f)
341 awaitClosed(t, conn)
342 require.NotPanics(t, func() { _ = f.sub.Latest() }, "Latest stays observable after Close")
343 })
344 }
345}
346
347func TestServicePreviewRouteValidation(t *testing.T) {
348 registration := domain.RemoteRegistration{Endpoint: "dev@host:22", Incarnation: [16]byte{1}, Generation: 1}
349 remoteTarget := previewTarget(0x11, "tab-alpha")
350 remoteTarget.Endpoint = "dev@host:22"
351 cases := []struct {
352 name string
353 mutate func(*ports.BrokerPreviewRequest)
354 }{
355 {name: "local route for remote target", mutate: func(r *ports.BrokerPreviewRequest) { r.Preview.Target = remoteTarget }},
356 {name: "remote route for local target", mutate: func(r *ports.BrokerPreviewRequest) {
357 r.Route = ports.BrokerPreviewRoute{Endpoint: registration.Endpoint, Registration: registration, Policy: poolPolicy()}
358 }},
359 {name: "remote route for another host", mutate: func(r *ports.BrokerPreviewRequest) {
360 other := remoteTarget
361 other.Endpoint = "other@host:22"
362 r.Route = ports.BrokerPreviewRoute{Endpoint: registration.Endpoint, Registration: registration, Policy: poolPolicy()}
363 r.Preview.Target = other
364 }},
365 {name: "remote route without registration", mutate: func(r *ports.BrokerPreviewRequest) {
366 r.Route = ports.BrokerPreviewRoute{Endpoint: registration.Endpoint, Policy: poolPolicy()}
367 r.Preview.Target = remoteTarget
368 }},
369 {name: "local route with registration", mutate: func(r *ports.BrokerPreviewRequest) { r.Route.Registration = registration }},
370 {name: "zero generation", mutate: func(r *ports.BrokerPreviewRequest) { r.Generation = 0 }},
371 {name: "invalid viewport", mutate: func(r *ports.BrokerPreviewRequest) { r.Preview.Width = 0 }},
372 }
373 f := newPreviewFixture(t, previewTarget(0x11, "tab-alpha"))
374 for _, tc := range cases {
375 t.Run(tc.name, func(t *testing.T) {
376 request := previewRequestFor(f.service, 5, previewTarget(0x11, "tab-alpha"))
377 tc.mutate(&request)
378 _, err := f.service.SubscribePreview(request)
379 require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
380 })
381 }
382 stale := previewRequestFor(f.service, 1, previewTarget(0x11, "tab-alpha"))
383 _, err := f.service.SubscribePreview(stale)
384 require.ErrorIs(t, err, ports.BrokerAdmissionStale, "a generation is never reused")
385}