1//go:build linux
2
3package app
4
5import (
6 "context"
7 "crypto/rand"
8 "errors"
9 "io"
10 "net/http"
11 "net/http/httptest"
12 "os"
13 "path/filepath"
14 "strings"
15 "sync"
16 "testing"
17 "time"
18
19 "github.com/coder/websocket"
20 "github.com/stretchr/testify/require"
21
22 "github.com/bnema/vev/internal/adapters/brokeripc"
23 "github.com/bnema/vev/internal/adapters/clock"
24 "github.com/bnema/vev/internal/adapters/daemonidentity"
25 "github.com/bnema/vev/internal/adapters/daemonmux"
26 "github.com/bnema/vev/internal/adapters/ipc"
27 "github.com/bnema/vev/internal/adapters/pty"
28 "github.com/bnema/vev/internal/adapters/snapshot"
29 "github.com/bnema/vev/internal/adapters/uiterm"
30 "github.com/bnema/vev/internal/adapters/webterm"
31 "github.com/bnema/vev/internal/domain"
32 "github.com/bnema/vev/internal/persist"
33 "github.com/bnema/vev/internal/platform"
34 "github.com/bnema/vev/internal/ports"
35 "github.com/bnema/vev/internal/protocol"
36 "github.com/bnema/vev/internal/usecase/daemon"
37 "github.com/bnema/vev/internal/usecase/recovery"
38 "github.com/bnema/vev/pkg/safedir"
39)
40
41type offlineClientFixture struct {
42 root string
43 socket string
44 streams <-chan struct{}
45 physical <-chan struct{}
46 physicalCount func() int
47 policy ports.BrokerPolicy
48 losePhysical func()
49 cancel context.CancelFunc
50 prodRuntime string
51 prodState string
52 // catalogue is the daemon's durable session catalogue when the fixture was
53 // started with persistence; nil otherwise.
54 catalogue *persist.Persister
55}
56
57func startOfflineClientFixture(t *testing.T) offlineClientFixture {
58 t.Helper()
59 // The default isolated production fixture shell prints one marker and then parks on read, so a
60 // client proves it reached a session without depending on shell echo of
61 // injected input.
62 return startOfflineClientFixtureWithShell(t, "/bin/sh", []string{"-c", "sleep 0.1; printf offline-ready; trap : TERM; while :; do read line || exit; done"})
63}
64
65// startOfflineClientFixtureWithShell starts the same isolated production broker with
66// an explicitly supplied daemon shell, so a browser test can drive real session
67// content and real session actions (echo and command output) through the
68// ordinary client path instead of a parked shell.
69func startOfflineClientFixtureWithShell(t *testing.T, shell string, shellArgs []string) offlineClientFixture {
70 t.Helper()
71 return startPersistentOfflineClientFixture(t, shell, shellArgs, false)
72}
73
74// startPersistentOfflineClientFixture starts the same isolated production broker but
75// additionally binds the daemon to real session persistence under the production daemon
76// state directory. It exists so a test can prove which lifecycle state the
77// daemon actually persists: a named session appears in the durable catalogue,
78// while an ephemeral one is never written.
79func startPersistentOfflineClientFixture(t *testing.T, shell string, shellArgs []string, persistEnabled bool) offlineClientFixture {
80 t.Helper()
81 layout := emptyProductionBrokerLayout(t, "")
82 root := layout.Root
83 prodRuntime, prodState := os.Getenv("XDG_RUNTIME_DIR"), os.Getenv("XDG_STATE_HOME")
84 identity, err := daemonidentity.LoadOrCreate(platform.StateDir())
85 require.NoError(t, err)
86 require.NoError(t, ensureProductionBrokerConfig())
87 route := daemonmux.SocketPath(ipc.SocketDir())
88 policy := localDaemonPolicy()
89 require.NoError(t, safedir.EnsurePrivate(ipc.SocketDir()))
90 rawListener, err := ipc.ListenMux(route)
91 require.NoError(t, err)
92 physicalSeen := make(chan struct{}, 8)
93 var physicalMu sync.Mutex
94 var physicalConnections []io.Closer
95 aggregate := daemonmux.NewAggregateListener()
96 binding, err := daemonmux.NewServerBinding(identity, ports.BrokerDaemonIncarnation{1}, policy)
97 require.NoError(t, err)
98 mux, err := daemonmux.NewServerSupervisor(aggregate, binding, daemonmux.DefaultMuxCeilings(), 8)
99 require.NoError(t, err)
100 ctx, cancel := context.WithCancel(context.Background())
101 go func() {
102 for {
103 raw, acceptErr := rawListener.Accept()
104 if acceptErr != nil {
105 return
106 }
107 physicalMu.Lock()
108 physicalConnections = append(physicalConnections, raw)
109 physicalMu.Unlock()
110 select {
111 case physicalSeen <- struct{}{}:
112 default:
113 }
114 go func() { _ = mux.Adopt(ctx, raw) }()
115 }
116 }()
117 streamSeen := make(chan struct{}, 16)
118 counting := &countingServerListener{ServerListener: aggregate, seen: streamSeen}
119 daemonOpts := []daemon.Option{daemon.WithShell(shell, shellArgs)}
120 var fixtureCatalogue *persist.Persister
121 if persistEnabled {
122 stateDir := platform.StateDir()
123 opened, err := persist.OpenOrCreate(stateDir)
124 require.NoError(t, err)
125 repository := snapshot.NewRepository(filepath.Join(stateDir, "snapshots"))
126 coordinator := recovery.NewCoordinator(opened.Catalogue, repository, rand.Reader)
127 daemonOpts = append(daemonOpts,
128 daemon.WithCatalogue(opened.Catalogue, opened.Records),
129 daemon.WithSnapshotRepository(repository),
130 daemon.WithRecoveryCoordinator(coordinator),
131 )
132 fixtureCatalogue = opened.Catalogue
133 }
134 d := daemon.New(pty.NewFactory(), clock.New(), discardLog(), daemonOpts...)
135 go func() { _ = d.Serve(ctx, counting) }()
136
137 // The client picker and registry must share wall-clock freshness semantics:
138 // a manual clock rooted at Unix zero would correctly make the observation
139 // stale to the real client clock before initial navigation resolves.
140 deps, ready := testBrokerServeDeps(clock.New())
141 brokerDone := runSandbox(ctx, brokerServeOptions{idleGrace: time.Hour}, deps)
142 var socket string
143 select {
144 case socket = <-ready:
145 case serveErr := <-brokerDone:
146 t.Fatalf("production broker exited before readiness: %v", serveErr)
147 case <-time.After(brokerTestWait):
148 t.Fatal("production broker did not become ready")
149 }
150 // The production broker becomes socket-ready before its first local daemon
151 // observation is committed. Product clients start from a committed local
152 // authority; wait for the fixture to provide the same contract so one-shot
153 // initial navigation is not consumed against an artificial empty catalogue.
154 probe, err := brokeripc.NewConnector(socket, brokeripc.Config{}).Connect(ctx)
155 require.NoError(t, err)
156 subscription, err := probe.Subscribe()
157 require.NoError(t, err)
158 require.Eventually(t, func() bool {
159 for _, observation := range probe.Snapshot().Daemons {
160 if observation.Local {
161 return true
162 }
163 }
164 select {
165 case <-subscription.Changed():
166 default:
167 }
168 return false
169 }, brokerTestWait, 5*time.Millisecond, "production broker never committed its local daemon authority")
170 subscription.Close()
171 require.NoError(t, probe.Close())
172 t.Cleanup(func() {
173 cancel()
174 _ = rawListener.Close()
175 _ = mux.Close()
176 _ = aggregate.Close()
177 select {
178 case <-brokerDone:
179 case <-time.After(brokerTestWait):
180 t.Error("production broker did not stop")
181 }
182 })
183 losePhysical := func() {
184 physicalMu.Lock()
185 connections := append([]io.Closer(nil), physicalConnections...)
186 physicalMu.Unlock()
187 for _, connection := range connections {
188 _ = connection.Close()
189 }
190 }
191 physicalCount := func() int {
192 physicalMu.Lock()
193 defer physicalMu.Unlock()
194 return len(physicalConnections)
195 }
196 return offlineClientFixture{
197 root: root, socket: socket, streams: streamSeen, physical: physicalSeen,
198 physicalCount: physicalCount, policy: policy, losePhysical: losePhysical,
199 cancel: cancel, prodRuntime: prodRuntime, prodState: prodState,
200 catalogue: fixtureCatalogue,
201 }
202}
203
204type countingServerListener struct {
205 ports.ServerListener
206 seen chan<- struct{}
207}
208
209func (l *countingServerListener) Accept() (ports.ServerConnection, error) {
210 connection, err := l.ServerListener.Accept()
211 if err == nil {
212 select {
213 case l.seen <- struct{}{}:
214 default:
215 }
216 }
217 return connection, err
218}
219
220func awaitLogicalStream(t *testing.T, streams <-chan struct{}) {
221 t.Helper()
222 select {
223 case <-streams:
224 case <-time.After(brokerTestWait):
225 t.Fatal("client did not reach a session through a broker logical stream")
226 }
227}
228
229func offlineSnapshotText(snapshot ports.UISnapshot) string {
230 var text strings.Builder
231 for _, cell := range snapshot.Cells {
232 if !cell.Continuation {
233 text.WriteString(cell.Text)
234 }
235 }
236 return text.String()
237}
238
239func awaitUITerminalMarker(t *testing.T, terminal *uiterm.Terminal, marker string) {
240 t.Helper()
241 deadline := time.NewTimer(brokerTestWait)
242 defer deadline.Stop()
243 for {
244 if snapshot, err := terminal.Snapshot(); err == nil && strings.Contains(offlineSnapshotText(snapshot), marker) {
245 return
246 }
247 select {
248 case <-terminal.Changes():
249 case <-deadline.C:
250 t.Fatalf("terminal never rendered session marker %q", marker)
251 }
252 }
253}
254
255// TestOfflineEphemeralSessionIsNotPersistedViaBroker proves the daemon-owned
256// broker fixture persists a named session but never an ephemeral one: the
257// durable catalogue holds exactly the named record after both lifecycles, so
258// the ephemeral session survives only while its daemon does.
259func TestOfflineEphemeralSessionIsNotPersistedViaBroker(t *testing.T) {
260 fixture := startPersistentOfflineClientFixture(t, "/bin/sh",
261 []string{"-c", "sleep 0.1; printf ephemeral-ready; trap : TERM; while :; do read line || exit; done"}, true)
262 require.NotNil(t, fixture.catalogue, "the persistent fixture must expose its durable catalogue")
263 ctx, cancel := context.WithTimeout(context.Background(), brokerTestWait)
264 defer cancel()
265
266 service, err := brokeripc.NewConnector(fixture.socket, brokeripc.Config{}).Connect(ctx)
267 require.NoError(t, err)
268 defer service.Close()
269
270 create := func(name string, admission ports.BrokerStreamAdmission, intent uint8) {
271 t.Helper()
272 streamID, err := service.NextStreamID()
273 require.NoError(t, err)
274 request := ports.BrokerOpenStreamRequest{
275 Purpose: ports.BrokerStreamAttachment, Stream: streamID, Local: true,
276 Policy: fixture.policy, Admission: admission, Name: name,
277 StartMode: ports.BrokerDaemonStartIfNeeded,
278 }
279 stream, err := service.OpenStream(ctx, request)
280 require.NoError(t, err)
281 require.NoError(t, stream.SendClient(protocol.Hello{
282 Version: protocol.Version, Intent: intent, Name: name,
283 Size: domain.Size{Cols: 80, Rows: 24}, EnvironmentPolicy: protocol.EnvironmentPolicyClientOwned,
284 }))
285 receiveOfflineWelcome(t, stream)
286 require.NoError(t, stream.Close())
287 }
288
289 create("persisted", ports.BrokerAdmissionCreateNamed, protocol.IntentNew)
290 create("", ports.BrokerAdmissionCreateEphemeral, protocol.IntentEphemeral)
291
292 // The daemon persists on its own cadence; the durable catalogue must settle
293 // to exactly the named record and never gain the ephemeral one.
294 require.Eventually(t, func() bool {
295 records, err := fixture.catalogue.Records()
296 require.NoError(t, err)
297 return len(records) == 1 && records[0].Name == "persisted"
298 }, brokerTestWait, brokerTestPollTick, "the durable catalogue must hold exactly the named session")
299
300 records, err := fixture.catalogue.Records()
301 require.NoError(t, err)
302 for _, record := range records {
303 require.NotEmpty(t, record.Name, "an ephemeral session must never be persisted")
304 }
305}
306
307func TestOfflineClientTerminalReachesSessionThroughBrokerIPC(t *testing.T) {
308 fixture := startOfflineClientFixture(t)
309 ctx, cancel := context.WithCancel(t.Context())
310 terminal, err := uiterm.New(ctx, domain.Geometry{Size: domain.Size{Cols: 80, Rows: 24}}, "")
311 require.NoError(t, err)
312 done := make(chan error, 1)
313 go func() {
314 done <- runOfflineClient(ctx, fixture.socket, terminal, nil, discardLog(), nil, nil)
315 }()
316 awaitLogicalStream(t, fixture.streams)
317 awaitUITerminalMarker(t, terminal, "offline-ready")
318 cancel()
319 select {
320 case err := <-done:
321 require.NoError(t, ignoreContextCancellation(err))
322 case <-time.After(brokerTestWait):
323 t.Fatal("offline terminal client did not stop")
324 }
325}
326
327// The sandbox UI-driver harness is exercised by
328// TestUIDriverSandboxHarnessUsesItsOwnConnector, TestUIDriverSandboxHarnessPickerOpensNothing,
329// TestUIDriverClientKeepsSharedDaemonOnEOF, and the shared
330// runUIDriverClient tests in ui_driver_client_test.go; the harness is the same
331// runUIDriverClient call production uses, so it is not re-tested here.
332
333func TestOfflineBrokerSharedTransportKeepsAttachmentsIndependent(t *testing.T) {
334 fixture := startOfflineClientFixture(t)
335 ctx, cancel := context.WithTimeout(context.Background(), brokerTestWait)
336 defer cancel()
337
338 service, err := brokeripc.NewConnector(fixture.socket, brokeripc.Config{}).Connect(ctx)
339 require.NoError(t, err)
340 defer service.Close()
341 connections := make([]ports.BrokerLogicalConnection, 3)
342 welcomes := make([]protocol.Welcome, 3)
343 for i := range connections {
344 streamID, err := service.NextStreamID()
345 require.NoError(t, err)
346 request := ports.BrokerOpenStreamRequest{Purpose: ports.BrokerStreamAttachment, Stream: streamID, Local: true, Policy: fixture.policy, StartMode: ports.BrokerDaemonStartIfNeeded}
347 if i == 0 {
348 request.Admission, request.Name = ports.BrokerAdmissionCreateNamed, "shared"
349 } else {
350 require.NotNil(t, welcomes[0].CommittedIdentity)
351 request.Admission, request.Target = ports.BrokerAdmissionExact, welcomes[0].CommittedIdentity.Target
352 }
353 connection, err := service.OpenStream(ctx, request)
354 require.NoError(t, err)
355 connections[i] = connection
356 hello := protocol.Hello{Version: protocol.Version, ClientID: [16]byte{9}, Size: domain.Size{Cols: 80 + i*10, Rows: 24 + i}}
357 if i == 0 {
358 hello.Intent, hello.Name = protocol.IntentNew, "shared"
359 } else {
360 hello.Intent, hello.Name, hello.ExactTarget = protocol.IntentAttach, "shared", &request.Target
361 }
362 require.NoError(t, connection.SendClient(hello))
363 welcomes[i] = receiveOfflineWelcome(t, connection)
364 }
365 for range connections {
366 awaitLogicalStream(t, fixture.streams)
367 }
368 // The fixture also performs one observation dial. The three attachment
369 // streams must add only one pooled carriage, never one carriage per client.
370 for range 2 {
371 select {
372 case <-fixture.physical:
373 case <-time.After(brokerTestWait):
374 t.Fatal("expected observation and pooled attachment transports")
375 }
376 }
377 baselinePhysical := fixture.physicalCount()
378 t.Cleanup(func() {
379 require.Equal(t, baselinePhysical, fixture.physicalCount(), "logical attachments opened another daemonmux carriage")
380 })
381
382 for i, welcome := range welcomes {
383 require.Equal(t, "shared", welcome.SessionName)
384 require.NotZero(t, welcome.ResumeToken, "attachment %d has no independent resume token", i)
385 for j := range i {
386 require.NotEqual(t, welcomes[j].ResumeToken, welcome.ResumeToken, "same client process identity collapsed attachments %d and %d", j, i)
387 }
388 }
389 // Resize remains attachment session traffic: each logical stream reports its
390 // own window after distinct claims rather than a broker-coalesced peer size.
391 sizes := make([]domain.Size, len(connections))
392 for i, connection := range connections {
393 sizes[i] = domain.Size{Cols: 101 + i*11, Rows: 31 + i}
394 require.NoError(t, connection.SendClient(protocol.Resize{Size: sizes[i]}))
395 }
396 for i, connection := range connections {
397 require.Equal(t, sizes[i], receiveOfflineOutputSize(t, connection, sizes[i]))
398 }
399 require.NoError(t, connections[0].SendClient(protocol.CommandRequest{Version: protocol.Version, RequestID: 6, Attached: true, Slug: "next-tab"}))
400 require.True(t, receiveOfflineCommandResult(t, connections[0], 6).Outcome == protocol.CommandSucceeded)
401 require.Equal(t, sizes[0], receiveOfflineOutputSize(t, connections[0], sizes[0]))
402 for _, connection := range connections {
403 require.NoError(t, connection.SendClient(protocol.CommandRequest{Version: protocol.Version, RequestID: 7, Attached: true, Slug: "next-tab"}))
404 result := receiveOfflineCommandResult(t, connection, 7)
405 require.True(t, result.Outcome == protocol.CommandSucceeded, result.Text)
406 }
407
408 require.NoError(t, connections[0].SendClient(protocol.Detach{}))
409 require.NoError(t, receiveOfflineEOF(t, connections[0]), "detach must end its logical stream with orderly EOF")
410 for i := 1; i < len(connections); i++ {
411 requestID := uint64(20 + i)
412 require.NoError(t, connections[i].SendClient(protocol.CommandRequest{Version: protocol.Version, RequestID: requestID, Attached: true, Slug: "next-tab"}))
413 require.True(t, receiveOfflineCommandResult(t, connections[i], requestID).Outcome == protocol.CommandSucceeded)
414 }
415 for i := 1; i < len(connections); i++ {
416 require.NoError(t, connections[i].Close())
417 }
418}
419
420func TestOfflineBrokerPhysicalLossSettlesEverySharedStream(t *testing.T) {
421 fixture := startOfflineClientFixture(t)
422 ctx, cancel := context.WithTimeout(context.Background(), brokerTestWait)
423 defer cancel()
424 service, err := brokeripc.NewConnector(fixture.socket, brokeripc.Config{}).Connect(ctx)
425 require.NoError(t, err)
426 defer service.Close()
427
428 var target protocol.ExactSessionTarget
429 connections := make([]ports.BrokerLogicalConnection, 3)
430 for i := range connections {
431 streamID, err := service.NextStreamID()
432 require.NoError(t, err)
433 request := ports.BrokerOpenStreamRequest{Purpose: ports.BrokerStreamAttachment, Stream: streamID, Local: true, Policy: fixture.policy, StartMode: ports.BrokerDaemonStartIfNeeded}
434 if i == 0 {
435 request.Admission, request.Name = ports.BrokerAdmissionCreateNamed, "loss"
436 } else {
437 request.Admission, request.Target = ports.BrokerAdmissionExact, target
438 }
439 connection, err := service.OpenStream(ctx, request)
440 require.NoError(t, err)
441 connections[i] = connection
442 hello := protocol.Hello{Version: protocol.Version, ClientID: [16]byte{3}, Size: domain.Size{Cols: 80 + i, Rows: 24}, Intent: protocol.IntentAttach, Name: "loss", ExactTarget: &target}
443 if i == 0 {
444 hello.Intent, hello.ExactTarget = protocol.IntentNew, nil
445 }
446 require.NoError(t, connection.SendClient(hello))
447 welcome := receiveOfflineWelcome(t, connection)
448 if i == 0 {
449 require.NotNil(t, welcome.CommittedIdentity)
450 target = welcome.CommittedIdentity.Target
451 }
452 }
453 fixture.losePhysical()
454 for i, connection := range connections {
455 select {
456 case <-connection.Done():
457 // Closing the accepted raw carriage is an orderly transport loss at
458 // this seam; the required invariant is that every stream settles.
459 case <-time.After(brokerTestWait):
460 t.Fatalf("stream %d survived physical transport loss", i)
461 }
462 }
463}
464
465func receiveOfflineWelcome(t *testing.T, connection ports.BrokerLogicalConnection) protocol.Welcome {
466 t.Helper()
467 message := receiveOfflineMessage(t, connection, "Welcome", func(message protocol.ServerMessage) bool {
468 _, ok := message.(protocol.Welcome)
469 return ok
470 })
471 return message.(protocol.Welcome)
472}
473
474func receiveOfflineOutputSize(t *testing.T, connection ports.BrokerLogicalConnection, expected domain.Size) domain.Size {
475 t.Helper()
476 message := receiveOfflineMessage(t, connection, "Output size", func(message protocol.ServerMessage) bool {
477 output, ok := message.(protocol.Output)
478 return ok && output.Size == expected
479 })
480 return message.(protocol.Output).Size
481}
482
483func receiveOfflineCommandResult(t *testing.T, connection ports.BrokerLogicalConnection, requestID uint64) protocol.CommandResult {
484 t.Helper()
485 message := receiveOfflineMessage(t, connection, "CommandResult", func(message protocol.ServerMessage) bool {
486 result, ok := message.(protocol.CommandResult)
487 return ok && result.RequestID == requestID
488 })
489 return message.(protocol.CommandResult)
490}
491
492func receiveOfflineEOF(t *testing.T, connection ports.BrokerLogicalConnection) error {
493 t.Helper()
494 result := make(chan error, 1)
495 go func() {
496 for {
497 _, err := connection.ReceiveServer()
498 if err != nil {
499 result <- err
500 return
501 }
502 }
503 }()
504 select {
505 case err := <-result:
506 if errors.Is(err, io.EOF) {
507 return nil
508 }
509 return err
510 case <-time.After(brokerTestWait):
511 return errors.New("timed out waiting for orderly EOF")
512 }
513}
514
515func receiveOfflineMessage(t *testing.T, connection ports.BrokerLogicalConnection, want string, match func(protocol.ServerMessage) bool) protocol.ServerMessage {
516 t.Helper()
517 type result struct {
518 message protocol.ServerMessage
519 err error
520 }
521 resultCh := make(chan result, 1)
522 go func() {
523 for {
524 message, err := connection.ReceiveServer()
525 if err != nil || match(message) {
526 resultCh <- result{message: message, err: err}
527 return
528 }
529 }
530 }()
531 select {
532 case received := <-resultCh:
533 require.NoError(t, received.err, "waiting for %s", want)
534 return received.message
535 case <-connection.Done():
536 t.Fatalf("connection settled while waiting for %s: %v", want, connection.Err())
537 case <-time.After(brokerTestWait):
538 _ = connection.Close()
539 t.Fatalf("timed out waiting for %s", want)
540 }
541 return nil
542}
543
544func TestOfflineClientBrowserReachesSessionThroughBrokerIPC(t *testing.T) {
545 fixture := startOfflineClientFixture(t)
546 ctx, cancel := context.WithCancel(t.Context())
547 defer cancel()
548 token := strings.Repeat("b", 43)
549 handler, err := webterm.NewServer(ctx, webterm.Settings{}, token, func(runCtx context.Context, terminal *webterm.Terminal) error {
550 return runOfflineClient(runCtx, fixture.socket, terminal, nil, discardLog(), nil, nil)
551 })
552 require.NoError(t, err)
553 server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { r.Host = webterm.Address; handler.ServeHTTP(w, r) }))
554 defer server.Close()
555 headers := http.Header{"Origin": {webterm.Origin}, "Cookie": {"vev-web-session=" + token}}
556 connection, _, err := websocket.Dial(ctx, "ws"+strings.TrimPrefix(server.URL, "http")+"/ws", &websocket.DialOptions{HTTPHeader: headers})
557 require.NoError(t, err)
558 defer connection.CloseNow()
559 connection.SetReadLimit(64 << 20)
560 awaitLogicalStream(t, fixture.streams)
561 readCtx, stopRead := context.WithTimeout(ctx, brokerTestWait)
562 defer stopRead()
563 // This proves authenticated browser transport, real broker logical-stream
564 // admission, and live browser frames. It does not prove post-Hello session
565 // output: webterm may publish its transactional full-screen snapshot before
566 // the fixture marker and does not guarantee another frame containing it.
567 _, frame, err := connection.Read(readCtx)
568 require.NoError(t, err)
569 require.NotEmpty(t, frame, "browser emitted no frame after broker logical-stream admission")
570 require.NoError(t, connection.CloseNow())
571 waited := make(chan struct{})
572 go func() {
573 handler.Wait()
574 close(waited)
575 }()
576 select {
577 case <-waited:
578 case <-time.After(brokerTestWait):
579 t.Fatal("browser handler did not stop")
580 }
581}