1package client
2
3import (
4 "context"
5 "errors"
6 "io"
7 "sync"
8 "testing"
9 "time"
10
11 "github.com/stretchr/testify/require"
12
13 "github.com/bnema/vev/internal/ports"
14 "github.com/bnema/vev/internal/protocol"
15)
16
17// Client control operations (Plan 001 P7, worker 3). Every case is
18// deterministic: a manually fired clock drives the one bound, scripted streams
19// stand in for the broker carriage, and no case sleeps.
20
21const brokerOpsTestEpoch = ports.BrokerEpoch(0x5a)
22
23// brokerOpsTestStream is one scripted broker logical stream. It records every
24// typed client message, delivers scripted server messages, and unblocks
25// ReceiveServer on Close exactly as the ports.BrokerLogicalConnection contract
26// requires.
27type brokerOpsTestStream struct {
28 mu sync.Mutex
29 sent []protocol.ClientMessage
30 sendErr error
31 recvErr error
32 incoming chan protocol.ServerMessage
33
34 closeOnce sync.Once
35 closed chan struct{}
36 closes int
37}
38
39func newBrokerOpsTestStream() *brokerOpsTestStream {
40 return &brokerOpsTestStream{
41 incoming: make(chan protocol.ServerMessage, 8),
42 closed: make(chan struct{}),
43 }
44}
45
46func (s *brokerOpsTestStream) failSend(err error) {
47 s.mu.Lock()
48 s.sendErr = err
49 s.mu.Unlock()
50}
51
52func (s *brokerOpsTestStream) failReceive(err error) {
53 s.mu.Lock()
54 s.recvErr = err
55 s.mu.Unlock()
56}
57
58func (s *brokerOpsTestStream) SendClient(message protocol.ClientMessage) error {
59 s.mu.Lock()
60 if s.sendErr != nil {
61 err := s.sendErr
62 s.mu.Unlock()
63 return err
64 }
65 s.sent = append(s.sent, message)
66 s.mu.Unlock()
67 return nil
68}
69
70func (s *brokerOpsTestStream) ReceiveServer() (protocol.ServerMessage, error) {
71 select {
72 case message := <-s.incoming:
73 return message, nil
74 default:
75 }
76 s.mu.Lock()
77 err := s.recvErr
78 s.mu.Unlock()
79 if err != nil {
80 return nil, err
81 }
82 select {
83 case message := <-s.incoming:
84 return message, nil
85 case <-s.closed:
86 return nil, io.EOF
87 }
88}
89
90func (s *brokerOpsTestStream) Capabilities() protocol.ConnectionCapabilities {
91 return protocol.ConnectionCapabilities{}
92}
93
94func (s *brokerOpsTestStream) LinkState() ports.LinkState { return ports.LinkStateConnected }
95
96func (s *brokerOpsTestStream) LinkEvents() <-chan ports.LinkEvent { return nil }
97
98func (s *brokerOpsTestStream) Close() error {
99 s.closeOnce.Do(func() {
100 s.mu.Lock()
101 s.closes++
102 s.mu.Unlock()
103 close(s.closed)
104 })
105 return nil
106}
107
108func (s *brokerOpsTestStream) Done() <-chan struct{} { return s.closed }
109
110func (s *brokerOpsTestStream) Err() error { return nil }
111
112func (s *brokerOpsTestStream) deliver(message protocol.ServerMessage) {
113 select {
114 case s.incoming <- message:
115 case <-s.closed:
116 }
117}
118
119func (s *brokerOpsTestStream) messages() []protocol.ClientMessage {
120 s.mu.Lock()
121 defer s.mu.Unlock()
122 return append([]protocol.ClientMessage(nil), s.sent...)
123}
124
125func (s *brokerOpsTestStream) sentCount() int {
126 s.mu.Lock()
127 defer s.mu.Unlock()
128 return len(s.sent)
129}
130
131func (s *brokerOpsTestStream) closeCount() int {
132 s.mu.Lock()
133 defer s.mu.Unlock()
134 return s.closes
135}
136
137func (s *brokerOpsTestStream) awaitDispatch(t *testing.T) {
138 t.Helper()
139 require.Eventually(t, func() bool { return s.sentCount() > 0 }, time.Second, time.Millisecond,
140 "the control request was never dispatched")
141}
142
143// brokerOpsTestService is a scripted BrokerService. It records every open, the
144// reconcile hints it received, and whether the borrowed service was closed.
145type brokerOpsTestService struct {
146 id ports.BrokerConnectionID
147 snapshot ports.BrokerSnapshot
148
149 mu sync.Mutex
150 opens []ports.BrokerOpenStreamRequest
151 reconciles []string
152 closes int
153 nextStream ports.BrokerStreamID
154 open func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error)
155}
156
157func newBrokerOpsTestService(id ports.BrokerConnectionID, snapshot ports.BrokerSnapshot) *brokerOpsTestService {
158 return &brokerOpsTestService{id: id, snapshot: snapshot}
159}
160
161func (s *brokerOpsTestService) ConnectionID() ports.BrokerConnectionID { return s.id }
162
163func (s *brokerOpsTestService) NextStreamID() (ports.BrokerStreamID, error) {
164 s.mu.Lock()
165 defer s.mu.Unlock()
166 s.nextStream++
167 return s.nextStream, nil
168}
169
170func (s *brokerOpsTestService) Done() <-chan struct{} { return make(chan struct{}) }
171func (s *brokerOpsTestService) Err() error { return nil }
172
173func (s *brokerOpsTestService) Snapshot() ports.BrokerSnapshot { return s.snapshot.Clone() }
174
175func (s *brokerOpsTestService) Subscribe() (ports.BrokerSubscription, error) {
176 return nil, errors.New("broker ops test service does not support subscriptions")
177}
178
179func (s *brokerOpsTestService) SubscribePreview(ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
180 return nil, errors.New("broker ops test service does not support preview")
181}
182
183func (s *brokerOpsTestService) OpenStream(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
184 s.mu.Lock()
185 s.opens = append(s.opens, request)
186 open := s.open
187 s.mu.Unlock()
188 if open == nil {
189 return nil, errors.New("broker ops test service does not support streams")
190 }
191 return open(ctx, request)
192}
193
194func (s *brokerOpsTestService) openedRequests() []ports.BrokerOpenStreamRequest {
195 s.mu.Lock()
196 defer s.mu.Unlock()
197 return append([]ports.BrokerOpenStreamRequest(nil), s.opens...)
198}
199
200func (s *brokerOpsTestService) reconcileHints() []string {
201 s.mu.Lock()
202 defer s.mu.Unlock()
203 return append([]string(nil), s.reconciles...)
204}
205
206func (s *brokerOpsTestService) closeCount() int {
207 s.mu.Lock()
208 defer s.mu.Unlock()
209 return s.closes
210}
211
212func (s *brokerOpsTestService) RequestReconcile(endpoint string) {
213 s.mu.Lock()
214 s.reconciles = append(s.reconciles, endpoint)
215 s.mu.Unlock()
216}
217
218func (s *brokerOpsTestService) Close() error {
219 s.mu.Lock()
220 s.closes++
221 s.mu.Unlock()
222 return nil
223}
224
225var _ ports.BrokerNavigator = (*brokerOpsTestService)(nil)
226
227func brokerOpsTestLocalSnapshot() ports.BrokerSnapshot {
228 now := time.Unix(1000, 0)
229 return ports.BrokerSnapshot{
230 Epoch: brokerOpsTestEpoch,
231 Revision: 1,
232 Daemons: []ports.BrokerDaemonObservation{pickerTestLocalObservation(now)},
233 }
234}
235
236func brokerOpsTestRemoteSnapshot() ports.BrokerSnapshot {
237 now := time.Unix(1000, 0)
238 return ports.BrokerSnapshot{
239 Epoch: brokerOpsTestEpoch,
240 Revision: 1,
241 Daemons: []ports.BrokerDaemonObservation{
242 pickerTestRemoteObservation("user@example.test", 1, 2, now),
243 },
244 }
245}
246
247func brokerOpsTestLocalRoute() BrokerOperationRoute {
248 return BrokerOperationRoute{Epoch: brokerOpsTestEpoch, Destination: ports.BrokerEndpointFence{Local: true}}
249}
250
251func brokerOpsTestRemoteRoute() BrokerOperationRoute {
252 return BrokerOperationRoute{
253 Epoch: brokerOpsTestEpoch,
254 Destination: ports.BrokerEndpointFence{Registration: pickerTestRegistration("user@example.test", 1)},
255 }
256}
257
258// brokerOpsTestOperation wires one executor over a scripted service whose every
259// open returns the supplied stream.
260func brokerOpsTestOperation(t *testing.T, snapshot ports.BrokerSnapshot, stream *brokerOpsTestStream) (*BrokerOperations, *brokerOpsTestService, *supervisorTestClock) {
261 t.Helper()
262 service := newBrokerOpsTestService(ports.BrokerConnectionID{7}, snapshot)
263 service.open = func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
264 return stream, nil
265 }
266 clock := newSupervisorTestClock()
267 ops, err := NewBrokerOperations(service, clock)
268 require.NoError(t, err)
269 return ops, service, clock
270}
271
272// TestBrokerOperationsConstructorRefusesMissingDependencies proves a nil or
273// typed-nil service or clock is refused rather than deferred to a panic.
274func TestBrokerOperationsConstructorRefusesMissingDependencies(t *testing.T) {
275 clock := newSupervisorTestClock()
276 service := newBrokerOpsTestService(ports.BrokerConnectionID{1}, brokerOpsTestLocalSnapshot())
277
278 var typedNilService *brokerOpsTestService
279 var typedNilClock *supervisorTestClock
280
281 tests := []struct {
282 name string
283 service ports.BrokerNavigator
284 clock ports.Clock
285 wantErr bool
286 }{
287 {name: "valid", service: service, clock: clock},
288 {name: "nil service", service: nil, clock: clock, wantErr: true},
289 {name: "typed nil service", service: typedNilService, clock: clock, wantErr: true},
290 {name: "nil clock", service: service, clock: nil, wantErr: true},
291 {name: "typed nil clock", service: service, clock: typedNilClock, wantErr: true},
292 }
293 for _, tt := range tests {
294 t.Run(tt.name, func(t *testing.T) {
295 ops, err := NewBrokerOperations(tt.service, tt.clock)
296 if tt.wantErr {
297 require.Error(t, err)
298 require.Nil(t, ops)
299 return
300 }
301 require.NoError(t, err)
302 require.NotNil(t, ops)
303 })
304 }
305}
306
307// TestBrokerOperationRouteValidation pins the closed route shape: a nonzero
308// epoch with a valid local or remote fence.
309func TestBrokerOperationRouteValidation(t *testing.T) {
310 remote := pickerTestRegistration("user@example.test", 1)
311 tests := []struct {
312 name string
313 route BrokerOperationRoute
314 wantErr bool
315 }{
316 {name: "local", route: BrokerOperationRoute{Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true}}},
317 {name: "remote", route: BrokerOperationRoute{Epoch: 3, Destination: ports.BrokerEndpointFence{Registration: remote}}},
318 {name: "zero epoch", route: BrokerOperationRoute{Destination: ports.BrokerEndpointFence{Local: true}}, wantErr: true},
319 {name: "zero destination", route: BrokerOperationRoute{Epoch: 3}, wantErr: true},
320 {name: "local carrying a registration", route: BrokerOperationRoute{Epoch: 3, Destination: ports.BrokerEndpointFence{Local: true, Registration: remote}}, wantErr: true},
321 }
322 for _, tt := range tests {
323 t.Run(tt.name, func(t *testing.T) {
324 err := tt.route.Validate()
325 if tt.wantErr {
326 require.Error(t, err)
327 return
328 }
329 require.NoError(t, err)
330 })
331 }
332}
333
334// TestBrokerOperationsPreDispatchRefusalsAreNotSent proves every refusal that
335// happens before the request reaches the wire wraps ErrBrokerOperationNotSent
336// and opens no stream.
337func TestBrokerOperationsPreDispatchRefusalsAreNotSent(t *testing.T) {
338 valid := protocol.CommandRequest{Version: protocol.Version, Slug: "next-tab"}
339
340 tests := []struct {
341 name string
342 call func(*testing.T, *BrokerOperations)
343 }{
344 {
345 name: "route with zero epoch",
346 call: func(t *testing.T, ops *BrokerOperations) {
347 _, err := ops.List(context.Background(), BrokerOperationRoute{Destination: ports.BrokerEndpointFence{Local: true}})
348 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
349 },
350 },
351 {
352 name: "destination missing from the snapshot",
353 call: func(t *testing.T, ops *BrokerOperations) {
354 route := BrokerOperationRoute{Epoch: brokerOpsTestEpoch, Destination: ports.BrokerEndpointFence{Registration: pickerTestRegistration("user@elsewhere.test", 3)}}
355 _, err := ops.List(context.Background(), route)
356 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
357 },
358 },
359 {
360 name: "route epoch does not match the current snapshot",
361 call: func(t *testing.T, ops *BrokerOperations) {
362 _, err := ops.List(context.Background(), BrokerOperationRoute{Epoch: brokerOpsTestEpoch + 1, Destination: ports.BrokerEndpointFence{Local: true}})
363 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
364 },
365 },
366 {
367 name: "already cancelled caller",
368 call: func(t *testing.T, ops *BrokerOperations) {
369 ctx, cancel := context.WithCancel(context.Background())
370 cancel()
371 _, err := ops.Command(ctx, brokerOpsTestLocalRoute(), valid)
372 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
373 require.ErrorIs(t, err, context.Canceled)
374 },
375 },
376 {
377 name: "command version mismatch",
378 call: func(t *testing.T, ops *BrokerOperations) {
379 _, err := ops.Command(context.Background(), brokerOpsTestLocalRoute(), protocol.CommandRequest{Version: protocol.Version + 1, Slug: "next-tab"})
380 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
381 },
382 },
383 {
384 name: "command carries a request ID",
385 call: func(t *testing.T, ops *BrokerOperations) {
386 request := valid
387 request.RequestID = 9
388 _, err := ops.Command(context.Background(), brokerOpsTestLocalRoute(), request)
389 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
390 },
391 },
392 {
393 name: "command is attached",
394 call: func(t *testing.T, ops *BrokerOperations) {
395 request := valid
396 request.Attached = true
397 _, err := ops.Command(context.Background(), brokerOpsTestLocalRoute(), request)
398 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
399 },
400 },
401 {
402 name: "command is self without an explicit target",
403 call: func(t *testing.T, ops *BrokerOperations) {
404 request := valid
405 request.Self = true
406 _, err := ops.Command(context.Background(), brokerOpsTestLocalRoute(), request)
407 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
408 },
409 },
410 {
411 name: "kill carries an invalid session name",
412 call: func(t *testing.T, ops *BrokerOperations) {
413 _, err := ops.Kill(context.Background(), brokerOpsTestLocalRoute(), "bad name")
414 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
415 },
416 },
417 }
418
419 for _, tt := range tests {
420 t.Run(tt.name, func(t *testing.T) {
421 stream := newBrokerOpsTestStream()
422 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
423 tt.call(t, ops)
424 require.Empty(t, service.openedRequests(), "a not-sent refusal must open no stream")
425 require.Equal(t, 0, stream.sentCount(), "a not-sent refusal must send nothing")
426 require.Empty(t, service.reconcileHints(), "a not-sent refusal must not request reconcile")
427 })
428 }
429}
430
431// TestBrokerOperationOpenFailureIsNotSent proves a refused stream open is a
432// definite not-sent outcome rather than an indeterminate one.
433func TestBrokerOperationOpenFailureIsNotSent(t *testing.T) {
434 service := newBrokerOpsTestService(ports.BrokerConnectionID{7}, brokerOpsTestLocalSnapshot())
435 service.open = func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
436 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible}
437 }
438 ops, err := NewBrokerOperations(service, newSupervisorTestClock())
439 require.NoError(t, err)
440
441 _, err = ops.Command(context.Background(), brokerOpsTestLocalRoute(), protocol.CommandRequest{Version: protocol.Version, Slug: "next-tab"})
442 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
443 require.Empty(t, service.reconcileHints())
444}
445
446// TestBrokerOperationsListUsesOneControlStream proves List opens exactly one
447// control stream, sends exactly one List, and returns the daemon's own
448// sessions rather than a fabricated or cached value.
449func TestBrokerOperationsListUsesOneControlStream(t *testing.T) {
450 tests := []struct {
451 name string
452 list func(*BrokerOperations, context.Context, BrokerOperationRoute) ([]protocol.SessionInfo, error)
453 startMode ports.BrokerDaemonStartMode
454 }{
455 {name: "list never starts the daemon", list: (*BrokerOperations).List, startMode: ports.BrokerDaemonExistingOnly},
456 {name: "list starting may start the daemon", list: (*BrokerOperations).ListStarting, startMode: ports.BrokerDaemonStartIfNeeded},
457 }
458 for _, tt := range tests {
459 t.Run(tt.name, func(t *testing.T) {
460 stream := newBrokerOpsTestStream()
461 stream.deliver(protocol.Sessions{Sessions: []protocol.SessionInfo{{Name: "alpha", State: protocol.SessionUp}}})
462 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
463
464 sessions, err := tt.list(ops, context.Background(), brokerOpsTestLocalRoute())
465 require.NoError(t, err)
466 require.Equal(t, []protocol.SessionInfo{{Name: "alpha", State: protocol.SessionUp}}, sessions)
467
468 requests := service.openedRequests()
469 require.Len(t, requests, 1)
470 request := requests[0]
471 require.Equal(t, ports.BrokerStreamControl, request.Purpose)
472 require.Equal(t, ports.BrokerStreamAdmission(0), request.Admission)
473 require.Empty(t, request.Name)
474 require.Equal(t, protocol.ExactSessionTarget{}, request.Target)
475 require.Empty(t, request.Env)
476 require.Equal(t, tt.startMode, request.StartMode)
477 require.True(t, request.Local)
478 require.Equal(t, service.ConnectionID(), request.Connection)
479 require.NotZero(t, request.Stream)
480 require.Equal(t, brokerOpsTestEpoch, request.Epoch)
481
482 sent := stream.messages()
483 require.Len(t, sent, 1)
484 require.Equal(t, protocol.List{}, sent[0])
485 require.Equal(t, 1, stream.closeCount(), "the owned stream is closed exactly once")
486 require.Equal(t, 0, service.closeCount(), "the borrowed service is never closed")
487 })
488 }
489}
490
491// TestBrokerOperationsListLostReplyIsError proves a lost or malformed listing
492// is an ordinary error and never an empty fabricated result.
493func TestBrokerOperationsListLostReplyIsError(t *testing.T) {
494 tests := []struct {
495 name string
496 prepare func(*brokerOpsTestStream)
497 }{
498 {
499 name: "wrong reply type",
500 prepare: func(s *brokerOpsTestStream) {
501 s.deliver(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded})
502 },
503 },
504 {
505 name: "EOF before a reply",
506 prepare: func(s *brokerOpsTestStream) { s.failReceive(io.EOF) },
507 },
508 }
509 for _, tt := range tests {
510 t.Run(tt.name, func(t *testing.T) {
511 stream := newBrokerOpsTestStream()
512 tt.prepare(stream)
513 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
514 sessions, err := ops.List(context.Background(), brokerOpsTestLocalRoute())
515 require.Error(t, err)
516 require.Nil(t, sessions)
517 require.Len(t, service.openedRequests(), 1, "the listing is never retried")
518 require.Empty(t, service.reconcileHints(), "a listing has no mutation to reconcile")
519 })
520 }
521}
522
523// TestBrokerOperationsCommandCorrelation proves a terminal, correlated,
524// well-formed command reply is returned verbatim while every wrong reply
525// becomes a synthetic, correlated OutcomeUnknown with a nil error.
526func TestBrokerOperationsCommandCorrelation(t *testing.T) {
527 tests := []struct {
528 name string
529 reply protocol.ServerMessage
530 wantResult protocol.CommandResult
531 wantSynth bool
532 }{
533 {
534 name: "succeeded verbatim",
535 reply: protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandSucceeded, Output: "ok\n"},
536 wantResult: protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandSucceeded, Output: "ok\n"},
537 },
538 {
539 name: "failed verbatim",
540 reply: protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandFailed, Code: 5, Text: "boom"},
541 wantResult: protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandFailed, Code: 5, Text: "boom"},
542 },
543 {
544 name: "wrong request ID is synthetic",
545 reply: protocol.CommandResult{RequestID: 42, Outcome: protocol.CommandSucceeded},
546 wantSynth: true,
547 },
548 {
549 name: "malformed success is synthetic",
550 reply: protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandSucceeded, Code: 5},
551 wantSynth: true,
552 },
553 {
554 name: "wrong reply type is synthetic",
555 reply: protocol.Sessions{},
556 wantSynth: true,
557 },
558 }
559 for _, tt := range tests {
560 t.Run(tt.name, func(t *testing.T) {
561 stream := newBrokerOpsTestStream()
562 stream.deliver(tt.reply)
563 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
564 result, err := ops.Command(context.Background(), brokerOpsTestLocalRoute(), protocol.CommandRequest{Version: protocol.Version, Slug: "next-tab"})
565 require.NoError(t, err)
566 if tt.wantSynth {
567 require.Equal(t, protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandOutcomeUnknown, Text: "command outcome unknown"}, result)
568 require.Len(t, service.reconcileHints(), 1, "an indeterminate command requests exactly one reconcile")
569 require.Len(t, service.openedRequests(), 1, "an indeterminate command is never replayed")
570 return
571 }
572 require.Equal(t, tt.wantResult, result)
573 require.Empty(t, service.reconcileHints())
574 })
575 }
576}
577
578// TestBrokerOperationsCommandDaemonUnknownIsVerbatim proves a daemon-reported
579// Unknown is returned as a terminal result (with one reconcile hint) and never
580// hidden behind a plain context error.
581func TestBrokerOperationsCommandDaemonUnknownIsVerbatim(t *testing.T) {
582 stream := newBrokerOpsTestStream()
583 stream.deliver(protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandOutcomeUnknown, Text: "daemon could not finish"})
584 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
585
586 result, err := ops.Command(context.Background(), brokerOpsTestLocalRoute(), protocol.CommandRequest{Version: protocol.Version, Slug: "next-tab"})
587 require.NoError(t, err)
588 require.Equal(t, protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandOutcomeUnknown, Text: "daemon could not finish"}, result)
589 require.Equal(t, []string{""}, service.reconcileHints())
590}
591
592// TestBrokerOperationsKillCorrelation proves the Kill-family correlation table:
593// terminal replies are verbatim, wrong or malformed replies become a synthetic
594// OutcomeUnknown, and partial kill-all failures are preserved.
595func TestBrokerOperationsKillCorrelation(t *testing.T) {
596 tests := []struct {
597 name string
598 call func(context.Context, *BrokerOperations, BrokerOperationRoute) (protocol.KillResult, error)
599 reply protocol.ServerMessage
600 wantResult protocol.KillResult
601 wantSynth bool
602 }{
603 {
604 name: "kill succeeded",
605 call: func(ctx context.Context, ops *BrokerOperations, route BrokerOperationRoute) (protocol.KillResult, error) {
606 return ops.Kill(ctx, route, "work")
607 },
608 reply: protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded},
609 wantResult: protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded},
610 },
611 {
612 name: "kill failed with a definite code",
613 call: func(ctx context.Context, ops *BrokerOperations, route BrokerOperationRoute) (protocol.KillResult, error) {
614 return ops.Kill(ctx, route, "ghost")
615 },
616 reply: protocol.KillResult{RequestID: 1, Outcome: protocol.KillFailed, Code: 4, Text: "no such session: ghost"},
617 wantResult: protocol.KillResult{RequestID: 1, Outcome: protocol.KillFailed, Code: 4, Text: "no such session: ghost"},
618 },
619 {
620 name: "kill all partial failures verbatim",
621 call: func(ctx context.Context, ops *BrokerOperations, route BrokerOperationRoute) (protocol.KillResult, error) {
622 return ops.KillAll(ctx, route)
623 },
624 reply: protocol.KillResult{RequestID: 1, Outcome: protocol.KillFailed, Code: 8, Text: "partial", Failures: []protocol.KillFailure{{Class: "stopped", Name: "work", Text: "boom"}}},
625 wantResult: protocol.KillResult{RequestID: 1, Outcome: protocol.KillFailed, Code: 8, Text: "partial", Failures: []protocol.KillFailure{{Class: "stopped", Name: "work", Text: "boom"}}},
626 },
627 {
628 name: "wrong request ID is synthetic",
629 call: func(ctx context.Context, ops *BrokerOperations, route BrokerOperationRoute) (protocol.KillResult, error) {
630 return ops.Kill(ctx, route, "work")
631 },
632 reply: protocol.KillResult{RequestID: 5, Outcome: protocol.KillSucceeded},
633 wantSynth: true,
634 },
635 {
636 name: "malformed failure is synthetic",
637 call: func(ctx context.Context, ops *BrokerOperations, route BrokerOperationRoute) (protocol.KillResult, error) {
638 return ops.Kill(ctx, route, "work")
639 },
640 reply: protocol.KillResult{RequestID: 1, Outcome: protocol.KillFailed},
641 wantSynth: true,
642 },
643 {
644 name: "wrong reply type is synthetic",
645 call: func(ctx context.Context, ops *BrokerOperations, route BrokerOperationRoute) (protocol.KillResult, error) {
646 return ops.KillAll(ctx, route)
647 },
648 reply: protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandSucceeded},
649 wantSynth: true,
650 },
651 }
652 for _, tt := range tests {
653 t.Run(tt.name, func(t *testing.T) {
654 stream := newBrokerOpsTestStream()
655 stream.deliver(tt.reply)
656 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
657 result, err := tt.call(context.Background(), ops, brokerOpsTestLocalRoute())
658 require.NoError(t, err)
659 if tt.wantSynth {
660 require.Equal(t, protocol.KillResult{RequestID: 1, Outcome: protocol.KillOutcomeUnknown, Text: "kill outcome unknown"}, result)
661 require.Len(t, service.reconcileHints(), 1)
662 require.Len(t, service.openedRequests(), 1, "an indeterminate kill is never replayed")
663 return
664 }
665 require.Equal(t, tt.wantResult, result)
666 require.Empty(t, service.reconcileHints())
667 })
668 }
669}
670
671// TestBrokerOperationsKillShapes proves Kill, KillAll, and StopDaemon carry the
672// exact v58 scope, name, and start-mode each semantic requires.
673func TestBrokerOperationsKillShapes(t *testing.T) {
674 tests := []struct {
675 name string
676 call func(context.Context, *BrokerOperations, BrokerOperationRoute) (protocol.KillResult, error)
677 wantScope protocol.KillScope
678 wantName string
679 wantStartMode ports.BrokerDaemonStartMode
680 }{
681 {
682 name: "kill session",
683 call: func(ctx context.Context, ops *BrokerOperations, route BrokerOperationRoute) (protocol.KillResult, error) {
684 return ops.Kill(ctx, route, "work")
685 },
686 wantScope: protocol.KillSession,
687 wantName: "work",
688 wantStartMode: ports.BrokerDaemonStartIfNeeded,
689 },
690 {
691 name: "kill all",
692 call: func(ctx context.Context, ops *BrokerOperations, route BrokerOperationRoute) (protocol.KillResult, error) {
693 return ops.KillAll(ctx, route)
694 },
695 wantScope: protocol.KillAll,
696 wantStartMode: ports.BrokerDaemonStartIfNeeded,
697 },
698 {
699 name: "stop daemon is existing only",
700 call: func(ctx context.Context, ops *BrokerOperations, route BrokerOperationRoute) (protocol.KillResult, error) {
701 return ops.StopDaemon(ctx, route)
702 },
703 wantScope: protocol.KillDaemon,
704 wantStartMode: ports.BrokerDaemonExistingOnly,
705 },
706 }
707 for _, tt := range tests {
708 t.Run(tt.name, func(t *testing.T) {
709 stream := newBrokerOpsTestStream()
710 stream.deliver(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded})
711 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
712
713 result, err := tt.call(context.Background(), ops, brokerOpsTestLocalRoute())
714 require.NoError(t, err)
715 require.Equal(t, protocol.KillSucceeded, result.Outcome)
716
717 requests := service.openedRequests()
718 require.Len(t, requests, 1)
719 require.Equal(t, ports.BrokerStreamControl, requests[0].Purpose)
720 require.Equal(t, tt.wantStartMode, requests[0].StartMode)
721
722 sent := stream.messages()
723 require.Len(t, sent, 1)
724 kill, ok := sent[0].(protocol.Kill)
725 require.True(t, ok)
726 require.Equal(t, tt.wantScope, kill.Scope)
727 require.Equal(t, tt.wantName, kill.Name)
728 require.Equal(t, uint64(1), kill.RequestID)
729 })
730 }
731}
732
733// TestBrokerOperationsStopDaemonLostAcknowledgementIsUnknown proves the
734// ExistingOnly stop reports OutcomeUnknown rather than a success when its
735// acknowledgement never arrives: a close or EOF is never a success.
736func TestBrokerOperationsStopDaemonLostAcknowledgementIsUnknown(t *testing.T) {
737 stream := newBrokerOpsTestStream()
738 stream.failReceive(io.EOF)
739 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
740
741 result, err := ops.StopDaemon(context.Background(), brokerOpsTestLocalRoute())
742 require.NoError(t, err)
743 require.Equal(t, protocol.KillOutcomeUnknown, result.Outcome)
744 require.Equal(t, uint64(1), result.RequestID)
745 require.Equal(t, ports.BrokerDaemonExistingOnly, service.openedRequests()[0].StartMode)
746 require.Len(t, service.reconcileHints(), 1)
747}
748
749// TestBrokerOperationsSendErrorIsOutcomeUnknown proves a send error may be a
750// partial delivery, so it becomes a synthetic OutcomeUnknown for a mutation.
751func TestBrokerOperationsSendErrorIsOutcomeUnknown(t *testing.T) {
752 stream := newBrokerOpsTestStream()
753 stream.failSend(errors.New("write: broken pipe"))
754 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
755
756 result, err := ops.Kill(context.Background(), brokerOpsTestLocalRoute(), "work")
757 require.NoError(t, err)
758 require.Equal(t, protocol.KillOutcomeUnknown, result.Outcome)
759 require.Equal(t, uint64(1), result.RequestID)
760 require.Len(t, service.reconcileHints(), 1)
761 require.Len(t, service.openedRequests(), 1, "a dispatched kill is never replayed")
762}
763
764// TestBrokerOperationsDispatchBoundsAreOutcomeUnknown proves a lapsed bound and
765// a cancellation after dispatch both become a synthetic OutcomeUnknown rather
766// than a not-sent error or a success.
767func TestBrokerOperationsDispatchBoundsAreOutcomeUnknown(t *testing.T) {
768 tests := []struct {
769 name string
770 trigger func(t *testing.T, stream *brokerOpsTestStream, clock *supervisorTestClock, cancel context.CancelFunc)
771 }{
772 {
773 name: "deadline after dispatch",
774 trigger: func(t *testing.T, stream *brokerOpsTestStream, clock *supervisorTestClock, _ context.CancelFunc) {
775 timer := clock.awaitTimer(t)
776 stream.awaitDispatch(t)
777 timer.fire()
778 },
779 },
780 {
781 name: "cancellation after dispatch",
782 trigger: func(t *testing.T, stream *brokerOpsTestStream, _ *supervisorTestClock, cancel context.CancelFunc) {
783 stream.awaitDispatch(t)
784 cancel()
785 },
786 },
787 }
788 for _, tt := range tests {
789 t.Run(tt.name, func(t *testing.T) {
790 stream := newBrokerOpsTestStream()
791 ops, service, clock := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
792 ctx, cancel := context.WithCancel(context.Background())
793 defer cancel()
794
795 type outcome struct {
796 result protocol.KillResult
797 err error
798 }
799 done := make(chan outcome, 1)
800 go func() {
801 result, err := ops.Kill(ctx, brokerOpsTestLocalRoute(), "work")
802 done <- outcome{result: result, err: err}
803 }()
804 tt.trigger(t, stream, clock, cancel)
805
806 select {
807 case got := <-done:
808 require.NoError(t, got.err)
809 require.Equal(t, protocol.KillOutcomeUnknown, got.result.Outcome)
810 require.Equal(t, uint64(1), got.result.RequestID)
811 case <-time.After(5 * time.Second):
812 t.Fatal("control operation did not settle")
813 }
814 require.Len(t, service.reconcileHints(), 1)
815 require.Len(t, service.openedRequests(), 1, "the operation is never replayed")
816 require.GreaterOrEqual(t, stream.closeCount(), 1, "the owned stream is closed to unblock the read")
817 })
818 }
819}
820
821// TestBrokerOperationsLateReplyCannotChangeVerdict proves a reply arriving after
822// the bound fired is dropped and never replaces the indeterminate verdict, and
823// that the timer was disarmed before return.
824func TestBrokerOperationsLateReplyCannotChangeVerdict(t *testing.T) {
825 stream := newBrokerOpsTestStream()
826 ops, _, clock := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
827
828 type outcome struct {
829 result protocol.KillResult
830 err error
831 }
832 done := make(chan outcome, 1)
833 go func() {
834 result, err := ops.Kill(context.Background(), brokerOpsTestLocalRoute(), "work")
835 done <- outcome{result: result, err: err}
836 }()
837 timer := clock.awaitTimer(t)
838 stream.awaitDispatch(t)
839 timer.fire()
840
841 var got outcome
842 select {
843 case got = <-done:
844 case <-time.After(5 * time.Second):
845 t.Fatal("control operation did not settle")
846 }
847 require.NoError(t, got.err)
848 require.Equal(t, protocol.KillOutcomeUnknown, got.result.Outcome)
849 require.True(t, timer.stopped(), "the bound timer is disarmed before return")
850
851 // A late success can never replace the settled verdict.
852 stream.deliver(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded})
853 require.Equal(t, protocol.KillOutcomeUnknown, got.result.Outcome)
854}
855
856// TestBrokerOperationsRemoteRouteUsesExactRegistration proves a remote
857// destination resolves policy from the exact snapshot observation and requests
858// reconcile against its endpoint.
859func TestBrokerOperationsRemoteRouteUsesExactRegistration(t *testing.T) {
860 stream := newBrokerOpsTestStream()
861 stream.deliver(protocol.KillResult{RequestID: 1, Outcome: protocol.KillOutcomeUnknown, Text: "daemon shutting down"})
862 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestRemoteSnapshot(), stream)
863
864 result, err := ops.StopDaemon(context.Background(), brokerOpsTestRemoteRoute())
865 require.NoError(t, err)
866 require.Equal(t, protocol.KillOutcomeUnknown, result.Outcome)
867
868 requests := service.openedRequests()
869 require.Len(t, requests, 1)
870 require.False(t, requests[0].Local)
871 require.Equal(t, "user@example.test", requests[0].Endpoint)
872 require.Equal(t, pickerTestRegistration("user@example.test", 1), requests[0].Registration)
873 require.Equal(t, pickerTestPolicy(), requests[0].Policy)
874
875 // A daemon-reported Unknown requests a reconcile of the exact endpoint.
876 require.Equal(t, []string{"user@example.test"}, service.reconcileHints())
877}
878
879// TestBrokerOperationsReplacedRegistrationIsNotSent proves a remote destination
880// whose registration was replaced is refused before dispatch rather than
881// resolved by endpoint alone.
882func TestBrokerOperationsReplacedRegistrationIsNotSent(t *testing.T) {
883 stream := newBrokerOpsTestStream()
884 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestRemoteSnapshot(), stream)
885
886 replaced := BrokerOperationRoute{
887 Epoch: brokerOpsTestEpoch,
888 Destination: ports.BrokerEndpointFence{Registration: pickerTestRegistration("user@example.test", 9)},
889 }
890 _, err := ops.List(context.Background(), replaced)
891 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
892 require.Empty(t, service.openedRequests())
893}
894
895// TestBrokerOperationsCommandCopiesArgsAtAdmission proves a dispatched command
896// cannot be mutated by a caller after admission and carries the normalized
897// shape.
898func TestBrokerOperationsCommandCopiesArgsAtAdmission(t *testing.T) {
899 stream := newBrokerOpsTestStream()
900 stream.deliver(protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandSucceeded})
901 ops, _, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
902
903 args := []string{"original"}
904 _, err := ops.Command(context.Background(), brokerOpsTestLocalRoute(), protocol.CommandRequest{Version: protocol.Version, Slug: "next-tab", Args: args})
905 require.NoError(t, err)
906 args[0] = "mutated"
907
908 sent := stream.messages()
909 require.Len(t, sent, 1)
910 command, ok := sent[0].(protocol.CommandRequest)
911 require.True(t, ok)
912 require.Equal(t, []string{"original"}, command.Args)
913 require.False(t, command.Attached)
914 require.False(t, command.Self)
915 require.Equal(t, uint64(1), command.RequestID)
916}
917
918// TestBrokerOperationsConcurrentStreamsDoNotCollide proves concurrent
919// operations each allocate a distinct stream identity, open exactly one stream,
920// and receive their own correlated reply.
921func TestBrokerOperationsConcurrentStreamsDoNotCollide(t *testing.T) {
922 const operations = 16
923
924 service := newBrokerOpsTestService(ports.BrokerConnectionID{7}, brokerOpsTestLocalSnapshot())
925 openGate := make(chan struct{})
926 service.open = func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
927 <-openGate
928 stream := newBrokerOpsTestStream()
929 stream.deliver(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded})
930 return stream, nil
931 }
932 ops, err := NewBrokerOperations(service, newSupervisorTestClock())
933 require.NoError(t, err)
934
935 var wg sync.WaitGroup
936 errs := make(chan error, operations)
937 for range operations {
938 wg.Add(1)
939 go func() {
940 defer wg.Done()
941 result, err := ops.Kill(context.Background(), brokerOpsTestLocalRoute(), "work")
942 if err != nil {
943 errs <- err
944 return
945 }
946 if result.Outcome != protocol.KillSucceeded || result.RequestID != 1 {
947 errs <- errors.New("uncorrelated concurrent kill result")
948 }
949 }()
950 }
951 close(openGate)
952 wg.Wait()
953 close(errs)
954 for err := range errs {
955 require.NoError(t, err)
956 }
957
958 requests := service.openedRequests()
959 require.Len(t, requests, operations)
960 seen := make(map[ports.BrokerStreamID]struct{}, operations)
961 for _, request := range requests {
962 require.NotZero(t, request.Stream)
963 _, dup := seen[request.Stream]
964 require.False(t, dup, "two concurrent operations named the same stream")
965 seen[request.Stream] = struct{}{}
966 }
967}
968
969// TestBrokerOperationsClosingOneStreamPreservesSibling proves cancelling one
970// operation closes only its own stream and never disturbs a concurrent
971// sibling's reply.
972func TestBrokerOperationsClosingOneStreamPreservesSibling(t *testing.T) {
973 cancelled := newBrokerOpsTestStream()
974 sibling := newBrokerOpsTestStream()
975 sibling.deliver(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded})
976
977 service := newBrokerOpsTestService(ports.BrokerConnectionID{7}, brokerOpsTestLocalSnapshot())
978 service.open = func(_ context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
979 if request.Stream == 1 {
980 return cancelled, nil
981 }
982 return sibling, nil
983 }
984 ops, err := NewBrokerOperations(service, newSupervisorTestClock())
985 require.NoError(t, err)
986
987 ctx, cancel := context.WithCancel(context.Background())
988 defer cancel()
989
990 cancelledDone := make(chan protocol.KillResult, 1)
991 go func() {
992 result, _ := ops.Kill(ctx, brokerOpsTestLocalRoute(), "work")
993 cancelledDone <- result
994 }()
995 cancelled.awaitDispatch(t)
996
997 siblingResult, err := ops.Kill(context.Background(), brokerOpsTestLocalRoute(), "other")
998 require.NoError(t, err)
999 require.Equal(t, protocol.KillSucceeded, siblingResult.Outcome)
1000
1001 cancel()
1002 select {
1003 case result := <-cancelledDone:
1004 require.Equal(t, protocol.KillOutcomeUnknown, result.Outcome)
1005 case <-time.After(5 * time.Second):
1006 t.Fatal("cancelled operation did not settle")
1007 }
1008 require.GreaterOrEqual(t, cancelled.closeCount(), 1)
1009 require.Equal(t, 1, sibling.closeCount(), "a sibling stream is closed only by its own call")
1010}
1011
1012// TestBrokerOperationsBorrowedServiceStaysOpen proves the executor never closes
1013// the service it borrows, across success and refusal.
1014func TestBrokerOperationsBorrowedServiceStaysOpen(t *testing.T) {
1015 stream := newBrokerOpsTestStream()
1016 stream.deliver(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded})
1017 ops, service, _ := brokerOpsTestOperation(t, brokerOpsTestLocalSnapshot(), stream)
1018
1019 _, err := ops.Kill(context.Background(), brokerOpsTestLocalRoute(), "work")
1020 require.NoError(t, err)
1021 _, err = ops.Kill(context.Background(), brokerOpsTestLocalRoute(), "bad name")
1022 require.Error(t, err)
1023
1024 require.Equal(t, 0, service.closeCount(), "the executor never closes the borrowed service")
1025}
1026
1027// TestBrokerOperationsConcurrentSafeAccess exercises mixed operations
1028// concurrently so the race detector proves the executor holds no shared mutable
1029// state.
1030func TestBrokerOperationsConcurrentSafeAccess(t *testing.T) {
1031 service := newBrokerOpsTestService(ports.BrokerConnectionID{7}, brokerOpsTestLocalSnapshot())
1032 service.open = func(_ context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1033 stream := newBrokerOpsTestStream()
1034 stream.deliver(protocol.Sessions{})
1035 stream.deliver(protocol.KillResult{RequestID: 1, Outcome: protocol.KillSucceeded})
1036 stream.deliver(protocol.CommandResult{RequestID: 1, Outcome: protocol.CommandSucceeded})
1037 return stream, nil
1038 }
1039 ops, err := NewBrokerOperations(service, newSupervisorTestClock())
1040 require.NoError(t, err)
1041
1042 var wg sync.WaitGroup
1043 for range 8 {
1044 wg.Add(3)
1045 go func() {
1046 defer wg.Done()
1047 _, _ = ops.List(context.Background(), brokerOpsTestLocalRoute())
1048 }()
1049 go func() {
1050 defer wg.Done()
1051 _, _ = ops.Kill(context.Background(), brokerOpsTestLocalRoute(), "work")
1052 }()
1053 go func() {
1054 defer wg.Done()
1055 _, _ = ops.Command(context.Background(), brokerOpsTestLocalRoute(), protocol.CommandRequest{Version: protocol.Version, Slug: "next-tab"})
1056 }()
1057 }
1058 wg.Wait()
1059}
1060
1061// TestBrokerOperationsAcquisitionBoundIsNotSent proves the single 10s bound
1062// also covers stream acquisition: an open that never completes because the
1063// bounded context lapses is refused as a definite not-sent outcome and
1064// dispatches nothing.
1065func TestBrokerOperationsAcquisitionBoundIsNotSent(t *testing.T) {
1066 require.Equal(t, 10*time.Second, brokerOperationTimeout)
1067
1068 service := newBrokerOpsTestService(ports.BrokerConnectionID{7}, brokerOpsTestLocalSnapshot())
1069 // A real adapter honors the bounded context; this one blocks until the bound
1070 // lapses, exactly as a socket that accepts but never admits would.
1071 service.open = func(ctx context.Context, _ ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
1072 <-ctx.Done()
1073 return nil, ctx.Err()
1074 }
1075 clock := newSupervisorTestClock()
1076 ops, err := NewBrokerOperations(service, clock)
1077 require.NoError(t, err)
1078
1079 done := make(chan error, 1)
1080 go func() {
1081 _, err := ops.Kill(context.Background(), brokerOpsTestLocalRoute(), "work")
1082 done <- err
1083 }()
1084 timer := clock.awaitTimer(t)
1085 require.Eventually(t, func() bool { return len(service.openedRequests()) == 1 }, time.Second, time.Millisecond)
1086 timer.fire()
1087
1088 select {
1089 case err := <-done:
1090 require.ErrorIs(t, err, ErrBrokerOperationNotSent)
1091 case <-time.After(5 * time.Second):
1092 t.Fatal("control operation did not settle")
1093 }
1094 require.Empty(t, service.reconcileHints(), "a not-sent outcome never requests reconcile")
1095}
1096
1097// TestBrokerOperationAuthorityExactSelection pins the helper directly,
1098// including that a local destination never resolves a remote observation and a
1099// remote destination never resolves the local one.
1100func TestBrokerOperationAuthorityExactSelection(t *testing.T) {
1101 local := brokerOpsTestLocalSnapshot()
1102 remote := brokerOpsTestRemoteSnapshot()
1103
1104 if _, ok := brokerOperationAuthority(local, ports.BrokerEndpointFence{Local: true}); !ok {
1105 t.Fatal("a local destination must resolve the local observation")
1106 }
1107 if _, ok := brokerOperationAuthority(remote, ports.BrokerEndpointFence{Local: true}); ok {
1108 t.Fatal("a local destination must not resolve against a remote-only snapshot")
1109 }
1110 if _, ok := brokerOperationAuthority(local, ports.BrokerEndpointFence{Registration: pickerTestRegistration("user@example.test", 1)}); ok {
1111 t.Fatal("a remote destination must not resolve against a local-only snapshot")
1112 }
1113 if _, ok := brokerOperationAuthority(remote, ports.BrokerEndpointFence{Registration: pickerTestRegistration("user@example.test", 1)}); !ok {
1114 t.Fatal("a remote destination must resolve its exact registration")
1115 }
1116}