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 // The session owns the carriage from here: drop it from pending in the
368 // same critical section, so a Register racing admit's own untrack never
369 // sees a stale pending carriage and refuses to retire.
370 delete(l.pending, transport)
371 l.sessions[session] = struct{}{}
372 l.mu.Unlock()
373 go session.run()
374 return session, nil
375}
376
377// Close stops accepting, unblocks a blocked Accept, cancels every in-flight
378// preamble and admission, closes every accepted carriage not yet owned by a
379// running session, closes every live session (ending each connection and
380// joining its workers), and unlinks only the socket inode this listener
381// created. It is idempotent and concurrent-safe.
382func (l *listener) Close() error {
383 l.closeOnce.Do(func() {
384 l.mu.Lock()
385 l.closed = true
386 l.mu.Unlock()
387 // Mark the listener closed before canceling, so an admission that wakes
388 // on the canceled context classifies its failure as ErrListenerClosed.
389 close(l.done)
390 l.cancel()
391 l.closeErr = l.mux.Close()
392 l.closePending()
393 // The accept loop exits on the closed carriage; every admission
394 // goroutine then finishes or drops its session because the listener is
395 // closed.
396 l.wg.Wait()
397 drained := make([]*serverSession, 0, cap(l.admitted))
398 for {
399 select {
400 case session := <-l.admitted:
401 drained = append(drained, session)
402 continue
403 default:
404 }
405 break
406 }
407 l.mu.Lock()
408 for session := range l.sessions {
409 drained = append(drained, session)
410 }
411 l.sessions = make(map[*serverSession]struct{})
412 l.mu.Unlock()
413 seen := make(map[*serverSession]struct{}, len(drained))
414 for _, session := range drained {
415 if _, duplicate := seen[session]; duplicate {
416 continue
417 }
418 seen[session] = struct{}{}
419 if err := session.Close(); err != nil {
420 l.closeErr = errors.Join(l.closeErr, err)
421 }
422 }
423 })
424 return l.closeErr
425}
426
427// Addr returns the bound endpoint path.
428func (l *listener) Addr() string {
429 if l == nil || l.mux == nil {
430 return ""
431 }
432 return l.mux.Addr()
433}
434
435// forgetSession drops one retired session from the live set.
436func (l *listener) forgetSession(session *serverSession) {
437 l.mu.Lock()
438 delete(l.sessions, session)
439 l.mu.Unlock()
440}
441
442func (l *listener) tryRetire(session *serverSession) bool {
443 l.mu.Lock()
444 defer l.mu.Unlock()
445 if l.closed || l.retiring || len(l.pending) != 0 || len(l.sessions) != 1 {
446 return false
447 }
448 if _, ok := l.sessions[session]; !ok {
449 return false
450 }
451 l.retiring = true
452 return true
453}
454
455func (l *listener) isClosed() bool {
456 if l == nil {
457 return true
458 }
459 select {
460 case <-l.done:
461 return true
462 default:
463 return false
464 }
465}
466
467// serverSession is one admitted broker connection: the brokerwire connection
468// state machine, the outbound frame path, one snapshot publisher, and the
469// bridged logical streams for exactly one client.
470type serverSession struct {
471 epoch ports.BrokerEpoch
472 scope brokerwire.Scope
473 conn *brokerwire.Connection
474 core ports.BrokerCoreService
475 transport wire.BoundedTransport
476 ceilings brokerwire.Ceilings
477 cfg Config
478 release func()
479 retire func() bool
480 closeListenerMux func()
481
482 ctx context.Context
483 cancel context.CancelFunc
484 wg sync.WaitGroup
485
486 subMu sync.Mutex
487 pub *publisher
488 preview ports.BrokerPreviewSubscription
489 previewGeneration ports.BrokerPreviewGeneration
490 previewForward *previewForwarder
491
492 streamsMu sync.Mutex
493 streams map[ports.BrokerStreamID]*serverStream
494
495 // onShutdown runs once, before done closes, so the owner can drop this
496 // session from its live set. It is set before run starts.
497 onShutdown func()
498
499 // registerDeadline is the absolute instant by which the client's Register
500 // must arrive: the accept-time handshake deadline that bounded the preamble
501 // and admission. Until Register is accepted the session is guarded by it, so
502 // a peer that stays silent after admission is settled instead of pinning the
503 // admitted core lease and the listener slot. Once Register is accepted the
504 // guard is stopped and the session runs on its own connection-lived context.
505 registerDeadline time.Time
506 // registered is set atomically the instant Register handling begins, before
507 // any connection state is touched. The pre-Register guard reads it when the
508 // registration deadline elapses, so a Register that arrives at the deadline
509 // boundary is never settled as a peer timeout even if the guard wins the
510 // race to the deadline.
511 registered atomic.Bool
512 // stopRegistration stops and joins the pre-Register guard. It is set by run
513 // and called once from the Register dispatch; a no-op for a zero deadline.
514 stopRegistration func()
515
516 errMu sync.Mutex
517 closeErr error
518 done chan struct{}
519}
520
521var _ ports.BrokerCoreService = (*serverSession)(nil)
522
523// registrationDeadline returns the absolute instant by which an accepted client
524// must send its Register: the accept-time handshake deadline that already
525// bounded the preamble and admission. The session extends that same budget over
526// the pre-Register wait, so a peer that completes the preamble and admission and
527// then stays silent cannot hold a core client lease or a listener slot past the
528// configured handshake budget. A missing deadline yields the zero time, which
529// disables the bound.
530func registrationDeadline(ctx context.Context) time.Time {
531 deadline, ok := ctx.Deadline()
532 if !ok {
533 return time.Time{}
534 }
535 return deadline
536}
537
538// newServerSession binds one admitted connection to its exact scope.
539//
540// registerDeadline is the absolute instant by which the client's Register must
541// arrive (the accept-time handshake deadline); the zero time disables the bound
542// and exists only for direct construction outside the listener.
543func newServerSession(epoch ports.BrokerEpoch, transport wire.BoundedTransport, ceilings brokerwire.Ceilings, core ports.BrokerCoreService, cfg Config, release func(), registerDeadline time.Time) (*serverSession, error) {
544 if epoch == 0 || transport == nil || core == nil {
545 return nil, ErrConfig
546 }
547 connection := core.ConnectionID()
548 if err := connection.Validate(); err != nil {
549 return nil, errors.Join(ErrConfig, err)
550 }
551 if err := ceilings.Validate(); err != nil {
552 return nil, errors.Join(ErrConfig, err)
553 }
554 conn, err := brokerwire.NewConnection(brokerwire.Scope{Epoch: epoch, Connection: connection})
555 if err != nil {
556 return nil, err
557 }
558 // The broker preamble was validated before this session was constructed,
559 // so the connection moves past New immediately and waits for exactly one
560 // Register.
561 if err := conn.SignalPreamble(); err != nil {
562 return nil, err
563 }
564 ctx, cancel := context.WithCancel(context.Background())
565 if release == nil {
566 release = func() {}
567 }
568 return &serverSession{
569 epoch: epoch,
570 scope: conn.Scope(),
571 conn: conn,
572 core: core,
573 transport: transport,
574 ceilings: ceilings,
575 cfg: cfg,
576 release: release,
577 ctx: ctx,
578 cancel: cancel,
579 streams: make(map[ports.BrokerStreamID]*serverStream),
580 registerDeadline: registerDeadline,
581 done: make(chan struct{}),
582 }, nil
583}
584
585// run is the connection's single reader. It decodes one whole broker envelope at
586// a time under the negotiated envelope and chunk ceilings; a malformed,
587// oversize, or wrong-direction frame, a protocol-ordering violation, or a
588// carriage error settles exactly this connection.
589func (s *serverSession) run() {
590 defer s.shutdown()
591 // The pre-Register phase shares the accept-time handshake deadline that
592 // bounded the preamble and admission. Arm it before the read loop and stop
593 // (and join) it as soon as Register is accepted, so a silent peer cannot pin
594 // the admitted core lease or the listener slot, and a healthy session is
595 // never disturbed by it.
596 s.stopRegistration = s.armRegistrationDeadline()
597 defer s.stopRegistration()
598 for {
599 envelope, err := s.transport.RecvBounded(s.ceilings.MaxReceiveEnvelopeBytes)
600 if err != nil {
601 s.setCloseErr(transportFailure(err))
602 return
603 }
604 message, err := brokerwire.DecodeClient(envelope.Payload, s.ceilings.MaxReceiveEnvelopeBytes, s.ceilings.StreamChunkLimit)
605 if err != nil {
606 s.setCloseErr(errors.Join(ErrMalformedFrame, err))
607 return
608 }
609 if err := s.dispatch(message); err != nil {
610 s.setCloseErr(err)
611 return
612 }
613 }
614}
615
616// armRegistrationDeadline enforces the accept-time registration budget on the
617// pre-Register phase. The peer must send its Register before the absolute
618// deadline that already bounded the preamble and admission; if it does not, the
619// guard settles the session, which closes the admitted core service and releases
620// the listener slot. The bound is derived from the session's own connection-lived
621// context, so shutdown cancels it too, and it is stopped as soon as Register is
622// accepted.
623//
624// The guard consults the registered marker before settling: a Register whose
625// handling already began is admitted even when the deadline elapses concurrently
626// with it, so the boundary between an on-time Register and a silent peer is the
627// marker, never a timer race.
628//
629// The returned stop function is idempotent and joins the guard goroutine, so an
630// accepted (or settled) session never leaves a timer goroutine behind. A zero
631// deadline disables the bound; only direct session construction uses that.
632func (s *serverSession) armRegistrationDeadline() func() {
633 if s.registerDeadline.IsZero() {
634 return func() {}
635 }
636 ctx, cancel := context.WithDeadline(s.ctx, s.registerDeadline)
637 stop := make(chan struct{})
638 done := make(chan struct{})
639 go func() {
640 defer close(done)
641 select {
642 case <-ctx.Done():
643 // Only the registration budget itself settles the session; a session
644 // cancel (shutdown) leaves the terminal cause to the reader. A
645 // Register already being handled makes the budget moot: the marker is
646 // set before any connection state is touched, so this check is the
647 // deadline boundary.
648 if errors.Is(ctx.Err(), context.DeadlineExceeded) && !s.registered.Load() {
649 s.abort(errors.Join(ErrRegistrationTimeout, context.DeadlineExceeded))
650 }
651 case <-stop:
652 }
653 }()
654 var once sync.Once
655 return func() {
656 once.Do(func() {
657 cancel()
658 close(stop)
659 <-done
660 })
661 }
662}
663
664// dispatch applies one client frame. A returned error is a protocol or carriage
665// violation that settles the connection; a refusal the peer can observe is sent
666// on the wire and returns nil.
667func (s *serverSession) dispatch(message brokerwire.ClientMessage) error {
668 switch m := message.(type) {
669 case brokerwire.Register:
670 // Mark registration as begun before touching connection state, so the
671 // pre-Register guard sharing the accept-time deadline can distinguish a
672 // Register arriving at the boundary from a silent peer and never settles
673 // this healthy connection.
674 s.registered.Store(true)
675 if err := s.conn.Register(); err != nil {
676 return errors.Join(ErrProtocol, err)
677 }
678 // The client met the accept-time registration budget: stop (and join)
679 // the pre-Register guard before answering, so this healthy connection is
680 // no longer bounded by the setup deadline.
681 s.stopRegistration()
682 retired := m.Build != s.cfg.Build && s.retire != nil && s.retire()
683 err := s.send(brokerwire.Registered{Epoch: s.epoch, Connection: s.scope.Connection, Build: s.cfg.Build, Retiring: retired})
684 if retired {
685 go func() {
686 s.closeListenerMux()
687 if s.cfg.OnRetire != nil {
688 s.cfg.OnRetire()
689 }
690 }()
691 }
692 return err
693 case brokerwire.Subscribe:
694 if !s.scopeMatches(m.Epoch, m.Connection) {
695 return s.refuseScope()
696 }
697 if err := s.conn.Subscribe(m.Generation); err != nil {
698 return errors.Join(ErrProtocol, err)
699 }
700 return s.retargetPublisher(m.Generation, m.Passive)
701 case brokerwire.Resync:
702 if !s.scopeMatches(m.Epoch, m.Connection) {
703 return s.refuseScope()
704 }
705 if err := s.conn.Resync(m.Generation); err != nil {
706 return errors.Join(ErrProtocol, err)
707 }
708 s.wakePublisher()
709 return nil
710 case brokerwire.Unsubscribe:
711 if !s.scopeMatches(m.Epoch, m.Connection) {
712 return s.refuseScope()
713 }
714 if err := s.conn.Unsubscribe(m.Generation); err != nil {
715 return errors.Join(ErrProtocol, err)
716 }
717 s.stopPublisher()
718 return nil
719 case brokerwire.AddHost:
720 if !s.scopeMatches(m.Epoch, m.Connection) {
721 return s.refuseScope()
722 }
723 return s.startMutation(m.Operation, brokerwire.MutationKindAddHost, func(ctx context.Context) (domain.RemoteRegistration, bool, error) {
724 registration, err := s.core.AddHost(ctx, m.Endpoint, m.Policy)
725 return registration, false, err
726 })
727 case brokerwire.RemoveHost:
728 if !s.scopeMatches(m.Epoch, m.Connection) {
729 return s.refuseScope()
730 }
731 return s.startMutation(m.Operation, brokerwire.MutationKindRemoveHost, func(ctx context.Context) (domain.RemoteRegistration, bool, error) {
732 removed, err := s.core.RemoveHost(ctx, m.Registration)
733 return domain.RemoteRegistration{}, removed, err
734 })
735 case brokerwire.UpdateHostPolicy:
736 if !s.scopeMatches(m.Epoch, m.Connection) {
737 return s.refuseScope()
738 }
739 return s.startMutation(m.Operation, brokerwire.MutationKindUpdateHostPolicy, func(ctx context.Context) (domain.RemoteRegistration, bool, error) {
740 registration, err := s.core.UpdateHostPolicy(ctx, m.Registration, m.Policy)
741 return registration, false, err
742 })
743 case brokerwire.Reconcile:
744 if !s.scopeMatches(m.Epoch, m.Connection) {
745 return s.refuseScope()
746 }
747 s.core.RequestReconcile(m.Registration.Endpoint)
748 return nil
749 case brokerwire.OpenStream:
750 if !s.scopeMatches(m.Epoch, m.Connection) {
751 return s.refuseStream(m.Stream, scopeRefusal())
752 }
753 s.startStream(m)
754 return nil
755 case brokerwire.ClientStreamData:
756 if !s.scopeMatches(m.Epoch, m.Connection) {
757 return s.refuseScope()
758 }
759 return s.deliverStreamData(m)
760 case brokerwire.CloseStream:
761 if !s.scopeMatches(m.Epoch, m.Connection) {
762 return s.refuseScope()
763 }
764 s.closeStreamByPeer(m.Stream)
765 return nil
766 case brokerwire.StreamWindowUpdate:
767 if !s.scopeMatches(m.Epoch, m.Connection) {
768 return s.refuseScope()
769 }
770 return s.grantStreamCredit(m)
771 case brokerwire.StartPreview:
772 if !s.scopeMatches(m.Epoch, m.Connection) {
773 return s.refuseScope()
774 }
775 return s.startPreview(m)
776 case brokerwire.CancelPreview:
777 if !s.scopeMatches(m.Epoch, m.Connection) {
778 return s.refuseScope()
779 }
780 s.subMu.Lock()
781 if s.preview != nil && m.Generation == s.previewGeneration {
782 s.preview.Close()
783 s.preview = nil
784 }
785 s.subMu.Unlock()
786 return nil
787 default:
788 return errors.Join(ErrProtocol, fmt.Errorf("brokeripc: unexpected client message %T", message))
789 }
790}
791
792// scopeMatches reports whether one frame carries the epoch and connection
793// identity assigned at accept. Every frame must: a mismatched frame is never
794// applied to this connection's state.
795func (s *serverSession) scopeMatches(epoch ports.BrokerEpoch, connection ports.BrokerConnectionID) bool {
796 return epoch == s.epoch && connection == s.scope.Connection
797}
798
799// refuseScope drops one frame whose scope is not this connection's. It is
800// deliberately not an error: the frame is fenced rather than obeyed, and the
801// connection keeps serving its own scope. A stale stream open is answered with
802// a typed stale refusal instead (see refuseStream).
803func (s *serverSession) refuseScope() error { return nil }
804
805// scopeRefusal is the typed refusal for a frame that does not carry this
806// connection's accepted scope: a stale admission the peer can classify.
807func scopeRefusal() error {
808 return errors.Join(ErrScopeMismatch, ports.BrokerAdmissionStale)
809}
810
811// refuseStream tells the peer one logical stream was refused, with the closed
812// admission or broker taxonomy carried on the detail.
813func (s *serverSession) refuseStream(stream ports.BrokerStreamID, err error) error {
814 detail := errorDetail(err)
815 return s.send(brokerwire.StreamClosed{
816 Epoch: s.epoch, Connection: s.scope.Connection, Stream: stream,
817 Error: detail, HasError: detail != (brokerwire.ErrorDetail{}),
818 })
819}
820
821// send encodes one server frame under the negotiated ceilings and writes it.
822func (s *serverSession) send(message brokerwire.ServerMessage) error {
823 payload, err := brokerwire.EncodeServer(message, s.ceilings.MaxReceiveEnvelopeBytes, s.ceilings.StreamChunkLimit)
824 if err != nil {
825 return err
826 }
827 return s.sendPayload(payload)
828}
829
830// sendPayload writes one already encoded server envelope.
831func (s *serverSession) sendPayload(payload []byte) error {
832 if err := s.transport.Send(wire.Envelope{Payload: payload}); err != nil {
833 return transportFailure(err)
834 }
835 return nil
836}
837
838// setCloseErr records the first terminal cause. Later causes never overwrite a
839// cause that already settled the connection, so a diagnostic from a worker that
840// observed the same terminal state cannot mask the original.
841func (s *serverSession) setCloseErr(err error) {
842 if err == nil {
843 return
844 }
845 s.errMu.Lock()
846 if s.closeErr == nil {
847 s.closeErr = err
848 }
849 s.errMu.Unlock()
850}
851
852// terminalErr reports the cause that settled the connection, or nil for an
853// orderly end.
854func (s *serverSession) terminalErr() error {
855 s.errMu.Lock()
856 defer s.errMu.Unlock()
857 return s.closeErr
858}
859
860// abort records a terminal cause from a worker goroutine and closes the
861// carriage so the reader observes it promptly.
862func (s *serverSession) abort(err error) {
863 s.setCloseErr(err)
864 _ = s.transport.Close()
865}
866
867// shutdown tears the session down exactly once, in a fixed order: cancel every
868// owned context, stop the publisher, close the outer carriage (which unblocks a
869// blocked read and any frame send), settle every bridged stream, close the
870// brokerwire connection state, join every worker, release the stream
871// resources, close the admitted core service, and release the listener's
872// client slot.
873//
874// The outer carriage closes before the bridged streams are settled because
875// settling a stream joins that stream's own carriage writer, and a writer
876// parked on a non-reading peer is only interrupted when the outer carriage
877// closes: settling streams first would wait on the peer instead of settling the
878// connection, so the session slot and done would never be released.
879func (s *serverSession) shutdown() {
880 s.cancel()
881 s.stopPublisher()
882 s.subMu.Lock()
883 if s.previewForward != nil {
884 s.previewForward.stop()
885 s.previewForward = nil
886 }
887 if s.preview != nil {
888 s.preview.Close()
889 s.preview = nil
890 }
891 s.subMu.Unlock()
892 _ = s.transport.Close()
893 s.closeStreams()
894 _ = s.conn.BeginClose()
895 s.wg.Wait()
896 _ = s.conn.FinishClose()
897 if s.core != nil {
898 if err := s.core.Close(); err != nil {
899 s.setCloseErr(err)
900 }
901 }
902 if s.release != nil {
903 s.release()
904 }
905 if s.onShutdown != nil {
906 s.onShutdown()
907 }
908 close(s.done)
909}
910
911// Close ends the connection and waits for every worker to stop. It is
912// idempotent and concurrent-safe. It reports a non-nil error only for a session
913// that ended on a protocol violation or an owned-resource failure: a peer
914// disconnect and a local close are orderly ends, so an owner draining accepted
915// sessions does not see a peer's departure as its own failure.
916func (s *serverSession) Close() error {
917 if s == nil {
918 return nil
919 }
920 _ = s.transport.Close()
921 <-s.done
922 err := s.terminalErr()
923 if orderlyDisconnect(err) {
924 return nil
925 }
926 return err
927}
928
929// orderlyDisconnect reports whether one terminal cause is a peer disconnect or a
930// local close rather than a failure. A registration timeout is a peer that
931// completed admission and then stayed silent: the session settles, but the
932// outcome is the peer's departure, so a listener draining its sessions never
933// reports it as its own failure.
934//
935// A peer that closes abruptly can surface a reset, a broken pipe on the next
936// write, or a local-close sentinel from a concurrently interrupted read instead
937// of a clean EOF; all are still
938// the peer's or this side's ordinary departure, never a sandbox failure. The
939// same holds for a frame write interrupted by this side's own carriage close.
940func orderlyDisconnect(err error) bool {
941 return err == nil ||
942 errors.Is(err, io.EOF) ||
943 errors.Is(err, syscall.ECONNRESET) ||
944 errors.Is(err, syscall.EPIPE) ||
945 errors.Is(err, net.ErrClosed) ||
946 errors.Is(err, fs.ErrClosed) ||
947 errors.Is(err, streamframe.ErrClosed) ||
948 errors.Is(err, ErrConnectionClosed) ||
949 errors.Is(err, ErrSessionClosed) ||
950 errors.Is(err, ErrRegistrationTimeout)
951}
952
953// ConnectionID returns the identity assigned to this connection at accept.
954func (s *serverSession) ConnectionID() ports.BrokerConnectionID { return s.scope.Connection }
955
956// NextStreamID allocates the next logical stream identity of the admitted core
957// connection. The wire identities are allocated by the calling service and are
958// carried to this listener unchanged; this delegating form exists so a composed
959// caller can use the session as a scoped ports.BrokerService. It never
960// re-allocates a wire identity and performs no I/O.
961func (s *serverSession) NextStreamID() (ports.BrokerStreamID, error) {
962 if s == nil {
963 return 0, ErrSessionClosed
964 }
965 return s.core.NextStreamID()
966}
967
968// Done closes exactly once when this session is terminal: the carriage failed,
969// the peer disconnected, or the owner closed it. Err is stable afterwards.
970func (s *serverSession) Done() <-chan struct{} { return s.done }
971
972// Err returns the cause that settled this session, or nil for an orderly end
973// (a peer disconnect, a registration timeout, or a local close). The underlying
974// terminal cause is never overwritten, so the result is stable once Done is
975// closed. A cause Close already reports as orderly is reported as nil here too,
976// so a client that observes Done/Err never mistakes a peer's ordinary departure
977// for a sandbox failure.
978func (s *serverSession) Err() error {
979 err := s.terminalErr()
980 if orderlyDisconnect(err) {
981 return nil
982 }
983 return err
984}
985
986// Snapshot returns the admitted core service's current publication.
987func (s *serverSession) Snapshot() ports.BrokerSnapshot { return s.core.Snapshot() }
988
989// Subscribe returns the admitted core service's subscription. The wire
990// protocol's Subscribe is driven by the connection's reader, which publishes
991// the broker core's newest snapshot for the client's generation.
992func (s *serverSession) Subscribe() (ports.BrokerSubscription, error) { return s.core.Subscribe() }
993func (s *serverSession) SubscribePreview(request ports.BrokerPreviewRequest) (ports.BrokerPreviewSubscription, error) {
994 return s.core.SubscribePreview(request)
995}
996
997// OpenStream delegates one scope-checked stream request to the admitted core
998// service. The connection's own reader drives the same core through
999// openStreamMessage, which additionally bridges the stream onto the wire; this
1000// method exists so a composed caller can use the session as a scoped
1001// ports.BrokerService. The stream identity is the one the caller already
1002// allocated (NextStreamID) and is transmitted unchanged: this listener never
1003// re-allocates a wire identity, whether the request arrives over the wire or
1004// through this in-process delegate.
1005func (s *serverSession) OpenEnvelopeStream(ctx context.Context, request ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
1006 if err := s.checkRequestScope(request.Epoch, request.Connection); err != nil {
1007 return nil, err
1008 }
1009 scoped := request
1010 scoped.Epoch = s.epoch
1011 scoped.Connection = s.scope.Connection
1012 return s.core.OpenEnvelopeStream(ctx, scoped)
1013}
1014
1015// CloseStream retires one scope-checked stream. An absent connection identity is
1016// filled from this session exactly like OpenStream, so a zero identity never
1017// reaches the core as a stale request, and a foreign one is still refused.
1018func (s *serverSession) CloseStream(connection ports.BrokerConnectionID, stream ports.BrokerStreamID) error {
1019 if err := s.checkRequestScope(s.epoch, connection); err != nil {
1020 return err
1021 }
1022 return s.core.CloseStream(s.scope.Connection, stream)
1023}
1024
1025// AddHost delegates one mutating operation to the admitted core service.
1026func (s *serverSession) AddHost(ctx context.Context, endpoint string, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
1027 return s.core.AddHost(ctx, endpoint, policy)
1028}
1029
1030// RemoveHost delegates one mutating operation to the admitted core service.
1031func (s *serverSession) RemoveHost(ctx context.Context, expected domain.RemoteRegistration) (bool, error) {
1032 return s.core.RemoveHost(ctx, expected)
1033}
1034
1035// UpdateHostPolicy delegates one mutating operation to the admitted core service.
1036func (s *serverSession) UpdateHostPolicy(ctx context.Context, expected domain.RemoteRegistration, policy ports.BrokerPolicy) (domain.RemoteRegistration, error) {
1037 return s.core.UpdateHostPolicy(ctx, expected, policy)
1038}
1039
1040// RequestReconcile delegates one reconcile hint to the admitted core service.
1041func (s *serverSession) RequestReconcile(endpoint string) { s.core.RequestReconcile(endpoint) }
1042
1043// checkRequestScope refuses a caller-supplied identity that is neither absent
1044// nor this connection's own, so a caller can never steer another connection's
1045// work through this session.
1046func (s *serverSession) checkRequestScope(epoch ports.BrokerEpoch, connection ports.BrokerConnectionID) error {
1047 if epoch != 0 && epoch != s.epoch {
1048 return errors.Join(ErrScopeMismatch, ports.BrokerError{Code: ports.BrokerErrorStaleEpoch})
1049 }
1050 if !connection.IsZero() && connection != s.scope.Connection {
1051 return errors.Join(ErrScopeMismatch, ports.BrokerAdmissionStale)
1052 }
1053 return nil
1054}