1// Server supervisor over accepted raw physical carriers.
2//
3// ServerSupervisor is the daemon-side owner of accepted daemonmux physical
4// connections. A daemon accept loop hands it one authenticated raw framed
5// carriage per connection; the supervisor completes the daemonmux physical
6// preamble under its immutable ServerBinding, then publishes a fully wired
7// typed listener to the daemon's AggregateListener. It performs no listener
8// policy of its own: the carriage must already be authenticated, and the
9// binding is the only authority the handshake enforces.
10//
11// Each accepted carriage becomes one owned supervisedPhysical: a Listener, its
12// Pump, and the framing bridge that carries it. The listener is always built
13// before its pump starts, because it registers the pump's single admission
14// observer and the pump refuses a late registration. The supervised child
15// implements ports.ServerListener by delegating Accept to that listener, and
16// its Close tears the listener, pump, and carrier down together, so the
17// AggregateListener never owns a half-closed child.
18//
19// The accepted physical policy is enforced on every inbound Open before fresh
20// stream admission: the supervised pump's engine is restricted to the binding
21// policy, so a stream that names a policy the daemon did not accept is refused
22// on its own stream and never reaches the daemon's accept stream.
23//
24// Admission is bounded: at most the configured number of connections may be
25// inside dial-handshake-registration at once, so a burst of raw carriers can
26// never start an unbounded number of handshakes or pump goroutines. Each child
27// is reaped on its own: when its pump reaches a terminal outcome - a lost
28// carriage, a closed listener, or a local teardown - one goroutine deregisters
29// it from the aggregate and closes it, without touching a sibling or the
30// supervisor. The supervisor's Close closes every owned child and joins the
31// reapers; it does not close the AggregateListener, which its caller owns.
32package daemonmux
33
34import (
35 "context"
36 "errors"
37 "fmt"
38 "sync"
39
40 "github.com/bnema/vev/internal/adapters/clock"
41 "github.com/bnema/vev/internal/ports"
42 "github.com/bnema/vev/internal/protocol"
43)
44
45// Server supervisor sentinels. They are typed so a caller can classify a
46// configuration, admission, or lifecycle outcome without matching on message
47// text.
48var (
49 // ErrSupervisorConfig reports a supervisor constructed without an
50 // aggregate listener or binding, with an invalid ceiling advertisement, or
51 // an Adopt call with a nil raw carriage.
52 ErrSupervisorConfig = errors.New("daemonmux: invalid server supervisor configuration")
53 // ErrSupervisorClosed reports a physical carriage offered to a supervisor
54 // that was closed, or a registration attempted after Close.
55 ErrSupervisorClosed = errors.New("daemonmux: server supervisor is closed")
56)
57
58// MaxSupervisedAdmissions bounds the accepted physical connections one
59// supervisor may have in handshake-or-registration at once. It mirrors the
60// per-connection stream ceiling, so a daemon never starts more handshakes than
61// one connection could multiplex.
62const MaxSupervisedAdmissions = int(MaxMuxStreams)
63
64// ServerSupervisor accepts authenticated raw physical carriage and publishes
65// owned typed listeners to one AggregateListener. It is safe for concurrent
66// use.
67type ServerSupervisor struct {
68 aggregate *AggregateListener
69 binding ServerBinding
70 ceilings MuxCeilings
71 clock ports.Clock
72
73 // slots bounds concurrent handshakes and admissions: one token is held
74 // from the start of Adopt until the child is registered and its pump
75 // started.
76 slots chan struct{}
77
78 // mu guards children and closed. children holds exactly the live owned
79 // children, so Close and the reapers agree on what still needs teardown.
80 mu sync.Mutex
81 children map[*supervisedPhysical]struct{}
82 closed bool
83
84 done chan struct{}
85 wg sync.WaitGroup
86 closeOnce sync.Once
87}
88
89// NewServerSupervisor returns a supervisor that publishes every accepted
90// physical connection to aggregate under binding, advertising ceilings in the
91// physical preamble. admissionLimit bounds concurrent handshakes and
92// admissions; a non-positive value falls back to MaxSupervisedAdmissions. A
93// nil aggregate, an invalid binding, or an invalid ceiling advertisement is
94// refused with ErrSupervisorConfig.
95func NewServerSupervisor(aggregate *AggregateListener, binding ServerBinding, ceilings MuxCeilings, admissionLimit int) (*ServerSupervisor, error) {
96 return newServerSupervisor(aggregate, binding, ceilings, admissionLimit, clock.New())
97}
98
99// newServerSupervisor builds a supervisor from an explicit time source, so a
100// test can drive the per-stream handshake deadline deterministically. The
101// exported constructor delegates with the wall clock.
102func newServerSupervisor(aggregate *AggregateListener, binding ServerBinding, ceilings MuxCeilings, admissionLimit int, timeSource ports.Clock) (*ServerSupervisor, error) {
103 if aggregate == nil {
104 return nil, ErrSupervisorConfig
105 }
106 if err := binding.Validate(); err != nil {
107 return nil, fmt.Errorf("%w: %w", ErrSupervisorConfig, err)
108 }
109 if err := ceilings.Validate(); err != nil {
110 return nil, fmt.Errorf("%w: %w", ErrSupervisorConfig, err)
111 }
112 if admissionLimit <= 0 {
113 admissionLimit = MaxSupervisedAdmissions
114 }
115 if timeSource == nil {
116 timeSource = clock.New()
117 }
118 return &ServerSupervisor{
119 aggregate: aggregate,
120 binding: binding,
121 ceilings: ceilings,
122 clock: timeSource,
123 slots: make(chan struct{}, admissionLimit),
124 children: make(map[*supervisedPhysical]struct{}),
125 done: make(chan struct{}),
126 }, nil
127}
128
129// Adopt accepts one authenticated raw physical carriage: it bounds admission,
130// completes the daemonmux physical preamble against the supervisor's binding,
131// negotiates the effective ceilings, builds the pump and its typed listener,
132// registers the owned child with the aggregate, and starts the pump. The
133// caller's ctx bounds the handshake; the started connection is detached from it
134// and owned by the supervisor.
135//
136// A failed preamble, a refused policy, or a closed supervisor closes the
137// carriage and registers nothing. After a successful return the caller's only
138// remaining role is to stop offering carriage (or Close the supervisor); the
139// supervisor owns the carriage from then on.
140func (s *ServerSupervisor) Adopt(ctx context.Context, raw RawFramedTransport) error {
141 if s == nil {
142 return ErrSupervisorConfig
143 }
144 if raw == nil {
145 return ErrSupervisorConfig
146 }
147 if ctx == nil {
148 ctx = context.Background()
149 }
150 if err := s.acquire(ctx); err != nil {
151 _ = raw.Close()
152 return err
153 }
154 defer s.release()
155
156 bridge, err := NewPreambleCarrier(raw)
157 if err != nil {
158 _ = raw.Close()
159 return fmt.Errorf("%w: %w", ErrSupervisorConfig, err)
160 }
161
162 result, err := RunServerHandshake(ctx, bridge, s.binding, s.ceilings)
163 if err != nil {
164 // The handshake already closed a refused carriage; Close is idempotent.
165 _ = bridge.Close()
166 return err
167 }
168 if err := bridge.Negotiate(result.Ceilings); err != nil {
169 _ = bridge.Close()
170 return err
171 }
172
173 pump, err := NewPump(bridge, DirectionClient, result.Ceilings)
174 if err != nil {
175 _ = bridge.Close()
176 return err
177 }
178 // Enforce the accepted physical policy before any fresh inbound Open can be
179 // admitted: the pump's engine must be restricted before the pump starts,
180 // and the listener below is built before that start too. The accepted entry
181 // carries both the exact policy the engine enforces and the locality origin
182 // the listener stamps on every admitted session connection.
183 accepted := ServerPolicyAdmission{Policy: result.Policy, Origin: result.Origin}
184 if err := pump.Engine().RestrictAdmissions(accepted.Policy); err != nil {
185 _ = pump.Close()
186 return err
187 }
188
189 listener, err := newListener(pump, protocol.HandshakeTimeout, MaxAcceptQueue, s.clock, accepted)
190 if err != nil {
191 _ = pump.Close()
192 return err
193 }
194
195 child := &supervisedPhysical{listener: listener, pump: pump}
196 child.start(context.WithoutCancel(ctx))
197 if err := s.register(child); err != nil {
198 _ = child.Close()
199 return err
200 }
201 return nil
202}
203
204// Close stops accepting physical carriage and tears every owned child down:
205// it deregisters and closes each child - listener, pump, and carrier together -
206// and joins every reaper. Close is idempotent and concurrent-safe. It leaves
207// the caller-owned AggregateListener open; the daemon closes that itself.
208func (s *ServerSupervisor) Close() error {
209 if s == nil {
210 return nil
211 }
212 s.closeOnce.Do(func() {
213 s.mu.Lock()
214 s.closed = true
215 children := make([]*supervisedPhysical, 0, len(s.children))
216 for child := range s.children {
217 children = append(children, child)
218 }
219 s.mu.Unlock()
220 close(s.done)
221 for _, child := range children {
222 // Remove closes the child and joins its aggregate forwarder. A
223 // child its own reaper already deregistered reports
224 // ErrAggregateUnknownChild and is closed below regardless.
225 _ = s.aggregate.Remove(child)
226 _ = child.Close()
227 s.forget(child)
228 }
229 s.wg.Wait()
230 })
231 return nil
232}
233
234// acquire takes one admission slot, blocking until one is free or ctx or the
235// supervisor ends. It refuses admission outright once the supervisor closed.
236func (s *ServerSupervisor) acquire(ctx context.Context) error {
237 s.mu.Lock()
238 closed := s.closed
239 s.mu.Unlock()
240 if closed {
241 return ErrSupervisorClosed
242 }
243 select {
244 case s.slots <- struct{}{}:
245 return nil
246 default:
247 }
248 select {
249 case s.slots <- struct{}{}:
250 return nil
251 case <-s.done:
252 return ErrSupervisorClosed
253 case <-ctx.Done():
254 return ctx.Err()
255 }
256}
257
258// release returns one admission slot.
259func (s *ServerSupervisor) release() { <-s.slots }
260
261// register takes ownership of one fully wired, already started child and
262// publishes it to the aggregate, then starts its reaper. It registers with the
263// aggregate and takes the wait-group reference under the same lock that checks
264// closed, so Close can never observe a child it did not also own and wait for,
265// and no child is ever registered after Close saw the supervisor closed. A
266// closed supervisor, or a closed aggregate, refuses the child, which its
267// caller closes.
268func (s *ServerSupervisor) register(child *supervisedPhysical) error {
269 s.mu.Lock()
270 defer s.mu.Unlock()
271 if s.closed {
272 return ErrSupervisorClosed
273 }
274 if err := s.aggregate.Register(child); err != nil {
275 return err
276 }
277 s.children[child] = struct{}{}
278 s.wg.Add(1)
279 go s.reap(child)
280 return nil
281}
282
283// forget drops one child from the owned set. It is idempotent.
284func (s *ServerSupervisor) forget(child *supervisedPhysical) {
285 s.mu.Lock()
286 delete(s.children, child)
287 s.mu.Unlock()
288}
289
290// reap watches one child's pump and closes the child once its carriage reaches
291// a terminal outcome, isolating the loss from every sibling and from the
292// supervisor. It deregisters the child from the aggregate first (a child whose
293// own forwarder already observed the terminal Accept reports
294// ErrAggregateUnknownChild) and then closes listener, pump, and carrier
295// together. It exits when Close wakes it through done.
296func (s *ServerSupervisor) reap(child *supervisedPhysical) {
297 defer s.wg.Done()
298 select {
299 case <-child.pump.Done():
300 case <-s.done:
301 }
302 _ = s.aggregate.Remove(child)
303 _ = child.Close()
304 s.forget(child)
305}
306
307// supervisedPhysical is one owned accepted physical connection. It implements
308// ports.ServerListener by delegating Accept to its typed Listener, so the
309// aggregate composes it exactly like any other physical listener, while Close
310// owns the whole triple: listener, pump, and carrier.
311type supervisedPhysical struct {
312 listener *Listener
313 pump *Pump
314
315 closeOnce sync.Once
316 closeErr error
317}
318
319var _ ports.ServerListener = (*supervisedPhysical)(nil)
320
321// start launches the child's pump. The listener was already constructed and
322// registered its admission observer, so no admitted start can be missed.
323func (c *supervisedPhysical) start(ctx context.Context) {
324 c.pump.Start(ctx)
325}
326
327// Addr names the child's carriage for ports.ServerListener.
328func (c *supervisedPhysical) Addr() string {
329 if c == nil || c.listener == nil {
330 return "daemonmux"
331 }
332 return c.listener.Addr()
333}
334
335// Accept returns the next typed connection from the child's listener.
336func (c *supervisedPhysical) Accept() (ports.ServerConnection, error) {
337 if c == nil || c.listener == nil {
338 return nil, ErrSupervisorConfig
339 }
340 return c.listener.Accept()
341}
342
343// Close tears the child down exactly once: it closes the listener (unblocking
344// and joining its watchdog), then the pump (cancelling the carriage goroutines
345// and closing the carrier). Every caller observes the same pump close error.
346func (c *supervisedPhysical) Close() error {
347 if c == nil {
348 return nil
349 }
350 c.closeOnce.Do(func() {
351 if c.listener != nil {
352 _ = c.listener.Close()
353 }
354 if c.pump != nil {
355 c.closeErr = c.pump.Close()
356 }
357 })
358 return c.closeErr
359}