1package broker
2
3import (
4 "context"
5 "errors"
6 "sync"
7 "testing"
8 "time"
9
10 "github.com/bnema/vev/internal/ports"
11 "github.com/stretchr/testify/require"
12)
13
14// newTestSupervisor builds a supervisor over a manual clock and always drains it
15// at test end so an unexpired run goroutine can never leak into another test.
16func newTestSupervisor(t testing.TB, grace time.Duration) (*Supervisor, *manualClock) {
17 t.Helper()
18 clock := newManualClock(time.Unix(0, 0))
19 s, err := NewSupervisor(clock, SupervisorConfig{IdleGrace: grace})
20 require.NoError(t, err)
21 // Ignore the error: a test may deliberately register a failing drain hook.
22 t.Cleanup(func() { _ = s.Close() })
23 return s, clock
24}
25
26// activeTimers reports the manual clock's live (armed, not yet fired) timers.
27func (c *manualClock) activeTimers() int {
28 c.mu.Lock()
29 defer c.mu.Unlock()
30 return len(c.timers)
31}
32
33func supervisorState(s *Supervisor) (clients, operations int, armed, closed bool) {
34 s.mu.Lock()
35 defer s.mu.Unlock()
36 return s.clients, s.operations, s.timerArmed, s.closed
37}
38
39func supervisorTimerArmed(s *Supervisor) bool {
40 _, _, armed, _ := supervisorState(s)
41 return armed
42}
43
44func requireSupervisorOpen(t *testing.T, s *Supervisor) {
45 t.Helper()
46 select {
47 case <-s.Done():
48 t.Fatal("supervisor shut down unexpectedly")
49 default:
50 }
51}
52
53func requireBrokerClosed(t *testing.T, err error) {
54 t.Helper()
55 require.Error(t, err)
56 var admission ports.BrokerAdmissionError
57 require.ErrorAs(t, err, &admission)
58 require.Equal(t, ports.BrokerAdmissionClosed, admission)
59}
60
61func requireBrokerInvalid(t *testing.T, err error) {
62 t.Helper()
63 require.Error(t, err)
64 var admission ports.BrokerAdmissionError
65 require.ErrorAs(t, err, &admission)
66 require.Equal(t, ports.BrokerAdmissionInvalid, admission)
67}
68
69type closeFunc func() error
70
71func (f closeFunc) Close() error { return f() }
72
73// blockingRunner stands in for a broker-owned background component such as the
74// observation Registry: it runs until cancellation and reports what it saw.
75type blockingRunner struct {
76 started chan struct{}
77 observed chan error
78}
79
80func newBlockingRunner() *blockingRunner {
81 return &blockingRunner{started: make(chan struct{}), observed: make(chan error, 1)}
82}
83
84func (r *blockingRunner) Run(ctx context.Context) {
85 close(r.started)
86 <-ctx.Done()
87 r.observed <- ctx.Err()
88}
89
90func TestSupervisorInitialIdleExpires(t *testing.T) {
91 s, clock := newTestSupervisor(t, time.Minute)
92 _, _, armed, closed := supervisorState(s)
93 require.True(t, armed, "the idle timer must be armed at construction")
94 require.False(t, closed)
95
96 clock.Advance(time.Minute - time.Nanosecond)
97 requireSupervisorOpen(t, s)
98
99 clock.Advance(time.Nanosecond)
100 await(t, s.Done())
101 require.ErrorIs(t, s.RootContext().Err(), context.Canceled)
102 require.False(t, supervisorTimerArmed(s))
103 require.Equal(t, 0, clock.activeTimers())
104
105 _, err := s.AdmitClient()
106 requireBrokerClosed(t, err)
107 _, err = s.AdmitOperation()
108 requireBrokerClosed(t, err)
109 requireBrokerClosed(t, s.Register("late", func(context.Context) error { return nil }))
110 requireBrokerClosed(t, s.RegisterRunner("late-runner", newBlockingRunner()))
111 requireBrokerClosed(t, s.RegisterCloseable("late-close", closeFunc(func() error { return nil })))
112}
113
114func TestSupervisorAdmissionBeforeDeadlineCancelsIdleShutdown(t *testing.T) {
115 s, clock := newTestSupervisor(t, time.Minute)
116 clock.Advance(30 * time.Second)
117
118 lease, err := s.AdmitClient()
119 require.NoError(t, err)
120 require.False(t, supervisorTimerArmed(s))
121
122 // Past the original deadline the broker is still open because the admission
123 // cancelled the pending idle shutdown.
124 clock.Advance(time.Hour)
125 requireSupervisorOpen(t, s)
126
127 lease.Release()
128 lease.Release() // idempotent
129 require.True(t, supervisorTimerArmed(s))
130
131 clock.Advance(time.Minute - time.Nanosecond)
132 requireSupervisorOpen(t, s)
133 clock.Advance(time.Nanosecond)
134 await(t, s.Done())
135}
136
137func TestSupervisorClientLeaseProtectsIdle(t *testing.T) {
138 s, clock := newTestSupervisor(t, time.Minute)
139 lease, err := s.AdmitClient()
140 require.NoError(t, err)
141
142 for i := 0; i < 3; i++ {
143 clock.Advance(time.Minute)
144 requireSupervisorOpen(t, s)
145 require.False(t, supervisorTimerArmed(s), "the timer must be inactive while a client lease is held")
146 }
147
148 lease.Release()
149 require.True(t, supervisorTimerArmed(s))
150 clock.Advance(time.Minute)
151 await(t, s.Done())
152}
153
154func TestSupervisorOperationLeaseProtectsIdle(t *testing.T) {
155 s, clock := newTestSupervisor(t, time.Minute)
156 lease, err := s.AdmitOperation()
157 require.NoError(t, err)
158
159 clock.Advance(10 * time.Minute)
160 requireSupervisorOpen(t, s)
161
162 lease.Release()
163 clock.Advance(time.Minute)
164 await(t, s.Done())
165}
166
167func TestSupervisorIdleTimerTracksBothLeaseKinds(t *testing.T) {
168 s, clock := newTestSupervisor(t, time.Minute)
169 require.True(t, supervisorTimerArmed(s))
170
171 client, err := s.AdmitClient()
172 require.NoError(t, err)
173 require.False(t, supervisorTimerArmed(s))
174
175 operation, err := s.AdmitOperation()
176 require.NoError(t, err)
177 require.False(t, supervisorTimerArmed(s))
178
179 // Releasing one kind leaves the other; the timer stays inactive until both
180 // counts are zero.
181 client.Release()
182 clients, operations, armed, closed := supervisorState(s)
183 require.Equal(t, 0, clients)
184 require.Equal(t, 1, operations)
185 require.False(t, armed)
186 require.False(t, closed)
187
188 operation.Release()
189 require.True(t, supervisorTimerArmed(s))
190
191 clock.Advance(time.Minute)
192 await(t, s.Done())
193}
194
195// TestSupervisorSupersededArmDoesNotCommit pins the guarantee behind the
196// concurrent race test: a fire from an arm that an admission/release pair has
197// superseded must be ignored so the broker still gets a full fresh grace.
198func TestSupervisorSupersededArmDoesNotCommit(t *testing.T) {
199 s, clock := newTestSupervisor(t, time.Minute)
200
201 s.mu.Lock()
202 stale := s.arm
203 s.mu.Unlock()
204
205 lease, err := s.AdmitClient()
206 require.NoError(t, err)
207 lease.Release()
208 require.True(t, supervisorTimerArmed(s))
209 require.False(t, s.tryCommitIdleExpiry(stale), "a fire from a superseded arm must not commit")
210 requireSupervisorOpen(t, s)
211
212 clock.Advance(time.Minute - time.Nanosecond)
213 requireSupervisorOpen(t, s)
214 clock.Advance(time.Nanosecond)
215 await(t, s.Done())
216}
217
218// TestSupervisorCurrentArmCommits pins the positive half: a fire from the arm the
219// run goroutine is still observing commits while the broker is idle.
220func TestSupervisorCurrentArmCommits(t *testing.T) {
221 s, _ := newTestSupervisor(t, time.Minute)
222 s.mu.Lock()
223 current := s.arm
224 s.mu.Unlock()
225 require.True(t, s.tryCommitIdleExpiry(current))
226 await(t, s.Done())
227}
228
229func TestSupervisorRepeatedLeaseCyclesStillExpire(t *testing.T) {
230 s, clock := newTestSupervisor(t, time.Minute)
231 for i := 0; i < 50; i++ {
232 lease, err := s.AdmitClient()
233 require.NoError(t, err)
234 lease.Release()
235 }
236
237 require.True(t, supervisorTimerArmed(s))
238 require.Equal(t, 1, clock.activeTimers(), "idle cycles must not accumulate timers")
239 clock.Advance(time.Minute)
240 await(t, s.Done())
241 require.Equal(t, 0, clock.activeTimers())
242}
243
244// TestSupervisorConcurrentExpiryAndAdmission races the idle expiry against a
245// burst of admissions. Because commitment and admission share the mutex, the
246// outcome is all-or-nothing: either the expiry committed first and every
247// admission is rejected with the typed broker-closed error, or an admission won
248// first and every admission is accepted and the broker stays open while a lease
249// is held.
250func TestSupervisorConcurrentExpiryAndAdmission(t *testing.T) {
251 const (
252 admitters = 8
253 rounds = 50
254 )
255 for round := 0; round < rounds; round++ {
256 s, clock := newTestSupervisor(t, time.Minute)
257
258 var mu sync.Mutex
259 var accepted []*Lease
260 var rejected []error
261 start := make(chan struct{})
262 var wg sync.WaitGroup
263 for i := 0; i < admitters; i++ {
264 wg.Add(1)
265 go func() {
266 defer wg.Done()
267 <-start
268 lease, err := s.AdmitClient()
269 mu.Lock()
270 if err != nil {
271 rejected = append(rejected, err)
272 } else {
273 accepted = append(accepted, lease)
274 }
275 mu.Unlock()
276 }()
277 }
278
279 clock.Advance(time.Minute) // fire the idle timer concurrently with admissions
280 close(start)
281 wg.Wait()
282
283 if len(accepted) > 0 {
284 require.Empty(t, rejected, "an accepted admission must prevent every later rejection")
285 requireSupervisorOpen(t, s)
286 } else {
287 require.Len(t, rejected, admitters)
288 for _, err := range rejected {
289 requireBrokerClosed(t, err)
290 }
291 await(t, s.Done())
292 }
293
294 for _, lease := range accepted {
295 lease.Release()
296 }
297 if len(accepted) > 0 {
298 clock.Advance(time.Minute)
299 await(t, s.Done())
300 }
301 require.Equal(t, 0, clock.activeTimers())
302 require.NoError(t, s.Close())
303 }
304}
305
306func TestSupervisorBlockedRunnerObservesCancellation(t *testing.T) {
307 s, clock := newTestSupervisor(t, time.Minute)
308 runner := newBlockingRunner()
309 require.NoError(t, s.RegisterRunner("registry", runner))
310 <-runner.started
311
312 clock.Advance(time.Minute)
313 select {
314 case observed := <-runner.observed:
315 require.Equal(t, context.Canceled, observed)
316 case <-time.After(3 * time.Second):
317 t.Fatal("runner never observed cancellation")
318 }
319 await(t, s.Done())
320 require.ErrorIs(t, s.RootContext().Err(), context.Canceled)
321}
322
323func TestSupervisorBlockedHookObservesCancellation(t *testing.T) {
324 s, clock := newTestSupervisor(t, time.Minute)
325 observed := make(chan error, 1)
326 require.NoError(t, s.Register("blocked", func(ctx context.Context) error {
327 <-ctx.Done()
328 observed <- ctx.Err()
329 return nil
330 }))
331
332 clock.Advance(time.Minute)
333 select {
334 case got := <-observed:
335 require.Equal(t, context.Canceled, got)
336 case <-time.After(3 * time.Second):
337 t.Fatal("hook never observed cancellation")
338 }
339 await(t, s.Done())
340}
341
342func TestSupervisorBackgroundComponentsDoNotPinIdle(t *testing.T) {
343 s, clock := newTestSupervisor(t, time.Minute)
344 runner := newBlockingRunner()
345 poolDrained := make(chan struct{})
346 require.NoError(t, s.RegisterRunner("registry", runner))
347 require.NoError(t, s.RegisterCloseable("pool", closeFunc(func() error {
348 close(poolDrained)
349 return nil
350 })))
351 <-runner.started
352
353 // Registering background components takes no lease: the timer stays armed
354 // and the broker still expires at the plain deadline.
355 require.True(t, supervisorTimerArmed(s))
356 clock.Advance(time.Minute)
357 await(t, s.Done())
358 await(t, poolDrained)
359 require.Equal(t, context.Canceled, <-runner.observed)
360}
361
362func TestSupervisorDrainsHooksInReverseRegistrationOrder(t *testing.T) {
363 s, clock := newTestSupervisor(t, time.Minute)
364 var mu sync.Mutex
365 var order []string
366 for _, name := range []string{"pool", "registry", "store"} {
367 name := name
368 require.NoError(t, s.Register(name, func(context.Context) error {
369 mu.Lock()
370 order = append(order, name)
371 mu.Unlock()
372 return nil
373 }))
374 }
375
376 clock.Advance(time.Minute)
377 await(t, s.Done())
378
379 mu.Lock()
380 defer mu.Unlock()
381 require.Equal(t, []string{"store", "registry", "pool"}, order)
382}
383
384func TestSupervisorCloseIsIdempotentAndForceCloses(t *testing.T) {
385 s, _ := newTestSupervisor(t, time.Hour)
386 lease, err := s.AdmitClient()
387 require.NoError(t, err)
388
389 const closers = 8
390 errs := make(chan error, closers)
391 var wg sync.WaitGroup
392 for i := 0; i < closers; i++ {
393 wg.Add(1)
394 go func() {
395 defer wg.Done()
396 errs <- s.Close()
397 }()
398 }
399 wg.Wait()
400 for i := 0; i < closers; i++ {
401 require.NoError(t, <-errs)
402 }
403
404 await(t, s.Done())
405 require.ErrorIs(t, s.RootContext().Err(), context.Canceled)
406 require.NoError(t, s.Close())
407
408 // A held lease that drains after the forced close must neither re-arm the
409 // timer nor change the terminal state.
410 lease.Release()
411 lease.Release()
412 clients, operations, armed, closed := supervisorState(s)
413 require.Equal(t, 0, clients)
414 require.Equal(t, 0, operations)
415 require.False(t, armed)
416 require.True(t, closed)
417}
418
419func TestSupervisorCloseReportsHookErrorsOnly(t *testing.T) {
420 s, _ := newTestSupervisor(t, time.Hour)
421 boom := errors.New("pool close failed")
422 require.NoError(t, s.RegisterCloseable("pool", closeFunc(func() error { return boom })))
423 require.NoError(t, s.Register("ok", func(context.Context) error { return nil }))
424
425 err := s.Close()
426 require.Error(t, err)
427 require.ErrorIs(t, err, boom)
428 // Repeated Close observes the exact same terminal error.
429 require.ErrorIs(t, s.Close(), boom)
430}
431
432func TestSupervisorRegistrationValidation(t *testing.T) {
433 s, _ := newTestSupervisor(t, time.Hour)
434 requireBrokerInvalid(t, s.Register("", func(context.Context) error { return nil }))
435 requireBrokerInvalid(t, s.Register("nil-hook", nil))
436 requireBrokerInvalid(t, s.RegisterRunner("nil-runner", nil))
437 requireBrokerInvalid(t, s.RegisterCloseable("nil-closeable", nil))
438 require.NoError(t, s.Register("hook", func(context.Context) error { return nil }))
439}
440
441// TestSupervisorIdleExpiryLeavesNoTimer pins that an idle expiry shuts the
442// supervisor down without leaving an armed timer behind.
443func TestSupervisorIdleExpiryLeavesNoTimer(t *testing.T) {
444 s, clock := newTestSupervisor(t, time.Minute)
445
446 clock.Advance(time.Minute)
447 await(t, s.Done())
448 await(t, s.stopped)
449
450 require.Equal(t, 0, clock.activeTimers())
451}
452
453// TestSupervisorCloseLeavesNoTimer pins that Close disarms the idle timer and
454// still reports shutdown through Done and stopped.
455func TestSupervisorCloseLeavesNoTimer(t *testing.T) {
456 s, clock := newTestSupervisor(t, time.Minute)
457
458 require.NoError(t, s.Close())
459 await(t, s.Done())
460 await(t, s.stopped)
461
462 require.Equal(t, 0, clock.activeTimers())
463}
464
465// TestSupervisorCloseRacingIdleExpiryDrainsOnce pins the atomic commit: whether
466// Close or an idle fire wins, the drain hook runs exactly once and every caller
467// observes the same terminal error.
468func TestSupervisorCloseRacingIdleExpiryDrainsOnce(t *testing.T) {
469 boom := errors.New("pool drain failed")
470 for range 25 {
471 s, clock := newTestSupervisor(t, time.Minute)
472
473 var mu sync.Mutex
474 drains := 0
475 require.NoError(t, s.RegisterCloseable("pool", closeFunc(func() error {
476 mu.Lock()
477 drains++
478 mu.Unlock()
479 return boom
480 })))
481
482 start := make(chan struct{})
483 closed := make(chan error, 1)
484 var wg sync.WaitGroup
485 wg.Add(2)
486 go func() {
487 defer wg.Done()
488 <-start
489 clock.Advance(time.Minute)
490 }()
491 go func() {
492 defer wg.Done()
493 <-start
494 closed <- s.Close()
495 }()
496 close(start)
497 wg.Wait()
498 await(t, s.Done())
499 await(t, s.stopped)
500
501 mu.Lock()
502 got := drains
503 mu.Unlock()
504 require.Equal(t, 1, got, "the drain hook must run exactly once")
505 require.ErrorIs(t, <-closed, boom)
506 require.ErrorIs(t, s.Close(), boom, "every Close caller observes the same terminal error")
507 require.Equal(t, 0, clock.activeTimers())
508 }
509}
510
511// TestSupervisorFailingHookDoesNotAbortRemainingDrains pins that a hook error is
512// collected rather than fatal: the remaining hooks still drain in reverse
513// registration order and the joined error surfaces through Close.
514func TestSupervisorFailingHookDoesNotAbortRemainingDrains(t *testing.T) {
515 s, clock := newTestSupervisor(t, time.Minute)
516 boom := errors.New("registry drain failed")
517 var mu sync.Mutex
518 var order []string
519 register := func(name string, err error) {
520 t.Helper()
521 require.NoError(t, s.Register(name, func(context.Context) error {
522 mu.Lock()
523 order = append(order, name)
524 mu.Unlock()
525 return err
526 }))
527 }
528 register("pool", nil)
529 register("registry", boom)
530 register("store", nil)
531
532 clock.Advance(time.Minute)
533 await(t, s.Done())
534 await(t, s.stopped)
535
536 mu.Lock()
537 defer mu.Unlock()
538 require.Equal(t, []string{"store", "registry", "pool"}, order, "a failing hook must not skip later drains")
539 require.ErrorIs(t, s.Close(), boom)
540}