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