1package broker
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "sync/atomic"
8 "testing"
9 "time"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13 portsmocks "github.com/bnema/vev/internal/ports/mocks"
14 "github.com/bnema/vev/internal/protocol"
15 "github.com/stretchr/testify/mock"
16 "github.com/stretchr/testify/require"
17)
18
19// Stateful fakes make terminal races and blocked I/O explicit; no real carriage.
20type poolResolver func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error)
21
22func (f poolResolver) ResolveDialTarget(c context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
23 return f(c, r)
24}
25
26type poolConnector func(context.Context, ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error)
27
28func (f poolConnector) Connect(c context.Context, r ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
29 return f(c, r)
30}
31
32type fakeLogical struct {
33 done chan struct{}
34 once sync.Once
35}
36
37func newFakeLogical() *fakeLogical { return &fakeLogical{done: make(chan struct{})} }
38func (f *fakeLogical) Done() <-chan struct{} { return f.done }
39func (f *fakeLogical) Err() error { return nil }
40func (f *fakeLogical) Close() error { f.once.Do(func() { close(f.done) }); return nil }
41func (f *fakeLogical) SendEnvelope([]byte) error { <-f.done; return errors.New("closed") }
42func (f *fakeLogical) RecvEnvelope() ([]byte, error) { <-f.done; return nil, errors.New("closed") }
43
44type fakePhysical struct {
45 endpoint ports.BrokerDialTarget
46 // identity, when set, overrides the endpoint's expected identity so a test
47 // can model a first-contact acquisition whose resolved endpoint binds none.
48 identity ports.BrokerDaemonIdentity
49 done chan struct{}
50 once sync.Once
51 open func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error)
52}
53
54func (f *fakePhysical) Identity() ports.BrokerDaemonIdentity {
55 if f.identity != "" {
56 return f.identity
57 }
58 return f.endpoint.ExpectedIdentity.Identity
59}
60func (f *fakePhysical) Policy() ports.BrokerPolicy { return f.endpoint.Policy }
61func (f *fakePhysical) Incarnation() ports.BrokerDaemonIncarnation {
62 return ports.BrokerDaemonIncarnation{1}
63}
64func (f *fakePhysical) FailureKind() domain.RemoteFailureKind { return domain.RemoteFailureTransport }
65func (f *fakePhysical) Done() <-chan struct{} { return f.done }
66func (f *fakePhysical) Err() error { return errors.New("physical loss") }
67func (f *fakePhysical) Close() error { f.once.Do(func() { close(f.done) }); return nil }
68func (f *fakePhysical) OpenStream(c context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
69 if f.open != nil {
70 return f.open(c, r)
71 }
72 return newFakeLogical(), nil
73}
74func poolPolicy() ports.BrokerPolicy {
75 return ports.BrokerPolicy{ProtocolVersion: 1, CatalogSchemaVersion: 1, Transport: "quic", Trust: "trusted", Launch: "explicit", Isolation: "user", EnvironmentPolicy: protocol.EnvironmentPolicyDaemonOwned}
76}
77func poolRequest(id ports.BrokerConnectionID, n uint64) ports.BrokerOpenStreamRequest {
78 return ports.BrokerOpenStreamRequest{Epoch: 1, Connection: id, Stream: ports.BrokerStreamID(n), Local: true, Purpose: ports.BrokerStreamControl, Policy: poolPolicy(), StartMode: ports.BrokerDaemonStartIfNeeded}
79}
80func setupPool(t testing.TB, connector poolConnector) (*Pool, *manualClock) {
81 t.Helper()
82 clock := newManualClock(time.Unix(0, 0))
83 resolver := portsmocks.NewMockBrokerRouteAuthority(t)
84 resolver.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
85 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: r.Policy, Address: "fake", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "canonical", Bound: true}}, nil
86 }).Maybe()
87 binder := portsmocks.NewMockBrokerIdentityBinder(t)
88 binder.EXPECT().BindAuthenticatedIdentity(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
89 return request.Identity, nil
90 }).Maybe()
91 p, err := NewPool(1, resolver, binder, connector, clock, PoolLimits{Physical: 2, Clients: 128, Streams: 128, StreamsPerClient: 128, Warm: 2, Idle: time.Minute})
92 require.NoError(t, err)
93 t.Cleanup(func() { require.NoError(t, p.Close()) })
94 return p, clock
95}
96func await(t *testing.T, ch <-chan struct{}) {
97 t.Helper()
98 select {
99 case <-ch:
100 case <-time.After(3 * time.Second):
101 t.Fatal("timed out")
102 }
103}
104
105// TestPoolRefusedResolutionNeverDials proves a refused resolution is terminal:
106// the pool returns the typed broker error and never asks the connector to open
107// a physical transport, so an unprovisioned or conflicting request can never
108// reach an address. The same ordering fences a refused local request.
109func TestPoolRefusedResolutionNeverDials(t *testing.T) {
110 resolver := portsmocks.NewMockBrokerRouteAuthority(t)
111 resolver.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).Return(ports.BrokerDialTarget{}, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy, Cause: errors.New("refused")}).Once()
112 connector := portsmocks.NewMockBrokerEndpointConnector(t)
113 binder := portsmocks.NewMockBrokerIdentityBinder(t)
114 pool, err := NewPool(1, resolver, binder, connector, newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 8, Streams: 8, StreamsPerClient: 8, Warm: 1, Idle: time.Minute})
115 require.NoError(t, err)
116 t.Cleanup(func() { require.NoError(t, pool.Close()) })
117
118 id, err := pool.RegisterClient()
119 require.NoError(t, err)
120 _, err = pool.OpenStream(context.Background(), poolRequest(id, 1))
121 require.Error(t, err)
122 var typed ports.BrokerError
123 require.ErrorAs(t, err, &typed)
124 require.Equal(t, ports.BrokerErrorConflictingPolicy, typed.Code)
125 connector.AssertNotCalled(t, "Connect", mock.Anything, mock.Anything)
126}
127
128// TestPoolOpenFailureWithTypedNilConnection proves a failed open that still
129// carries a typed-nil logical connection is not closed, and never panics the
130// serving goroutine. A concrete-typed adapter returning a nil pointer directly
131// produces an interface that compares unequal to nil while holding no receiver.
132func TestPoolOpenFailureWithTypedNilConnection(t *testing.T) {
133 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
134 return &fakePhysical{
135 endpoint: e,
136 done: make(chan struct{}),
137 open: func(context.Context, ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
138 var absent *fakeLogical
139 return absent, errors.New("open failed")
140 },
141 }, nil
142 })
143 id, err := p.RegisterClient()
144 require.NoError(t, err)
145
146 stream, err := p.OpenStream(context.Background(), poolRequest(id, 1))
147 require.Error(t, err)
148 require.True(t, stream == nil, "a failed open must not yield a stream")
149 var typed ports.BrokerError
150 require.ErrorAs(t, err, &typed)
151}
152
153func TestPoolHundredStreamsLossAndIsolation(t *testing.T) {
154 var calls atomic.Int32
155 var physical *fakePhysical
156 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
157 calls.Add(1)
158 physical = &fakePhysical{endpoint: e, done: make(chan struct{})}
159 return physical, nil
160 })
161 streams := make([]ports.BrokerEnvelopeStream, 100)
162 var wg sync.WaitGroup
163 for i := range streams {
164 id, err := p.RegisterClient()
165 require.NoError(t, err)
166 wg.Add(1)
167 go func(i int) {
168 defer wg.Done()
169 s, e := p.OpenStream(context.Background(), poolRequest(id, 1))
170 if e != nil {
171 t.Error(e)
172 return
173 }
174 streams[i] = s
175 }(i)
176 }
177 wg.Wait()
178 require.EqualValues(t, 1, calls.Load())
179 blocked := make(chan struct{})
180 go func() { _ = streams[0].SendEnvelope(nil); close(blocked) }()
181 require.NoError(t, streams[0].Close())
182 await(t, blocked)
183 select {
184 case <-streams[1].Done():
185 t.Fatal("sibling closed")
186 default:
187 }
188 require.NoError(t, physical.Close())
189 for _, s := range streams[1:] {
190 await(t, s.Done())
191 var loss ports.BrokerError
192 require.ErrorAs(t, s.Err(), &loss)
193 require.Equal(t, ports.BrokerErrorAttachmentLost, loss.Code)
194 }
195}
196
197func TestPoolAdmissionAndNonreuse(t *testing.T) {
198 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
199 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
200 })
201 p.limits.StreamsPerClient = 1
202 id, err := p.RegisterClient()
203 require.NoError(t, err)
204 s, err := p.OpenStream(context.Background(), poolRequest(id, 1))
205 require.NoError(t, err)
206 _, err = p.OpenStream(context.Background(), poolRequest(id, 2))
207 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
208 require.NoError(t, s.Close())
209 _, err = p.OpenStream(context.Background(), poolRequest(id, 2))
210 require.ErrorIs(t, err, ports.BrokerAdmissionStale)
211 s, err = p.OpenStream(context.Background(), poolRequest(id, 3))
212 require.NoError(t, err)
213 p.CloseClient(id)
214 await(t, s.Done())
215 _, err = p.OpenStream(context.Background(), poolRequest(id, 4))
216 require.ErrorIs(t, err, ports.BrokerAdmissionStale)
217}
218
219// TestPoolStreamWindowAdmission pins the bounded anti-replay window at the pool
220// boundary: a concurrent lower identity that was allocated first is admitted
221// after a higher one, a duplicate is refused, and an identity evicted past the
222// window is stale rather than merely duplicate.
223func TestPoolStreamWindowAdmission(t *testing.T) {
224 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
225 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
226 })
227 p.limits.StreamsPerClient = ports.BrokerStreamWindowSize + 8
228 id, err := p.RegisterClient()
229 require.NoError(t, err)
230
231 // 2 then 1: the lower identity was allocated first, so it is admitted.
232 second, err := p.OpenStream(context.Background(), poolRequest(id, 2))
233 require.NoError(t, err)
234 defer second.Close()
235 first, err := p.OpenStream(context.Background(), poolRequest(id, 1))
236 require.NoError(t, err)
237 defer first.Close()
238
239 // A duplicate inside the window is stale.
240 _, err = p.OpenStream(context.Background(), poolRequest(id, 1))
241 require.ErrorIs(t, err, ports.BrokerAdmissionStale)
242 // Push the newest admitted identity past the window, then replay a low one.
243 high, err := p.OpenStream(context.Background(), poolRequest(id, ports.BrokerStreamWindowSize+2))
244 require.NoError(t, err)
245 defer high.Close()
246 _, err = p.OpenStream(context.Background(), poolRequest(id, 1))
247 require.ErrorIs(t, err, ports.BrokerAdmissionStale, "an identity evicted from the window is stale")
248}
249
250// TestPoolPendingAcquisitionSeparatesStartMode proves an ExistingOnly
251// acquisition is never coalesced with a spawn-capable acquisition to the same
252// unbound endpoint: while the first acquisition is still in flight, the second
253// starts its own instead of waiting on it, and each reaches the connector with
254// its own authorization.
255func TestPoolPendingAcquisitionSeparatesStartMode(t *testing.T) {
256 var calls atomic.Int32
257 modes := make(chan ports.BrokerDaemonStartMode, 4)
258 entered := make(chan struct{}, 4)
259 release := make(chan struct{})
260 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
261 calls.Add(1)
262 modes <- e.StartMode
263 entered <- struct{}{}
264 <-release
265 return &fakePhysical{endpoint: e, identity: "canonical", done: make(chan struct{})}, nil
266 })
267 // The endpoint is first-contact: it binds no identity, so acquisition
268 // coalescing is decided by the pending key, which is exactly where the start
269 // mode has to participate.
270 p.routes = poolResolver(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
271 return ports.BrokerDialTarget{
272 Fence: ports.BrokerEndpointFence{Registration: r.Registration}, Policy: r.Policy,
273 Address: "fake", StartMode: r.StartMode,
274 ExpectedIdentity: ports.BrokerExpectedIdentity{},
275 }, nil
276 })
277 id, _ := p.RegisterClient()
278
279 existing := poolRequest(id, 1)
280 existing.Local = false
281 existing.Endpoint = "user@host"
282 existing.Registration = domain.RemoteRegistration{Endpoint: "user@host", Incarnation: [16]byte{1}, Generation: 1}
283 existing.StartMode = ports.BrokerDaemonExistingOnly
284 startable := existing
285 startable.Stream = 2
286 startable.StartMode = ports.BrokerDaemonStartIfNeeded
287
288 firstErr := make(chan error, 1)
289 go func() { _, err := p.OpenStream(context.Background(), existing); firstErr <- err }()
290 await(t, entered)
291 secondErr := make(chan error, 1)
292 go func() { _, err := p.OpenStream(context.Background(), startable); secondErr <- err }()
293 // Both acquisitions must reach the connector while the first is still in
294 // flight; a coalesced implementation would leave the second waiting here.
295 await(t, entered)
296 close(release)
297 require.NoError(t, <-firstErr)
298 require.NoError(t, <-secondErr)
299
300 require.EqualValues(t, 2, calls.Load(), "a spawn-capable request must not reuse an ExistingOnly acquisition")
301 got := map[ports.BrokerDaemonStartMode]bool{}
302 for range 2 {
303 select {
304 case mode := <-modes:
305 got[mode] = true
306 case <-time.After(3 * time.Second):
307 t.Fatal("the connector never observed both acquisitions")
308 }
309 }
310 require.True(t, got[ports.BrokerDaemonExistingOnly], "the ExistingOnly authorization reaches the transport")
311 require.True(t, got[ports.BrokerDaemonStartIfNeeded], "the start-if-needed authorization reaches the transport")
312}
313
314func TestPoolCoalescedCancellation(t *testing.T) {
315 entered, proceed := make(chan struct{}), make(chan struct{})
316 var calls atomic.Int32
317 p, _ := setupPool(t, func(ctx context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
318 calls.Add(1)
319 close(entered)
320 select {
321 case <-ctx.Done():
322 return nil, ctx.Err()
323 case <-proceed:
324 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
325 }
326 })
327 id1, _ := p.RegisterClient()
328 id2, _ := p.RegisterClient()
329 ctx, cancel := context.WithCancel(context.Background())
330 result := make(chan error, 1)
331 go func() { _, err := p.OpenStream(ctx, poolRequest(id1, 1)); result <- err }()
332 await(t, entered)
333 result2 := make(chan ports.BrokerEnvelopeStream, 1)
334 go func() {
335 s, err := p.OpenStream(context.Background(), poolRequest(id2, 1))
336 if err != nil {
337 t.Error(err)
338 }
339 result2 <- s
340 }()
341 require.Eventually(t, func() bool {
342 p.mu.Lock()
343 defer p.mu.Unlock()
344 for _, e := range p.entries {
345 return e.refs == 2
346 }
347 return false
348 }, time.Second, time.Millisecond)
349 cancel()
350 require.ErrorIs(t, <-result, context.Canceled)
351 close(proceed)
352 s := <-result2
353 require.NotNil(t, s)
354 require.EqualValues(t, 1, calls.Load())
355 require.NoError(t, s.Close())
356}
357
358func TestPoolAbandonedAttemptAndShutdown(t *testing.T) {
359 entered, exited := make(chan struct{}), make(chan struct{})
360 p, _ := setupPool(t, func(ctx context.Context, _ ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
361 close(entered)
362 <-ctx.Done()
363 close(exited)
364 return nil, ctx.Err()
365 })
366 id, _ := p.RegisterClient()
367 ctx, cancel := context.WithCancel(context.Background())
368 result := make(chan error, 1)
369 go func() { _, err := p.OpenStream(ctx, poolRequest(id, 1)); result <- err }()
370 await(t, entered)
371 cancel()
372 require.ErrorIs(t, <-result, context.Canceled)
373 await(t, exited)
374 require.NoError(t, p.Close())
375 _, err := p.RegisterClient()
376 require.ErrorIs(t, err, ports.BrokerAdmissionClosed)
377}
378
379func TestPoolIdleEviction(t *testing.T) {
380 physicals := make(chan *fakePhysical, 2)
381 p, clock := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
382 f := &fakePhysical{endpoint: e, done: make(chan struct{})}
383 physicals <- f
384 return f, nil
385 })
386 id, _ := p.RegisterClient()
387 s, err := p.OpenStream(context.Background(), poolRequest(id, 1))
388 require.NoError(t, err)
389 f := <-physicals
390 require.NoError(t, s.Close())
391 // The watcher re-arms its idle timer each loop, so an early Advance can
392 // land before the re-arm. Keep advancing until the eviction is observed.
393 require.Eventually(t, func() bool {
394 clock.Advance(time.Minute)
395 select {
396 case <-f.done:
397 return true
398 default:
399 return false
400 }
401 }, 3*time.Second, time.Millisecond)
402 require.NoError(t, p.Close())
403}
404
405func TestPoolPolicyAndIdentityVerification(t *testing.T) {
406 for _, mode := range []string{"identity", "policy", "resolver"} {
407 t.Run(mode, func(t *testing.T) {
408 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
409 if mode == "identity" {
410 e.ExpectedIdentity.Identity = "impostor"
411 }
412 if mode == "policy" {
413 e.Policy.Trust = "other"
414 }
415 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
416 })
417 if mode == "resolver" {
418 p.routes = poolResolver(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
419 policy := r.Policy
420 policy.Trust = "other"
421 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: policy, Address: "fake", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "canonical", Bound: true}}, nil
422 })
423 }
424 id, _ := p.RegisterClient()
425 _, err := p.OpenStream(context.Background(), poolRequest(id, 1))
426 var typed ports.BrokerError
427 require.ErrorAs(t, err, &typed)
428 })
429 }
430}
431
432func BenchmarkPoolOpenClose(b *testing.B) {
433 p, _ := setupPool(b, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
434 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
435 })
436 id, _ := p.RegisterClient()
437 b.ReportAllocs()
438 b.ResetTimer()
439 for i := 0; i < b.N; i++ {
440 s, err := p.OpenStream(context.Background(), poolRequest(id, uint64(i+1)))
441 if err != nil {
442 b.Fatal(err)
443 }
444 _ = s.Close()
445 }
446}
447
448func TestPoolAliasesAndPolicyPartition(t *testing.T) {
449 var calls atomic.Int32
450 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
451 calls.Add(1)
452 return &fakePhysical{endpoint: e, done: make(chan struct{})}, nil
453 })
454 id, _ := p.RegisterClient()
455 for i, alias := range []string{"user@first", "user@second"} {
456 r := poolRequest(id, uint64(i+1))
457 r.Local = false
458 r.Endpoint = alias
459 r.Registration = domain.RemoteRegistration{Endpoint: alias, Incarnation: [16]byte{1}, Generation: 1}
460 s, err := p.OpenStream(context.Background(), r)
461 require.NoError(t, err)
462 defer s.Close()
463 }
464 require.EqualValues(t, 1, calls.Load())
465 r := poolRequest(id, 3)
466 r.Policy.Isolation = "other"
467 s, err := p.OpenStream(context.Background(), r)
468 require.NoError(t, err)
469 defer s.Close()
470 require.EqualValues(t, 2, calls.Load())
471 r.Stream = 4
472 r.Policy.Trust = "third"
473 _, err = p.OpenStream(context.Background(), r)
474 require.ErrorIs(t, err, ports.BrokerAdmissionLimit)
475}
476
477func TestPoolPendingOpenCancellationDoesNotHoldLock(t *testing.T) {
478 entered := make(chan struct{})
479 p, _ := setupPool(t, func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
480 return &fakePhysical{endpoint: e, done: make(chan struct{}), open: func(ctx context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
481 if r.Stream == 1 {
482 close(entered)
483 <-ctx.Done()
484 return nil, ctx.Err()
485 }
486 return newFakeLogical(), nil
487 }}, nil
488 })
489 id, _ := p.RegisterClient()
490 result := make(chan error, 1)
491 go func() { _, err := p.OpenStream(context.Background(), poolRequest(id, 1)); result <- err }()
492 await(t, entered)
493 s, err := p.OpenStream(context.Background(), poolRequest(id, 2))
494 require.NoError(t, err)
495 require.NoError(t, p.CloseStream(id, 1))
496 require.ErrorIs(t, <-result, context.Canceled)
497 select {
498 case <-s.Done():
499 t.Fatal("sibling canceled")
500 default:
501 }
502 require.NoError(t, p.Close())
503 await(t, s.Done())
504}