1package client
2
3import (
4 "context"
5 "errors"
6 "io"
7 "math"
8 "runtime"
9 "strings"
10 "sync"
11 "testing"
12 "time"
13
14 "github.com/stretchr/testify/require"
15
16 "github.com/bnema/vev/internal/domain"
17 "github.com/bnema/vev/internal/domain/terminalcap"
18 "github.com/bnema/vev/internal/ports"
19)
20
21// supervisorTestClock is a manually fired ports.Clock. NewTimer publishes each
22// timer on a buffered channel so a test can fire it deterministically without a
23// sleep, and Stop is recorded so a test can prove every retry timer was
24// retired.
25type supervisorTestClock struct {
26 mu sync.Mutex
27 now time.Time
28 timers []*supervisorTestTimer
29 created chan *supervisorTestTimer
30}
31
32func newSupervisorTestClock() *supervisorTestClock {
33 return &supervisorTestClock{now: time.Unix(1000, 0), created: make(chan *supervisorTestTimer, 32)}
34}
35
36func (c *supervisorTestClock) Now() time.Time {
37 c.mu.Lock()
38 defer c.mu.Unlock()
39 return c.now
40}
41
42func (c *supervisorTestClock) NewTimer(delay time.Duration) ports.Timer {
43 timer := &supervisorTestTimer{ch: make(chan time.Time, 1), delay: delay}
44 c.mu.Lock()
45 c.timers = append(c.timers, timer)
46 c.mu.Unlock()
47 select {
48 case c.created <- timer:
49 default:
50 }
51 return timer
52}
53
54func (c *supervisorTestClock) awaitTimer(t *testing.T) *supervisorTestTimer {
55 t.Helper()
56 select {
57 case timer := <-c.created:
58 return timer
59 case <-time.After(5 * time.Second):
60 t.Fatal("supervisor did not schedule a retry timer")
61 return nil
62 }
63}
64
65func (c *supervisorTestClock) timerCount() int {
66 c.mu.Lock()
67 defer c.mu.Unlock()
68 return len(c.timers)
69}
70
71func (c *supervisorTestClock) liveTimers() int {
72 c.mu.Lock()
73 defer c.mu.Unlock()
74 live := 0
75 for _, timer := range c.timers {
76 if !timer.stopped() {
77 live++
78 }
79 }
80 return live
81}
82
83func (c *supervisorTestClock) allStopped() bool {
84 c.mu.Lock()
85 defer c.mu.Unlock()
86 for _, timer := range c.timers {
87 if !timer.stopped() {
88 return false
89 }
90 }
91 return true
92}
93
94type supervisorTestTimer struct {
95 ch chan time.Time
96 delay time.Duration
97 mu sync.Mutex
98 stoppedFlag bool
99}
100
101func (t *supervisorTestTimer) C() <-chan time.Time { return t.ch }
102
103func (t *supervisorTestTimer) Reset(time.Duration) bool { return false }
104
105func (t *supervisorTestTimer) Stop() bool {
106 t.mu.Lock()
107 defer t.mu.Unlock()
108 if t.stoppedFlag {
109 return false
110 }
111 t.stoppedFlag = true
112 return true
113}
114
115func (t *supervisorTestTimer) stopped() bool {
116 t.mu.Lock()
117 defer t.mu.Unlock()
118 return t.stoppedFlag
119}
120
121// fire delivers the timer without blocking, so a test may fire before or after
122// the supervisor reaches its select.
123func (t *supervisorTestTimer) fire() {
124 select {
125 case t.ch <- time.Now():
126 default:
127 }
128}
129
130// supervisorTestTerminal records raw-mode enters and restores. Its reader is
131// shared so tests can model input EOF and a live terminal.
132type supervisorTestTerminal struct {
133 in io.Reader
134 enterErr error
135
136 mu sync.Mutex
137 restores int
138}
139
140func newSupervisorTestTerminal(in io.Reader) *supervisorTestTerminal {
141 return &supervisorTestTerminal{in: in}
142}
143
144func (t *supervisorTestTerminal) EnableKittyKeyboard(int) error { return nil }
145
146func (t *supervisorTestTerminal) EnterRaw() (func() error, error) {
147 if t.enterErr != nil {
148 return nil, t.enterErr
149 }
150 return func() error {
151 t.mu.Lock()
152 t.restores++
153 t.mu.Unlock()
154 return nil
155 }, nil
156}
157
158func (t *supervisorTestTerminal) Geometry() (domain.Geometry, error) {
159 return domain.Geometry{}, nil
160}
161
162func (t *supervisorTestTerminal) ResizeEvents() <-chan domain.Geometry { return nil }
163
164func (t *supervisorTestTerminal) In() io.Reader { return t.in }
165
166func (t *supervisorTestTerminal) Out() io.Writer { return io.Discard }
167
168func (t *supervisorTestTerminal) Flush() error { return nil }
169
170func (t *supervisorTestTerminal) restoreCount() int {
171 t.mu.Lock()
172 defer t.mu.Unlock()
173 return t.restores
174}
175
176// supervisorTestConnector scripts connection attempts and records concurrency,
177// so a test can prove attempts never overlap.
178type supervisorTestConnector struct {
179 connect func(ctx context.Context, call int) (ports.BrokerNavigator, error)
180
181 mu sync.Mutex
182 calls int
183 inFlight int
184 maxInFlight int
185 started chan int
186}
187
188func newSupervisorTestConnector(connect func(ctx context.Context, call int) (ports.BrokerNavigator, error)) *supervisorTestConnector {
189 return &supervisorTestConnector{connect: connect, started: make(chan int, 32)}
190}
191
192func (c *supervisorTestConnector) Connect(ctx context.Context) (ports.BrokerNavigator, error) {
193 c.mu.Lock()
194 c.calls++
195 call := c.calls
196 c.inFlight++
197 if c.inFlight > c.maxInFlight {
198 c.maxInFlight = c.inFlight
199 }
200 c.mu.Unlock()
201 select {
202 case c.started <- call:
203 default:
204 }
205 defer func() {
206 c.mu.Lock()
207 c.inFlight--
208 c.mu.Unlock()
209 }()
210 if c.connect == nil {
211 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "no script"}
212 }
213 return c.connect(ctx, call)
214}
215
216func (c *supervisorTestConnector) callCount() int {
217 c.mu.Lock()
218 defer c.mu.Unlock()
219 return c.calls
220}
221
222func (c *supervisorTestConnector) maxConcurrent() int {
223 c.mu.Lock()
224 defer c.mu.Unlock()
225 return c.maxInFlight
226}
227
228func (c *supervisorTestConnector) inFlightCount() int {
229 c.mu.Lock()
230 defer c.mu.Unlock()
231 return c.inFlight
232}
233
234func (c *supervisorTestConnector) awaitStart(t *testing.T) int {
235 t.Helper()
236 select {
237 case call := <-c.started:
238 return call
239 case <-time.After(5 * time.Second):
240 t.Fatal("supervisor did not start a connection attempt")
241 return 0
242 }
243}
244
245// supervisorTestHub is one broker publication source. It conforms to
246// ports.BrokerSubscription: each subscriber owns its own capacity-one channel,
247// a publication offers a non-blocking wake to every live subscriber and never
248// replaces or closes a channel, and a subscriber re-reads the newest snapshot
249// on wake. A slow or parked subscriber therefore can never block the
250// publisher or busy-spin.
251type supervisorTestHub struct {
252 mu sync.Mutex
253 snapshot ports.BrokerSnapshot
254 subs map[chan struct{}]struct{}
255}
256
257func newSupervisorTestHub() *supervisorTestHub {
258 return &supervisorTestHub{subs: make(map[chan struct{}]struct{})}
259}
260
261func (h *supervisorTestHub) publish(snapshot ports.BrokerSnapshot) {
262 h.mu.Lock()
263 defer h.mu.Unlock()
264 h.snapshot = snapshot
265 for changed := range h.subs {
266 select {
267 case changed <- struct{}{}:
268 default:
269 }
270 }
271}
272
273func (h *supervisorTestHub) current() ports.BrokerSnapshot {
274 h.mu.Lock()
275 defer h.mu.Unlock()
276 return h.snapshot
277}
278
279func (h *supervisorTestHub) subscribe(closeFn func()) ports.BrokerSubscription {
280 changed := make(chan struct{}, 1)
281 h.mu.Lock()
282 h.subs[changed] = struct{}{}
283 h.mu.Unlock()
284 return supervisorTestSubscription{changed: changed, closeFn: func() {
285 h.mu.Lock()
286 delete(h.subs, changed)
287 h.mu.Unlock()
288 if closeFn != nil {
289 closeFn()
290 }
291 }}
292}
293
294type supervisorTestSubscription struct {
295 changed chan struct{}
296 closeFn func()
297}
298
299func (s supervisorTestSubscription) Changed() <-chan struct{} { return s.changed }
300
301func (s supervisorTestSubscription) Close() {
302 if s.closeFn != nil {
303 s.closeFn()
304 }
305}
306
307// supervisorTestService is one scripted BrokerService. Close is observable so a
308// test can prove a retired or late service was closed, and Close can be held open
309// so a test can order a parent cancellation strictly before a settle decision that
310// runs after the connection is released.
311type supervisorTestService struct {
312 id ports.BrokerConnectionID
313 hub *supervisorTestHub
314
315 mu sync.Mutex
316 err error
317 doneOnce sync.Once
318 done chan struct{}
319
320 closeOnce sync.Once
321 closed chan struct{}
322 subClosed chan struct{}
323
324 // closeGate, when non-nil, holds Close until releaseClose is called, and
325 // closeEntered reports that Close began waiting on it.
326 closeGate chan struct{}
327 closeEntered chan struct{}
328
329 // openStream scripts logical-stream admission for the P5.3b attachment
330 // path. When nil, OpenStream keeps the historical refusal. openCalls
331 // records every exact request the supervisor admitted.
332 openStream func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error)
333 openCalls []ports.BrokerOpenStreamRequest
334 nextStream ports.BrokerStreamID
335 reconciles []string
336}
337
338func newSupervisorTestService(id ports.BrokerConnectionID) *supervisorTestService {
339 return &supervisorTestService{
340 id: id,
341 hub: newSupervisorTestHub(),
342 done: make(chan struct{}),
343 closed: make(chan struct{}),
344 subClosed: make(chan struct{}, 1),
345 }
346}
347
348func (s *supervisorTestService) ConnectionID() ports.BrokerConnectionID { return s.id }
349
350// NextStreamID is the connection's only stream allocator, exactly like the
351// production service: the tested supervisor reads every identity here.
352func (s *supervisorTestService) NextStreamID() (ports.BrokerStreamID, error) {
353 s.mu.Lock()
354 defer s.mu.Unlock()
355 s.nextStream++
356 return s.nextStream, nil
357}
358
359func (s *supervisorTestService) Done() <-chan struct{} { return s.done }
360
361func (s *supervisorTestService) Err() error {
362 s.mu.Lock()
363 defer s.mu.Unlock()
364 return s.err
365}
366
367func (s *supervisorTestService) Snapshot() ports.BrokerSnapshot { return s.hub.current() }
368
369func (s *supervisorTestService) Subscribe() (ports.BrokerSubscription, error) {
370 return s.hub.subscribe(func() {
371 select {
372 case s.subClosed <- struct{}{}:
373 default:
374 }
375 }), nil
376}
377
378func (s *supervisorTestService) SubscribePreview(ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
379 return nil, errors.New("supervisor test service does not support preview")
380}
381
382func (s *supervisorTestService) OpenStream(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
383 s.mu.Lock()
384 s.openCalls = append(s.openCalls, request)
385 open := s.openStream
386 s.mu.Unlock()
387 if open == nil {
388 return nil, errors.New("supervisor test service does not support streams")
389 }
390 return open(ctx, request)
391}
392
393// setOpenStream installs the logical-stream admission all P5.3b tests drive.
394func (s *supervisorTestService) setOpenStream(open func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error)) {
395 s.mu.Lock()
396 defer s.mu.Unlock()
397 s.openStream = open
398}
399
400// openedRequests returns a copy of every exact stream request admitted so far.
401func (s *supervisorTestService) openedRequests() []ports.BrokerOpenStreamRequest {
402 s.mu.Lock()
403 defer s.mu.Unlock()
404 return append([]ports.BrokerOpenStreamRequest(nil), s.openCalls...)
405}
406
407func (s *supervisorTestService) RequestReconcile(endpoint string) {
408 s.mu.Lock()
409 s.reconciles = append(s.reconciles, endpoint)
410 s.mu.Unlock()
411}
412
413// reconcileHints returns every re-observation hint the supervisor asked for.
414func (s *supervisorTestService) reconcileHints() []string {
415 s.mu.Lock()
416 defer s.mu.Unlock()
417 return append([]string(nil), s.reconciles...)
418}
419
420func (s *supervisorTestService) Close() error {
421 s.closeOnce.Do(func() {
422 if s.closeGate != nil {
423 select {
424 case s.closeEntered <- struct{}{}:
425 default:
426 }
427 <-s.closeGate
428 }
429 close(s.closed)
430 s.terminate(nil)
431 })
432 return nil
433}
434
435// holdClose makes Close block until releaseClose is called, so a test can order a
436// parent cancellation strictly before a settle decision that runs after the
437// connection is released. closingCh reports that the release was entered.
438func (s *supervisorTestService) holdClose() {
439 s.closeGate = make(chan struct{})
440 s.closeEntered = make(chan struct{}, 1)
441}
442
443// releaseClose releases a held Close.
444func (s *supervisorTestService) releaseClose() {
445 if s.closeGate != nil {
446 close(s.closeGate)
447 }
448}
449
450// closingCh reports that the held release was entered.
451func (s *supervisorTestService) closingCh() <-chan struct{} { return s.closeEntered }
452
453// terminate removes the connection, recording the first cause exactly once.
454func (s *supervisorTestService) terminate(err error) {
455 if err != nil {
456 s.mu.Lock()
457 if s.err == nil {
458 s.err = err
459 }
460 s.mu.Unlock()
461 }
462 s.doneOnce.Do(func() { close(s.done) })
463}
464
465// lose models an unexplained broker-side loss.
466func (s *supervisorTestService) lose(err error) { s.terminate(err) }
467
468func (s *supervisorTestService) publish(epoch ports.BrokerEpoch, revision ports.BrokerRevision) {
469 s.hub.publish(ports.BrokerSnapshot{Epoch: epoch, Revision: revision})
470}
471
472// publishSnapshot publishes one complete broker snapshot, as a real service
473// does, so a test can deliver a second-and-later catalogue publication.
474func (s *supervisorTestService) publishSnapshot(snapshot ports.BrokerSnapshot) {
475 s.hub.publish(snapshot)
476}
477
478func (s *supervisorTestService) closedCh() <-chan struct{} { return s.closed }
479
480func mustSupervisor(t *testing.T, cfg SupervisorConfig) *Supervisor {
481 t.Helper()
482 if cfg.SessionEnvironment.Provenance == SessionEnvironmentUnspecified {
483 cfg.SessionEnvironment = SessionEnvironment{Provenance: SessionEnvironmentLocalCLI}
484 }
485 sup, err := NewSupervisor(cfg)
486 require.NoError(t, err)
487 return sup
488}
489
490// supervisorTestReader blocks until unblocked, so a test controls exactly when
491// the owned terminal input reaches EOF.
492type supervisorTestReader struct {
493 mu sync.Mutex
494 done chan struct{}
495}
496
497func newSupervisorTestReader() *supervisorTestReader {
498 return &supervisorTestReader{done: make(chan struct{})}
499}
500
501func (r *supervisorTestReader) Read([]byte) (int, error) {
502 <-r.done
503 return 0, io.EOF
504}
505
506func (r *supervisorTestReader) unblock() {
507 r.mu.Lock()
508 defer r.mu.Unlock()
509 select {
510 case <-r.done:
511 default:
512 close(r.done)
513 }
514}
515
516func TestSupervisorReducerTransitions(t *testing.T) {
517 transient := errors.New("unavailable")
518 incompatible := ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "version"}
519 exited := ports.BrokerError{Code: ports.BrokerErrorExplicitExit, Text: "exit"}
520 picker := State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected}
521
522 tests := []struct {
523 name string
524 state State
525 event supervisorEvent
526 want State
527 }{
528 {
529 name: "begin attempt claims a generation and connects",
530 state: picker,
531 event: supervisorEvent{kind: supervisorBeginAttempt},
532 want: State{Presentation: PresentPicker, Connectivity: ConnectivityConnectingBroker, Attempt: 0, Generation: 1},
533 },
534 {
535 name: "ready resets the retry cadence",
536 state: State{Presentation: PresentPicker, Connectivity: ConnectivityConnectingBroker, Attempt: 3, Generation: 4, Err: transient},
537 event: supervisorEvent{kind: supervisorReady},
538 want: State{Presentation: PresentPicker, Connectivity: ConnectivityReady, Attempt: 0, Generation: 4},
539 },
540 {
541 name: "transient failure waits with a bumped attempt",
542 state: State{Presentation: PresentPicker, Connectivity: ConnectivityConnectingBroker, Attempt: 0, Generation: 2},
543 event: supervisorEvent{kind: supervisorTransientFailure, err: transient},
544 want: State{Presentation: PresentPicker, Connectivity: ConnectivityRetryWait, Attempt: 1, Generation: 2, Err: transient},
545 },
546 {
547 name: "broker loss returns to picker retry wait",
548 state: State{Presentation: PresentPicker, Connectivity: ConnectivityReady, Attempt: 0, Generation: 5},
549 event: supervisorEvent{kind: supervisorBrokerLoss, err: transient},
550 want: State{Presentation: PresentPicker, Connectivity: ConnectivityRetryWait, Attempt: 1, Generation: 5, Err: transient, ReadyLost: true},
551 },
552 {
553 name: "non-retryable failure disconnects and stays on the picker",
554 state: State{Presentation: PresentPicker, Connectivity: ConnectivityConnectingBroker, Attempt: 1, Generation: 3},
555 event: supervisorEvent{kind: supervisorNonRetryable, err: incompatible},
556 want: State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected, Attempt: 1, Generation: 3, Err: incompatible},
557 },
558 {
559 name: "begin attempt with a pending navigation presents connecting, not the picker",
560 state: picker,
561 event: supervisorEvent{kind: supervisorBeginAttempt, navigating: true},
562 want: State{Presentation: PresentConnecting, Connectivity: ConnectivityConnectingBroker, Generation: 1},
563 },
564 {
565 name: "retry wait keeps a pending navigation connecting",
566 state: State{Presentation: PresentConnecting, Connectivity: ConnectivityConnectingBroker, Generation: 2},
567 event: supervisorEvent{kind: supervisorTransientFailure, err: transient},
568 want: State{Presentation: PresentConnecting, Connectivity: ConnectivityRetryWait, Attempt: 1, Generation: 2, Err: transient},
569 },
570 {
571 name: "non-retryable failure during a pending navigation falls back to the picker",
572 state: State{Presentation: PresentConnecting, Connectivity: ConnectivityConnectingBroker, Generation: 3},
573 event: supervisorEvent{kind: supervisorNonRetryable, err: incompatible},
574 want: State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected, Generation: 3, Err: incompatible},
575 },
576 {
577 name: "a navigation that settles without attaching returns to the picker",
578 state: State{Presentation: PresentConnecting, Connectivity: ConnectivityReady, Generation: 4},
579 event: supervisorEvent{kind: supervisorNavigationSettled},
580 want: State{Presentation: PresentPicker, Connectivity: ConnectivityReady, Generation: 4},
581 },
582 {
583 name: "navigation settled leaves an unrelated presentation alone",
584 state: State{Presentation: PresentAttached, Connectivity: ConnectivityReady, Generation: 4},
585 event: supervisorEvent{kind: supervisorNavigationSettled},
586 want: State{Presentation: PresentAttached, Connectivity: ConnectivityReady, Generation: 4},
587 },
588 {
589 name: "terminal failure terminates",
590 state: State{Presentation: PresentPicker, Connectivity: ConnectivityReady, Generation: 7},
591 event: supervisorEvent{kind: supervisorTerminal, err: exited},
592 want: State{Presentation: PresentTerminating, Connectivity: ConnectivityDisconnected, Generation: 7, Err: exited},
593 },
594 }
595
596 for _, tt := range tests {
597 t.Run(tt.name, func(t *testing.T) {
598 require.Equal(t, tt.want, reduceSupervisor(tt.state, tt.event))
599 })
600 }
601}
602
603func TestSupervisorBackoffCadence(t *testing.T) {
604 zero := func() float64 { return 0 }
605 half := func() float64 { return 0.5 }
606 clamped := func() float64 { return 1.5 }
607 nan := func() float64 { return math.NaN() }
608 negInf := func() float64 { return math.Inf(-1) }
609 posInf := func() float64 { return math.Inf(1) }
610
611 tests := []struct {
612 name string
613 attempt uint64
614 jitter func() float64
615 want time.Duration
616 }{
617 {"no attempt has no delay", 0, zero, 0},
618 {"first failure halves 100ms", 1, zero, 50 * time.Millisecond},
619 {"second failure halves 200ms", 2, zero, 100 * time.Millisecond},
620 {"third failure halves 400ms", 3, zero, 200 * time.Millisecond},
621 {"fourth failure halves 800ms", 4, zero, 400 * time.Millisecond},
622 {"fifth failure halves 1.6s", 5, zero, 800 * time.Millisecond},
623 {"cap reached at 2s", 6, zero, time.Second},
624 {"cap stays at 2s", 9, zero, time.Second},
625 {"equal jitter reaches three quarters", 1, half, 75 * time.Millisecond},
626 {"equal jitter scales with the cap", 6, half, 1500 * time.Millisecond},
627 {"full jitter reaches the cap", 1, clamped, 100 * time.Millisecond},
628 {"full jitter stays capped", 6, clamped, 2 * time.Second},
629 {"NaN jitter clamps to the minimum half", 1, nan, 50 * time.Millisecond},
630 {"NaN jitter stays at half the cap", 6, nan, time.Second},
631 {"negative jitter clamps to the minimum half", 1, negInf, 50 * time.Millisecond},
632 {"negative jitter stays at half the cap", 6, negInf, time.Second},
633 {"positive infinity clamps to the cap", 1, posInf, 100 * time.Millisecond},
634 {"positive infinity stays capped", 6, posInf, 2 * time.Second},
635 }
636
637 for _, tt := range tests {
638 t.Run(tt.name, func(t *testing.T) {
639 require.Equal(t, tt.want, supervisorBackoffDelay(tt.attempt, tt.jitter))
640 })
641 }
642}
643
644func TestSupervisorPickerWithoutBroker(t *testing.T) {
645 ctx, cancel := context.WithCancel(context.Background())
646 defer cancel()
647
648 clock := newSupervisorTestClock()
649 reader := newSupervisorTestReader()
650 t.Cleanup(reader.unblock)
651 terminal := newSupervisorTestTerminal(reader)
652 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
653 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker down"}
654 })
655
656 var (
657 renderMu sync.Mutex
658 renders []State
659 )
660 sup := mustSupervisor(t, SupervisorConfig{
661 Connector: connector,
662 Terminal: terminal,
663 Clock: clock,
664 Jitter: func() float64 { return 0 },
665 Render: func(state State) {
666 renderMu.Lock()
667 renders = append(renders, state)
668 renderMu.Unlock()
669 },
670 })
671
672 runErr := make(chan error, 1)
673 go func() { runErr <- sup.Run(ctx) }()
674
675 first := clock.awaitTimer(t)
676 require.Equal(t, 50*time.Millisecond, first.delay, "the first retry is the equal-jitter half of 100ms")
677 require.Equal(t, 1, connector.awaitStart(t))
678 require.Equal(t, State{Presentation: PresentPicker, Connectivity: ConnectivityRetryWait, Attempt: 1, Generation: 1, Err: ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker down"}}, sup.State())
679
680 first.fire()
681 second := clock.awaitTimer(t)
682 require.Equal(t, 100*time.Millisecond, second.delay, "the second retry doubles the cap")
683 require.Equal(t, 2, connector.awaitStart(t))
684 require.Equal(t, PresentPicker, sup.State().Presentation, "the picker stays up while disconnected")
685 require.Equal(t, ConnectivityRetryWait, sup.State().Connectivity)
686 require.Equal(t, uint64(2), sup.State().Generation)
687
688 cancel()
689 require.ErrorIs(t, <-runErr, context.Canceled)
690 require.Equal(t, 1, terminal.restoreCount())
691 require.Equal(t, 1, connector.maxConcurrent())
692
693 renderMu.Lock()
694 defer renderMu.Unlock()
695 require.NotEmpty(t, renders)
696 require.Equal(t, PresentTerminating, renders[len(renders)-1].Presentation)
697 for _, state := range renders[:len(renders)-1] {
698 require.Equal(t, PresentPicker, state.Presentation, "P5.1a never leaves the picker")
699 }
700}
701
702func TestSupervisorPickerPublicationRequestsRepaint(t *testing.T) {
703 ctx, cancel := context.WithCancel(context.Background())
704 defer cancel()
705
706 clock := newSupervisorTestClock()
707 reader := newSupervisorTestReader()
708 t.Cleanup(reader.unblock)
709 service := newSupervisorTestService(ports.BrokerConnectionID{1})
710 service.publish(1, 1)
711 renders := make(chan State, 16)
712
713 sup := mustSupervisor(t, SupervisorConfig{
714 Connector: newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
715 return service, nil
716 }),
717 Terminal: newSupervisorTestTerminal(reader),
718 Clock: clock,
719 Picker: newPickerController(clock, pickerTestFreshness, terminalcap.ColorCapabilities{Mode: terminalcap.TrueColor}),
720 Render: func(state State) {
721 renders <- state
722 },
723 })
724
725 runErr := make(chan error, 1)
726 go func() { runErr <- sup.Run(ctx) }()
727 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
728 for {
729 select {
730 case <-renders:
731 default:
732 goto drained
733 }
734 }
735
736drained:
737 service.publish(1, 2)
738 select {
739 case state := <-renders:
740 require.Equal(t, PresentPicker, state.Presentation)
741 require.Equal(t, ConnectivityReady, state.Connectivity)
742 case <-time.After(5 * time.Second):
743 t.Fatal("picker publication did not request a repaint")
744 }
745
746 cancel()
747 require.ErrorIs(t, <-runErr, context.Canceled)
748}
749
750func TestSupervisorBrokerLossReconnectsWithFreshGeneration(t *testing.T) {
751 ctx, cancel := context.WithCancel(context.Background())
752 defer cancel()
753
754 clock := newSupervisorTestClock()
755 reader := newSupervisorTestReader()
756 t.Cleanup(reader.unblock)
757 terminal := newSupervisorTestTerminal(reader)
758
759 first := newSupervisorTestService(ports.BrokerConnectionID{1})
760 first.publish(1, 1)
761 second := newSupervisorTestService(ports.BrokerConnectionID{2})
762
763 connector := newSupervisorTestConnector(func(_ context.Context, call int) (ports.BrokerNavigator, error) {
764 switch call {
765 case 1:
766 return first, nil
767 default:
768 return second, nil
769 }
770 })
771
772 sup := mustSupervisor(t, SupervisorConfig{
773 Connector: connector,
774 Terminal: terminal,
775 Clock: clock,
776 Jitter: func() float64 { return 0 },
777 })
778
779 runErr := make(chan error, 1)
780 go func() { runErr <- sup.Run(ctx) }()
781
782 require.Equal(t, 1, connector.awaitStart(t))
783 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
784 require.Equal(t, uint64(1), sup.State().Generation)
785 require.Equal(t, uint64(0), sup.State().Attempt)
786
787 // The idle broker connection is lost: the supervisor retires it and waits
788 // the first cadence before reconnecting.
789 first.lose(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "idle loss"})
790 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityRetryWait }, 5*time.Second, time.Millisecond)
791 require.Equal(t, uint64(1), sup.State().Attempt)
792 select {
793 case <-first.closedCh():
794 default:
795 t.Fatal("the lost connection was not retired")
796 }
797 select {
798 case <-first.subClosed:
799 default:
800 t.Fatal("the lost subscription was not retired")
801 }
802
803 retry := clock.awaitTimer(t)
804 require.Equal(t, 50*time.Millisecond, retry.delay)
805 retry.fire()
806 require.Equal(t, 2, connector.awaitStart(t))
807
808 // Wait until the replacement is waiting for its first publication.
809 require.Eventually(t, func() bool {
810 return sup.State().Connectivity == ConnectivityConnectingBroker && sup.State().Generation == 2
811 }, 5*time.Second, time.Millisecond)
812
813 // A publication on the retired connection must never make the replacement
814 // Ready; only the replacement's own fresh epoch does.
815 first.publish(1, 2)
816 require.Never(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 50*time.Millisecond, 5*time.Millisecond)
817
818 second.publish(2, 1)
819 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
820 require.Equal(t, uint64(2), sup.State().Generation)
821 require.Equal(t, uint64(0), sup.State().Attempt, "readiness resets the cadence")
822
823 cancel()
824 require.ErrorIs(t, <-runErr, context.Canceled)
825 require.Equal(t, 1, terminal.restoreCount())
826 select {
827 case <-second.closedCh():
828 default:
829 t.Fatal("the active connection was not retired on shutdown")
830 }
831}
832
833func TestSupervisorCancellationClosesLateConnection(t *testing.T) {
834 ctx, cancel := context.WithCancel(context.Background())
835
836 clock := newSupervisorTestClock()
837 reader := newSupervisorTestReader()
838 t.Cleanup(reader.unblock)
839 terminal := newSupervisorTestTerminal(reader)
840
841 release := make(chan struct{})
842 late := newSupervisorTestService(ports.BrokerConnectionID{9})
843 late.publish(1, 1)
844 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
845 <-release
846 return late, nil
847 })
848
849 sup := mustSupervisor(t, SupervisorConfig{
850 Connector: connector,
851 Terminal: terminal,
852 Clock: clock,
853 Jitter: func() float64 { return 0 },
854 })
855
856 runErr := make(chan error, 1)
857 go func() { runErr <- sup.Run(ctx) }()
858 require.Equal(t, 1, connector.awaitStart(t))
859
860 // Cancel while the connector ignores cancellation, then let it return a
861 // live service: the supervisor must close the late superseded success.
862 cancel()
863 close(release)
864
865 require.ErrorIs(t, <-runErr, context.Canceled)
866 select {
867 case <-late.closedCh():
868 default:
869 t.Fatal("a late superseded success must be closed")
870 }
871 require.Equal(t, 1, terminal.restoreCount())
872 require.Zero(t, connector.inFlightCount())
873}
874
875func TestSupervisorIncompatibleStopsReconnecting(t *testing.T) {
876 ctx, cancel := context.WithCancel(context.Background())
877 defer cancel()
878
879 clock := newSupervisorTestClock()
880 reader := newSupervisorTestReader()
881 t.Cleanup(reader.unblock)
882 terminal := newSupervisorTestTerminal(reader)
883
884 incompatible := ports.BrokerError{Code: ports.BrokerErrorIncompatible, Text: "protocol"}
885 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
886 return nil, incompatible
887 })
888 notified := make(chan error, 4)
889 sup := mustSupervisor(t, SupervisorConfig{
890 Connector: connector,
891 Terminal: terminal,
892 Clock: clock,
893 Jitter: func() float64 { return 0 },
894 Notify: func(_ State, err error) { notified <- err },
895 })
896
897 runErr := make(chan error, 1)
898 go func() { runErr <- sup.Run(ctx) }()
899
900 select {
901 case err := <-notified:
902 var typed ports.BrokerError
903 require.ErrorAs(t, err, &typed)
904 require.Equal(t, ports.BrokerErrorIncompatible, typed.Code)
905 case <-time.After(5 * time.Second):
906 t.Fatal("the incompatible failure was not surfaced")
907 }
908
909 require.Equal(t, State{Presentation: PresentPicker, Connectivity: ConnectivityDisconnected, Generation: 1, Err: incompatible}, sup.State())
910 require.Never(t, func() bool { return connector.callCount() > 1 }, 50*time.Millisecond, 5*time.Millisecond)
911 require.Zero(t, clock.timerCount(), "a non-retryable failure schedules no retry timer")
912
913 cancel()
914 require.ErrorIs(t, <-runErr, context.Canceled)
915 require.Equal(t, 1, terminal.restoreCount())
916}
917
918func TestSupervisorTerminalEOFRestoresRawOnce(t *testing.T) {
919 clock := newSupervisorTestClock()
920 terminal := newSupervisorTestTerminal(strings.NewReader(""))
921 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
922 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "down"}
923 })
924
925 sup := mustSupervisor(t, SupervisorConfig{
926 Connector: connector,
927 Terminal: terminal,
928 Clock: clock,
929 Jitter: func() float64 { return 0 },
930 })
931
932 require.NoError(t, sup.Run(context.Background()), "a clean terminal EOF is an orderly exit")
933 require.Equal(t, 1, terminal.restoreCount())
934 require.Equal(t, PresentTerminating, sup.State().Presentation)
935}
936
937func TestSupervisorRawModePrecedesBroker(t *testing.T) {
938 clock := newSupervisorTestClock()
939 reader := newSupervisorTestReader()
940 t.Cleanup(reader.unblock)
941 terminal := newSupervisorTestTerminal(reader)
942
943 enteredAfterRaw := make(chan bool, 1)
944 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
945 enteredAfterRaw <- terminal.restoreCount() == 0
946 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "down"}
947 })
948
949 sup := mustSupervisor(t, SupervisorConfig{
950 Connector: connector,
951 Terminal: terminal,
952 Clock: clock,
953 Jitter: func() float64 { return 0 },
954 })
955
956 ctx, cancel := context.WithCancel(context.Background())
957 runErr := make(chan error, 1)
958 go func() { runErr <- sup.Run(ctx) }()
959 require.Equal(t, 1, connector.awaitStart(t))
960 require.True(t, <-enteredAfterRaw, "raw mode must be entered before the first broker connection")
961 cancel()
962 <-runErr
963}
964
965func TestSupervisorSingleAttemptAndTimer(t *testing.T) {
966 ctx, cancel := context.WithCancel(context.Background())
967
968 clock := newSupervisorTestClock()
969 reader := newSupervisorTestReader()
970 t.Cleanup(reader.unblock)
971 terminal := newSupervisorTestTerminal(reader)
972 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
973 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "down"}
974 })
975
976 sup := mustSupervisor(t, SupervisorConfig{
977 Connector: connector,
978 Terminal: terminal,
979 Clock: clock,
980 Jitter: func() float64 { return 0 },
981 })
982
983 runErr := make(chan error, 1)
984 go func() { runErr <- sup.Run(ctx) }()
985
986 first := clock.awaitTimer(t)
987 first.fire()
988 second := clock.awaitTimer(t)
989 require.Equal(t, 100*time.Millisecond, second.delay)
990 second.fire()
991 clock.awaitTimer(t)
992
993 require.Equal(t, 1, connector.maxConcurrent(), "attempts never overlap")
994 require.Equal(t, 3, connector.callCount())
995 require.Equal(t, 1, clock.liveTimers(), "exactly one retry timer is live")
996 require.Zero(t, connector.inFlightCount(), "the connector call has returned before the next wait")
997
998 cancel()
999 require.ErrorIs(t, <-runErr, context.Canceled)
1000 require.Eventually(t, clock.allStopped, time.Second, time.Millisecond, "every retry timer is retired")
1001 require.Equal(t, terminal.restoreCount(), 1)
1002 require.Zero(t, connector.inFlightCount())
1003}
1004
1005// supervisorCancellationRaceRounds is how many times each cancellation race is
1006// replayed. Every round orders the parent cancellation strictly before the run's
1007// settle decision by construction, so the rounds are deterministic rather than
1008// timing-dependent; the repetition runs both settle points many times over.
1009const supervisorCancellationRaceRounds = 100
1010
1011// TestSupervisorCancellationRacingSettlementIsTerminal pins the parent
1012// cancellation fence at both settle points: an attempt that settles at the same
1013// instant as the parent cancellation is retired rather than adopted, and an
1014// established-connection loss that settles at that instant is never turned into a
1015// retry. Every round must end terminating with ctx.Err without scheduling a retry
1016// cadence, rendering ConnectivityRetryWait, or notifying a connectivity failure
1017// for a settle the parent had already given up on.
1018//
1019// The ordering is enforced, never hoped for: each round parks the settle at a
1020// point the round itself releases (the state lock the run decides adoption under,
1021// or the connection release the loss settle runs after) and cancels there, so the
1022// decision sees the cancellation whichever branch the run's select took.
1023func TestSupervisorCancellationRacingSettlementIsTerminal(t *testing.T) {
1024 transient := ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "raced"}
1025 exited := ports.BrokerError{Code: ports.BrokerErrorExplicitExit, Text: "raced exit"}
1026
1027 attempts := []struct {
1028 name string
1029 attempt func() (*supervisorTestService, error)
1030 }{
1031 {name: "retryable failure", attempt: func() (*supervisorTestService, error) { return nil, transient }},
1032 {
1033 name: "ready connection",
1034 attempt: func() (*supervisorTestService, error) {
1035 service := newSupervisorTestService(ports.BrokerConnectionID{3})
1036 service.publish(1, 1)
1037 return service, nil
1038 },
1039 },
1040 {name: "terminal failure", attempt: func() (*supervisorTestService, error) { return nil, exited }},
1041 }
1042
1043 for _, tc := range attempts {
1044 t.Run("attempt "+tc.name, func(t *testing.T) {
1045 // One carrier thread for the round: yielding after the attempt is
1046 // released then runs that goroutine to its settlement before this one
1047 // resumes, so the outcome is delivered while the parent is still live
1048 // and the run's select takes the settled attempt rather than ctx.Done.
1049 defer runtime.GOMAXPROCS(runtime.GOMAXPROCS(1))
1050 for round := 0; round < supervisorCancellationRaceRounds; round++ {
1051 release := make(chan struct{})
1052 var raced *supervisorTestService
1053 race := startCancellationRace(t, func(context.Context, int) (ports.BrokerNavigator, error) {
1054 <-release
1055 service, err := tc.attempt()
1056 raced = service
1057 if service == nil {
1058 return nil, err
1059 }
1060 return service, err
1061 })
1062 require.Equal(t, 1, race.connector.awaitStart(t), "round %d", round)
1063
1064 // Take the state lock the run decides adoption under, release the
1065 // attempt, let it settle, and cancel the parent while still holding
1066 // the lock: the decision is ordered strictly after the cancellation
1067 // by construction.
1068 race.supervisor.mu.Lock()
1069 close(release)
1070 runtime.Gosched()
1071 race.cancel()
1072 race.supervisor.mu.Unlock()
1073
1074 race.assertSettledAsCancelled(t, round, false)
1075 if raced != nil {
1076 select {
1077 case <-raced.closedCh():
1078 default:
1079 t.Fatalf("round %d: an attempt that raced cancellation must be retired, not adopted", round)
1080 }
1081 }
1082 }
1083 })
1084 }
1085
1086 t.Run("established loss", func(t *testing.T) {
1087 for round := 0; round < supervisorCancellationRaceRounds; round++ {
1088 losing := newSupervisorTestService(ports.BrokerConnectionID{4})
1089 losing.publish(1, 1)
1090 losing.holdClose()
1091 race := startCancellationRace(t, func(context.Context, int) (ports.BrokerNavigator, error) {
1092 return losing, nil
1093 })
1094 require.Equal(t, 1, race.connector.awaitStart(t), "round %d", round)
1095 require.Eventually(t, func() bool { return race.supervisor.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond, "round %d", round)
1096
1097 // The loss wakes the run while the parent is still live, and its
1098 // connection release then parks the run before the settle decision, so
1099 // the cancellation below is ordered strictly before that decision.
1100 losing.lose(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "raced loss"})
1101 select {
1102 case <-losing.closingCh():
1103 case <-time.After(5 * time.Second):
1104 t.Fatalf("round %d: the run never released the lost connection", round)
1105 }
1106 race.cancel()
1107 losing.releaseClose()
1108
1109 race.assertSettledAsCancelled(t, round, true)
1110 select {
1111 case <-losing.closedCh():
1112 case <-time.After(5 * time.Second):
1113 t.Fatalf("round %d: the lost connection must be retired", round)
1114 }
1115 }
1116 })
1117}
1118
1119// cancellationRace is one scripted round of the parent-cancellation race: the
1120// running supervisor with its fake clock, terminal, and recorder, plus the run's
1121// result channel. A round cancels the parent itself, at the point that orders the
1122// cancellation before the settle decision under test.
1123type cancellationRace struct {
1124 cancel context.CancelFunc
1125 clock *supervisorTestClock
1126 terminal *supervisorTestTerminal
1127 recorder *cancellationRaceRecorder
1128 connector *supervisorTestConnector
1129 supervisor *Supervisor
1130 runErr chan error
1131}
1132
1133// startCancellationRace starts one supervisor run over a scripted connector. The
1134// reader is shared and never unblocked until the test ends, so the run can only
1135// end through cancellation.
1136func startCancellationRace(t *testing.T, connect func(ctx context.Context, call int) (ports.BrokerNavigator, error)) *cancellationRace {
1137 t.Helper()
1138 ctx, cancel := context.WithCancel(context.Background())
1139 reader := newSupervisorTestReader()
1140 t.Cleanup(reader.unblock)
1141
1142 race := &cancellationRace{
1143 cancel: cancel,
1144 clock: newSupervisorTestClock(),
1145 terminal: newSupervisorTestTerminal(reader),
1146 recorder: &cancellationRaceRecorder{},
1147 connector: newSupervisorTestConnector(connect),
1148 runErr: make(chan error, 1),
1149 }
1150 race.supervisor = mustSupervisor(t, SupervisorConfig{
1151 Connector: race.connector,
1152 Terminal: race.terminal,
1153 Clock: race.clock,
1154 Jitter: func() float64 { return 0 },
1155 Render: race.recorder.render,
1156 Notify: race.recorder.notify,
1157 })
1158 go func() { race.runErr <- race.supervisor.Run(ctx) }()
1159 return race
1160}
1161
1162// assertSettledAsCancelled asserts one round's cancellation invariants: the run
1163// ends terminating with ctx.Err, raw mode is restored exactly once, no retry
1164// cadence was scheduled or notified, and every earlier render is the attempt that
1165// was still in flight. established marks a round whose connection was ready
1166// before the race, so its earlier ready render is expected.
1167func (r *cancellationRace) assertSettledAsCancelled(t *testing.T, round int, established bool) {
1168 t.Helper()
1169 require.ErrorIs(t, <-r.runErr, context.Canceled, "round %d: parent cancellation must settle the run", round)
1170 renders, notified := r.recorder.recorded()
1171 require.Empty(t, notified, "round %d: cancellation must not notify a connectivity failure", round)
1172 require.Zero(t, r.clock.timerCount(), "round %d: cancellation must not schedule a retry timer", round)
1173 require.Equal(t, 1, r.terminal.restoreCount(), "round %d: raw mode must be restored exactly once", round)
1174 require.NotEmpty(t, renders, "round %d: the run must render its state", round)
1175 require.Equal(t, PresentTerminating, renders[len(renders)-1].Presentation, "round %d: the run must end terminating", round)
1176 for _, state := range renders[:len(renders)-1] {
1177 require.Equal(t, PresentPicker, state.Presentation, "round %d: the run must stay in the picker", round)
1178 require.NotEqual(t, ConnectivityRetryWait, state.Connectivity,
1179 "round %d: no retry cadence may be published after the cancellation", round)
1180 if established {
1181 // The connection was ready before the race, so its ready render is the
1182 // only settled state the round expects.
1183 continue
1184 }
1185 require.Equal(t, ConnectivityConnectingBroker, state.Connectivity,
1186 "round %d: no settled connection may be published for the cancelled attempt", round)
1187 }
1188}
1189
1190// cancellationRaceRecorder records what one round's supervisor published.
1191type cancellationRaceRecorder struct {
1192 mu sync.Mutex
1193 renders []State
1194 notified []error
1195}
1196
1197func (r *cancellationRaceRecorder) render(state State) {
1198 r.mu.Lock()
1199 r.renders = append(r.renders, state)
1200 r.mu.Unlock()
1201}
1202
1203func (r *cancellationRaceRecorder) notify(_ State, err error) {
1204 r.mu.Lock()
1205 r.notified = append(r.notified, err)
1206 r.mu.Unlock()
1207}
1208
1209func (r *cancellationRaceRecorder) recorded() ([]State, []error) {
1210 r.mu.Lock()
1211 defer r.mu.Unlock()
1212 return append([]State(nil), r.renders...), append([]error(nil), r.notified...)
1213}
1214
1215// supervisorNilSub is a typed-nil subscription: it satisfies the interface but
1216// is nil underneath, so only the reflection guard keeps the supervisor from
1217// dereferencing it.
1218type supervisorNilSub struct{}
1219
1220func (*supervisorNilSub) Changed() <-chan struct{} { return nil }
1221func (*supervisorNilSub) Close() {}
1222
1223// supervisorNilSubService overrides Subscribe so a test can exercise the
1224// supervisor's nil-subscription guard without a live subscription.
1225type supervisorNilSubService struct {
1226 *supervisorTestService
1227 sub ports.BrokerSubscription
1228}
1229
1230func (s *supervisorNilSubService) Subscribe() (ports.BrokerSubscription, error) {
1231 return s.sub, nil
1232}
1233
1234// TestSupervisorDefensiveGenerationFenceRejectsSupersededAttempt pins the
1235// defensive generation fence directly: a settled attempt whose generation is not
1236// the current one is retired and discarded, while the current generation is
1237// adopted unchanged. Run joins every attempt before it advances, so no
1238// superseded completion can reach adoptAttempt in the production driver today;
1239// pinning the fence here keeps a future concurrent driver from dropping it.
1240func TestSupervisorDefensiveGenerationFenceRejectsSupersededAttempt(t *testing.T) {
1241 clock := newSupervisorTestClock()
1242 terminal := newSupervisorTestTerminal(newSupervisorTestReader())
1243 sup := mustSupervisor(t, SupervisorConfig{
1244 Connector: newSupervisorTestConnector(nil),
1245 Terminal: terminal,
1246 Clock: clock,
1247 Jitter: func() float64 { return 0 },
1248 })
1249
1250 // Claim a generation exactly as Run does before it launches the connector.
1251 sup.transition(supervisorEvent{kind: supervisorBeginAttempt})
1252 current := sup.State().Generation
1253 require.Equal(t, uint64(1), current)
1254
1255 service := newSupervisorTestService(ports.BrokerConnectionID{1})
1256 subClosed := make(chan struct{})
1257 stale := supervisorAttempt{
1258 generation: current - 1,
1259 service: service,
1260 sub: supervisorTestSubscription{closeFn: func() { close(subClosed) }},
1261 }
1262 adopted, cause := sup.adoptAttempt(context.Background(), stale)
1263 require.False(t, adopted, "a superseded generation must not be adopted")
1264 require.NoError(t, cause, "a superseded generation is not a cancellation")
1265 // Run retires every attempt it does not adopt.
1266 stale.retire()
1267 select {
1268 case <-service.closedCh():
1269 default:
1270 t.Fatal("a superseded service must be closed")
1271 }
1272 select {
1273 case <-subClosed:
1274 default:
1275 t.Fatal("a superseded subscription must be closed")
1276 }
1277
1278 // The current generation is adopted, so its live service is left untouched.
1279 live := newSupervisorTestService(ports.BrokerConnectionID{2})
1280 adopted, cause = sup.adoptAttempt(context.Background(), supervisorAttempt{generation: current, service: live})
1281 require.True(t, adopted, "the current generation must be adopted")
1282 require.NoError(t, cause)
1283 select {
1284 case <-live.closedCh():
1285 t.Fatal("the current attempt must be adopted, not retired")
1286 default:
1287 }
1288}
1289
1290// TestSupervisorNormalizesUntypedConnectFailure proves an untyped connect
1291// failure becomes a typed unavailable failure that preserves its cause, and that
1292// the same typed value reaches the notifier.
1293func TestSupervisorNormalizesUntypedConnectFailure(t *testing.T) {
1294 ctx, cancel := context.WithCancel(context.Background())
1295 defer cancel()
1296
1297 clock := newSupervisorTestClock()
1298 reader := newSupervisorTestReader()
1299 t.Cleanup(reader.unblock)
1300 terminal := newSupervisorTestTerminal(reader)
1301
1302 cause := errors.New("dial: connection refused")
1303 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
1304 return nil, cause
1305 })
1306 notified := make(chan error, 4)
1307 sup := mustSupervisor(t, SupervisorConfig{
1308 Connector: connector,
1309 Terminal: terminal,
1310 Clock: clock,
1311 Jitter: func() float64 { return 0 },
1312 Notify: func(_ State, err error) { notified <- err },
1313 })
1314
1315 runErr := make(chan error, 1)
1316 go func() { runErr <- sup.Run(ctx) }()
1317 clock.awaitTimer(t)
1318
1319 var typed ports.BrokerError
1320 require.ErrorAs(t, sup.State().Err, &typed)
1321 require.Equal(t, ports.BrokerErrorUnavailable, typed.Code)
1322 require.ErrorIs(t, typed.Cause, cause, "the untyped cause is preserved")
1323
1324 select {
1325 case err := <-notified:
1326 var surfaced ports.BrokerError
1327 require.ErrorAs(t, err, &surfaced)
1328 require.Equal(t, ports.BrokerErrorUnavailable, surfaced.Code)
1329 require.ErrorIs(t, surfaced.Cause, cause)
1330 case <-time.After(5 * time.Second):
1331 t.Fatal("the normalized failure was not surfaced")
1332 }
1333
1334 cancel()
1335 require.ErrorIs(t, <-runErr, context.Canceled)
1336 require.Equal(t, 1, terminal.restoreCount())
1337}
1338
1339// TestSupervisorNormalizesUntypedLoss proves an untyped established-connection
1340// loss is normalized to a typed unavailable failure that preserves its cause.
1341func TestSupervisorNormalizesUntypedLoss(t *testing.T) {
1342 ctx, cancel := context.WithCancel(context.Background())
1343 defer cancel()
1344
1345 clock := newSupervisorTestClock()
1346 reader := newSupervisorTestReader()
1347 t.Cleanup(reader.unblock)
1348 terminal := newSupervisorTestTerminal(reader)
1349
1350 service := newSupervisorTestService(ports.BrokerConnectionID{1})
1351 service.publish(1, 1)
1352 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
1353 return service, nil
1354 })
1355 sup := mustSupervisor(t, SupervisorConfig{
1356 Connector: connector,
1357 Terminal: terminal,
1358 Clock: clock,
1359 Jitter: func() float64 { return 0 },
1360 })
1361
1362 runErr := make(chan error, 1)
1363 go func() { runErr <- sup.Run(ctx) }()
1364 require.Equal(t, 1, connector.awaitStart(t))
1365 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityReady }, 5*time.Second, time.Millisecond)
1366
1367 cause := errors.New("carriage reset")
1368 service.lose(cause)
1369 require.Eventually(t, func() bool { return sup.State().Connectivity == ConnectivityRetryWait }, 5*time.Second, time.Millisecond)
1370
1371 var typed ports.BrokerError
1372 require.ErrorAs(t, sup.State().Err, &typed)
1373 require.Equal(t, ports.BrokerErrorUnavailable, typed.Code)
1374 require.ErrorIs(t, typed.Cause, cause, "the untyped loss cause is preserved")
1375
1376 cancel()
1377 require.ErrorIs(t, <-runErr, context.Canceled)
1378}
1379
1380// TestSupervisorNilSubscriptionIsTypedUnavailable proves a service that reports
1381// no subscription (a nil interface or a typed nil) is closed and treated as a
1382// typed unavailable failure instead of panicking on the first Changed call.
1383func TestSupervisorNilSubscriptionIsTypedUnavailable(t *testing.T) {
1384 cases := []struct {
1385 name string
1386 sub ports.BrokerSubscription
1387 }{
1388 {"nil interface", nil},
1389 {"typed nil", (*supervisorNilSub)(nil)},
1390 }
1391 for _, tc := range cases {
1392 t.Run(tc.name, func(t *testing.T) {
1393 ctx, cancel := context.WithCancel(context.Background())
1394 defer cancel()
1395
1396 clock := newSupervisorTestClock()
1397 reader := newSupervisorTestReader()
1398 t.Cleanup(reader.unblock)
1399 terminal := newSupervisorTestTerminal(reader)
1400
1401 base := newSupervisorTestService(ports.BrokerConnectionID{7})
1402 svc := &supervisorNilSubService{supervisorTestService: base, sub: tc.sub}
1403 connector := newSupervisorTestConnector(func(context.Context, int) (ports.BrokerNavigator, error) {
1404 return svc, nil
1405 })
1406 sup := mustSupervisor(t, SupervisorConfig{
1407 Connector: connector,
1408 Terminal: terminal,
1409 Clock: clock,
1410 Jitter: func() float64 { return 0 },
1411 })
1412
1413 runErr := make(chan error, 1)
1414 go func() { runErr <- sup.Run(ctx) }()
1415 clock.awaitTimer(t)
1416
1417 var typed ports.BrokerError
1418 require.ErrorAs(t, sup.State().Err, &typed)
1419 require.Equal(t, ports.BrokerErrorUnavailable, typed.Code)
1420 select {
1421 case <-base.closedCh():
1422 default:
1423 t.Fatal("a service that returned no subscription must be closed")
1424 }
1425
1426 cancel()
1427 require.ErrorIs(t, <-runErr, context.Canceled)
1428 require.Equal(t, 1, terminal.restoreCount())
1429 })
1430 }
1431}
1432
1433// TestSupervisorTestHubModelsSubscriptionContract pins the fake against
1434// ports.BrokerSubscription: each subscriber owns a distinct capacity-one
1435// channel that is never replaced or closed on publish, and a publication is a
1436// non-blocking offer so a parked subscriber can never busy-spin the publisher.
1437func TestSupervisorTestHubModelsSubscriptionContract(t *testing.T) {
1438 hub := newSupervisorTestHub()
1439 first := hub.subscribe(nil)
1440 second := hub.subscribe(nil)
1441 require.NotEqual(t, first.Changed(), second.Changed(), "each subscriber owns its own channel")
1442
1443 hub.publish(ports.BrokerSnapshot{Epoch: 1, Revision: 1})
1444 for name, sub := range map[string]ports.BrokerSubscription{"first": first, "second": second} {
1445 select {
1446 case <-sub.Changed():
1447 default:
1448 t.Fatalf("subscriber %s must receive the publication", name)
1449 }
1450 }
1451
1452 // A second publication while a wake is still buffered is a non-blocking
1453 // drop, and the channel identity is never replaced.
1454 channel := first.Changed()
1455 hub.publish(ports.BrokerSnapshot{Epoch: 1, Revision: 2})
1456 hub.publish(ports.BrokerSnapshot{Epoch: 1, Revision: 3})
1457 require.Equal(t, channel, first.Changed())
1458 select {
1459 case <-first.Changed():
1460 default:
1461 t.Fatal("a publication after the drained wake must offer a new wake")
1462 }
1463
1464 first.Close()
1465 hub.publish(ports.BrokerSnapshot{Epoch: 1, Revision: 4})
1466 select {
1467 case <-first.Changed():
1468 t.Fatal("a closed subscription must not receive further wakes")
1469 default:
1470 }
1471 require.Equal(t, ports.BrokerRevision(4), hub.current().Revision)
1472}
1473
1474// Raw-mode Ctrl-C is input, not SIGINT. It must work before a broker exists,
1475// including during a blocked connect and after permanent or transient failures.
1476func TestSupervisorPickerCloseWhileOffline(t *testing.T) {
1477 for _, phase := range []string{"connecting", "retry", "nonretryable"} {
1478 t.Run(phase, func(t *testing.T) {
1479 reader := newPickerChunkReader()
1480 defer reader.close()
1481 terminal := newSupervisorTestTerminal(reader)
1482 clock := newSupervisorTestClock()
1483 reached := make(chan struct{}, 1)
1484 connector := newSupervisorTestConnector(func(ctx context.Context, _ int) (ports.BrokerNavigator, error) {
1485 if phase == "connecting" {
1486 reached <- struct{}{}
1487 <-ctx.Done()
1488 return nil, ctx.Err()
1489 }
1490 code := ports.BrokerErrorUnavailable
1491 if phase == "nonretryable" {
1492 code = ports.BrokerErrorIncompatible
1493 }
1494 return nil, ports.BrokerError{Code: code, Text: "test refusal"}
1495 })
1496 s := mustSupervisor(t, SupervisorConfig{
1497 Connector: connector, Terminal: terminal, Clock: clock,
1498 Picker: NewPicker(clock, 0, terminalcap.ColorCapabilities{Mode: terminalcap.TrueColor}),
1499 Notify: func(State, error) { reached <- struct{}{} },
1500 })
1501 ctx, cancel := context.WithCancel(context.Background())
1502 defer cancel()
1503 result := make(chan error, 1)
1504 go func() { result <- s.Run(ctx) }()
1505 select {
1506 case <-reached:
1507 case <-time.After(5 * time.Second):
1508 t.Fatal("did not reach test phase")
1509 }
1510 reader.push([]byte{3})
1511 select {
1512 case err := <-result:
1513 require.NoError(t, err)
1514 case <-time.After(5 * time.Second):
1515 t.Fatal("Ctrl-C did not exit with stdin still open")
1516 }
1517 require.Equal(t, 1, terminal.restoreCount())
1518 require.True(t, clock.allStopped())
1519 })
1520 }
1521}