feat(broker): add idle lifecycle supervisor

commit: e8cd8f6034413cb47a0e2661767303d32cf56fa5
author: brice · 36d ago · 2026-09-16 12:40 UTC
parent: 75fae3a
signature: GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email

2 files · +948 -0

A internal/usecase/broker/lifecycle.go +408 -0

@@ -0,0 +1,408 @@
	1	+	package broker
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"fmt"
	7	+	"sync"
	8	+	"time"
	9	+
	10	+	"github.com/bnema/vev/internal/ports"
	11	+	)
	12	+
	13	+	// Lifecycle supervisor (Plan 001 P2.4, not activated).
	14	+	//
	15	+	// The supervisor is the single owner of the broker's idle lifetime. Exactly two
	16	+	// kinds of work can pin the broker open:
	17	+	//
	18	+	// - an admitted client lease: one accepted broker client connection, and
	19	+	// - an in-flight user operation lease: one operation a user asked the broker
	20	+	// to perform while it is admitted (for example opening a stream or editing
	21	+	// the configured-host set).
	22	+	//
	23	+	// Nothing else pins. Work the broker schedules for itself -- observation
	24	+	// retries, durable snapshot writes, opening the store, or keeping physical
	25	+	// transports warm in the idle pool -- is deliberately not represented here and
	26	+	// can never extend the broker's life: a broker with no admitted client and no
	27	+	// in-flight user operation shuts down on schedule even while those components
	28	+	// run. Registered background components therefore receive the root context and
	29	+	// an ordered drain hook, but never a lease.
	30	+	//
	31	+	// Idle shutdown is committed atomically. The idle timer is active exactly when
	32	+	// both lease counts are zero. A lease admitted before the commit stops the timer
	33	+	// and keeps the broker open; once the commit wins, every later admission is
	34	+	// refused with the typed, non-terminal ports.BrokerAdmissionClosed so the caller
	35	+	// reconnects to a fresh broker instead of treating the exit as fatal.
	36	+	//
	37	+	// Arming installs a fresh timer and claims a new arm number, and admission
	38	+	// disarms it under the same mutex. The run goroutine therefore commits only for
	39	+	// the arm it is still observing: a fire from a superseded arm can never shut the
	40	+	// broker down out of turn, so an admission that wins the race always leaves the
	41	+	// broker open for a full grace.
	42	+	//
	43	+	// Shutdown itself runs in a fixed order: the supervisor marks itself closed and
	44	+	// stops/drains the idle timer under its mutex, cancels the root context, then
	45	+	// drains every registered hook outside the mutex in reverse registration order
	46	+	// (so a component registered later, which may depend on an earlier one, is
	47	+	// drained first). The supervisor joins every hook before Done closes. Close runs
	48	+	// the same sequence and is idempotent and concurrent-safe.
	49	+
	50	+	// DefaultIdleGrace is the idle grace granted before a broker with no admitted
	51	+	// client lease and no in-flight user operation lease shuts itself down.
	52	+	const DefaultIdleGrace = 5 * time.Minute
	53	+
	54	+	// SupervisorConfig configures a Supervisor. The zero value is valid and selects
	55	+	// DefaultIdleGrace.
	56	+	type SupervisorConfig struct {
	57	+	// IdleGrace overrides DefaultIdleGrace when positive.
	58	+	IdleGrace time.Duration
	59	+	}
	60	+
	61	+	// ShutdownHook drains one registered broker-owned component during shutdown.
	62	+	// It runs after the root context is cancelled and must return once the
	63	+	// component has stopped; the supervisor joins every hook before Done closes.
	64	+	// Hooks run outside the supervisor mutex, in reverse registration order. A hook
	65	+	// must not call Close, which would wait for the very shutdown running it.
	66	+	type ShutdownHook func(ctx context.Context) error
	67	+
	68	+	// Runner is the minimal seam for a broker-owned background component that runs
	69	+	// until its context is cancelled (for example the observation Registry). Run
	70	+	// must return once ctx is cancelled so the supervisor can join it.
	71	+	type Runner interface {
	72	+	Run(ctx context.Context)
	73	+	}
	74	+
	75	+	// Closeable is the minimal seam for a broker-owned resource closed at shutdown
	76	+	// (for example the physical-transport Pool). The supervisor never inspects pool
	77	+	// or registry internals; it only asks them to drain.
	78	+	type Closeable interface {
	79	+	Close() error
	80	+	}
	81	+
	82	+	// Lease is an idempotent admission handle. Release may be called any number of
	83	+	// times, from any goroutine, and after the supervisor has shut down; only the
	84	+	// first call releases the lease.
	85	+	type Lease struct {
	86	+	once sync.Once
	87	+	release func()
	88	+	}
	89	+
	90	+	// Release returns the lease exactly once. A nil Lease is a no-op.
	91	+	func (l *Lease) Release() {
	92	+	if l == nil \|\| l.release == nil {
	93	+	return
	94	+	}
	95	+	l.once.Do(l.release)
	96	+	}
	97	+
	98	+	type leaseKind uint8
	99	+
	100	+	const (
	101	+	leaseClient leaseKind = iota + 1
	102	+	leaseOperation
	103	+	)
	104	+
	105	+	type registeredHook struct {
	106	+	name string
	107	+	hook ShutdownHook
	108	+	}
	109	+
	110	+	// Supervisor owns the broker idle lifetime, the root context handed to
	111	+	// background components, and the ordered shutdown drain.
	112	+	type Supervisor struct {
	113	+	clock ports.Clock
	114	+	idle time.Duration
	115	+
	116	+	root context.Context
	117	+	cancel context.CancelFunc
	118	+
	119	+	// ready is closed once the idle timer is armed and the run goroutine is
	120	+	// accepting lifecycle signals. done is closed only after the root context is
	121	+	// cancelled and every hook has drained. stopped is closed as the run
	122	+	// goroutine exits. closing wakes the run goroutine on the committed shutdown,
	123	+	// and wake nudges it whenever a lease changes the timer state.
	124	+	ready chan struct{}
	125	+	done chan struct{}
	126	+	stopped chan struct{}
	127	+	closing chan struct{}
	128	+	wake chan struct{}
	129	+
	130	+	mu sync.Mutex
	131	+	closed bool
	132	+	committed bool
	133	+	clients int
	134	+	operations int
	135	+	timer ports.Timer
	136	+	timerArmed bool
	137	+	arm uint64
	138	+	hooks []registeredHook
	139	+	shutdownErr error
	140	+	}
	141	+
	142	+	// NewSupervisor starts the idle lifetime. The idle timer is armed immediately:
	143	+	// a broker that serves nothing shuts down one idle grace after construction.
	144	+	func NewSupervisor(clock ports.Clock, cfg SupervisorConfig) (*Supervisor, error) {
	145	+	if nilDependency(clock) {
	146	+	return nil, errors.New("broker: supervisor requires a clock")
	147	+	}
	148	+	grace := cfg.IdleGrace
	149	+	if grace <= 0 {
	150	+	grace = DefaultIdleGrace
	151	+	}
	152	+	root, cancel := context.WithCancel(context.Background())
	153	+	s := &Supervisor{
	154	+	clock: clock, idle: grace, root: root, cancel: cancel,
	155	+	ready: make(chan struct{}), done: make(chan struct{}),
	156	+	stopped: make(chan struct{}), closing: make(chan struct{}),
	157	+	wake: make(chan struct{}, 1),
	158	+	}
	159	+	go s.run()
	160	+	<-s.ready
	161	+	return s, nil
	162	+	}
	163	+
	164	+	// RootContext returns the supervisor-owned root context. Background components
	165	+	// must derive their work from it and must stop when it is cancelled.
	166	+	func (s *Supervisor) RootContext() context.Context { return s.root }
	167	+
	168	+	// Done is closed once the root context is cancelled and every registered hook
	169	+	// has been drained and joined.
	170	+	func (s *Supervisor) Done() <-chan struct{} { return s.done }
	171	+
	172	+	// Register admits one background component with an ordered drain hook. The
	173	+	// component must derive its work from RootContext and must stop when that
	174	+	// context is cancelled; hook must return once it has. Registration never grants
	175	+	// a lease, so a registered component cannot pin the broker open. Register is
	176	+	// refused with ports.BrokerAdmissionClosed after shutdown commits and with
	177	+	// ports.BrokerAdmissionInvalid for an empty name or nil hook.
	178	+	func (s *Supervisor) Register(name string, hook ShutdownHook) error {
	179	+	if name == "" \|\| hook == nil {
	180	+	return ports.BrokerAdmissionInvalid
	181	+	}
	182	+	s.mu.Lock()
	183	+	defer s.mu.Unlock()
	184	+	if s.closed {
	185	+	return ports.BrokerAdmissionClosed
	186	+	}
	187	+	s.hooks = append(s.hooks, registeredHook{name: name, hook: hook})
	188	+	return nil
	189	+	}
	190	+
	191	+	// RegisterRunner starts a broker-owned background component under the root
	192	+	// context and joins it at shutdown: its drain hook waits for Run to return. The
	193	+	// component must therefore honor cancellation promptly. Registration is refused
	194	+	// exactly as Register is.
	195	+	func (s *Supervisor) RegisterRunner(name string, runner Runner) error {
	196	+	if nilDependency(runner) {
	197	+	return ports.BrokerAdmissionInvalid
	198	+	}
	199	+	drained := make(chan struct{})
	200	+	if err := s.Register(name, func(context.Context) error {
	201	+	<-drained
	202	+	return nil
	203	+	}); err != nil {
	204	+	return err
	205	+	}
	206	+	ctx := s.root
	207	+	go func() {
	208	+	defer close(drained)
	209	+	runner.Run(ctx)
	210	+	}()
	211	+	return nil
	212	+	}
	213	+
	214	+	// RegisterCloseable registers a broker-owned resource to be closed after the
	215	+	// root context is cancelled. Its Close error is reported by Close and never
	216	+	// aborts the remaining drains. Registration is refused exactly as Register is.
	217	+	func (s *Supervisor) RegisterCloseable(name string, closeable Closeable) error {
	218	+	if nilDependency(closeable) {
	219	+	return ports.BrokerAdmissionInvalid
	220	+	}
	221	+	return s.Register(name, func(context.Context) error { return closeable.Close() })
	222	+	}
	223	+
	224	+	// AdmitClient admits one client-connection lease. It returns
	225	+	// ports.BrokerAdmissionClosed once shutdown has committed.
	226	+	func (s Supervisor) AdmitClient() (Lease, error) { return s.admit(leaseClient) }
	227	+
	228	+	// AdmitOperation admits one in-flight user-operation lease. It returns
	229	+	// ports.BrokerAdmissionClosed once shutdown has committed.
	230	+	func (s Supervisor) AdmitOperation() (Lease, error) { return s.admit(leaseOperation) }
	231	+
	232	+	// Close commits shutdown and waits for the root context to be cancelled and
	233	+	// every hook to drain. It is idempotent and concurrent-safe: every caller
	234	+	// observes the same terminal state and the same joined hook error.
	235	+	func (s *Supervisor) Close() error {
	236	+	s.mu.Lock()
	237	+	s.commitLocked()
	238	+	s.mu.Unlock()
	239	+	<-s.stopped
	240	+	s.mu.Lock()
	241	+	defer s.mu.Unlock()
	242	+	return s.shutdownErr
	243	+	}
	244	+
	245	+	func (s Supervisor) admit(kind leaseKind) (Lease, error) {
	246	+	s.mu.Lock()
	247	+	if s.closed {
	248	+	s.mu.Unlock()
	249	+	return nil, ports.BrokerAdmissionClosed
	250	+	}
	251	+	switch kind {
	252	+	case leaseClient:
	253	+	s.clients++
	254	+	case leaseOperation:
	255	+	s.operations++
	256	+	}
	257	+	// An admission that wins the commit race disarms the pending idle shutdown.
	258	+	s.disarmTimerLocked()
	259	+	s.mu.Unlock()
	260	+	s.signal()
	261	+	return &Lease{release: func() { s.release(kind) }}, nil
	262	+	}
	263	+
	264	+	func (s *Supervisor) release(kind leaseKind) {
	265	+	s.mu.Lock()
	266	+	switch kind {
	267	+	case leaseClient:
	268	+	if s.clients > 0 {
	269	+	s.clients--
	270	+	}
	271	+	case leaseOperation:
	272	+	if s.operations > 0 {
	273	+	s.operations--
	274	+	}
	275	+	}
	276	+	if !s.closed && s.clients == 0 && s.operations == 0 {
	277	+	s.armTimerLocked()
	278	+	}
	279	+	s.mu.Unlock()
	280	+	s.signal()
	281	+	}
	282	+
	283	+	// signal nudges the run goroutine to re-read the timer state. Signals coalesce.
	284	+	func (s *Supervisor) signal() {
	285	+	select {
	286	+	case s.wake <- struct{}{}:
	287	+	default:
	288	+	}
	289	+	}
	290	+
	291	+	// run owns the idle timer and is the sole shutdown executor. It waits on the
	292	+	// current arm's timer and finalizes exactly once, either when an un-superseded
	293	+	// arm fires while the broker is still idle or when Close commits the shutdown.
	294	+	func (s *Supervisor) run() {
	295	+	defer close(s.stopped)
	296	+
	297	+	s.mu.Lock()
	298	+	s.armTimerLocked()
	299	+	s.mu.Unlock()
	300	+	close(s.ready)
	301	+
	302	+	for {
	303	+	s.mu.Lock()
	304	+	timer, arm := s.timer, s.arm
	305	+	s.mu.Unlock()
	306	+
	307	+	var fire <-chan time.Time
	308	+	if timer != nil {
	309	+	fire = timer.C()
	310	+	}
	311	+	select {
	312	+	case <-s.closing:
	313	+	s.finalize()
	314	+	return
	315	+	case <-s.wake:
	316	+	continue
	317	+	case <-fire:
	318	+	if s.tryCommitIdleExpiry(arm) {
	319	+	s.finalize()
	320	+	return
	321	+	}
	322	+	}
	323	+	}
	324	+	}
	325	+
	326	+	// tryCommitIdleExpiry reports whether a fire from arm must commit the idle
	327	+	// shutdown. It returns true only for the run goroutine that must finalize: a
	328	+	// superseded or already disarmed arm, a committed supervisor, or a still-held
	329	+	// lease all make the fire a no-op.
	330	+	func (s *Supervisor) tryCommitIdleExpiry(arm uint64) bool {
	331	+	s.mu.Lock()
	332	+	defer s.mu.Unlock()
	333	+	if s.closed \|\| !s.timerArmed \|\| s.arm != arm {
	334	+	return false
	335	+	}
	336	+	if s.clients > 0 \|\| s.operations > 0 {
	337	+	return false
	338	+	}
	339	+	return s.commitLocked()
	340	+	}
	341	+
	342	+	// commitLocked marks the supervisor closed and disarms the idle timer. It
	343	+	// returns true exactly once, for the caller that committed. Because admission
	344	+	// and commitment share the mutex, an admission either disarms the pending expiry
	345	+	// before the commit or observes the closed state after it.
	346	+	func (s *Supervisor) commitLocked() bool {
	347	+	if s.committed {
	348	+	return false
	349	+	}
	350	+	s.committed = true
	351	+	s.closed = true
	352	+	s.disarmTimerLocked()
	353	+	close(s.closing)
	354	+	return true
	355	+	}
	356	+
	357	+	// finalize cancels the root context, drains every hook outside the mutex in
	358	+	// reverse registration order, and only then closes Done. It runs exactly once,
	359	+	// on the run goroutine.
	360	+	func (s *Supervisor) finalize() {
	361	+	s.cancel()
	362	+	s.mu.Lock()
	363	+	hooks := append([]registeredHook(nil), s.hooks...)
	364	+	s.mu.Unlock()
	365	+
	366	+	var errs []error
	367	+	for i := len(hooks) - 1; i >= 0; i-- {
	368	+	if err := hooks[i].hook(s.root); err != nil {
	369	+	errs = append(errs, fmt.Errorf("broker: %s shutdown: %w", hooks[i].name, err))
	370	+	}
	371	+	}
	372	+
	373	+	s.mu.Lock()
	374	+	s.shutdownErr = errors.Join(errs...)
	375	+	s.mu.Unlock()
	376	+	close(s.done)
	377	+	}
	378	+
	379	+	// armTimerLocked installs a fresh idle timer and claims a new arm number. A
	380	+	// fresh timer per idle cycle lets the run goroutine attribute a fire to the arm
	381	+	// that produced it, so a fire from a superseded arm is ignored instead of
	382	+	// shutting the broker down out of turn. Callers hold mu and have established
	383	+	// that both lease counts are zero.
	384	+	func (s *Supervisor) armTimerLocked() {
	385	+	if s.timerArmed \|\| s.committed {
	386	+	return
	387	+	}
	388	+	s.timer = s.clock.NewTimer(s.idle)
	389	+	s.timerArmed = true
	390	+	s.arm++
	391	+	}
	392	+
	393	+	// disarmTimerLocked stops and drains the idle timer, superseding its arm,
	394	+	// supporting buffered timer adapters with pre-Go-1.23 reset semantics. Callers
	395	+	// hold mu.
	396	+	func (s *Supervisor) disarmTimerLocked() {
	397	+	s.timerArmed = false
	398	+	if s.timer == nil {
	399	+	return
	400	+	}
	401	+	if !s.timer.Stop() {
	402	+	select {
	403	+	case <-s.timer.C():
	404	+	default:
	405	+	}
	406	+	}
	407	+	s.timer = nil
	408	+	}

@@ -0,0 +1,408 @@
	1	+	package broker
	2	+
	3	+	import (
	4	+	"context"
	5	+	"errors"
	6	+	"fmt"
	7	+	"sync"
	8	+	"time"
	9	+
	10	+	"github.com/bnema/vev/internal/ports"
	11	+	)
	12	+
	13	+	// Lifecycle supervisor (Plan 001 P2.4, not activated).
	14	+	//
	15	+	// The supervisor is the single owner of the broker's idle lifetime. Exactly two
	16	+	// kinds of work can pin the broker open:
	17	+	//
	18	+	// - an admitted client lease: one accepted broker client connection, and
	19	+	// - an in-flight user operation lease: one operation a user asked the broker
	20	+	// to perform while it is admitted (for example opening a stream or editing
	21	+	// the configured-host set).
	22	+	//
	23	+	// Nothing else pins. Work the broker schedules for itself -- observation
	24	+	// retries, durable snapshot writes, opening the store, or keeping physical
	25	+	// transports warm in the idle pool -- is deliberately not represented here and
	26	+	// can never extend the broker's life: a broker with no admitted client and no
	27	+	// in-flight user operation shuts down on schedule even while those components
	28	+	// run. Registered background components therefore receive the root context and
	29	+	// an ordered drain hook, but never a lease.
	30	+	//
	31	+	// Idle shutdown is committed atomically. The idle timer is active exactly when
	32	+	// both lease counts are zero. A lease admitted before the commit stops the timer
	33	+	// and keeps the broker open; once the commit wins, every later admission is
	34	+	// refused with the typed, non-terminal ports.BrokerAdmissionClosed so the caller
	35	+	// reconnects to a fresh broker instead of treating the exit as fatal.
	36	+	//
	37	+	// Arming installs a fresh timer and claims a new arm number, and admission
	38	+	// disarms it under the same mutex. The run goroutine therefore commits only for
	39	+	// the arm it is still observing: a fire from a superseded arm can never shut the
	40	+	// broker down out of turn, so an admission that wins the race always leaves the
	41	+	// broker open for a full grace.
	42	+	//
	43	+	// Shutdown itself runs in a fixed order: the supervisor marks itself closed and
	44	+	// stops/drains the idle timer under its mutex, cancels the root context, then
	45	+	// drains every registered hook outside the mutex in reverse registration order
	46	+	// (so a component registered later, which may depend on an earlier one, is
	47	+	// drained first). The supervisor joins every hook before Done closes. Close runs
	48	+	// the same sequence and is idempotent and concurrent-safe.
	49	+
	50	+	// DefaultIdleGrace is the idle grace granted before a broker with no admitted
	51	+	// client lease and no in-flight user operation lease shuts itself down.
	52	+	const DefaultIdleGrace = 5 * time.Minute
	53	+
	54	+	// SupervisorConfig configures a Supervisor. The zero value is valid and selects
	55	+	// DefaultIdleGrace.
	56	+	type SupervisorConfig struct {
	57	+	// IdleGrace overrides DefaultIdleGrace when positive.
	58	+	IdleGrace time.Duration
	59	+	}
	60	+
	61	+	// ShutdownHook drains one registered broker-owned component during shutdown.
	62	+	// It runs after the root context is cancelled and must return once the
	63	+	// component has stopped; the supervisor joins every hook before Done closes.
	64	+	// Hooks run outside the supervisor mutex, in reverse registration order. A hook
	65	+	// must not call Close, which would wait for the very shutdown running it.
	66	+	type ShutdownHook func(ctx context.Context) error
	67	+
	68	+	// Runner is the minimal seam for a broker-owned background component that runs
	69	+	// until its context is cancelled (for example the observation Registry). Run
	70	+	// must return once ctx is cancelled so the supervisor can join it.
	71	+	type Runner interface {
	72	+	Run(ctx context.Context)
	73	+	}
	74	+
	75	+	// Closeable is the minimal seam for a broker-owned resource closed at shutdown
	76	+	// (for example the physical-transport Pool). The supervisor never inspects pool
	77	+	// or registry internals; it only asks them to drain.
	78	+	type Closeable interface {
	79	+	Close() error
	80	+	}
	81	+
	82	+	// Lease is an idempotent admission handle. Release may be called any number of
	83	+	// times, from any goroutine, and after the supervisor has shut down; only the
	84	+	// first call releases the lease.
	85	+	type Lease struct {
	86	+	once sync.Once
	87	+	release func()
	88	+	}
	89	+
	90	+	// Release returns the lease exactly once. A nil Lease is a no-op.
	91	+	func (l *Lease) Release() {
	92	+	if l == nil \|\| l.release == nil {
	93	+	return
	94	+	}
	95	+	l.once.Do(l.release)
	96	+	}
	97	+
	98	+	type leaseKind uint8
	99	+
	100	+	const (
	101	+	leaseClient leaseKind = iota + 1
	102	+	leaseOperation
	103	+	)
	104	+
	105	+	type registeredHook struct {
	106	+	name string
	107	+	hook ShutdownHook
	108	+	}
	109	+
	110	+	// Supervisor owns the broker idle lifetime, the root context handed to
	111	+	// background components, and the ordered shutdown drain.
	112	+	type Supervisor struct {
	113	+	clock ports.Clock
	114	+	idle time.Duration
	115	+
	116	+	root context.Context
	117	+	cancel context.CancelFunc
	118	+
	119	+	// ready is closed once the idle timer is armed and the run goroutine is
	120	+	// accepting lifecycle signals. done is closed only after the root context is
	121	+	// cancelled and every hook has drained. stopped is closed as the run
	122	+	// goroutine exits. closing wakes the run goroutine on the committed shutdown,
	123	+	// and wake nudges it whenever a lease changes the timer state.
	124	+	ready chan struct{}
	125	+	done chan struct{}
	126	+	stopped chan struct{}
	127	+	closing chan struct{}
	128	+	wake chan struct{}
	129	+
	130	+	mu sync.Mutex
	131	+	closed bool
	132	+	committed bool
	133	+	clients int
	134	+	operations int
	135	+	timer ports.Timer
	136	+	timerArmed bool
	137	+	arm uint64
	138	+	hooks []registeredHook
	139	+	shutdownErr error
	140	+	}
	141	+
	142	+	// NewSupervisor starts the idle lifetime. The idle timer is armed immediately:
	143	+	// a broker that serves nothing shuts down one idle grace after construction.
	144	+	func NewSupervisor(clock ports.Clock, cfg SupervisorConfig) (*Supervisor, error) {
	145	+	if nilDependency(clock) {
	146	+	return nil, errors.New("broker: supervisor requires a clock")
	147	+	}
	148	+	grace := cfg.IdleGrace
	149	+	if grace <= 0 {
	150	+	grace = DefaultIdleGrace
	151	+	}
	152	+	root, cancel := context.WithCancel(context.Background())
	153	+	s := &Supervisor{
	154	+	clock: clock, idle: grace, root: root, cancel: cancel,
	155	+	ready: make(chan struct{}), done: make(chan struct{}),
	156	+	stopped: make(chan struct{}), closing: make(chan struct{}),
	157	+	wake: make(chan struct{}, 1),
	158	+	}
	159	+	go s.run()
	160	+	<-s.ready
	161	+	return s, nil
	162	+	}
	163	+
	164	+	// RootContext returns the supervisor-owned root context. Background components
	165	+	// must derive their work from it and must stop when it is cancelled.
	166	+	func (s *Supervisor) RootContext() context.Context { return s.root }
	167	+
	168	+	// Done is closed once the root context is cancelled and every registered hook
	169	+	// has been drained and joined.
	170	+	func (s *Supervisor) Done() <-chan struct{} { return s.done }
	171	+
	172	+	// Register admits one background component with an ordered drain hook. The
	173	+	// component must derive its work from RootContext and must stop when that
	174	+	// context is cancelled; hook must return once it has. Registration never grants
	175	+	// a lease, so a registered component cannot pin the broker open. Register is
	176	+	// refused with ports.BrokerAdmissionClosed after shutdown commits and with
	177	+	// ports.BrokerAdmissionInvalid for an empty name or nil hook.
	178	+	func (s *Supervisor) Register(name string, hook ShutdownHook) error {
	179	+	if name == "" \|\| hook == nil {
	180	+	return ports.BrokerAdmissionInvalid
	181	+	}
	182	+	s.mu.Lock()
	183	+	defer s.mu.Unlock()
	184	+	if s.closed {
	185	+	return ports.BrokerAdmissionClosed
	186	+	}
	187	+	s.hooks = append(s.hooks, registeredHook{name: name, hook: hook})
	188	+	return nil
	189	+	}
	190	+
	191	+	// RegisterRunner starts a broker-owned background component under the root
	192	+	// context and joins it at shutdown: its drain hook waits for Run to return. The
	193	+	// component must therefore honor cancellation promptly. Registration is refused
	194	+	// exactly as Register is.
	195	+	func (s *Supervisor) RegisterRunner(name string, runner Runner) error {
	196	+	if nilDependency(runner) {
	197	+	return ports.BrokerAdmissionInvalid
	198	+	}
	199	+	drained := make(chan struct{})
	200	+	if err := s.Register(name, func(context.Context) error {
	201	+	<-drained
	202	+	return nil
	203	+	}); err != nil {
	204	+	return err
	205	+	}
	206	+	ctx := s.root
	207	+	go func() {
	208	+	defer close(drained)
	209	+	runner.Run(ctx)
	210	+	}()
	211	+	return nil
	212	+	}
	213	+
	214	+	// RegisterCloseable registers a broker-owned resource to be closed after the
	215	+	// root context is cancelled. Its Close error is reported by Close and never
	216	+	// aborts the remaining drains. Registration is refused exactly as Register is.
	217	+	func (s *Supervisor) RegisterCloseable(name string, closeable Closeable) error {
	218	+	if nilDependency(closeable) {
	219	+	return ports.BrokerAdmissionInvalid
	220	+	}
	221	+	return s.Register(name, func(context.Context) error { return closeable.Close() })
	222	+	}
	223	+
	224	+	// AdmitClient admits one client-connection lease. It returns
	225	+	// ports.BrokerAdmissionClosed once shutdown has committed.
	226	+	func (s Supervisor) AdmitClient() (Lease, error) { return s.admit(leaseClient) }
	227	+
	228	+	// AdmitOperation admits one in-flight user-operation lease. It returns
	229	+	// ports.BrokerAdmissionClosed once shutdown has committed.
	230	+	func (s Supervisor) AdmitOperation() (Lease, error) { return s.admit(leaseOperation) }
	231	+
	232	+	// Close commits shutdown and waits for the root context to be cancelled and
	233	+	// every hook to drain. It is idempotent and concurrent-safe: every caller
	234	+	// observes the same terminal state and the same joined hook error.
	235	+	func (s *Supervisor) Close() error {
	236	+	s.mu.Lock()
	237	+	s.commitLocked()
	238	+	s.mu.Unlock()
	239	+	<-s.stopped
	240	+	s.mu.Lock()
	241	+	defer s.mu.Unlock()
	242	+	return s.shutdownErr
	243	+	}
	244	+
	245	+	func (s Supervisor) admit(kind leaseKind) (Lease, error) {
	246	+	s.mu.Lock()
	247	+	if s.closed {
	248	+	s.mu.Unlock()
	249	+	return nil, ports.BrokerAdmissionClosed
	250	+	}
	251	+	switch kind {
	252	+	case leaseClient:
	253	+	s.clients++
	254	+	case leaseOperation:
	255	+	s.operations++
	256	+	}
	257	+	// An admission that wins the commit race disarms the pending idle shutdown.
	258	+	s.disarmTimerLocked()
	259	+	s.mu.Unlock()
	260	+	s.signal()
	261	+	return &Lease{release: func() { s.release(kind) }}, nil
	262	+	}
	263	+
	264	+	func (s *Supervisor) release(kind leaseKind) {
	265	+	s.mu.Lock()
	266	+	switch kind {
	267	+	case leaseClient:
	268	+	if s.clients > 0 {
	269	+	s.clients--
	270	+	}
	271	+	case leaseOperation:
	272	+	if s.operations > 0 {
	273	+	s.operations--
	274	+	}
	275	+	}
	276	+	if !s.closed && s.clients == 0 && s.operations == 0 {
	277	+	s.armTimerLocked()
	278	+	}
	279	+	s.mu.Unlock()
	280	+	s.signal()
	281	+	}
	282	+
	283	+	// signal nudges the run goroutine to re-read the timer state. Signals coalesce.
	284	+	func (s *Supervisor) signal() {
	285	+	select {
	286	+	case s.wake <- struct{}{}:
	287	+	default:
	288	+	}
	289	+	}
	290	+
	291	+	// run owns the idle timer and is the sole shutdown executor. It waits on the
	292	+	// current arm's timer and finalizes exactly once, either when an un-superseded
	293	+	// arm fires while the broker is still idle or when Close commits the shutdown.
	294	+	func (s *Supervisor) run() {
	295	+	defer close(s.stopped)
	296	+
	297	+	s.mu.Lock()
	298	+	s.armTimerLocked()
	299	+	s.mu.Unlock()
	300	+	close(s.ready)
	301	+
	302	+	for {
	303	+	s.mu.Lock()
	304	+	timer, arm := s.timer, s.arm
	305	+	s.mu.Unlock()
	306	+
	307	+	var fire <-chan time.Time
	308	+	if timer != nil {
	309	+	fire = timer.C()
	310	+	}
	311	+	select {
	312	+	case <-s.closing:
	313	+	s.finalize()
	314	+	return
	315	+	case <-s.wake:
	316	+	continue
	317	+	case <-fire:
	318	+	if s.tryCommitIdleExpiry(arm) {
	319	+	s.finalize()
	320	+	return
	321	+	}
	322	+	}
	323	+	}
	324	+	}
	325	+
	326	+	// tryCommitIdleExpiry reports whether a fire from arm must commit the idle
	327	+	// shutdown. It returns true only for the run goroutine that must finalize: a
	328	+	// superseded or already disarmed arm, a committed supervisor, or a still-held
	329	+	// lease all make the fire a no-op.
	330	+	func (s *Supervisor) tryCommitIdleExpiry(arm uint64) bool {
	331	+	s.mu.Lock()
	332	+	defer s.mu.Unlock()
	333	+	if s.closed \|\| !s.timerArmed \|\| s.arm != arm {
	334	+	return false
	335	+	}
	336	+	if s.clients > 0 \|\| s.operations > 0 {
	337	+	return false
	338	+	}
	339	+	return s.commitLocked()
	340	+	}
	341	+
	342	+	// commitLocked marks the supervisor closed and disarms the idle timer. It
	343	+	// returns true exactly once, for the caller that committed. Because admission
	344	+	// and commitment share the mutex, an admission either disarms the pending expiry
	345	+	// before the commit or observes the closed state after it.
	346	+	func (s *Supervisor) commitLocked() bool {
	347	+	if s.committed {
	348	+	return false
	349	+	}
	350	+	s.committed = true
	351	+	s.closed = true
	352	+	s.disarmTimerLocked()
	353	+	close(s.closing)
	354	+	return true
	355	+	}
	356	+
	357	+	// finalize cancels the root context, drains every hook outside the mutex in
	358	+	// reverse registration order, and only then closes Done. It runs exactly once,
	359	+	// on the run goroutine.
	360	+	func (s *Supervisor) finalize() {
	361	+	s.cancel()
	362	+	s.mu.Lock()
	363	+	hooks := append([]registeredHook(nil), s.hooks...)
	364	+	s.mu.Unlock()
	365	+
	366	+	var errs []error
	367	+	for i := len(hooks) - 1; i >= 0; i-- {
	368	+	if err := hooks[i].hook(s.root); err != nil {
	369	+	errs = append(errs, fmt.Errorf("broker: %s shutdown: %w", hooks[i].name, err))
	370	+	}
	371	+	}
	372	+
	373	+	s.mu.Lock()
	374	+	s.shutdownErr = errors.Join(errs...)
	375	+	s.mu.Unlock()
	376	+	close(s.done)
	377	+	}
	378	+
	379	+	// armTimerLocked installs a fresh idle timer and claims a new arm number. A
	380	+	// fresh timer per idle cycle lets the run goroutine attribute a fire to the arm
	381	+	// that produced it, so a fire from a superseded arm is ignored instead of
	382	+	// shutting the broker down out of turn. Callers hold mu and have established
	383	+	// that both lease counts are zero.
	384	+	func (s *Supervisor) armTimerLocked() {
	385	+	if s.timerArmed \|\| s.committed {
	386	+	return
	387	+	}
	388	+	s.timer = s.clock.NewTimer(s.idle)
	389	+	s.timerArmed = true
	390	+	s.arm++
	391	+	}
	392	+
	393	+	// disarmTimerLocked stops and drains the idle timer, superseding its arm,
	394	+	// supporting buffered timer adapters with pre-Go-1.23 reset semantics. Callers
	395	+	// hold mu.
	396	+	func (s *Supervisor) disarmTimerLocked() {
	397	+	s.timerArmed = false
	398	+	if s.timer == nil {
	399	+	return
	400	+	}
	401	+	if !s.timer.Stop() {
	402	+	select {
	403	+	case <-s.timer.C():
	404	+	default:
	405	+	}
	406	+	}
	407	+	s.timer = nil
	408	+	}

A internal/usecase/broker/lifecycle_test.go +540 -0

@@ -0,0 +1,540 @@
	1	+	package broker
	2	+
	3	+	import (
	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.
	16	+	func 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.
	27	+	func (c *manualClock) activeTimers() int {
	28	+	c.mu.Lock()
	29	+	defer c.mu.Unlock()
	30	+	return len(c.timers)
	31	+	}
	32	+
	33	+	func 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	+
	39	+	func supervisorTimerArmed(s *Supervisor) bool {
	40	+	_, _, armed, _ := supervisorState(s)
	41	+	return armed
	42	+	}
	43	+
	44	+	func 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	+
	53	+	func 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	+
	61	+	func 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	+
	69	+	type closeFunc func() error
	70	+
	71	+	func (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.
	75	+	type blockingRunner struct {
	76	+	started chan struct{}
	77	+	observed chan error
	78	+	}
	79	+
	80	+	func newBlockingRunner() *blockingRunner {
	81	+	return &blockingRunner{started: make(chan struct{}), observed: make(chan error, 1)}
	82	+	}
	83	+
	84	+	func (r *blockingRunner) Run(ctx context.Context) {
	85	+	close(r.started)
	86	+	<-ctx.Done()
	87	+	r.observed <- ctx.Err()
	88	+	}
	89	+
	90	+	func 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	+
	114	+	func 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	+
	137	+	func 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	+
	154	+	func 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	+
	167	+	func 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.
	198	+	func 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.
	220	+	func 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	+
	229	+	func 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.
	250	+	func 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	+
	306	+	func 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	+
	323	+	func 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	+
	342	+	func 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	+
	362	+	func 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	+
	384	+	func 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	+
	419	+	func 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	+
	432	+	func 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.
	443	+	func 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.
	455	+	func 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.
	468	+	func 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.
	514	+	func 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	+	}

@@ -0,0 +1,540 @@
	1	+	package broker
	2	+
	3	+	import (
	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.
	16	+	func 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.
	27	+	func (c *manualClock) activeTimers() int {
	28	+	c.mu.Lock()
	29	+	defer c.mu.Unlock()
	30	+	return len(c.timers)
	31	+	}
	32	+
	33	+	func 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	+
	39	+	func supervisorTimerArmed(s *Supervisor) bool {
	40	+	_, _, armed, _ := supervisorState(s)
	41	+	return armed
	42	+	}
	43	+
	44	+	func 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	+
	53	+	func 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	+
	61	+	func 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	+
	69	+	type closeFunc func() error
	70	+
	71	+	func (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.
	75	+	type blockingRunner struct {
	76	+	started chan struct{}
	77	+	observed chan error
	78	+	}
	79	+
	80	+	func newBlockingRunner() *blockingRunner {
	81	+	return &blockingRunner{started: make(chan struct{}), observed: make(chan error, 1)}
	82	+	}
	83	+
	84	+	func (r *blockingRunner) Run(ctx context.Context) {
	85	+	close(r.started)
	86	+	<-ctx.Done()
	87	+	r.observed <- ctx.Err()
	88	+	}
	89	+
	90	+	func 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	+
	114	+	func 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	+
	137	+	func 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	+
	154	+	func 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	+
	167	+	func 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.
	198	+	func 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.
	220	+	func 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	+
	229	+	func 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.
	250	+	func 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	+
	306	+	func 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	+
	323	+	func 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	+
	342	+	func 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	+
	362	+	func 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	+
	384	+	func 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	+
	419	+	func 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	+
	432	+	func 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.
	443	+	func 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.
	455	+	func 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.
	468	+	func 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.
	514	+	func 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	+	}