1package broker
2
3import (
4 "context"
5 "sync/atomic"
6 "testing"
7 "time"
8
9 "github.com/bnema/vev/internal/ports"
10 "github.com/stretchr/testify/require"
11)
12
13func waitPoolEmpty(t *testing.T, p *Pool) {
14 t.Helper()
15 require.Eventually(t, func() bool { p.mu.Lock(); defer p.mu.Unlock(); return len(p.entries) == 0 }, time.Second, time.Millisecond)
16}
17
18func TestPoolIdleEvictionAndReconnect(t *testing.T) {
19 var calls atomic.Int32
20 p, clock := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
21 calls.Add(1)
22 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
23 })
24 id, _ := p.RegisterClient()
25 s, err := p.OpenStream(context.Background(), poolRequest(id, 1))
26 require.NoError(t, err)
27 require.NoError(t, s.Close())
28 require.Eventually(t, func() bool { clock.mu.Lock(); defer clock.mu.Unlock(); return len(clock.timers) == 1 }, time.Second, time.Millisecond)
29 clock.Advance(time.Minute)
30 waitPoolEmpty(t, p)
31 s, err = p.OpenStream(context.Background(), poolRequest(id, 2))
32 require.NoError(t, err)
33 require.EqualValues(t, 2, calls.Load())
34 require.NoError(t, s.Close())
35}
36
37type retiringPhysical struct {
38 *fakePhysical
39 closing, allowClose chan struct{}
40}
41
42func (f *retiringPhysical) Close() error {
43 close(f.closing)
44 <-f.allowClose
45 return f.fakePhysical.Close()
46}
47
48func TestPoolRetiringKeyUntilCloseCompletes(t *testing.T) {
49 var calls atomic.Int32
50 var first *retiringPhysical
51 allow := make(chan struct{})
52 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
53 f := &fakePhysical{endpoint: e, done: make(chan struct{})}
54 if calls.Add(1) == 1 {
55 first = &retiringPhysical{fakePhysical: f, closing: make(chan struct{}), allowClose: allow}
56 return first, nil
57 }
58 return f, nil
59 })
60 defer func() {
61 select {
62 case <-allow:
63 default:
64 close(allow)
65 }
66 }()
67 id, _ := p.RegisterClient()
68 s, err := p.OpenStream(context.Background(), poolRequest(id, 1))
69 require.NoError(t, err)
70 require.NoError(t, first.fakePhysical.Close())
71 await(t, first.closing)
72 await(t, s.Done())
73 _, err = p.OpenStream(context.Background(), poolRequest(id, 2))
74 var typed ports.BrokerError
75 require.ErrorAs(t, err, &typed)
76 require.Equal(t, ports.BrokerErrorUnavailable, typed.Code)
77 require.EqualValues(t, 1, calls.Load())
78 close(allow)
79 waitPoolEmpty(t, p)
80 s, err = p.OpenStream(context.Background(), poolRequest(id, 3))
81 require.NoError(t, err)
82 require.EqualValues(t, 2, calls.Load())
83 require.NoError(t, s.Close())
84}
85
86func TestPoolFailedConnectRetry(t *testing.T) {
87 var calls atomic.Int32
88 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
89 if calls.Add(1) == 1 {
90 return nil, context.DeadlineExceeded
91 }
92 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
93 })
94 id, _ := p.RegisterClient()
95 _, err := p.OpenStream(context.Background(), poolRequest(id, 1))
96 require.ErrorIs(t, err, context.DeadlineExceeded)
97 waitPoolEmpty(t, p)
98 _, err = p.OpenStream(context.Background(), poolRequest(id, 1))
99 require.ErrorIs(t, err, ports.BrokerAdmissionStale)
100 s, err := p.OpenStream(context.Background(), poolRequest(id, 2))
101 require.NoError(t, err)
102 require.NoError(t, s.Close())
103 require.EqualValues(t, 2, calls.Load())
104}