1package broker
2
3import (
4 "context"
5 "errors"
6 "sync"
7
8 "github.com/bnema/vev/internal/domain"
9 "github.com/bnema/vev/internal/ports"
10)
11
12// Offline broker composition.
13//
14// Authority admits one accepted client connection to the broker core and
15// returns the connection-scoped ports.BrokerCoreService for exactly that
16// connection. It is the structural implementation of ports.BrokerAuthority,
17// the admission seam the broker IPC listener (internal/adapters/brokeripc)
18// consumes and which this use case must not import: admission obtains one
19// supervisor client lease and one pool connection
20// identity, and rolls both back on any failure so a refused admission never
21// pins the broker open or leaks a client slot. The admission context bounds
22// admission only and is never retained; the admitted service owns a
23// connection-lived context derived from the supervisor root.
24//
25// Registry construction determines observation and membership policy. The
26// authority preserves those guards and exposes admitted mutations through the
27// connection-scoped service.
28
29// Authority admits one accepted client connection to the broker core. It owns
30// no lifetime of its own: the supervisor owns the broker lifetime, the registry
31// owns snapshots, and the pool owns transports. An admitted Service is bound to
32// exactly one pool client identity and one broker epoch.
33//
34// It is the structural implementation of ports.BrokerAuthority, the admission
35// seam the broker IPC listener consumes; this use case never imports that
36// adapter.
37type Authority struct {
38 epoch ports.BrokerEpoch
39 registry *Registry
40 pool *Pool
41 supervisor *Supervisor
42 clock ports.Clock
43 codec ports.SessionCodec
44}
45
46var _ ports.BrokerAuthority = (*Authority)(nil)
47
48// NewAuthority composes one connection-scoped broker authority over an existing
49// registry, pool, and supervisor. All three must be live and share epoch.
50//
51// The caller owns the dependencies' lifecycles. In particular it must start the
52// registry's single Registry.Run to own and drain the durable writer, and must
53// cancel that run context and join Run before the store is closed. NewAuthority
54// neither starts nor settles Run: serving connections without a live Run leaks
55// the writer and never flushes the newest staged publication, and a registry
56// that was never run leaves its durable writer undrained.
57func NewAuthority(epoch ports.BrokerEpoch, registry *Registry, pool *Pool, supervisor *Supervisor, codec ports.SessionCodec, clocks ...ports.Clock) (*Authority, error) {
58 if epoch == 0 || nilDependency(registry) || nilDependency(pool) || nilDependency(supervisor) {
59 return nil, errors.New("broker: invalid authority dependencies")
60 }
61 if registry.epoch != epoch || pool.epoch != epoch {
62 return nil, errors.New("broker: authority epoch mismatch")
63 }
64 clk := registry.clock
65 if len(clocks) > 0 {
66 clk = clocks[0]
67 }
68 if nilDependency(clk) {
69 return nil, errors.New("broker: invalid authority clock")
70 }
71 if nilDependency(codec) {
72 return nil, errors.New("broker: authority requires a session codec")
73 }
74 return &Authority{epoch: epoch, registry: registry, pool: pool, supervisor: supervisor, clock: clk, codec: codec}, nil
75}
76
77// AdmitClient admits exactly one accepted client connection. The supervisor
78// client lease and the pool connection identity are acquired together; if
79// either fails, the other is released before returning, so admission either
80// commits both or neither. ctx bounds admission only and is never retained: the
81// returned service derives its own connection-lived context from the supervisor
82// root, so a caller's setup deadline cannot disturb an admitted connection.
83func (a *Authority) AdmitClient(ctx context.Context) (ports.BrokerCoreService, error) {
84 if err := ctx.Err(); err != nil {
85 return nil, err
86 }
87 lease, err := a.supervisor.AdmitClient()
88 if err != nil {
89 return nil, err
90 }
91 id, err := a.pool.RegisterClient()
92 if err != nil {
93 lease.Release()
94 return nil, err
95 }
96 previewID, err := a.pool.RegisterClient()
97 if err != nil {
98 a.pool.CloseClient(id)
99 lease.Release()
100 return nil, err
101 }
102 if err := ctx.Err(); err != nil {
103 a.pool.CloseClient(previewID)
104 a.pool.CloseClient(id)
105 lease.Release()
106 return nil, err
107 }
108 return newService(serviceConfig{
109 epoch: a.epoch, id: id, previewID: previewID, registry: a.registry,
110 pool: a.pool, supervisor: a.supervisor, lease: lease, clock: a.clock, codec: a.codec,
111 }), nil
112}
113
114// Service is one admitted client connection: the core's service for exactly one
115// pool connection identity and broker epoch. It implements ports.BrokerCoreService.
116type Service struct {
117 epoch ports.BrokerEpoch
118 id ports.BrokerConnectionID
119 previewID ports.BrokerConnectionID
120 registry *Registry
121 pool *Pool
122 supervisor *Supervisor
123 lease *Lease
124 clock ports.Clock
125 codec ports.SessionCodec
126
127 // ctx owns everything this connection started: in-flight setup opens and
128 // live streams. It is derived from the supervisor root, never from the
129 // admission context, so broker shutdown and Close both abort owned work.
130 ctx context.Context
131 cancel context.CancelFunc
132
133 mu sync.Mutex
134 closed bool
135 subs map[*serviceSubscription]struct{}
136 preview *servicePreviewSubscription
137 previewGeneration ports.BrokerPreviewGeneration
138 // nextStream is this connection's strictly increasing, never-zero logical
139 // stream ID allocator. It is the only allocator on this connection: the
140 // supervisor and the broker operations each read one identity here and carry
141 // it on OpenStream rather than relying on an implicit allocation there. An
142 // allocated identity is consumed even when the open it names is refused.
143 nextStream ports.BrokerStreamID
144 previewStream ports.BrokerStreamID
145 // wg counts in-flight stream opens and membership mutations so Close drains
146 // them, and their operation leases, before releasing connection resources.
147 wg sync.WaitGroup
148
149 closeOnce sync.Once
150
151 // done closes exactly once when this connection is terminal, either because
152 // the broker root shut down or because Close settled it locally. Err is
153 // stable afterwards (nil for an orderly local Close). stopRoot detaches the
154 // root watcher once Close has committed so a closed connection never leaves
155 // a root callback behind.
156 done chan struct{}
157 stopRoot func() bool
158 termOnce sync.Once
159 termErr error
160}
161
162var _ ports.BrokerCoreService = (*Service)(nil)
163
164type serviceConfig struct {
165 epoch ports.BrokerEpoch
166 id, previewID ports.BrokerConnectionID
167 registry *Registry
168 pool *Pool
169 supervisor *Supervisor
170 lease *Lease
171 clock ports.Clock
172 codec ports.SessionCodec
173}
174
175func (c serviceConfig) validate() error {
176 if c.epoch == 0 || c.id == (ports.BrokerConnectionID{}) || c.previewID == (ports.BrokerConnectionID{}) || c.id == c.previewID ||
177 nilDependency(c.registry) || nilDependency(c.pool) || nilDependency(c.supervisor) || nilDependency(c.lease) || nilDependency(c.clock) || nilDependency(c.codec) {
178 return errors.New("broker: invalid service dependencies")
179 }
180 return nil
181}
182
183func newService(c serviceConfig) *Service {
184 if err := c.validate(); err != nil {
185 panic(err) // Only the admitted authority constructs a service after acquiring both IDs and lease.
186 }
187 ctx, cancel := context.WithCancel(c.supervisor.RootContext())
188 s := &Service{
189 epoch: c.epoch, id: c.id, previewID: c.previewID, registry: c.registry, pool: c.pool, supervisor: c.supervisor, lease: c.lease, clock: c.clock, codec: c.codec,
190 ctx: ctx, cancel: cancel, subs: make(map[*serviceSubscription]struct{}),
191 done: make(chan struct{}),
192 }
193 // Broker shutdown ends every admitted connection. The watcher records the
194 // loss as the terminal cause unless a local Close has already settled the
195 // connection, so a connection closed by its owner reports nil and one lost
196 // to a broker shutdown reports the typed loss.
197 s.stopRoot = context.AfterFunc(c.supervisor.RootContext(), func() {
198 s.terminalize(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker shutdown", Cause: context.Canceled})
199 })
200 return s
201}
202
203// ConnectionID returns the pool identity assigned at admission. Clients carry
204// it on stream operations so a stale request can be fenced.
205func (s *Service) ConnectionID() ports.BrokerConnectionID { return s.id }
206
207// NextStreamID allocates this connection's next logical stream identity. It is
208// thread-safe, strictly monotone, and never zero, performs no I/O, and refuses
209// a closed connection or an exhausted counter rather than wrapping. It is the
210// only allocation seam on this connection: supervisor and BrokerOperations both
211// allocate here and carry the resulting identity on the request.
212func (s *Service) NextStreamID() (ports.BrokerStreamID, error) {
213 if s == nil {
214 return 0, ports.BrokerAdmissionClosed
215 }
216 s.mu.Lock()
217 defer s.mu.Unlock()
218 if s.closed {
219 return 0, ports.BrokerAdmissionClosed
220 }
221 if s.nextStream == ^ports.BrokerStreamID(0) {
222 return 0, ports.BrokerAdmissionLimit
223 }
224 s.nextStream++
225 return s.nextStream, nil
226}
227
228// Done closes exactly once when this connection is terminal: the broker root
229// shut down or Close settled it locally. Err is stable afterwards.
230func (s *Service) Done() <-chan struct{} { return s.done }
231
232// Err returns this connection's terminal cause: nil for an orderly local Close,
233// otherwise the broker-side loss that settled it. The cause is recorded exactly
234// once and never overwritten, so it is stable once Done is closed.
235func (s *Service) Err() error {
236 s.mu.Lock()
237 defer s.mu.Unlock()
238 return s.termErr
239}
240
241// terminalize records the terminal cause and closes Done exactly once. The
242// first cause wins, so a race between a broker-root loss and a local Close
243// yields one stable, observable outcome.
244func (s *Service) terminalize(err error) {
245 s.termOnce.Do(func() {
246 s.mu.Lock()
247 s.termErr = err
248 s.mu.Unlock()
249 close(s.done)
250 })
251}
252
253// Snapshot delegates to the registry's current immutable publication.
254func (s *Service) Snapshot() ports.BrokerSnapshot { return s.registry.Snapshot() }
255
256// Subscribe delegates to the registry and tracks the returned subscription so
257// Close releases every subscription this connection still owns. A closed
258// connection refuses new subscriptions. Opening a subscription also records
259// one unit of registry demand (Registry.SetDemand(true)): while a client
260// watches the snapshot, the registry observes on the demand cadence. Without
261// a subscriber it performs no scheduled observations. Close balances demand.
262func (s *Service) Subscribe() (ports.BrokerSubscription, error) {
263 return s.subscribe(true)
264}
265
266// SubscribePassive is Subscribe for a read-only snapshot reader (vev ls, vev
267// host list): it receives every publication but records no demand, so reading
268// state never triggers an observation.
269func (s *Service) SubscribePassive() (ports.BrokerSubscription, error) {
270 return s.subscribe(false)
271}
272
273func (s *Service) subscribe(demand bool) (ports.BrokerSubscription, error) {
274 s.mu.Lock()
275 if s.closed {
276 s.mu.Unlock()
277 return nil, ports.BrokerAdmissionClosed
278 }
279 sub := &serviceSubscription{service: s, inner: s.registry.Subscribe(), demand: demand}
280 s.subs[sub] = struct{}{}
281 if demand {
282 s.registry.SetDemand(true)
283 }
284 s.mu.Unlock()
285 return sub, nil
286}
287
288// OpenStream opens one logical stream through the pool on behalf of the user.
289// The request is fenced to this connection's exact epoch and identity before
290// the pool sees it, and one operation lease pins the broker only while setup is
291// in flight: a stream that opens successfully outlives its setup lease, and
292// Close or broker shutdown still aborts anything that has not finished.
293func (s *Service) OpenEnvelopeStream(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
294 if err := s.scope(&request); err != nil {
295 return nil, err
296 }
297 // Service admission owns validation of caller input before the pool sees it.
298 if err := request.Validate(); err != nil {
299 return nil, errors.Join(ports.BrokerAdmissionInvalid, err)
300 }
301 s.mu.Lock()
302 if s.closed {
303 s.mu.Unlock()
304 return nil, ports.BrokerAdmissionClosed
305 }
306 s.wg.Add(1)
307 s.mu.Unlock()
308 defer s.wg.Done()
309
310 operation, err := s.supervisor.AdmitOperation()
311 if err != nil {
312 return nil, err
313 }
314 defer operation.Release()
315
316 openCtx, release := s.openContext(ctx)
317 stream, err := s.pool.OpenStream(openCtx, request)
318 if err != nil {
319 release()
320 return nil, err
321 }
322 // A local control or attachment stream is a natural signal that the local
323 // daemon's own catalogue may be about to change (a session request is
324 // being issued, or an attachment claim is being taken): mark a local
325 // re-probe pending as soon as the stream opens, not only once it ends, so
326 // a session created by this very attach appears in `ls --all`/other
327 // pickers promptly, even while the client stays attached. An observation
328 // stream never triggers this: it is the probe traffic itself, and
329 // re-arming from it would starve the schedule instead of catching up to
330 // it.
331 if request.Local && (request.Purpose == ports.BrokerStreamControl || request.Purpose == ports.BrokerStreamAttachment) {
332 s.registry.RequestProbe("")
333 }
334 // Detach the connection-scoped context link once the pool reports the
335 // stream terminal, so a long-lived connection never accumulates one link
336 // per opened stream. Nothing here ends the stream: setup completion and the
337 // operation lease's release both happen while it keeps running.
338 go func() {
339 <-stream.Done()
340 release()
341 // The stream ending is the same signal again (a session was created,
342 // killed, or its attachment state flipped over the stream's lifetime):
343 // request one more local re-probe so a change made just before detach
344 // is not left stale for up to the retry/fresh interval.
345 if request.Local && (request.Purpose == ports.BrokerStreamControl || request.Purpose == ports.BrokerStreamAttachment) {
346 s.registry.RequestProbe("")
347 }
348 }()
349 return stream, nil
350}
351
352// CloseStream retires one stream of this connection. A foreign connection
353// identity is refused as stale and never retires another connection's stream.
354func (s *Service) CloseStream(connection ports.BrokerConnectionID, stream ports.BrokerStreamID) error {
355 if !connection.IsZero() && connection != s.id {
356 return ports.BrokerAdmissionStale
357 }
358 return s.pool.CloseStream(s.id, stream)
359}
360
361// AddHost delegates one admitted membership mutation to the registry.
362func (s *Service) AddHost(ctx context.Context, endpoint string, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
363 return serviceMutation(s, ctx, func(ctx context.Context) (domain.RemoteRegistration, error) {
364 return s.registry.AddHost(ctx, endpoint, policy)
365 })
366}
367
368// RemoveHost delegates one admitted exact-registration removal to the registry.
369func (s *Service) RemoveHost(ctx context.Context, expected domain.RemoteRegistration) (bool, error) {
370 return serviceMutation(s, ctx, func(ctx context.Context) (bool, error) {
371 return s.registry.RemoveHost(ctx, expected)
372 })
373}
374
375// UpdateHostPolicy delegates one admitted exact-registration policy update.
376func (s *Service) UpdateHostPolicy(ctx context.Context, expected domain.RemoteRegistration, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
377 return serviceMutation(s, ctx, func(ctx context.Context) (domain.RemoteRegistration, error) {
378 return s.registry.UpdateHostPolicy(ctx, expected, policy)
379 })
380}
381
382func serviceMutation[T any](s *Service, ctx context.Context, mutate func(context.Context) (T, error)) (T, error) {
383 var zero T
384 s.mu.Lock()
385 if s.closed {
386 s.mu.Unlock()
387 return zero, ports.BrokerAdmissionClosed
388 }
389 s.wg.Add(1)
390 s.mu.Unlock()
391 defer s.wg.Done()
392
393 operation, err := s.supervisor.AdmitOperation()
394 if err != nil {
395 return zero, err
396 }
397 defer operation.Release()
398
399 operationCtx, release := s.openContext(ctx)
400 defer release()
401 value, err := mutate(operationCtx)
402 return value, membershipError(err)
403}
404
405func membershipError(err error) error {
406 if err == nil {
407 return nil
408 }
409 code := ports.BrokerErrorCode(0)
410 text := ""
411 var brokerError ports.BrokerError
412 switch {
413 case errors.Is(err, ports.ErrBrokerHostConflict):
414 code, text = ports.BrokerErrorHostConflict, "host registration conflict"
415 case errors.Is(err, ports.ErrBrokerMembershipImmutable):
416 code, text = ports.BrokerErrorMembershipImmutable, "broker membership is immutable"
417 case errors.As(err, &brokerError) && brokerError.Code == ports.BrokerErrorConflictingPolicy:
418 code, text = ports.BrokerErrorConflictingPolicy, "host policy conflicts with existing policy"
419 default:
420 var unknown ports.BrokerStoreOutcomeUnknownError
421 var unknownPointer *ports.BrokerStoreOutcomeUnknownError
422 if errors.As(err, &unknown) || errors.As(err, &unknownPointer) {
423 code, text = ports.BrokerErrorOutcomeUnknown, "mutation outcome is unknown"
424 }
425 }
426 if code == 0 {
427 return err
428 }
429 return ports.BrokerError{Code: code, Text: text, Cause: err}
430}
431
432// RequestReconcile schedules one coalesced observation for a configured
433// endpoint. The registry validates membership and drops unknown endpoints, so
434// this hint cannot expand authority or start an attachment.
435func (s *Service) RequestReconcile(endpoint string) {
436 s.mu.Lock()
437 closed := s.closed
438 s.mu.Unlock()
439 if closed {
440 return
441 }
442 s.registry.RequestProbe(endpoint)
443}
444
445// Close cancels and drains every owned resource exactly once: in-flight setup
446// opens and live streams stop, subscriptions close, the pool client is retired,
447// and the supervisor client lease is released. It is idempotent and
448// concurrent-safe; every caller observes the same terminal state.
449func (s *Service) Close() error {
450 s.closeOnce.Do(func() {
451 s.mu.Lock()
452 s.closed = true
453 subs := make([]*serviceSubscription, 0, len(s.subs))
454 for sub := range s.subs {
455 subs = append(subs, sub)
456 }
457 preview := s.preview
458 s.mu.Unlock()
459
460 s.cancel()
461 if preview != nil {
462 preview.Close()
463 }
464 s.wg.Wait()
465 for _, sub := range subs {
466 sub.Close()
467 }
468 s.pool.CloseClient(s.previewID)
469 s.pool.CloseClient(s.id)
470 s.lease.Release()
471 // The connection is locally and orderly closed: detach the broker-root
472 // watcher so it can neither fire later nor overwrite the clean outcome,
473 // then publish the terminal state.
474 if s.stopRoot != nil {
475 s.stopRoot()
476 }
477 s.terminalize(nil)
478 })
479 return nil
480}
481
482// scope fences one request to this connection's exact epoch and identity,
483// filling an absent identity with this connection's own. A foreign epoch or
484// connection is refused before it can steer another connection's work.
485func (s *Service) scope(request *ports.BrokerOpenStreamRequest) error {
486 if request.Epoch != 0 && request.Epoch != s.epoch {
487 return ports.BrokerError{Code: ports.BrokerErrorStaleEpoch}
488 }
489 if !request.Connection.IsZero() && request.Connection != s.id {
490 return ports.BrokerAdmissionStale
491 }
492 request.Epoch = s.epoch
493 request.Connection = s.id
494 return nil
495}
496
497// openContext derives the context that owns one stream: the caller's context
498// bounds the request, and the connection context bounds the service, so the
499// stream and its setup end when either does. The returned release detaches the
500// connection link once the stream has settled.
501func (s *Service) openContext(ctx context.Context) (context.Context, func()) {
502 openCtx, cancel := context.WithCancel(ctx)
503 stop := context.AfterFunc(s.ctx, cancel)
504 return openCtx, func() { stop(); cancel() }
505}
506
507// serviceSubscription is one connection-owned subscription. Closing it releases
508// the registry subscription and forgets it on the owning service.
509type serviceSubscription struct {
510 service *Service
511 inner ports.BrokerSubscription
512 demand bool
513 once sync.Once
514}
515
516func (s *serviceSubscription) Changed() <-chan struct{} { return s.inner.Changed() }
517
518func (s *serviceSubscription) Close() {
519 s.once.Do(func() {
520 s.inner.Close()
521 s.service.mu.Lock()
522 delete(s.service.subs, s)
523 if s.demand {
524 s.service.registry.SetDemand(false)
525 }
526 s.service.mu.Unlock()
527 })
528}