1package broker
2
3import (
4 "context"
5 "errors"
6 "testing"
7 "time"
8
9 "github.com/bnema/vev/internal/domain"
10 "github.com/bnema/vev/internal/ports"
11 "github.com/stretchr/testify/require"
12)
13
14func TestPoolAdapterFailures(t *testing.T) {
15 for _, stage := range []string{"resolver", "connector", "open"} {
16 for _, tc := range []struct {
17 name string
18 err error
19 code ports.BrokerErrorCode
20 }{
21 {"unknown", errors.New("adapter failure"), ports.BrokerErrorUnavailable},
22 {"cancel", context.Canceled, ports.BrokerErrorCancelled},
23 {"deadline", context.DeadlineExceeded, ports.BrokerErrorTimeout},
24 {"semantic", ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy}, ports.BrokerErrorConflictingPolicy},
25 } {
26 t.Run(stage+"/"+tc.name, func(t *testing.T) {
27 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
28 if stage == "connector" {
29 return nil, tc.err
30 }
31 return &fakePhysical{endpoint: e, done: make(chan struct{}), open: func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
32 return nil, tc.err
33 }}, nil
34 })
35 if stage == "resolver" {
36 p.routes = poolResolver(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
37 return ports.BrokerDialTarget{}, tc.err
38 })
39 }
40 id, _ := p.RegisterClient()
41 _, err := p.OpenStream(context.Background(), poolRequest(id, 1))
42 require.ErrorIs(t, err, tc.err)
43 var typed ports.BrokerError
44 require.ErrorAs(t, err, &typed)
45 require.Equal(t, tc.code, typed.Code)
46 })
47 }
48 }
49}
50
51func TestPoolRejectedPaths(t *testing.T) {
52 for _, name := range []string{"closed", "epoch", "invalid endpoint", "nil raw", "retiring"} {
53 t.Run(name, func(t *testing.T) {
54 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
55 return &fakePhysical{endpoint: e, done: make(chan struct{}), open: func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
56 return nil, nil
57 }}, nil
58 })
59 id, _ := p.RegisterClient()
60 req := poolRequest(id, 1)
61 switch name {
62 case "closed":
63 require.NoError(t, p.Close())
64 case "epoch":
65 req.Epoch++
66 case "invalid endpoint":
67 p.routes = poolResolver(func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
68 return ports.BrokerDialTarget{}, nil
69 })
70 case "retiring":
71 key := poolKey{"canonical", poolPolicy()}
72 p.entries[key] = &poolEntry{key: key, retiring: true}
73 }
74 _, err := p.OpenStream(context.Background(), req)
75 require.Error(t, err)
76 if name == "closed" {
77 require.ErrorIs(t, err, ports.BrokerAdmissionClosed)
78 _, err = p.RegisterClient()
79 require.ErrorIs(t, err, ports.BrokerAdmissionClosed)
80 } else {
81 var typed ports.BrokerError
82 require.ErrorAs(t, err, &typed)
83 want := ports.BrokerErrorIncompatible
84 if name == "epoch" {
85 want = ports.BrokerErrorStaleEpoch
86 }
87 if name == "retiring" {
88 want = ports.BrokerErrorUnavailable
89 }
90 require.Equal(t, want, typed.Code)
91 }
92 p.mu.Lock()
93 require.Zero(t, p.streams)
94 p.mu.Unlock()
95 })
96 }
97}
98
99func TestPoolRedirectRejectsUnavailableWinnerWithoutResurrectingRefs(t *testing.T) {
100 for _, state := range []string{"removed", "retiring"} {
101 t.Run(state, func(t *testing.T) {
102 p, _ := setupPool(t, func(context.Context, ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) { panic("unused") })
103 key := poolKey{"canonical", poolPolicy()}
104 winner := &poolEntry{key: key, refs: 2}
105 loser := &poolEntry{redirect: winner, refs: 1}
106 reservation := &reservation{entry: loser}
107 if state == "retiring" {
108 winner.retiring = true
109 p.entries[key] = winner
110 }
111
112 got, err := p.redirectReservation(loser, reservation)
113 require.Nil(t, got)
114 var typed ports.BrokerError
115 require.ErrorAs(t, err, &typed)
116 require.Equal(t, ports.BrokerErrorUnavailable, typed.Code)
117 require.Equal(t, 1, loser.refs)
118 require.Equal(t, 2, winner.refs)
119 require.Same(t, loser, reservation.entry)
120 })
121 }
122}
123
124func TestPoolPostOpenPhysicalLoss(t *testing.T) {
125 raw := newFakeLogical()
126 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
127 f := &fakePhysical{endpoint: e, done: make(chan struct{})}
128 f.open = func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
129 _ = f.Close()
130 return raw, nil
131 }
132 return f, nil
133 })
134 id, _ := p.RegisterClient()
135 _, err := p.OpenStream(context.Background(), poolRequest(id, 1))
136 var lost ports.BrokerStreamLost
137 require.ErrorAs(t, err, &lost)
138 require.NoError(t, lost.Validate())
139 require.Equal(t, domain.RemoteFailureTransport, lost.Cause)
140 await(t, raw.Done())
141}
142
143func TestPoolTerminalPrecedence(t *testing.T) {
144 for _, name := range []string{"physical over logical", "request over physical", "shutdown over physical"} {
145 t.Run(name, func(t *testing.T) {
146 p, _ := setupPool(t, func(context.Context, ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
147 panic("unused")
148 })
149 ctx, cancel := context.WithCancel(context.Background())
150 defer cancel()
151 f := &fakePhysical{done: make(chan struct{})}
152 _ = f.Close()
153 e := &poolEntry{ctx: context.Background(), physical: f}
154 if name == "request over physical" {
155 cancel()
156 }
157 if name == "shutdown over physical" {
158 p.cancel()
159 }
160 err := p.streamTerminal(ctx, e, poolRequest(ports.BrokerConnectionID{1}, 1), nil)
161 if name == "physical over logical" {
162 var lost ports.BrokerStreamLost
163 require.ErrorAs(t, err, &lost)
164 require.NoError(t, lost.Validate())
165 } else {
166 var typed ports.BrokerError
167 require.ErrorAs(t, err, &typed)
168 require.Equal(t, ports.BrokerErrorCancelled, typed.Code)
169 }
170 })
171 }
172}
173
174// TestPoolRetiredEntryNamesReason proves a stream ended by an entry retirement
175// carries the retirement reason, not a bare attachment_lost.
176func TestPoolRetiredEntryNamesReason(t *testing.T) {
177 cases := []struct {
178 name string
179 reason error
180 }{
181 {"idle", errEntryIdle},
182 {"evicted", errEntryEvicted},
183 {"abandoned", errEntryAbandoned},
184 }
185 for _, tc := range cases {
186 t.Run(tc.name, func(t *testing.T) {
187 p, _ := setupPool(t, func(context.Context, ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
188 panic("unused")
189 })
190 ctx, cancel := context.WithCancelCause(context.Background())
191 e := &poolEntry{ctx: ctx, cancel: cancel, physical: &fakePhysical{done: make(chan struct{})}}
192 e.retire(tc.reason)()
193
194 err := p.streamTerminal(context.Background(), e, poolRequest(ports.BrokerConnectionID{1}, 1), nil)
195 var typed ports.BrokerError
196 require.ErrorAs(t, err, &typed)
197 require.Equal(t, ports.BrokerErrorAttachmentLost, typed.Code)
198 require.Equal(t, tc.reason.Error(), typed.Text)
199 require.NoError(t, typed.Validate())
200 require.ErrorIs(t, err, tc.reason)
201 })
202 }
203}
204
205// Reset deliberately does not drain, matching older buffered timer adapters.
206type bufferedPoolTimer struct {
207 ch chan time.Time
208 active bool
209}
210
211func (t *bufferedPoolTimer) C() <-chan time.Time { return t.ch }
212func (t *bufferedPoolTimer) Stop() bool { was := t.active; t.active = false; return was }
213func (t *bufferedPoolTimer) Reset(time.Duration) bool { was := t.active; t.active = true; return was }
214func TestPoolTimerStopDrain(t *testing.T) {
215 for _, name := range []string{"active", "expired buffered", "expired consumed", "stopped"} {
216 t.Run(name, func(t *testing.T) {
217 timer := &bufferedPoolTimer{ch: make(chan time.Time, 1), active: name == "active"}
218 if name == "expired buffered" {
219 timer.ch <- time.Time{}
220 }
221 stopPoolTimer(timer)
222 timer.Reset(time.Minute)
223 select {
224 case <-timer.C():
225 t.Fatal("stale expiry survived reset")
226 default:
227 }
228 stopPoolTimer(timer)
229 require.False(t, timer.active)
230 })
231 }
232}
233
234type closeErrorLogical struct {
235 *fakeLogical
236 err error
237}
238
239func (l *closeErrorLogical) Close() error { _ = l.fakeLogical.Close(); return l.err }
240func TestPooledStreamPreservesCloseError(t *testing.T) {
241 cause := errors.New("close failed")
242 calls := 0
243 s := &pooledStream{BrokerEnvelopeStream: &closeErrorLogical{newFakeLogical(), cause}, done: make(chan struct{}), release: func() { calls++ }}
244 require.ErrorIs(t, s.Close(), cause)
245 require.ErrorIs(t, s.Close(), cause)
246 require.Equal(t, 1, calls)
247}
248
249func TestPoolCaps(t *testing.T) {
250 for _, name := range []string{"clients", "streams", "physical"} {
251 t.Run(name, func(t *testing.T) {
252 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
253 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
254 })
255 p.limits.Clients = 1
256 p.limits.Streams = 1
257 p.limits.Physical = 1
258 id, _ := p.RegisterClient()
259 if name == "clients" {
260 _, err := p.RegisterClient()
261 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
262 return
263 }
264 s, err := p.OpenStream(context.Background(), poolRequest(id, 1))
265 require.NoError(t, err)
266 defer s.Close()
267 if name == "physical" {
268 p.limits.Streams = 2
269 p.routes = poolResolver(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
270 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: r.Policy, Address: "route", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "other", Bound: true}}, nil
271 })
272 }
273 _, err = p.OpenStream(context.Background(), poolRequest(id, 2))
274 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
275 })
276 }
277}
278
279func TestPoolShutdownStreamsCancelled(t *testing.T) {
280 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
281 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
282 })
283 id, _ := p.RegisterClient()
284 s, err := p.OpenStream(context.Background(), poolRequest(id, 1))
285 require.NoError(t, err)
286 require.NoError(t, p.Close())
287 await(t, s.Done())
288 var typed ports.BrokerError
289 require.ErrorAs(t, s.Err(), &typed)
290 require.Equal(t, ports.BrokerErrorCancelled, typed.Code)
291}