1package broker
2
3import (
4 "context"
5 "sync"
6 "testing"
7 "time"
8
9 "github.com/bnema/vev/internal/ports"
10 portsmocks "github.com/bnema/vev/internal/ports/mocks"
11 "github.com/stretchr/testify/mock"
12 "github.com/stretchr/testify/require"
13)
14
15// warmHarness records every physical the pool dials per identity. The pool
16// ports are mocks; fakePhysical stays a stateful fake because its Done channel
17// models transport lifetime across pool workers.
18type warmHarness struct {
19 pool *Pool
20 clock *manualClock
21 client ports.BrokerConnectionID
22 mu sync.Mutex
23 target ports.BrokerDaemonIdentity
24 dials map[ports.BrokerDaemonIdentity][]*fakePhysical
25 stream uint64
26}
27
28func newWarmHarness(t *testing.T, warm int, idle time.Duration) *warmHarness {
29 t.Helper()
30 h := &warmHarness{clock: newManualClock(time.Unix(0, 0)), target: "alpha", dials: make(map[ports.BrokerDaemonIdentity][]*fakePhysical)}
31 routes := portsmocks.NewMockBrokerRouteAuthority(t)
32 routes.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
33 h.mu.Lock()
34 defer h.mu.Unlock()
35 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: r.Policy, Address: "fake", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: h.target, Bound: true}}, nil
36 }).Maybe()
37 binder := portsmocks.NewMockBrokerIdentityBinder(t)
38 binder.EXPECT().BindAuthenticatedIdentity(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, r ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
39 return r.Identity, nil
40 }).Maybe()
41 connector := portsmocks.NewMockBrokerEndpointConnector(t)
42 connector.EXPECT().Connect(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
43 f := &fakePhysical{endpoint: e, done: make(chan struct{})}
44 h.mu.Lock()
45 h.dials[e.ExpectedIdentity.Identity] = append(h.dials[e.ExpectedIdentity.Identity], f)
46 h.mu.Unlock()
47 return f, nil
48 }).Maybe()
49 pool, err := NewPool(1, routes, binder, connector, h.clock, PoolLimits{Physical: 4, Clients: 4, Streams: 16, StreamsPerClient: 16, Warm: warm, Idle: idle})
50 require.NoError(t, err)
51 h.pool = pool
52 t.Cleanup(func() { require.NoError(t, pool.Close()) })
53 h.client, err = pool.RegisterClient()
54 require.NoError(t, err)
55 return h
56}
57
58// visit opens and closes one logical stream to identity, as a client that
59// attaches to a remote and then navigates away does.
60func (h *warmHarness) visit(t *testing.T, identity ports.BrokerDaemonIdentity) {
61 t.Helper()
62 h.mu.Lock()
63 h.target = identity
64 h.stream++
65 stream := h.stream
66 h.mu.Unlock()
67 s, err := h.pool.OpenStream(context.Background(), poolRequest(h.client, stream))
68 require.NoError(t, err)
69 require.NoError(t, s.Close())
70}
71
72func (h *warmHarness) physicals(identity ports.BrokerDaemonIdentity) []*fakePhysical {
73 h.mu.Lock()
74 defer h.mu.Unlock()
75 return append([]*fakePhysical(nil), h.dials[identity]...)
76}
77
78func closed(f *fakePhysical) bool {
79 select {
80 case <-f.done:
81 return true
82 default:
83 return false
84 }
85}
86
87// reap advances the fake clock one grace at a time until f retires. Repeating
88// the advance tolerates a worker that re-arms after a late wake; each step is
89// a bounded fake-clock move, never a sleep.
90func (h *warmHarness) reap(t *testing.T, f *fakePhysical, grace time.Duration) {
91 t.Helper()
92 require.Eventually(t, func() bool { h.clock.Advance(grace); return closed(f) }, 3*time.Second, time.Millisecond)
93}
94
95func TestPoolWarmReuse(t *testing.T) {
96 const grace = 5 * time.Minute
97 tests := []struct {
98 name string
99 warm int
100 idle time.Duration
101 // run drives local→remote→local→remote style visits and returns the
102 // dial count expected per identity afterwards.
103 run func(t *testing.T, h *warmHarness)
104 want map[ports.BrokerDaemonIdentity]int
105 }{
106 {
107 name: "reuse within grace",
108 warm: 2, idle: grace,
109 run: func(t *testing.T, h *warmHarness) {
110 h.visit(t, "alpha")
111 h.clock.Advance(grace / 2)
112 h.visit(t, "alpha")
113 require.False(t, closed(h.physicals("alpha")[0]))
114 },
115 want: map[ports.BrokerDaemonIdentity]int{"alpha": 1},
116 },
117 {
118 name: "reap after grace",
119 warm: 2, idle: grace,
120 run: func(t *testing.T, h *warmHarness) {
121 h.visit(t, "alpha")
122 h.reap(t, h.physicals("alpha")[0], grace)
123 require.Eventually(t, func() bool { _, ok := h.pool.SharedPhysical("alpha", poolPolicy()); return !ok }, 3*time.Second, time.Millisecond)
124 require.Eventually(t, func() bool { h.pool.mu.Lock(); defer h.pool.mu.Unlock(); return len(h.pool.entries) == 0 }, 3*time.Second, time.Millisecond)
125 h.visit(t, "alpha")
126 },
127 want: map[ports.BrokerDaemonIdentity]int{"alpha": 2},
128 },
129 {
130 name: "no age bound keeps warm without timers",
131 warm: 2, idle: 0,
132 run: func(t *testing.T, h *warmHarness) {
133 h.visit(t, "alpha")
134 h.clock.Advance(24 * time.Hour)
135 h.visit(t, "alpha")
136 h.clock.mu.Lock()
137 timers := len(h.clock.timers)
138 h.clock.mu.Unlock()
139 require.Zero(t, timers)
140 },
141 want: map[ports.BrokerDaemonIdentity]int{"alpha": 1},
142 },
143 {
144 name: "bound evicts least recently idled",
145 warm: 1, idle: 0,
146 run: func(t *testing.T, h *warmHarness) {
147 h.visit(t, "alpha")
148 h.visit(t, "beta")
149 await(t, h.physicals("alpha")[0].done)
150 require.False(t, closed(h.physicals("beta")[0]))
151 h.visit(t, "beta")
152 },
153 want: map[ports.BrokerDaemonIdentity]int{"alpha": 1, "beta": 1},
154 },
155 {
156 name: "active streams do not count toward the bound",
157 warm: 1, idle: 0,
158 run: func(t *testing.T, h *warmHarness) {
159 h.visit(t, "alpha")
160 h.mu.Lock()
161 h.target = "beta"
162 h.stream++
163 stream := h.stream
164 h.mu.Unlock()
165 held, err := h.pool.OpenStream(context.Background(), poolRequest(h.client, stream))
166 require.NoError(t, err)
167 h.visit(t, "alpha")
168 require.False(t, closed(h.physicals("alpha")[0]))
169 require.NoError(t, held.Close())
170 await(t, h.physicals("alpha")[0].done)
171 },
172 want: map[ports.BrokerDaemonIdentity]int{"alpha": 1, "beta": 1},
173 },
174 {
175 name: "zero warm retires on release",
176 warm: 0, idle: 0,
177 run: func(t *testing.T, h *warmHarness) {
178 h.visit(t, "alpha")
179 await(t, h.physicals("alpha")[0].done)
180 require.Eventually(t, func() bool { h.pool.mu.Lock(); defer h.pool.mu.Unlock(); return len(h.pool.entries) == 0 }, 3*time.Second, time.Millisecond)
181 h.visit(t, "alpha")
182 },
183 want: map[ports.BrokerDaemonIdentity]int{"alpha": 2},
184 },
185 }
186 for _, tt := range tests {
187 t.Run(tt.name, func(t *testing.T) {
188 h := newWarmHarness(t, tt.warm, tt.idle)
189 tt.run(t, h)
190 for identity, want := range tt.want {
191 require.Len(t, h.physicals(identity), want, "dials for %s", identity)
192 }
193 })
194 }
195}
196
197func TestPoolAdoptProbePhysical(t *testing.T) {
198 for _, tt := range []struct {
199 name string
200 warm int
201 idle time.Duration
202 want bool
203 }{
204 {name: "warm probe is reused", warm: 2, want: true},
205 {name: "warm probe has idle deadline", warm: 2, idle: time.Minute, want: true},
206 {name: "warm disabled refuses probe", warm: 0},
207 } {
208 t.Run(tt.name, func(t *testing.T) {
209 h := newWarmHarness(t, tt.warm, tt.idle)
210 physical := &fakePhysical{endpoint: ports.BrokerDialTarget{Policy: poolPolicy(), ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "alpha", Bound: true}}, done: make(chan struct{})}
211 require.Equal(t, tt.want, h.pool.AdoptPhysical(physical, poolPolicy()))
212 if !tt.want {
213 require.NoError(t, physical.Close())
214 return
215 }
216 shared, ok := h.pool.SharedPhysical("alpha", poolPolicy())
217 require.True(t, ok)
218 require.Same(t, physical, shared)
219 h.visit(t, "alpha")
220 require.Empty(t, h.physicals("alpha"), "attach must not redial after a first probe")
221 require.False(t, h.pool.AdoptPhysical(&fakePhysical{endpoint: physical.endpoint, done: make(chan struct{})}, poolPolicy()), "duplicate probe cannot replace the attachment")
222 if tt.idle > 0 {
223 h.reap(t, physical, tt.idle)
224 } else {
225 require.NoError(t, h.pool.Close())
226 }
227 require.True(t, closed(physical))
228 })
229 }
230}
231
232func TestPoolCloseRetiresWarmTransports(t *testing.T) {
233 h := newWarmHarness(t, 2, 0)
234 h.visit(t, "alpha")
235 h.visit(t, "beta")
236 require.NoError(t, h.pool.Close())
237 for _, identity := range []ports.BrokerDaemonIdentity{"alpha", "beta"} {
238 require.True(t, closed(h.physicals(identity)[0]), identity)
239 }
240 _, ok := h.pool.SharedPhysical("alpha", poolPolicy())
241 require.False(t, ok)
242}
243
244func TestPoolLocalOpenDoesNotWaitForProbeAdoption(t *testing.T) {
245 h := newWarmHarness(t, 2, time.Minute)
246 h.visit(t, "local-daemon")
247
248 probe := &fakePhysical{
249 endpoint: ports.BrokerDialTarget{Policy: poolPolicy()},
250 identity: "remote-daemon",
251 done: make(chan struct{}),
252 }
253 start := make(chan struct{})
254 adopted := make(chan bool, 1)
255 opened := make(chan error, 1)
256 go func() {
257 <-start
258 adopted <- h.pool.AdoptPhysical(probe, poolPolicy())
259 }()
260 go func() {
261 <-start
262 h.mu.Lock()
263 h.target = "local-daemon"
264 h.stream++
265 stream := h.stream
266 h.mu.Unlock()
267 logical, err := h.pool.OpenStream(context.Background(), poolRequest(h.client, stream))
268 if err == nil {
269 err = logical.Close()
270 }
271 opened <- err
272 }()
273 close(start)
274
275 select {
276 case err := <-opened:
277 require.NoError(t, err, "local OpenStream must progress while another goroutine adopts a remote probe transport")
278 case <-time.After(3 * time.Second):
279 t.Fatal("local OpenStream blocked while probe adoption was in progress")
280 }
281 select {
282 case ok := <-adopted:
283 require.True(t, ok, "the authenticated remote physical should be adopted")
284 case <-time.After(3 * time.Second):
285 t.Fatal("probe transport adoption did not complete")
286 }
287}
288
289func TestPoolSharedPhysical(t *testing.T) {
290 tests := []struct {
291 name string
292 identity ports.BrokerDaemonIdentity
293 policy func() ports.BrokerPolicy
294 want bool
295 }{
296 {name: "warm exact key", identity: "alpha", policy: poolPolicy, want: true},
297 {name: "unknown identity", identity: "beta", policy: poolPolicy},
298 {name: "different policy", identity: "alpha", policy: func() ports.BrokerPolicy { p := poolPolicy(); p.Trust = "other"; return p }},
299 }
300 for _, tt := range tests {
301 t.Run(tt.name, func(t *testing.T) {
302 h := newWarmHarness(t, 1, 0)
303 h.visit(t, "alpha")
304 physical, ok := h.pool.SharedPhysical(tt.identity, tt.policy())
305 require.Equal(t, tt.want, ok)
306 if tt.want {
307 require.Same(t, h.physicals("alpha")[0], physical)
308 // Borrowing neither dials nor refreshes warm order: beta still
309 // evicts alpha.
310 h.visit(t, "beta")
311 await(t, h.physicals("alpha")[0].done)
312 require.Len(t, h.physicals("alpha"), 1)
313 }
314 })
315 }
316}
317
318func TestNewPoolWarmLimits(t *testing.T) {
319 tests := []struct {
320 name string
321 warm int
322 idle time.Duration
323 wantErr bool
324 }{
325 {name: "defaults off", warm: 0, idle: 0},
326 {name: "bounded with grace", warm: 8, idle: time.Minute},
327 {name: "negative warm", warm: -1, wantErr: true},
328 {name: "negative idle", warm: 1, idle: -time.Second, wantErr: true},
329 }
330 for _, tt := range tests {
331 t.Run(tt.name, func(t *testing.T) {
332 pool, err := NewPool(1, portsmocks.NewMockBrokerRouteAuthority(t), portsmocks.NewMockBrokerIdentityBinder(t), portsmocks.NewMockBrokerEndpointConnector(t), newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 1, Streams: 1, StreamsPerClient: 1, Warm: tt.warm, Idle: tt.idle})
333 if tt.wantErr {
334 require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
335 return
336 }
337 require.NoError(t, err)
338 require.NoError(t, pool.Close())
339 })
340 }
341}