1package broker
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "sync"
8 "time"
9
10 "github.com/bnema/vev/internal/ports"
11)
12
13// Lifecycle supervisor.
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.
52const DefaultIdleGrace = 5 * time.Minute
53
54// SupervisorConfig configures a Supervisor. The zero value is valid and selects
55// DefaultIdleGrace.
56type 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.
66type 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.
71type 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.
78type 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.
85type 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.
91func (l *Lease) Release() {
92 if l == nil || l.release == nil {
93 return
94 }
95 l.once.Do(l.release)
96}
97
98type leaseKind uint8
99
100const (
101 leaseClient leaseKind = iota + 1
102 leaseOperation
103)
104
105type 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.
112type 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.
144func 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.
166func (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.
170func (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.
178func (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.
195func (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.
217func (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.
226func (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.
230func (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.
235func (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
245func (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
264func (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.
284func (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.
294func (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.
330func (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.
346func (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.
360func (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.
384func (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.
396func (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}