1// Physical pump over an abstract framed carrier.
2//
3// Pump is the in-memory physical half of one daemonmux connection. It
4// composes the stateless directional codec (codec.go), the StreamEngine
5// (engine_state.go), the outbound fair Scheduler (scheduler.go), and the
6// publish-once terminalState (terminal.go) over a private FramedCarrier
7// abstraction, and drives exactly one reader goroutine and one writer
8// goroutine. It owns no socket, listener, sessionwire conversion, or stream
9// framing: carriage framing is the carrier's, and the pump never imports a
10// concrete physical implementation.
11//
12// The reader strict-decodes every inbound directional envelope with the
13// negotiated MuxCeilings and applies it to the engine: Open/Data/Close/Reset
14// when the inbound direction is DirectionClient (the local side is the
15// daemon), Opened/Refused/Data/Close/Reset when it is DirectionServer (the
16// local side is the broker client). The writer dequeues the scheduler's
17// fairly ordered, already-encoded immutable envelopes and sends them
18// verbatim; it never re-encodes, mutates, or reorders them. Between frames it
19// blocks on a coalescing wake channel that every successful Send signals
20// after enqueuing, so the writer never polls or sleeps and a lost wakeup is
21// impossible: a signal raised while the writer drains is re-tested by the
22// next dequeue. Flush is the explicit write barrier: it waits until the
23// scheduler is empty and no send is in flight (or the context or the
24// connection ends), because an accepted Send only means the frame is queued,
25// never that it reached the carrier. Close is not a flush; a caller that must
26// publish a final frame flushes first.
27//
28// The reader retires the outbound half of a stream the peer terminalizes: a
29// successful inbound Close or Reset resets the matching scheduler record, so
30// queued outbound frames are discarded rather than sent after the peer
31// stopped reading. A queue overflow in the outbound scheduler is settled
32// symmetrically: the overflowing stream's engine record is reset to free its
33// admission slot and exactly one terminal Reset is scheduled for the peer.
34//
35// Failures are split into two classes, exactly as the engine does. A
36// malformed outer envelope (a strict-scan, direction, or bound refusal of a
37// complete payload) and any carrier Receive/Send error are physical-fatal:
38// the pump publishes its own terminal state first, then the engine's, so
39// Done closes before every stream is terminalized, and the carrier is closed
40// to unblock the peer goroutine. A stream-local semantic, reference, or queue
41// refusal (an engine error that leaves the physical connection healthy)
42// schedules exactly one Reset for the offending stream, settles that stream
43// locally, and lets every sibling keep flowing: one stream's refusal never
44// terminalizes the physical connection or another stream.
45//
46// Cancellation is an orderly close. Cancelling the context passed to Start
47// (the carrier returns the context error from its blocked Send or Receive) or
48// calling Close stops both goroutines, closes the carrier, joins the reader
49// and writer, and publishes an orderly outcome: a nil Err and
50// RemoteFailureNone. A physical failure that already won keeps its cause; the
51// pump's terminal authority and the engine's are both first-wins, and the
52// pump serializes its two publications so they can never disagree. Close
53// unblocks an in-flight carrier Send or Receive through the adapter
54// prompt-close contract and joins both goroutines bounded by it: a carrier
55// whose Close promptly releases its blocked I/O lets Close return as soon as
56// the reader and writer observe the release.
57//
58// The pump itself performs no admission and no outbound encoding policy: a
59// locally initiated stream is admitted through Engine (Open) and confirmed
60// through Engine (Opened) by its owner before Send queues the matching frame,
61// exactly as the scheduler's admission gate requires.
62package daemonmux
63
64import (
65 "context"
66 "errors"
67 "log/slog"
68 "sync"
69
70 "github.com/bnema/vev/internal/domain"
71 "github.com/bnema/vev/internal/ports"
72)
73
74// FramedCarrier is the private physical carriage one Pump drives: one
75// complete framed application envelope in each direction over one connection.
76// Send writes one immutable envelope and reports the write result; it must not
77// retain or mutate the payload. Receive blocks for the next complete envelope;
78// it returns the receiver's own copy. Both must return promptly once their
79// context is done, which is what makes context cancellation an orderly close
80// without a concurrent Close. Close is safe to call concurrently with Send,
81// Receive, and itself, and it must unblock an in-flight Send or Receive
82// promptly: that is the adapter prompt-close contract Pump.Close relies on to
83// join both goroutines in bounded time.
84type FramedCarrier interface {
85 Send(ctx context.Context, payload []byte) error
86 Receive(ctx context.Context) ([]byte, error)
87 Close() error
88}
89
90// ErrPumpConfig reports a pump constructed without a carrier or with an
91// unknown inbound direction.
92var ErrPumpConfig = errors.New("daemonmux: invalid pump configuration")
93
94// ErrDataNeedsCredit reports a Data frame handed to Send: stream data must go
95// through SendData so it is sent under the stream's flow-control credit.
96var ErrDataNeedsCredit = errors.New("daemonmux: stream data must be sent with SendData")
97
98// ErrAdmissionObserverLate reports an admission-observer registration attempted
99// after Start. The reader goroutine is already running and may have admitted an
100// inbound Open, so a late observer would silently miss it; a consumer must
101// register before Start.
102var ErrAdmissionObserverLate = errors.New("daemonmux: admission observer registered after start")
103
104// Pump is the physical pump of one daemonmux connection. It is safe for
105// concurrent use: the engine and scheduler it owns are themselves
106// concurrency-safe, and Send, Close, Start, and the accessors may be called
107// from any goroutine. It performs no stream framing and owns no socket,
108// listener, or session conversion.
109type Pump struct {
110 carrier FramedCarrier
111 engine *StreamEngine
112 scheduler *Scheduler
113 ceilings MuxCeilings
114 inbound EnvelopeDirection
115 local EnvelopeDirection
116 terminal *terminalState
117
118 // wake coalesces outbound wakeups: one pending token means the writer
119 // must re-check the scheduler, and a signal raised while a token is
120 // already pending is redundant rather than lost.
121 wake chan struct{}
122
123 // flushMu guards the flush bookkeeping. inflight counts frames the writer
124 // has dequeued but not finished sending, and flushCh is closed and replaced
125 // whenever the pump may have become idle (no in-flight send and an empty
126 // scheduler), so Flush can wait for the transition without polling. The
127 // scheduler lock is always taken after flushMu, never before it.
128 flushMu sync.Mutex
129 inflight int
130 flushCh chan struct{}
131
132 // watchMu guards the per-stream inbound watch table. Each admitted stream
133 // a consumer is watching owns one channel that the reader closes and
134 // replaces whenever it applies an inbound frame for that stream, so a
135 // consumer can block for new inbound data or a stream state change without
136 // polling. The table only holds entries for streams a consumer created with
137 // Watch and are dropped again by ReleaseWatch, so it stays bounded by the
138 // number of live logical connections rather than by every ID ever opened.
139 watchMu sync.Mutex
140 watch map[PhysicalStreamID]chan struct{}
141
142 // admitMu guards admitFn, the optional inbound-admission observer. The
143 // reader reads it on every admitted Open; its owner sets it before Start.
144 admitMu sync.Mutex
145 admitFn func(Open)
146
147 mu sync.Mutex
148 started bool
149 closed bool
150 cancel context.CancelFunc
151
152 startOnce sync.Once
153 closeOnce sync.Once
154 stopOnce sync.Once
155
156 // termMu serializes the pump-level and engine-level terminal
157 // publications so both always carry the same outcome and only the first
158 // publisher applies.
159 termMu sync.Mutex
160 settled bool
161
162 readerDone chan struct{}
163 writerDone chan struct{}
164 closeErr error
165}
166
167// NewPump returns a pump over carrier that reads inbound envelopes of the
168// given direction and writes the opposite one. The negotiated MuxCeilings are
169// validated and copied: they build the engine with NewStreamEngineWithCeilings
170// and the scheduler (gated on that engine) with NewSchedulerWithCeilings, so
171// the pump enforces exactly the ceilings one physical connection negotiated
172// for its lifetime. An invalid advertisement is refused with ErrInvalidCeilings
173// and a missing carrier or unknown direction with ErrPumpConfig. Goroutines
174// start on Start.
175func NewPump(carrier FramedCarrier, inbound EnvelopeDirection, ceilings MuxCeilings) (*Pump, error) {
176 if carrier == nil {
177 return nil, ErrPumpConfig
178 }
179 if inbound != DirectionClient && inbound != DirectionServer {
180 return nil, ErrPumpConfig
181 }
182 if err := ceilings.Validate(); err != nil {
183 return nil, err
184 }
185 engine, err := NewStreamEngineWithCeilings(ceilings)
186 if err != nil {
187 return nil, err
188 }
189 scheduler, err := NewSchedulerWithCeilings(engine, ceilings)
190 if err != nil {
191 return nil, err
192 }
193 return &Pump{
194 carrier: carrier,
195 engine: engine,
196 scheduler: scheduler,
197 ceilings: ceilings,
198 inbound: inbound,
199 local: oppositeDirection(inbound),
200 terminal: newTerminalState(),
201 wake: make(chan struct{}, 1),
202 flushCh: make(chan struct{}),
203 readerDone: make(chan struct{}),
204 writerDone: make(chan struct{}),
205 }, nil
206}
207
208// Start launches the reader and writer goroutines under ctx. It is
209// idempotent: only the first call takes effect, so a later call with a
210// different context is ignored rather than starting a second pair of
211// goroutines. A nil context is replaced with context.Background. Start after
212// Close does nothing, because the connection already reached its terminal
213// outcome.
214func (p *Pump) Start(ctx context.Context) {
215 if p == nil {
216 return
217 }
218 p.startOnce.Do(func() {
219 if ctx == nil {
220 ctx = context.Background()
221 }
222 p.mu.Lock()
223 if p.closed {
224 p.mu.Unlock()
225 return
226 }
227 runCtx, cancel := context.WithCancel(ctx)
228 p.cancel = cancel
229 p.started = true
230 p.mu.Unlock()
231 go p.readLoop(runCtx)
232 go p.writeLoop(runCtx)
233 })
234}
235
236// Close orderly closes the pump: it cancels the run context, closes the
237// carrier through the adapter prompt-close contract to unblock the reader and
238// writer, joins both goroutines, and publishes an orderly terminal outcome
239// unless a physical failure already won. It returns the carrier's Close error.
240// Close is idempotent and safe to call concurrently; every caller observes the
241// same result. A Close before Start never starts goroutines and never joins.
242//
243// Close is not a flush: it cancels the writer rather than draining it, so
244// frames accepted by Send but not yet written are dropped. A caller that must
245// publish a final frame (for example the last Close) calls Flush before Close;
246// a caller that races Close against the writer may lose the queued frame.
247func (p *Pump) Close() error {
248 if p == nil {
249 return ErrPumpConfig
250 }
251 p.closeOnce.Do(func() {
252 p.mu.Lock()
253 p.closed = true
254 started := p.started
255 p.mu.Unlock()
256 p.stop()
257 if started {
258 <-p.readerDone
259 <-p.writerDone
260 }
261 p.settleOrderly()
262 })
263 return p.closeErr
264}
265
266// Done returns the pump's terminal channel: closed exactly once, before every
267// stream of the connection is terminalized.
268func (p *Pump) Done() <-chan struct{} {
269 if p == nil || p.terminal == nil {
270 return nil
271 }
272 return p.terminal.Done()
273}
274
275// Err returns the pump's terminal cause: nil while open and for an orderly
276// Close or context cancellation.
277func (p *Pump) Err() error {
278 if p == nil || p.terminal == nil {
279 return nil
280 }
281 return p.terminal.Err()
282}
283
284// FailureKind returns the pump's terminal classification: RemoteFailureNone
285// while open and for an orderly Close.
286func (p *Pump) FailureKind() domain.RemoteFailureKind {
287 if p == nil || p.terminal == nil {
288 return domain.RemoteFailureNone
289 }
290 return p.terminal.FailureKind()
291}
292
293// Engine returns the stream engine the pump drives. The reader applies every
294// inbound envelope to it, and a stream's owner uses it to admit a locally
295// initiated stream and to consume inbound chunks with Take.
296func (p *Pump) Engine() *StreamEngine {
297 if p == nil {
298 return nil
299 }
300 return p.engine
301}
302
303// Take removes the oldest queued inbound chunk of one stream. It is the
304// pump's convenience delegation to the engine and reports false when the
305// stream is unknown or its queue is empty. Draining the last queued chunk of a
306// stream that already received an orderly Close terminalizes it here, and the
307// pump rotates that stream's watch channel so a consumer blocked on the queue
308// transition wakes and observes the end instead of waiting for a frame that
309// can never arrive.
310func (p *Pump) Take(physical PhysicalStreamID) ([]byte, bool) {
311 if p == nil || p.engine == nil {
312 return nil, false
313 }
314 chunk, ok, finalized := p.engine.take(physical)
315 if finalized {
316 p.signalWatch(physical)
317 }
318 if ok {
319 p.returnCredit(physical)
320 }
321 return chunk, ok
322}
323
324// returnCredit grants consumed inbound credit back to the peer once a batch
325// is due. A refusal for a settled stream is dropped; an aggregate overflow
326// resets the stream like any outbound overflow, because the lost grant would
327// otherwise leave the peer's writer short of credit forever.
328func (p *Pump) returnCredit(physical PhysicalStreamID) {
329 grant := p.engine.takeReturn(physical)
330 if grant == 0 || p.isTerminal() {
331 return
332 }
333 _ = p.enqueue(WindowUpdate{Physical: physical, Credit: grant})
334}
335
336// SendData sends one stream data chunk under the stream's flow-control
337// credit. It blocks while the peer has not granted enough credit, so a slow
338// network or a slow remote consumer slows the writer instead of resetting the
339// stream. It returns when the chunk is queued, the stream or the connection
340// settles, or cancel closes.
341func (p *Pump) SendData(physical PhysicalStreamID, chunk []byte, cancel <-chan struct{}) error {
342 if p == nil {
343 return ErrPumpConfig
344 }
345 for {
346 if p.isTerminal() {
347 return ErrPhysicalClosed
348 }
349 ok, wake, err := p.engine.reserveSend(physical, len(chunk))
350 if err != nil {
351 return err
352 }
353 if ok {
354 break
355 }
356 select {
357 case <-wake:
358 case <-cancel:
359 return ErrLogicalClosed
360 case <-p.Done():
361 return ErrPhysicalClosed
362 }
363 }
364 if p.isTerminal() {
365 p.engine.refundSend(physical, len(chunk))
366 return ErrPhysicalClosed
367 }
368 if err := p.enqueue(Data{Physical: physical, Data: chunk}); err != nil {
369 p.engine.refundSend(physical, len(chunk))
370 return err
371 }
372 return nil
373}
374
375// Ceilings returns the negotiated ceilings the pump enforces. The value is
376// immutable for the pump's lifetime.
377func (p *Pump) Ceilings() MuxCeilings {
378 if p == nil {
379 return MuxCeilings{}
380 }
381 return p.ceilings
382}
383
384// Inbound returns the envelope direction the pump's reader decodes.
385func (p *Pump) Inbound() EnvelopeDirection {
386 if p == nil {
387 return 0
388 }
389 return p.inbound
390}
391
392// Local returns the envelope direction the pump's writer emits: the opposite
393// of Inbound.
394func (p *Pump) Local() EnvelopeDirection {
395 if p == nil {
396 return 0
397 }
398 return p.local
399}
400
401// OnAdmitted registers fn to observe every inbound logical stream the pump's
402// reader admits: it runs once per admitted Open, immediately after the engine
403// accepted it and before any later frame for that stream can be applied (and
404// before the opening side is confirmed with Opened). It carries the decoded
405// admission request, which the reader never mutates afterwards.
406//
407// fn runs on the reader goroutine, so it must not block: it is the one
408// goroutine that applies every inbound frame of the physical connection. The
409// daemon-side Listener registers its accept-queue admission here. Registering
410// a second observer replaces the first - one physical connection has exactly
411// one daemon-side admission consumer - and a nil fn clears it.
412//
413// The observer must be registered before the first inbound Open, that is before
414// Start: a registration attempted on a pump that already started fails with
415// ErrAdmissionObserverLate rather than silently missing the Opens the reader
416// may already have admitted. A nil pump fails with ErrPumpConfig.
417func (p *Pump) OnAdmitted(fn func(Open)) error {
418 if p == nil {
419 return ErrPumpConfig
420 }
421 p.mu.Lock()
422 defer p.mu.Unlock()
423 if p.started {
424 return ErrAdmissionObserverLate
425 }
426 p.admitMu.Lock()
427 p.admitFn = fn
428 p.admitMu.Unlock()
429 return nil
430}
431
432// notifyAdmitted runs the registered admission observer, if any.
433func (p *Pump) notifyAdmitted(message Open) {
434 p.admitMu.Lock()
435 fn := p.admitFn
436 p.admitMu.Unlock()
437 if fn != nil {
438 fn(message)
439 }
440}
441
442// Watch returns a channel the pump's reader closes and replaces whenever it
443// applies an inbound frame for the physical stream: Open, Opened, Refused,
444// Data, Close, or Reset. A consumer blocks on it to wait for new inbound data
445// or a stream state change without polling. The channel is stable until the
446// next applied frame, so a consumer must re-read the stream after every wake
447// and must also select on the stream's terminal channel and the connection's
448// Done to observe a terminal outcome. A zero, unknown, or evicted identity,
449// and a stream that already reached its terminal state, return an
450// already-closed channel and retain no watch entry, so the watch table stays
451// bounded by the number of live logical connections. Watch is safe for
452// concurrent use and the same channel is returned to every caller until the
453// next applied frame.
454func (p *Pump) Watch(physical PhysicalStreamID) <-chan struct{} {
455 if p == nil || p.engine == nil || physical.Validate() != nil {
456 return closedChannel()
457 }
458 p.watchMu.Lock()
459 defer p.watchMu.Unlock()
460 if status, ok := p.engine.Status(physical); !ok || status.State == StreamTerminal {
461 // No further frame can ever be applied for this stream: drop any stale
462 // entry and hand back a closed channel instead of retaining a watch
463 // nobody will ever signal.
464 delete(p.watch, physical)
465 return closedChannel()
466 }
467 if ch, ok := p.watch[physical]; ok {
468 return ch
469 }
470 ch := make(chan struct{})
471 if p.watch == nil {
472 p.watch = make(map[PhysicalStreamID]chan struct{})
473 }
474 p.watch[physical] = ch
475 return ch
476}
477
478// cause returns the retained publish-once terminal authority of one admitted
479// stream, or nil when the pump is unset or the identity was never admitted.
480// The pointer outlives the engine's bounded record, so a stream's owner
481// captures it at open and keeps reporting the exact terminal cause even after
482// eviction.
483func (p *Pump) cause(physical PhysicalStreamID) *terminalState {
484 if p == nil || p.engine == nil {
485 return nil
486 }
487 cause, _ := p.engine.cause(physical)
488 return cause
489}
490
491// ReleaseWatch drops the watch channel of one terminal or abandoned physical
492// stream so the watch table stays bounded by the number of live logical
493// connections. It is safe to call concurrently with Watch; a later Watch of
494// an already-terminal stream returns a closed channel without retaining an
495// entry.
496func (p *Pump) ReleaseWatch(physical PhysicalStreamID) {
497 if p == nil {
498 return
499 }
500 p.watchMu.Lock()
501 defer p.watchMu.Unlock()
502 delete(p.watch, physical)
503}
504
505// signalWatch closes and replaces the current watch channel of one stream so
506// every blocked watcher wakes and re-reads the engine. It is a no-op for a
507// stream nobody watches. The caller must not hold the watch lock, and the
508// engine lock is never held when it runs.
509func (p *Pump) signalWatch(physical PhysicalStreamID) {
510 if p == nil || physical.Validate() != nil {
511 return
512 }
513 p.watchMu.Lock()
514 defer p.watchMu.Unlock()
515 ch, ok := p.watch[physical]
516 if !ok {
517 return
518 }
519 close(ch)
520 p.watch[physical] = make(chan struct{})
521}
522
523// closedChannel returns a shared, already-closed channel for watchers of an
524// unknown or terminal stream.
525var closedWatchChan = func() <-chan struct{} {
526 ch := make(chan struct{})
527 close(ch)
528 return ch
529}()
530
531func closedChannel() <-chan struct{} { return closedWatchChan }
532
533// watchSettled reports whether a channel returned by Watch marks a terminal or
534// unknown stream. Only that case returns the shared closed channel: a live
535// stream's watch channel is also closed, by signalWatch, whenever a frame
536// arrives, so being closed alone never means the stream ended.
537func watchSettled(ch <-chan struct{}) bool { return ch == closedWatchChan }
538
539// watchClosed reports whether a signal channel is already closed. Use
540// watchSettled, not this, to tell whether a Watch channel marks a settled
541// stream.
542func watchClosed(ch <-chan struct{}) bool {
543 select {
544 case <-ch:
545 return true
546 default:
547 return false
548 }
549}
550
551// Send encodes one outbound message with the negotiated ceilings, queues it
552// on the fair scheduler, and wakes the writer. The message's direction must
553// match the pump's local side: DirectionClient accepts ClientMessage and
554// DirectionServer accepts ServerMessage; the shared Data, Close, and Reset
555// values satisfy both. A wrong-direction value fails with ErrWrongDirection
556// before anything is encoded or queued, a message the codec refuses propagates
557// the codec's error, and a pump that already reached its terminal outcome
558// fails with ErrPhysicalClosed. Data is refused with ErrDataNeedsCredit:
559// stream data goes through SendData. On success the encoded bytes are
560// immutable and owned by the writer.
561//
562// An accepted Send only means the frame is queued, never that it reached the
563// wire: the writer drains asynchronously. Call Flush to wait until every
564// accepted frame has been handed to the carrier.
565//
566// A stream-local scheduler refusal propagates the scheduler's error. A queue
567// overflow is additionally isolated: the stream's engine record is reset to
568// free its admission slot and exactly one terminal Reset is scheduled so the
569// peer learns the stream was aborted; the caller still observes the overflow
570// error for the frame it tried to send.
571func (p *Pump) Send(message any) error {
572 if p == nil {
573 return ErrPumpConfig
574 }
575 if message == nil {
576 return ErrInvalidMessage
577 }
578 switch message.(type) {
579 case Data, *Data:
580 return ErrDataNeedsCredit
581 }
582 if p.isTerminal() {
583 return ErrPhysicalClosed
584 }
585 return p.enqueue(message)
586}
587
588// enqueue queues one validated outbound message on the scheduler in the
589// pump's local direction and isolates a stream-local overflow.
590func (p *Pump) enqueue(message any) error {
591 switch p.local {
592 case DirectionClient:
593 client, ok := message.(ClientMessage)
594 if !ok {
595 return ErrWrongDirection
596 }
597 physical, _, err := clientFrame(client)
598 if err != nil {
599 return err
600 }
601 return p.admitOutbound(physical, func() error { return p.scheduler.EnqueueClient(client) })
602 case DirectionServer:
603 server, ok := message.(ServerMessage)
604 if !ok {
605 return ErrWrongDirection
606 }
607 physical, _, err := serverFrame(server)
608 if err != nil {
609 return err
610 }
611 return p.admitOutbound(physical, func() error { return p.scheduler.EnqueueServer(server) })
612 default:
613 return ErrPumpConfig
614 }
615}
616
617// Flush blocks until every frame the scheduler accepted has been handed to the
618// carrier and no send is in flight, or until ctx is done. Send accepted means
619// queued only, so Flush is the explicit barrier that makes a frame observable
620// on the wire before the connection is closed. It returns ctx.Err() when the
621// deadline or cancellation wins, and the pump's terminal outcome when the
622// connection became terminal before the scheduler drained (a physical failure
623// carries its cause; an orderly Close carries ErrPhysicalClosed when frames
624// were still queued). It never polls: the writer signals an idle transition.
625func (p *Pump) Flush(ctx context.Context) error {
626 if p == nil {
627 return ErrPumpConfig
628 }
629 if ctx == nil {
630 ctx = context.Background()
631 }
632 for {
633 p.flushMu.Lock()
634 if p.inflight == 0 && !p.scheduler.Pending() {
635 p.flushMu.Unlock()
636 return nil
637 }
638 wait := p.flushCh
639 p.flushMu.Unlock()
640 select {
641 case <-wait:
642 case <-ctx.Done():
643 return ctx.Err()
644 case <-p.terminal.Done():
645 p.flushMu.Lock()
646 idle := p.inflight == 0 && !p.scheduler.Pending()
647 p.flushMu.Unlock()
648 if idle {
649 return nil
650 }
651 if err := p.Err(); err != nil {
652 return err
653 }
654 return ErrPhysicalClosed
655 }
656 }
657}
658
659// admitOutbound queues one already-typed outbound frame, isolates a queue
660// overflow, and wakes the writer only when the scheduler accepted the frame,
661// so a refused frame never produces a spurious wakeup.
662func (p *Pump) admitOutbound(physical PhysicalStreamID, admit func() error) error {
663 if err := admit(); err != nil {
664 p.handleOutboundRefusal(physical, err)
665 return err
666 }
667 p.signal()
668 return nil
669}
670
671// handleOutboundRefusal isolates one refused outbound frame. A queue overflow
672// is settled exactly like an inbound stream-local refusal: the stream's engine
673// record is reset to free its admission slot and exactly one terminal Reset is
674// scheduled towards the peer. Every other refusal (a settled or unadmitted
675// identity) is left untouched.
676func (p *Pump) handleOutboundRefusal(physical PhysicalStreamID, err error) {
677 if errors.Is(err, ErrSchedulerFull) {
678 p.scheduleReset(physical, err)
679 }
680}
681
682// signalFlushLocked broadcasts one possible idle transition to every Flush
683// waiter by closing the current channel and installing a fresh one. The caller
684// holds flushMu.
685func (p *Pump) signalFlushLocked() {
686 if p.flushCh != nil {
687 close(p.flushCh)
688 }
689 p.flushCh = make(chan struct{})
690}
691
692// maybeSignalFlush broadcasts an idle transition when the scheduler is empty
693// and no send is in flight, so a pending Flush can return. It is called after
694// the writer finishes a send and after a scheduler reset discards queued
695// frames.
696func (p *Pump) maybeSignalFlush() {
697 p.flushMu.Lock()
698 if p.inflight == 0 && !p.scheduler.Pending() {
699 p.signalFlushLocked()
700 }
701 p.flushMu.Unlock()
702}
703
704// signal raises one coalesced writer wakeup without ever blocking.
705func (p *Pump) signal() {
706 select {
707 case p.wake <- struct{}{}:
708 default:
709 }
710}
711
712// readLoop is the reader goroutine: it receives one complete framed envelope
713// at a time, strict-decodes the inbound direction, and applies it to the
714// engine. A carrier Receive error that is not the local cancellation is
715// physical-fatal; so is a malformed outer envelope. Either way Done closes
716// before any stream is terminalized and the peer goroutine is unblocked.
717func (p *Pump) readLoop(ctx context.Context) {
718 defer p.settleOrderly()
719 defer close(p.readerDone)
720 for {
721 if ctx.Err() != nil {
722 return
723 }
724 payload, err := p.carrier.Receive(ctx)
725 if err != nil {
726 if ctx.Err() != nil {
727 return
728 }
729 p.settleFailure(domain.RemoteFailureTransport, err)
730 return
731 }
732 if fatal := p.handleInbound(payload); fatal != nil {
733 p.settleFailure(domain.RemoteFailureInvalidResponse, fatal)
734 return
735 }
736 }
737}
738
739// writeLoop is the writer goroutine: it dequeues the scheduler's fairly
740// ordered immutable envelopes and sends them verbatim, then blocks on the
741// coalescing wake channel until more work arrives or the run context ends. It
742// never polls, sleeps, or reorders.
743func (p *Pump) writeLoop(ctx context.Context) {
744 defer p.settleOrderly()
745 defer close(p.writerDone)
746 for {
747 p.flushMu.Lock()
748 envelope, ok := p.scheduler.Dequeue()
749 if ok {
750 p.inflight++
751 }
752 p.flushMu.Unlock()
753 if ok {
754 err := p.carrier.Send(ctx, envelope.Bytes)
755 p.flushMu.Lock()
756 if p.inflight > 0 {
757 p.inflight--
758 }
759 idle := p.inflight == 0 && !p.scheduler.Pending()
760 if idle {
761 p.signalFlushLocked()
762 }
763 p.flushMu.Unlock()
764 if err != nil {
765 if ctx.Err() != nil {
766 return
767 }
768 p.settleFailure(domain.RemoteFailureTransport, err)
769 return
770 }
771 continue
772 }
773 // The scheduler is empty and no send is in flight: release any Flush
774 // waiter before blocking on the coalescing wake channel.
775 p.maybeSignalFlush()
776 select {
777 case <-p.wake:
778 case <-ctx.Done():
779 return
780 }
781 }
782}
783
784// handleInbound strict-decodes one complete inbound payload in the pump's
785// direction and applies it to the engine. It returns the physical-fatal cause
786// of a malformed outer envelope, or nil when the payload was applied, when a
787// stream-local refusal was isolated with one Reset, or when the physical
788// connection was already terminal.
789func (p *Pump) handleInbound(payload []byte) error {
790 switch p.inbound {
791 case DirectionClient:
792 message, err := DecodeClient(payload, p.ceilings.MaxReceiveEnvelopeBytes, p.ceilings.StreamChunkLimit)
793 if err != nil {
794 return err
795 }
796 return p.applyClient(message)
797 case DirectionServer:
798 message, err := DecodeServer(payload, p.ceilings.MaxReceiveEnvelopeBytes, p.ceilings.StreamChunkLimit)
799 if err != nil {
800 return err
801 }
802 return p.applyServer(message)
803 default:
804 return ErrPumpConfig
805 }
806}
807
808// applyClient applies one broker-to-daemon message to the engine.
809func (p *Pump) applyClient(message ClientMessage) error {
810 switch m := message.(type) {
811 case Open:
812 return p.applyOpen(m)
813 case *Open:
814 if m == nil {
815 return ErrInvalidMessage
816 }
817 return p.applyClient(*m)
818 case Data:
819 return p.applyData(m)
820 case *Data:
821 if m == nil {
822 return ErrInvalidMessage
823 }
824 return p.applyData(*m)
825 case Close:
826 disposition, err := p.engine.Close(m)
827 return p.applyStreamTerminal(m.Physical, disposition, err)
828 case *Close:
829 if m == nil {
830 return ErrInvalidMessage
831 }
832 return p.applyClient(*m)
833 case Reset:
834 disposition, err := p.engine.Reset(m)
835 p.logPeerReset(m, disposition)
836 return p.applyStreamTerminal(m.Physical, disposition, err)
837 case *Reset:
838 if m == nil {
839 return ErrInvalidMessage
840 }
841 return p.applyClient(*m)
842 case WindowUpdate:
843 return p.applyWindowUpdate(m)
844 case *WindowUpdate:
845 if m == nil {
846 return ErrInvalidMessage
847 }
848 return p.applyClient(*m)
849 default:
850 return ErrWrongDirection
851 }
852}
853
854// applyOpen admits one incoming logical stream and notifies the admission
855// observer exactly when the engine accepted it. A refusal never notifies: a
856// stream that was never admitted has no daemon-side consumer. The observer runs
857// on the reader goroutine, between admission and any later frame for that
858// stream.
859//
860// The two refusal classes are deliberately different. A duplicate or replayed
861// Open (an identity at or below the high-water mark the engine already owns)
862// is a protocol violation that must not disturb the stream it names: it is
863// ignored, so it neither resets nor alters the existing live or retired stream,
864// and it never produces an outbound frame. A fresh Open the engine refused
865// before recording anything (capacity, a closed physical connection, or an
866// invalid authority) is signalled with exactly one typed Mux Refused carrying
867// its admission code; no stream is recorded for it, and a sibling continues
868// unaffected.
869func (p *Pump) applyOpen(message Open) error {
870 err := p.engine.Open(message)
871 switch {
872 case err == nil:
873 p.notifyAdmitted(message)
874 return nil
875 case errors.Is(err, ErrStreamIDReused):
876 return nil
877 default:
878 p.refuseFreshOpen(message.Ref, err)
879 return nil
880 }
881}
882
883// refuseFreshOpen signals the peer that one fresh Open was refused before the
884// engine admitted anything. It schedules exactly one typed Mux Refused through
885// the scheduler's fresh-refusal exception, because the engine recorded no
886// stream to confirm, and it records no engine stream itself. A pump that
887// already reached its terminal outcome, or a Refused the scheduler refuses,
888// leaves the peer unsignalled rather than faulting the shared physical
889// connection.
890func (p *Pump) refuseFreshOpen(ref StreamRef, cause error) {
891 if p == nil || p.isTerminal() {
892 return
893 }
894 refused := Refused{Ref: ref, Error: openRefusalDetail(cause)}
895 if err := p.scheduler.Refuse(refused); err != nil {
896 return
897 }
898 p.signal()
899}
900
901// applyServer applies one daemon-to-broker message to the engine.
902func (p *Pump) applyServer(message ServerMessage) error {
903 switch m := message.(type) {
904 case Opened:
905 return p.applyStreamError(m.Ref.Physical, p.engine.Opened(m))
906 case *Opened:
907 if m == nil {
908 return ErrInvalidMessage
909 }
910 return p.applyServer(*m)
911 case Refused:
912 return p.applyStreamError(m.Ref.Physical, p.engine.Refused(m))
913 case *Refused:
914 if m == nil {
915 return ErrInvalidMessage
916 }
917 return p.applyServer(*m)
918 case Data:
919 return p.applyData(m)
920 case *Data:
921 if m == nil {
922 return ErrInvalidMessage
923 }
924 return p.applyData(*m)
925 case Close:
926 disposition, err := p.engine.Close(m)
927 return p.applyStreamTerminal(m.Physical, disposition, err)
928 case *Close:
929 if m == nil {
930 return ErrInvalidMessage
931 }
932 return p.applyServer(*m)
933 case Reset:
934 disposition, err := p.engine.Reset(m)
935 p.logPeerReset(m, disposition)
936 return p.applyStreamTerminal(m.Physical, disposition, err)
937 case *Reset:
938 if m == nil {
939 return ErrInvalidMessage
940 }
941 return p.applyServer(*m)
942 case WindowUpdate:
943 return p.applyWindowUpdate(m)
944 case *WindowUpdate:
945 if m == nil {
946 return ErrInvalidMessage
947 }
948 return p.applyServer(*m)
949 default:
950 return ErrWrongDirection
951 }
952}
953
954// applyStreamTerminal applies one inbound Close or Reset. When the engine
955// accepted it, the matching outbound scheduler record is reset so queued
956// outbound frames for the peer-terminalized stream are discarded rather than
957// sent after the peer stopped reading it, and the record is retired. A refused
958// terminal falls back to the stream-local refusal path.
959func (p *Pump) applyStreamTerminal(physical PhysicalStreamID, disposition StreamDisposition, err error) error {
960 p.signalWatch(physical)
961 if err != nil {
962 return p.applyStreamError(physical, err)
963 }
964 if disposition == StreamAccepted {
965 p.scheduler.Reset(physical)
966 p.maybeSignalFlush()
967 }
968 return nil
969}
970
971// logPeerReset records one stream the peer aborted, with the reason it sent.
972// A late reset for an already settled stream is not logged.
973func (p *Pump) logPeerReset(message Reset, disposition StreamDisposition) {
974 if disposition != StreamAccepted || !message.HasError {
975 return
976 }
977 slog.Warn("daemonmux_stream_reset", "side", p.local.String(), "stream", uint64(message.Physical), "initiator", "peer", "kind", message.Error.FailureKind.String(), "code", message.Error.Code.String(), "cause", message.Error.Text)
978}
979
980// applyWindowUpdate grows one stream's send credit. A credit violation resets
981// only that stream.
982func (p *Pump) applyWindowUpdate(message WindowUpdate) error {
983 _, err := p.engine.WindowUpdate(message)
984 if err == nil {
985 return nil
986 }
987 return p.applyStreamError(message.Physical, err)
988}
989
990// applyData queues one inbound chunk and classifies the engine's refusal.
991func (p *Pump) applyData(message Data) error {
992 _, err := p.engine.Data(message)
993 return p.applyStreamError(message.Physical, err)
994}
995
996// applyStreamError classifies one engine refusal. A stream-local refusal
997// schedules exactly one Reset for the offending stream and returns nil so the
998// reader continues; the physical connection stays open. ErrPhysicalClosed only
999// occurs once the physical connection is already terminal, so the reader stops
1000// without a second outcome. A nil refusal applies nothing.
1001func (p *Pump) applyStreamError(physical PhysicalStreamID, err error) error {
1002 p.signalWatch(physical)
1003 if err == nil {
1004 return nil
1005 }
1006 if errors.Is(err, ErrPhysicalClosed) {
1007 return nil
1008 }
1009 p.scheduleReset(physical, err)
1010 return nil
1011}
1012
1013// scheduleReset isolates one stream-local refusal: it settles the offending
1014// stream locally, freeing its admission slot, and schedules exactly one Reset
1015// frame towards the peer. An identity the engine never admitted cannot be
1016// signalled and is left alone. The engine settlement is always applied first
1017// and unconditionally, so the slot is freed even when the terminal frame
1018// cannot be scheduled; the scheduler enqueue result decides only whether the
1019// peer can be signalled, so a stream that already drained its one Reset is not
1020// reset twice and a repeated refusal stays idempotent. Siblings and the
1021// physical connection are untouched.
1022func (p *Pump) scheduleReset(physical PhysicalStreamID, cause error) {
1023 if physical.Validate() != nil {
1024 return
1025 }
1026 if _, known := p.engine.Status(physical); !known {
1027 return
1028 }
1029 detail := resetErrorDetail(cause)
1030 reset := Reset{Physical: physical, Error: detail, HasError: true}
1031 slog.Warn("daemonmux_stream_reset", "side", p.local.String(), "stream", uint64(physical), "initiator", "local", "kind", detail.FailureKind.String(), "cause", cause)
1032 _, _ = p.engine.Reset(reset)
1033 var err error
1034 if p.local == DirectionServer {
1035 err = p.scheduler.EnqueueServer(reset)
1036 } else {
1037 err = p.scheduler.EnqueueClient(reset)
1038 }
1039 if err != nil {
1040 // The peer cannot be signalled (the record is sealed or the identity
1041 // is not schedulable); the stream is already settled locally.
1042 return
1043 }
1044 p.signal()
1045}
1046
1047// settleFailure publishes one physical failure outcome: the pump's terminal
1048// state first, then the engine's, so Done closes before every stream is
1049// terminalized, and then closes the carrier to unblock the peer goroutine.
1050// The two publications are serialized and first-wins, so a concurrent orderly
1051// close or a second failure can never make them disagree.
1052func (p *Pump) settleFailure(kind domain.RemoteFailureKind, cause error) {
1053 p.termMu.Lock()
1054 if !p.settled {
1055 p.settled = true
1056 slog.Warn("daemonmux_physical_failed", "side", p.local.String(), "kind", kind.String(), "live_streams", p.engine.Live(), "cause", cause)
1057 p.terminal.Fail(kind, cause)
1058 p.engine.Fail(kind, cause)
1059 }
1060 p.termMu.Unlock()
1061 p.stop()
1062}
1063
1064// settleOrderly publishes the orderly local outcome unless a failure already
1065// won, then closes the carrier. The pump's terminal state is published before
1066// the engine's, exactly as a failure publishes it.
1067func (p *Pump) settleOrderly() {
1068 p.termMu.Lock()
1069 if !p.settled {
1070 p.settled = true
1071 p.terminal.Close()
1072 p.engine.Terminate()
1073 }
1074 p.termMu.Unlock()
1075 p.stop()
1076}
1077
1078// stop cancels the run context exactly once and closes the carrier exactly
1079// once, unblocking an in-flight Send or Receive so both goroutines can exit.
1080func (p *Pump) stop() {
1081 p.stopOnce.Do(func() {
1082 p.mu.Lock()
1083 cancel := p.cancel
1084 p.mu.Unlock()
1085 if cancel != nil {
1086 cancel()
1087 }
1088 p.closeErr = p.carrier.Close()
1089 })
1090}
1091
1092// isTerminal reports whether the pump already reached its terminal outcome.
1093func (p *Pump) isTerminal() bool {
1094 select {
1095 case <-p.terminal.Done():
1096 return true
1097 default:
1098 return false
1099 }
1100}
1101
1102// oppositeDirection returns the direction a pump writes when it reads inbound.
1103func oppositeDirection(inbound EnvelopeDirection) EnvelopeDirection {
1104 if inbound == DirectionClient {
1105 return DirectionServer
1106 }
1107 return DirectionClient
1108}
1109
1110// openRefusalDetail builds the typed refusal one fresh Open carries. The
1111// admission code is the closed taxonomy the broker classifies without matching
1112// message text: capacity is a limit (1), a closed physical connection is closed
1113// (2), and every other refusal is an invalid request (4). The text is a fixed,
1114// bounded, presentation-safe summary; the underlying diagnostic cause stays
1115// local and never travels on this wire.
1116func openRefusalDetail(cause error) ErrorDetail {
1117 detail := ErrorDetail{
1118 Code: ports.BrokerErrorUnavailable,
1119 AdmissionCode: 4,
1120 Text: "invalid open request",
1121 FailureKind: domain.RemoteFailureNone,
1122 }
1123 switch {
1124 case errors.Is(cause, ErrTooManyStreams):
1125 detail.AdmissionCode = 1
1126 detail.Text = "stream limit reached"
1127 case errors.Is(cause, ErrPhysicalClosed):
1128 detail.AdmissionCode = 2
1129 detail.Text = "physical connection closed"
1130 }
1131 return detail
1132}
1133
1134// resetErrorDetail builds the typed detail one scheduled Reset carries. The
1135// cause is always one of the package's bounded sentinels, so its text is
1136// presentation-safe and within the display bound; a queue overflow is a
1137// transport classification and every other stream-local refusal is an invalid
1138// response.
1139func resetErrorDetail(cause error) ErrorDetail {
1140 kind := domain.RemoteFailureInvalidResponse
1141 if errors.Is(cause, ErrSchedulerFull) {
1142 // The local aggregate safety net: a transport-class abort, not an
1143 // invalid peer frame.
1144 kind = domain.RemoteFailureTransport
1145 }
1146 return ErrorDetail{
1147 Code: ports.BrokerErrorUnavailable,
1148 Text: cause.Error(),
1149 FailureKind: kind,
1150 }
1151}