1package brokeripc
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "io"
8 "io/fs"
9 "net"
10 "sync"
11 "sync/atomic"
12 "syscall"
13 "time"
14
15 "github.com/bnema/vev/internal/adapters/brokerwire"
16 "github.com/bnema/vev/internal/adapters/ipc"
17 "github.com/bnema/vev/internal/adapters/streamframe"
18 "github.com/bnema/vev/internal/domain"
19 "github.com/bnema/vev/internal/ports"
20 "github.com/bnema/vev/internal/protocol/wire"
21)
22
23// Broker IPC listener and per-connection session.
24//
25// The listener owns one per-user AF_UNIX endpoint through the private IPC
26// carriage: owner-only directory and 0600 socket, same-user kernel peer
27// credentials on accept, race-safe stale-socket recovery, foreign-path refusal,
28// and idempotent Close that unlinks only the socket inode it created. It caps
29// concurrent accepted clients with one slot acquired before every accept, so
30// the bound is enforced before a socket is admitted.
31//
32// Accept completes the broker preamble and admits the client to the broker core
33// before it returns, so the returned ports.BrokerService knows the
34// BrokerConnectionID assigned to exactly that connection. The session then
35// runs the brokerwire conversation for that connection in its own goroutines:
36// one reader, one snapshot publisher, one relay pair per logical stream, and
37// one goroutine per admitted mutating operation. Every one of those is
38// per-connection, so a slow or stalled client blocks only its own work.
39
40// Listen binds the private per-user broker endpoint at path and returns a
41// ports.BrokerListener that admits same-user clients only.
42//
43// path must be an absolute, cleaned path inside an owner-only directory; the
44// listener creates or validates that parent as 0700 and tightens the bound
45// socket to 0600. A path that already exists and is not a socket is refused
46// without being removed, a stale socket left by a dead owner is recovered
47// race-safely, and a live owner is reported as ipc.ErrDaemonRunning instead of
48// being evicted.
49//
50// authority is the ports.BrokerAuthority admission seam: the listener calls
51// AdmitClient once per accepted connection under the handshake budget and binds
52// the returned service to exactly that connection. That service's ConnectionID
53// is the identity the session stamps on every frame; the broker use case
54// supplies the implementation, and this adapter consumes it without
55// constructing a broker itself. See ports.BrokerAuthority for the
56// admission-context contract.
57func Listen(path string, epoch ports.BrokerEpoch, authority ports.BrokerAuthority, cfg Config, opts ...ipc.Option) (ports.BrokerListener, error) {
58 return listen(path, epoch, authority, cfg, ipc.SameUserPeerVerifier(), opts...)
59}
60
61// listen is Listen with an injected peer verifier, so a test can observe a
62// deterministic same-user refusal without root. A nil verifier falls back to
63// the platform default.
64func listen(path string, epoch ports.BrokerEpoch, authority ports.BrokerAuthority, cfg Config, verify ipc.PeerVerifier, opts ...ipc.Option) (ports.BrokerListener, error) {
65 cfg = cfg.withDefaults()
66 if err := cfg.validate(); err != nil {
67 return nil, err
68 }
69 if epoch == 0 || authority == nil {
70 return nil, ErrConfig
71 }
72 mux, err := ipc.ListenMuxWithPeerVerifier(path, verify, opts...)
73 if err != nil {
74 return nil, err
75 }
76 l := &listener{
77 epoch: epoch,
78 authority: authority,
79 cfg: cfg,
80 mux: mux,
81 slots: make(chan struct{}, cfg.MaxClients),
82 admitted: make(chan *serverSession, cfg.MaxClients),
83 done: make(chan struct{}),
84 fatal: make(chan struct{}),
85 sessions: make(map[*serverSession]struct{}),
86 pending: make(map[wire.BoundedTransport]struct{}),
87 }
88 l.ctx, l.cancel = context.WithCancel(context.Background())
89 l.wg.Add(1)
90 go l.acceptLoop()
91 return l, nil
92}
93
94// listener implements ports.BrokerListener over one bound private carriage.
95//
96// The listener owns its accept loop: each accepted carriage completes its
97// preamble and broker admission in its own goroutine, bounded by one client slot
98// acquired before the accept. A client that stalls during the preamble therefore
99// occupies a slot and a goroutine, never the loop, so no stalled handshake can
100// delay another client's admission. Admitted sessions are published through a
101// bounded FIFO to Accept.
102//
103// The listener owns a listener-scoped context, canceled by Close, that bounds
104// every in-flight preamble and admission. Each accepted carriage is also
105// tracked until a running session owns it, so Close can close a carriage whose
106// framing I/O would otherwise stay blocked. Together those make Close drop an
107// already-accepted client immediately instead of waiting out HandshakeTimeout.
108type listener struct {
109 epoch ports.BrokerEpoch
110 authority ports.BrokerAuthority
111 cfg Config
112 mux ipc.MuxListener
113 slots chan struct{}
114 admitted chan *serverSession
115
116 ctx context.Context
117 cancel context.CancelFunc
118
119 wg sync.WaitGroup
120 fatal chan struct{}
121
122 mu sync.Mutex
123 closed bool
124 retiring bool
125 sessions map[*serverSession]struct{}
126 pending map[wire.BoundedTransport]struct{}
127 lastRefuse error
128 fatalCause error
129 fatalOnce sync.Once
130 closeOnce sync.Once
131 closeErr error
132 done chan struct{}
133}
134
135var _ ports.BrokerListener = (*listener)(nil)
136
137// acceptLoop accepts one verified same-user carriage at a time, bounded by the
138// client slots, and hands each accepted carriage to its own admission goroutine.
139// It stops when the listener closes or the carriage reports a terminal accept
140// failure.
141func (l *listener) acceptLoop() {
142 defer l.wg.Done()
143 for {
144 select {
145 case l.slots <- struct{}{}:
146 case <-l.done:
147 return
148 }
149 // The slot is released exactly once, whether admission fails here or the
150 // session reaches its terminal outcome later.
151 release := sync.OnceFunc(func() { <-l.slots })
152 if l.isClosed() {
153 release()
154 return
155 }
156 transport, err := l.mux.Accept()
157 if err != nil {
158 release()
159 if l.isClosed() {
160 return
161 }
162 if errors.Is(err, ipc.ErrMuxPeerRejected) {
163 // A refused peer is isolated to that connection: it is
164 // recorded for diagnostics and admission continues.
165 l.confirmRefusal(err)
166 continue
167 }
168 // Any other carriage failure stops admission deterministically;
169 // Accept reports it instead of spinning on a broken listener.
170 l.failAccept(err)
171 return
172 }
173 l.trackPending(transport)
174 l.wg.Add(1)
175 go func() {
176 defer l.wg.Done()
177 l.admit(transport, release)
178 }()
179 }
180}
181
182// failAccept records the one terminal accept failure and wakes every blocked
183// Accept. The terminal cause is stored separately from the diagnostic last
184// refusal, so an in-flight admission refusal can never overwrite the cause that
185// ended acceptance.
186func (l *listener) failAccept(err error) {
187 l.mu.Lock()
188 if l.fatalCause == nil {
189 l.fatalCause = err
190 }
191 l.mu.Unlock()
192 l.fatalOnce.Do(func() {
193 close(l.fatal)
194 })
195}
196
197// fatalErr returns the terminal accept failure, if any. It is never the
198// diagnostic last refusal.
199func (l *listener) fatalErr() error {
200 l.mu.Lock()
201 defer l.mu.Unlock()
202 return l.fatalCause
203}
204
205// confirmRefusal records the most recent refused or failed admission. It is
206// diagnostic only: a refusal is isolated to the peer that caused it.
207func (l *listener) confirmRefusal(err error) {
208 if err == nil {
209 return
210 }
211 l.mu.Lock()
212 l.lastRefuse = err
213 l.mu.Unlock()
214}
215
216// admit completes the preamble and broker admission for one accepted carriage
217// and publishes the running session. Slot ownership transfers to the session on
218// success.
219func (l *listener) admit(transport wire.BoundedTransport, release func()) {
220 session, err := l.startSession(transport, release)
221 l.untrackPending(transport)
222 if err != nil {
223 release()
224 l.confirmRefusal(err)
225 return
226 }
227 select {
228 case l.admitted <- session:
229 case <-l.done:
230 _ = session.Close()
231 }
232}
233
234// trackPending records one accepted carriage that no running session owns yet,
235// so Close can close it directly and unblock its framing I/O.
236func (l *listener) trackPending(transport wire.BoundedTransport) {
237 l.mu.Lock()
238 l.pending[transport] = struct{}{}
239 l.mu.Unlock()
240}
241
242// untrackPending drops one accepted carriage whose admission has finished.
243func (l *listener) untrackPending(transport wire.BoundedTransport) {
244 l.mu.Lock()
245 delete(l.pending, transport)
246 l.mu.Unlock()
247}
248
249// closePending closes every accepted carriage not yet owned by a running
250// session, so a client stalled in framing or admission is disconnected at once
251// rather than waiting out HandshakeTimeout.
252func (l *listener) closePending() {
253 l.mu.Lock()
254 pending := make([]wire.BoundedTransport, 0, len(l.pending))
255 for transport := range l.pending {
256 pending = append(pending, transport)
257 }
258 l.mu.Unlock()
259 for _, transport := range pending {
260 _ = transport.Close()
261 }
262}
263
264// classifyAdmissionFailure maps one preamble or authority failure. Once the
265// listener is closing, an accepted client that never reached admission was
266// dropped by the close, so the failure is reported as ErrListenerClosed. A
267// genuine handshake timeout with the listener still open keeps its
268// context.DeadlineExceeded classification.
269func (l *listener) classifyAdmissionFailure(err error) error {
270 if l.isClosed() {
271 return errors.Join(ErrListenerClosed, err)
272 }
273 return err
274}
275
276// Accept returns the next fully admitted client session, in accept order, and
277// blocks until one is available, the listener closes, or the carriage reports a
278// terminal accept failure. The returned ports.BrokerService is bound to exactly
279// the connection admitted at accept; Close on the listener ends it.
280//
281// A peer the carriage refuses (a non-same-user credential, an unreadable
282// credential, or an unsupported build) or a client that fails the preamble or
283// admission is isolated: that peer is dropped and the listener keeps serving.
284// The most recent refusal is available to diagnostics through refusal.
285func (l *listener) Accept() (ports.BrokerCoreService, error) {
286 if l == nil {
287 return nil, ErrListenerClosed
288 }
289 select {
290 case session := <-l.admitted:
291 return session, nil
292 case <-l.fatal:
293 select {
294 case session := <-l.admitted:
295 return session, nil
296 default:
297 }
298 return nil, l.fatalErr()
299 default:
300 }
301 select {
302 case session := <-l.admitted:
303 return session, nil
304 case <-l.fatal:
305 select {
306 case session := <-l.admitted:
307 return session, nil
308 default:
309 }
310 return nil, l.fatalErr()
311 case <-l.done:
312 select {
313 case session := <-l.admitted:
314 return session, nil
315 default:
316 }
317 return nil, ErrListenerClosed
318 }
319}
320
321// startSession completes the preamble and admission for one accepted carriage
322// and starts the session. The admission context is bounded to the handshake
323// budget and is canceled as soon as admission returns or when the listener
324// closes, so the admitted service must not retain it (see
325// ports.BrokerAuthority) and Close can drop a client still in preamble or
326// admission. Slot ownership transfers to the session on success.
327func (l *listener) startSession(transport wire.BoundedTransport, release func()) (*serverSession, error) {
328 // The handshake budget is bounded by the listener-scoped context, so
329 // Listener.Close cancels a preamble or admission that is still waiting
330 // instead of leaving it to run out HandshakeTimeout.
331 ctx, cancel := context.WithTimeout(l.ctx, l.cfg.HandshakeTimeout)
332 defer cancel()
333 ceilings, err := runServerPreamble(ctx, transport, brokerwire.DefaultCeilings())
334 if err != nil {
335 _ = transport.Close()
336 return nil, l.classifyAdmissionFailure(err)
337 }
338 core, err := l.authority.AdmitClient(ctx)
339 if err != nil {
340 _ = transport.Close()
341 return nil, l.classifyAdmissionFailure(err)
342 }
343 if core == nil {
344 _ = transport.Close()
345 return nil, fmt.Errorf("%w: authority admitted a nil service", ErrConfig)
346 }
347 session, err := newServerSession(l.epoch, transport, ceilings, core, l.cfg, release, registrationDeadline(ctx))
348 if err != nil {
349 _ = core.Close()
350 _ = transport.Close()
351 return nil, err
352 }
353 // The session deregisters itself once it reaches its terminal state, so a
354 // long-lived listener never accumulates retired connections.
355 session.onShutdown = func() { l.forgetSession(session) }
356 session.retire = func() bool { return l.tryRetire(session) }
357 session.closeListenerMux = func() { _ = l.mux.Close() }
358 l.mu.Lock()
359 if l.closed || l.retiring {
360 l.mu.Unlock()
361 // The run loop was never started, so Close (which waits on the run
362 // loop's done channel) would wait forever. Tear the unstarted session
363 // down directly instead.
364 session.shutdown()
365 return nil, ErrListenerClosed
366 }
367 l.sessions[session] = struct{}{}
368 l.mu.Unlock()
369 go session.run()
370 return session, nil
371}
372
373// Close stops accepting, unblocks a blocked Accept, cancels every in-flight
374// preamble and admission, closes every accepted carriage not yet owned by a
375// running session, closes every live session (ending each connection and
376// joining its workers), and unlinks only the socket inode this listener
377// created. It is idempotent and concurrent-safe.
378func (l *listener) Close() error {
379 l.closeOnce.Do(func() {
380 l.mu.Lock()
381 l.closed = true
382 l.mu.Unlock()
383 // Mark the listener closed before canceling, so an admission that wakes
384 // on the canceled context classifies its failure as ErrListenerClosed.
385 close(l.done)
386 l.cancel()
387 l.closeErr = l.mux.Close()
388 l.closePending()
389 // The accept loop exits on the closed carriage; every admission
390 // goroutine then finishes or drops its session because the listener is
391 // closed.
392 l.wg.Wait()
393 drained := make([]*serverSession, 0, cap(l.admitted))
394 for {
395 select {
396 case session := <-l.admitted:
397 drained = append(drained, session)
398 continue
399 default:
400 }
401 break
402 }
403 l.mu.Lock()
404 for session := range l.sessions {
405 drained = append(drained, session)
406 }
407 l.sessions = make(map[*serverSession]struct{})
408 l.mu.Unlock()
409 seen := make(map[*serverSession]struct{}, len(drained))
410 for _, session := range drained {
411 if _, duplicate := seen[session]; duplicate {
412 continue
413 }
414 seen[session] = struct{}{}
415 if err := session.Close(); err != nil {
416 l.closeErr = errors.Join(l.closeErr, err)
417 }
418 }
419 })
420 return l.closeErr
421}
422
423// Addr returns the bound endpoint path.
424func (l *listener) Addr() string {
425 if l == nil || l.mux == nil {
426 return ""
427 }
428 return l.mux.Addr()
429}
430
431// forgetSession drops one retired session from the live set.
432func (l *listener) forgetSession(session *serverSession) {
433 l.mu.Lock()
434 delete(l.sessions, session)
435 l.mu.Unlock()
436}
437
438func (l *listener) tryRetire(session *serverSession) bool {
439 l.mu.Lock()
440 defer l.mu.Unlock()
441 if l.closed || l.retiring || len(l.pending) != 0 || len(l.sessions) != 1 {
442 return false
443 }
444 if _, ok := l.sessions[session]; !ok {
445 return false
446 }
447 l.retiring = true
448 return true
449}
450
451func (l *listener) isClosed() bool {
452 if l == nil {
453 return true
454 }
455 select {
456 case <-l.done:
457 return true
458 default:
459 return false
460 }
461}
462
463// serverSession is one admitted broker connection: the brokerwire connection
464// state machine, the outbound frame path, one snapshot publisher, and the
465// bridged logical streams for exactly one client.
466type serverSession struct {
467 epoch ports.BrokerEpoch
468 scope brokerwire.Scope
469 conn *brokerwire.Connection
470 core ports.BrokerCoreService
471 transport wire.BoundedTransport
472 ceilings brokerwire.Ceilings
473 cfg Config
474 release func()
475 retire func() bool
476 closeListenerMux func()
477
478 ctx context.Context
479 cancel context.CancelFunc
480 wg sync.WaitGroup
481
482 subMu sync.Mutex
483 pub *publisher
484 preview ports.BrokerPreviewSubscription
485 previewGeneration ports.BrokerPreviewGeneration
486 previewForward *previewForwarder
487
488 streamsMu sync.Mutex
489 streams map[ports.BrokerStreamID]*serverStream
490
491 // onShutdown runs once, before done closes, so the owner can drop this
492 // session from its live set. It is set before run starts.
493 onShutdown func()
494
495 // registerDeadline is the absolute instant by which the client's Register
496 // must arrive: the accept-time handshake deadline that bounded the preamble
497 // and admission. Until Register is accepted the session is guarded by it, so
498 // a peer that stays silent after admission is settled instead of pinning the
499 // admitted core lease and the listener slot. Once Register is accepted the
500 // guard is stopped and the session runs on its own connection-lived context.
501 registerDeadline time.Time
502 // registered is set atomically the instant Register handling begins, before
503 // any connection state is touched. The pre-Register guard reads it when the
504 // registration deadline elapses, so a Register that arrives at the deadline
505 // boundary is never settled as a peer timeout even if the guard wins the
506 // race to the deadline.
507 registered atomic.Bool
508 // stopRegistration stops and joins the pre-Register guard. It is set by run
509 // and called once from the Register dispatch; a no-op for a zero deadline.
510 stopRegistration func()
511
512 errMu sync.Mutex
513 closeErr error
514 done chan struct{}
515}
516
517var _ ports.BrokerCoreService = (*serverSession)(nil)
518
519// registrationDeadline returns the absolute instant by which an accepted client
520// must send its Register: the accept-time handshake deadline that already
521// bounded the preamble and admission. The session extends that same budget over
522// the pre-Register wait, so a peer that completes the preamble and admission and
523// then stays silent cannot hold a core client lease or a listener slot past the
524// configured handshake budget. A missing deadline yields the zero time, which
525// disables the bound.
526func registrationDeadline(ctx context.Context) time.Time {
527 deadline, ok := ctx.Deadline()
528 if !ok {
529 return time.Time{}
530 }
531 return deadline
532}
533
534// newServerSession binds one admitted connection to its exact scope.
535//
536// registerDeadline is the absolute instant by which the client's Register must
537// arrive (the accept-time handshake deadline); the zero time disables the bound
538// and exists only for direct construction outside the listener.
539func newServerSession(epoch ports.BrokerEpoch, transport wire.BoundedTransport, ceilings brokerwire.Ceilings, core ports.BrokerCoreService, cfg Config, release func(), registerDeadline time.Time) (*serverSession, error) {
540 if epoch == 0 || transport == nil || core == nil {
541 return nil, ErrConfig
542 }
543 connection := core.ConnectionID()
544 if err := connection.Validate(); err != nil {
545 return nil, errors.Join(ErrConfig, err)
546 }
547 if err := ceilings.Validate(); err != nil {
548 return nil, errors.Join(ErrConfig, err)
549 }
550 conn, err := brokerwire.NewConnection(brokerwire.Scope{Epoch: epoch, Connection: connection})
551 if err != nil {
552 return nil, err
553 }
554 // The broker preamble was validated before this session was constructed,
555 // so the connection moves past New immediately and waits for exactly one
556 // Register.
557 if err := conn.SignalPreamble(); err != nil {
558 return nil, err
559 }
560 ctx, cancel := context.WithCancel(context.Background())
561 if release == nil {
562 release = func() {}
563 }
564 return &serverSession{
565 epoch: epoch,
566 scope: conn.Scope(),
567 conn: conn,
568 core: core,
569 transport: transport,
570 ceilings: ceilings,
571 cfg: cfg,
572 release: release,
573 ctx: ctx,
574 cancel: cancel,
575 streams: make(map[ports.BrokerStreamID]*serverStream),
576 registerDeadline: registerDeadline,
577 done: make(chan struct{}),
578 }, nil
579}
580
581// run is the connection's single reader. It decodes one whole broker envelope at
582// a time under the negotiated envelope and chunk ceilings; a malformed,
583// oversize, or wrong-direction frame, a protocol-ordering violation, or a
584// carriage error settles exactly this connection.
585func (s *serverSession) run() {
586 defer s.shutdown()
587 // The pre-Register phase shares the accept-time handshake deadline that
588 // bounded the preamble and admission. Arm it before the read loop and stop
589 // (and join) it as soon as Register is accepted, so a silent peer cannot pin
590 // the admitted core lease or the listener slot, and a healthy session is
591 // never disturbed by it.
592 s.stopRegistration = s.armRegistrationDeadline()
593 defer s.stopRegistration()
594 for {
595 envelope, err := s.transport.RecvBounded(s.ceilings.MaxReceiveEnvelopeBytes)
596 if err != nil {
597 s.setCloseErr(transportFailure(err))
598 return
599 }
600 message, err := brokerwire.DecodeClient(envelope.Payload, s.ceilings.MaxReceiveEnvelopeBytes, s.ceilings.StreamChunkLimit)
601 if err != nil {
602 s.setCloseErr(errors.Join(ErrMalformedFrame, err))
603 return
604 }
605 if err := s.dispatch(message); err != nil {
606 s.setCloseErr(err)
607 return
608 }
609 }
610}
611
612// armRegistrationDeadline enforces the accept-time registration budget on the
613// pre-Register phase. The peer must send its Register before the absolute
614// deadline that already bounded the preamble and admission; if it does not, the
615// guard settles the session, which closes the admitted core service and releases
616// the listener slot. The bound is derived from the session's own connection-lived
617// context, so shutdown cancels it too, and it is stopped as soon as Register is
618// accepted.
619//
620// The guard consults the registered marker before settling: a Register whose
621// handling already began is admitted even when the deadline elapses concurrently
622// with it, so the boundary between an on-time Register and a silent peer is the
623// marker, never a timer race.
624//
625// The returned stop function is idempotent and joins the guard goroutine, so an
626// accepted (or settled) session never leaves a timer goroutine behind. A zero
627// deadline disables the bound; only direct session construction uses that.
628func (s *serverSession) armRegistrationDeadline() func() {
629 if s.registerDeadline.IsZero() {
630 return func() {}
631 }
632 ctx, cancel := context.WithDeadline(s.ctx, s.registerDeadline)
633 stop := make(chan struct{})
634 done := make(chan struct{})
635 go func() {
636 defer close(done)
637 select {
638 case <-ctx.Done():
639 // Only the registration budget itself settles the session; a session
640 // cancel (shutdown) leaves the terminal cause to the reader. A
641 // Register already being handled makes the budget moot: the marker is
642 // set before any connection state is touched, so this check is the
643 // deadline boundary.
644 if errors.Is(ctx.Err(), context.DeadlineExceeded) && !s.registered.Load() {
645 s.abort(errors.Join(ErrRegistrationTimeout, context.DeadlineExceeded))
646 }
647 case <-stop:
648 }
649 }()
650 var once sync.Once
651 return func() {
652 once.Do(func() {
653 cancel()
654 close(stop)
655 <-done
656 })
657 }
658}
659
660// dispatch applies one client frame. A returned error is a protocol or carriage
661// violation that settles the connection; a refusal the peer can observe is sent
662// on the wire and returns nil.
663func (s *serverSession) dispatch(message brokerwire.ClientMessage) error {
664 switch m := message.(type) {
665 case brokerwire.Register:
666 // Mark registration as begun before touching connection state, so the
667 // pre-Register guard sharing the accept-time deadline can distinguish a
668 // Register arriving at the boundary from a silent peer and never settles
669 // this healthy connection.
670 s.registered.Store(true)
671 if err := s.conn.Register(); err != nil {
672 return errors.Join(ErrProtocol, err)
673 }
674 // The client met the accept-time registration budget: stop (and join)
675 // the pre-Register guard before answering, so this healthy connection is
676 // no longer bounded by the setup deadline.
677 s.stopRegistration()
678 retired := m.Build != s.cfg.Build && s.retire != nil && s.retire()
679 err := s.send(brokerwire.Registered{Epoch: s.epoch, Connection: s.scope.Connection, Build: s.cfg.Build, Retiring: retired})
680 if retired {
681 go func() {
682 s.closeListenerMux()
683 if s.cfg.OnRetire != nil {
684 s.cfg.OnRetire()
685 }
686 }()
687 }
688 return err
689 case brokerwire.Subscribe:
690 if !s.scopeMatches(m.Epoch, m.Connection) {
691 return s.refuseScope()
692 }
693 if err := s.conn.Subscribe(m.Generation); err != nil {
694 return errors.Join(ErrProtocol, err)
695 }
696 return s.retargetPublisher(m.Generation, m.Passive)
697 case brokerwire.Resync:
698 if !s.scopeMatches(m.Epoch, m.Connection) {
699 return s.refuseScope()
700 }
701 if err := s.conn.Resync(m.Generation); err != nil {
702 return errors.Join(ErrProtocol, err)
703 }
704 s.wakePublisher()
705 return nil
706 case brokerwire.Unsubscribe:
707 if !s.scopeMatches(m.Epoch, m.Connection) {
708 return s.refuseScope()
709 }
710 if err := s.conn.Unsubscribe(m.Generation); err != nil {
711 return errors.Join(ErrProtocol, err)
712 }
713 s.stopPublisher()
714 return nil
715 case brokerwire.AddHost:
716 if !s.scopeMatches(m.Epoch, m.Connection) {
717 return s.refuseScope()
718 }
719 return s.startMutation(m.Operation, brokerwire.MutationKindAddHost, func(ctx context.Context) (domain.RemoteRegistration, bool, error) {
720 registration, err := s.core.AddHost(ctx, m.Endpoint, m.Policy)
721 return registration, false, err
722 })
723 case brokerwire.RemoveHost:
724 if !s.scopeMatches(m.Epoch, m.Connection) {
725 return s.refuseScope()
726 }
727 return s.startMutation(m.Operation, brokerwire.MutationKindRemoveHost, func(ctx context.Context) (domain.RemoteRegistration, bool, error) {
728 removed, err := s.core.RemoveHost(ctx, m.Registration)
729 return domain.RemoteRegistration{}, removed, err
730 })
731 case brokerwire.UpdateHostPolicy:
732 if !s.scopeMatches(m.Epoch, m.Connection) {
733 return s.refuseScope()
734 }
735 return s.startMutation(m.Operation, brokerwire.MutationKindUpdateHostPolicy, func(ctx context.Context) (domain.RemoteRegistration, bool, error) {
736 registration, err := s.core.UpdateHostPolicy(ctx, m.Registration, m.Policy)
737 return registration, false, err
738 })
739 case brokerwire.Reconcile:
740 if !s.scopeMatches(m.Epoch, m.Connection) {
741 return s.refuseScope()
742 }
743 s.core.RequestReconcile(m.Registration.Endpoint)
744 return nil
745 case brokerwire.OpenStream:
746 if !s.scopeMatches(m.Epoch, m.Connection) {
747 return s.refuseStream(m.Stream, scopeRefusal())
748 }
749 s.startStream(m)
750 return nil
751 case brokerwire.ClientStreamData:
752 if !s.scopeMatches(m.Epoch, m.Connection) {
753 return s.refuseScope()
754 }
755 return s.deliverStreamData(m)
756 case brokerwire.CloseStream:
757 if !s.scopeMatches(m.Epoch, m.Connection) {
758 return s.refuseScope()
759 }
760 s.closeStreamByPeer(m.Stream)
761 return nil
762 case brokerwire.StartPreview:
763 if !s.scopeMatches(m.Epoch, m.Connection) {
764 return s.refuseScope()
765 }
766 return s.startPreview(m)
767 case brokerwire.CancelPreview:
768 if !s.scopeMatches(m.Epoch, m.Connection) {
769 return s.refuseScope()
770 }
771 s.subMu.Lock()
772 if s.preview != nil && m.Generation == s.previewGeneration {
773 s.preview.Close()
774 s.preview = nil
775 }
776 s.subMu.Unlock()
777 return nil
778 default:
779 return errors.Join(ErrProtocol, fmt.Errorf("brokeripc: unexpected client message %T", message))
780 }
781}
782
783// scopeMatches reports whether one frame carries the epoch and connection
784// identity assigned at accept. Every frame must: a mismatched frame is never
785// applied to this connection's state.
786func (s *serverSession) scopeMatches(epoch ports.BrokerEpoch, connection ports.BrokerConnectionID) bool {
787 return epoch == s.epoch && connection == s.scope.Connection
788}
789
790// refuseScope drops one frame whose scope is not this connection's. It is
791// deliberately not an error: the frame is fenced rather than obeyed, and the
792// connection keeps serving its own scope. A stale stream open is answered with
793// a typed stale refusal instead (see refuseStream).
794func (s *serverSession) refuseScope() error { return nil }
795
796// scopeRefusal is the typed refusal for a frame that does not carry this
797// connection's accepted scope: a stale admission the peer can classify.
798func scopeRefusal() error {
799 return errors.Join(ErrScopeMismatch, ports.BrokerAdmissionStale)
800}
801
802// refuseStream tells the peer one logical stream was refused, with the closed
803// admission or broker taxonomy carried on the detail.
804func (s *serverSession) refuseStream(stream ports.BrokerStreamID, err error) error {
805 detail := errorDetail(err)
806 return s.send(brokerwire.StreamClosed{
807 Epoch: s.epoch, Connection: s.scope.Connection, Stream: stream,
808 Error: detail, HasError: detail != (brokerwire.ErrorDetail{}),
809 })
810}
811
812// send encodes one server frame under the negotiated ceilings and writes it.
813func (s *serverSession) send(message brokerwire.ServerMessage) error {
814 payload, err := brokerwire.EncodeServer(message, s.ceilings.MaxReceiveEnvelopeBytes, s.ceilings.StreamChunkLimit)
815 if err != nil {
816 return err
817 }
818 return s.sendPayload(payload)
819}
820
821// sendPayload writes one already encoded server envelope.
822func (s *serverSession) sendPayload(payload []byte) error {
823 if err := s.transport.Send(wire.Envelope{Payload: payload}); err != nil {
824 return transportFailure(err)
825 }
826 return nil
827}
828
829// setCloseErr records the first terminal cause. Later causes never overwrite a
830// cause that already settled the connection, so a diagnostic from a worker that
831// observed the same terminal state cannot mask the original.
832func (s *serverSession) setCloseErr(err error) {
833 if err == nil {
834 return
835 }
836 s.errMu.Lock()
837 if s.closeErr == nil {
838 s.closeErr = err
839 }
840 s.errMu.Unlock()
841}
842
843// terminalErr reports the cause that settled the connection, or nil for an
844// orderly end.
845func (s *serverSession) terminalErr() error {
846 s.errMu.Lock()
847 defer s.errMu.Unlock()
848 return s.closeErr
849}
850
851// abort records a terminal cause from a worker goroutine and closes the
852// carriage so the reader observes it promptly.
853func (s *serverSession) abort(err error) {
854 s.setCloseErr(err)
855 _ = s.transport.Close()
856}
857
858// shutdown tears the session down exactly once, in a fixed order: cancel every
859// owned context, stop the publisher, close the outer carriage (which unblocks a
860// blocked read and any frame send), settle every bridged stream, close the
861// brokerwire connection state, join every worker, release the stream
862// resources, close the admitted core service, and release the listener's
863// client slot.
864//
865// The outer carriage closes before the bridged streams are settled because
866// settling a stream joins that stream's own carriage writer, and a writer
867// parked on a non-reading peer is only interrupted when the outer carriage
868// closes: settling streams first would wait on the peer instead of settling the
869// connection, so the session slot and done would never be released.
870func (s *serverSession) shutdown() {
871 s.cancel()
872 s.stopPublisher()
873 s.subMu.Lock()
874 if s.previewForward != nil {
875 s.previewForward.stop()
876 s.previewForward = nil
877 }
878 if s.preview != nil {
879 s.preview.Close()
880 s.preview = nil
881 }
882 s.subMu.Unlock()
883 _ = s.transport.Close()
884 s.closeStreams()
885 _ = s.conn.BeginClose()
886 s.wg.Wait()
887 _ = s.conn.FinishClose()
888 if s.core != nil {
889 if err := s.core.Close(); err != nil {
890 s.setCloseErr(err)
891 }
892 }
893 if s.release != nil {
894 s.release()
895 }
896 if s.onShutdown != nil {
897 s.onShutdown()
898 }
899 close(s.done)
900}
901
902// Close ends the connection and waits for every worker to stop. It is
903// idempotent and concurrent-safe. It reports a non-nil error only for a session
904// that ended on a protocol violation or an owned-resource failure: a peer
905// disconnect and a local close are orderly ends, so an owner draining accepted
906// sessions does not see a peer's departure as its own failure.
907func (s *serverSession) Close() error {
908 if s == nil {
909 return nil
910 }
911 _ = s.transport.Close()
912 <-s.done
913 err := s.terminalErr()
914 if orderlyDisconnect(err) {
915 return nil
916 }
917 return err
918}
919
920// orderlyDisconnect reports whether one terminal cause is a peer disconnect or a
921// local close rather than a failure. A registration timeout is a peer that
922// completed admission and then stayed silent: the session settles, but the
923// outcome is the peer's departure, so a listener draining its sessions never
924// reports it as its own failure.
925//
926// A peer that closes abruptly can surface a reset, a broken pipe on the next
927// write, or a local-close sentinel from a concurrently interrupted read instead
928// of a clean EOF; all are still
929// the peer's or this side's ordinary departure, never a sandbox failure. The
930// same holds for a frame write interrupted by this side's own carriage close.
931func orderlyDisconnect(err error) bool {
932 return err == nil ||
933 errors.Is(err, io.EOF) ||
934 errors.Is(err, syscall.ECONNRESET) ||
935 errors.Is(err, syscall.EPIPE) ||
936 errors.Is(err, net.ErrClosed) ||
937 errors.Is(err, fs.ErrClosed) ||
938 errors.Is(err, streamframe.ErrClosed) ||
939 errors.Is(err, ErrConnectionClosed) ||
940 errors.Is(err, ErrSessionClosed) ||
941 errors.Is(err, ErrRegistrationTimeout)
942}
943
944// ConnectionID returns the identity assigned to this connection at accept.
945func (s *serverSession) ConnectionID() ports.BrokerConnectionID { return s.scope.Connection }
946
947// NextStreamID allocates the next logical stream identity of the admitted core
948// connection. The wire identities are allocated by the calling service and are
949// carried to this listener unchanged; this delegating form exists so a composed
950// caller can use the session as a scoped ports.BrokerService. It never
951// re-allocates a wire identity and performs no I/O.
952func (s *serverSession) NextStreamID() (ports.BrokerStreamID, error) {
953 if s == nil {
954 return 0, ErrSessionClosed
955 }
956 return s.core.NextStreamID()
957}
958
959// Done closes exactly once when this session is terminal: the carriage failed,
960// the peer disconnected, or the owner closed it. Err is stable afterwards.
961func (s *serverSession) Done() <-chan struct{} { return s.done }
962
963// Err returns the cause that settled this session, or nil for an orderly end
964// (a peer disconnect, a registration timeout, or a local close). The underlying
965// terminal cause is never overwritten, so the result is stable once Done is
966// closed. A cause Close already reports as orderly is reported as nil here too,
967// so a client that observes Done/Err never mistakes a peer's ordinary departure
968// for a sandbox failure.
969func (s *serverSession) Err() error {
970 err := s.terminalErr()
971 if orderlyDisconnect(err) {
972 return nil
973 }
974 return err
975}
976
977// Snapshot returns the admitted core service's current publication.
978func (s *serverSession) Snapshot() ports.BrokerSnapshot { return s.core.Snapshot() }
979
980// Subscribe returns the admitted core service's subscription. The wire
981// protocol's Subscribe is driven by the connection's reader, which publishes
982// the broker core's newest snapshot for the client's generation.
983func (s *serverSession) Subscribe() (ports.BrokerSubscription, error) { return s.core.Subscribe() }
984func (s *serverSession) SubscribePreview(request ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
985 return s.core.SubscribePreview(request)
986}
987
988// OpenStream delegates one scope-checked stream request to the admitted core
989// service. The connection's own reader drives the same core through
990// openStreamMessage, which additionally bridges the stream onto the wire; this
991// method exists so a composed caller can use the session as a scoped
992// ports.BrokerService. The stream identity is the one the caller already
993// allocated (NextStreamID) and is transmitted unchanged: this listener never
994// re-allocates a wire identity, whether the request arrives over the wire or
995// through this in-process delegate.
996func (s *serverSession) OpenEnvelopeStream(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
997 if err := s.checkRequestScope(request.Epoch, request.Connection); err != nil {
998 return nil, err
999 }
1000 scoped := request
1001 scoped.Epoch = s.epoch
1002 scoped.Connection = s.scope.Connection
1003 return s.core.OpenEnvelopeStream(ctx, scoped)
1004}
1005
1006// CloseStream retires one scope-checked stream. An absent connection identity is
1007// filled from this session exactly like OpenStream, so a zero identity never
1008// reaches the core as a stale request, and a foreign one is still refused.
1009func (s *serverSession) CloseStream(connection ports.BrokerConnectionID, stream ports.BrokerStreamID) error {
1010 if err := s.checkRequestScope(s.epoch, connection); err != nil {
1011 return err
1012 }
1013 return s.core.CloseStream(s.scope.Connection, stream)
1014}
1015
1016// AddHost delegates one mutating operation to the admitted core service.
1017func (s *serverSession) AddHost(ctx context.Context, endpoint string, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
1018 return s.core.AddHost(ctx, endpoint, policy)
1019}
1020
1021// RemoveHost delegates one mutating operation to the admitted core service.
1022func (s *serverSession) RemoveHost(ctx context.Context, expected domain.RemoteRegistration) (bool, error) {
1023 return s.core.RemoveHost(ctx, expected)
1024}
1025
1026// UpdateHostPolicy delegates one mutating operation to the admitted core service.
1027func (s *serverSession) UpdateHostPolicy(ctx context.Context, expected domain.RemoteRegistration, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
1028 return s.core.UpdateHostPolicy(ctx, expected, policy)
1029}
1030
1031// RequestReconcile delegates one reconcile hint to the admitted core service.
1032func (s *serverSession) RequestReconcile(endpoint string) { s.core.RequestReconcile(endpoint) }
1033
1034// checkRequestScope refuses a caller-supplied identity that is neither absent
1035// nor this connection's own, so a caller can never steer another connection's
1036// work through this session.
1037func (s *serverSession) checkRequestScope(epoch ports.BrokerEpoch, connection ports.BrokerConnectionID) error {
1038 if epoch != 0 && epoch != s.epoch {
1039 return errors.Join(ErrScopeMismatch, ports.BrokerError{Code: ports.BrokerErrorStaleEpoch})
1040 }
1041 if !connection.IsZero() && connection != s.scope.Connection {
1042 return errors.Join(ErrScopeMismatch, ports.BrokerAdmissionStale)
1043 }
1044 return nil
1045}