1// In-memory stream state and admission.
2//
3// StreamEngine owns the admission and lifecycle of the logical streams
4// multiplexed over one physical daemon connection. It performs no I/O and
5// carries no framing, pump, socket, scheduler, or writer-fairness policy: it
6// holds the stream table, the strictly-increasing physical stream identity
7// fence, the per-stream lifecycle, the bounded inbound queues, and the
8// terminal publication that fences the physical connection and every stream
9// it owns.
10//
11// One physical connection admits at most the negotiated stream ceiling
12// (MuxCeilings.MaxStreams, 1..128) pending plus live streams: opening, open,
13// or closing - a closing stream keeps its slot until Take drains its preserved
14// data. A fresh incoming Open must carry a nonzero physical stream ID that
15// is strictly above every ID already admitted; an ID is never reused,
16// including after its stream reached a terminal state. Traffic at or below
17// the high-water mark that is no longer live belongs to a retired stream and
18// is ignored; traffic above the high-water mark was never opened and is
19// refused. The high-water mark alone fences a retired identity, so it keeps
20// classifying late traffic even after its record was evicted past the
21// retention bound.
22//
23// Every stream walks one exact lifecycle: opening (admitted, not yet
24// confirmed), open (confirmed), closing (an orderly inbound Close arrived
25// while the stream still held unread data), and terminal (closed, reset,
26// refused, or lost with the physical connection). Data is queued only while
27// open. An orderly Close that finds the inbound queue empty terminalizes the
28// stream immediately; a Close that finds already-queued data marks the stream
29// closing instead, so the consumer keeps draining exactly the chunks that were
30// accepted before the Close while no new chunk is accepted, and the stream
31// reaches terminal - releasing its admission slot and Done - only once Take
32// drains the last queued chunk. Reset is an abnormal, stream-local termination
33// that always discards the queue immediately. Neither Close nor Reset ever
34// terminalizes the physical connection or a sibling stream.
35//
36// Every stream is credit flow controlled in both directions. Each side starts
37// with the same per-stream window (MuxCeilings.StreamWindow) of send credit; a
38// Data frame costs its payload plus a fixed overhead (chunkCredit), and the
39// receiver returns credit with WindowUpdate as its consumer takes chunks. A
40// conforming sender therefore waits for credit instead of overrunning the
41// receiver, so a slow network or a slow consumer only slows the stream.
42//
43// The inbound queue is bounded by that same window, and bytes are charged
44// against every budget before the chunk is copied into the queue: the stream
45// window per stream, the negotiated aggregate ceiling
46// (MuxCeilings.MaxAggregateBytes) across the connection, and the negotiated
47// chunk ceiling (MuxCeilings.StreamChunkLimit) per chunk. A chunk above the
48// negotiated chunk ceiling is refused statelessly with ErrTooLarge and mutates
49// nothing, exactly as the codec refuses it. A chunk beyond the window is a
50// peer flow-control violation: it resets exactly that stream and releases the
51// bytes it held; sibling streams keep their budget and keep progressing.
52//
53// Terminal publication is deliberately ordered: a physical terminal outcome
54// publishes its own Done first and only then terminalizes every live stream
55// independently, so a reader woken by any stream always observes the physical
56// connection already terminal. A stream that already reached its own outcome
57// keeps it.
58//
59// A settled stream stays in the table so its reference, state, Done, and cause
60// remain observable; only an opening, open, or closing stream holds a slot
61// against the admission ceiling, and a closing stream retains that slot until
62// Take drains its preserved data. Retired records are themselves bounded: the
63// engine retains at most MaxRetiredStreamRecords settled records and evicts the
64// oldest first, so a long-lived connection cannot accumulate an unbounded
65// stream table. An evicted identity is still classified as retired rather than
66// future because the high-water mark is never rolled back.
67package daemonmux
68
69import (
70 "errors"
71 "sync"
72
73 "github.com/bnema/vev/internal/domain"
74 "github.com/bnema/vev/internal/ports"
75)
76
77// Stream queue bounds. The stream-count and aggregate-bytes ceilings are the
78// negotiated maxima from limits.go (MaxMuxStreams, MaxMuxAggregateBytes); the
79// per-stream bound is the flow-control window (MuxCeilings.StreamWindow).
80const (
81 // MaxRetiredStreamRecords bounds the settled stream records one engine
82 // retains so a late frame is still classified as retired instead of
83 // mistaken for a future stream. It mirrors the negotiated stream ceiling
84 // and is evicted oldest-first, so a long-lived connection's stream table
85 // stays bounded by the live ceiling plus this retention bound.
86 MaxRetiredStreamRecords = 128
87)
88
89// Stream admission and lifecycle sentinels. They are typed so a caller can
90// classify a refused frame without matching on message text.
91var (
92 // ErrInvalidStream reports a missing or zero physical stream identity.
93 ErrInvalidStream = errors.New("daemonmux: invalid physical stream")
94 // ErrStreamIDReused reports a physical stream identity that is not
95 // strictly above every admitted identity: IDs strictly increase and are
96 // never reused, including after a stream reached a terminal state.
97 ErrStreamIDReused = errors.New("daemonmux: stream ID is not fresh")
98 // ErrFutureStream reports a frame for a physical stream identity that was
99 // never opened. The physical connection's high-water mark fences it.
100 ErrFutureStream = errors.New("daemonmux: stream was never opened")
101 // ErrTooManyStreams reports the pending-plus-live stream ceiling of one
102 // physical connection.
103 ErrTooManyStreams = errors.New("daemonmux: too many live streams")
104 // ErrStreamState reports an event that is illegal in the stream's current
105 // state.
106 ErrStreamState = errors.New("daemonmux: invalid stream state")
107 // ErrStreamRefMismatch reports an Opened or Refused whose full stream
108 // reference does not match the one recorded at admission.
109 ErrStreamRefMismatch = errors.New("daemonmux: stream reference mismatch")
110 // ErrStreamQueueFull reports an inbound chunk the peer sent beyond the
111 // credit it was granted, or beyond the aggregate budget. It is a peer
112 // flow-control violation: the offending stream is reset and its budget
113 // released.
114 ErrStreamQueueFull = errors.New("daemonmux: stream receive window exceeded")
115 // ErrStreamCredit reports a WindowUpdate that would grow the send credit
116 // past the stream window: a peer flow-control violation.
117 ErrStreamCredit = errors.New("daemonmux: stream credit exceeds window")
118 // errStreamSettled reports a local send on a stream that already reached
119 // its terminal state.
120 errStreamSettled = errors.New("daemonmux: stream is settled")
121 // ErrPhysicalClosed reports admission or stream work presented to a
122 // physical connection that already reached its terminal outcome.
123 ErrPhysicalClosed = errors.New("daemonmux: physical connection is closed")
124 // ErrOpenPolicyRefused reports a fresh inbound Open whose requested policy
125 // does not exactly match the accepted physical policy an engine was
126 // restricted to. The stream is refused before it is recorded or admitted,
127 // so it never holds an admission slot and never reaches a consumer.
128 ErrOpenPolicyRefused = errors.New("daemonmux: open policy refused")
129)
130
131// StreamState is the closed lifecycle of one logical stream over a physical
132// daemon connection.
133type StreamState uint8
134
135const (
136 // StreamOpening is an admitted stream the opening side has not confirmed.
137 StreamOpening StreamState = iota + 1
138 // StreamOpen is a confirmed stream that accepts data.
139 StreamOpen
140 // StreamClosing is a stream whose opening side sent an orderly Close while
141 // inbound data was still queued. It accepts no new data; its already-queued
142 // chunks stay readable until Take drains them, at which point the stream
143 // reaches terminal and releases its slot.
144 StreamClosing
145 // StreamTerminal is a settled stream: no further frame is applied.
146 StreamTerminal
147)
148
149func (s StreamState) String() string {
150 switch s {
151 case StreamOpening:
152 return "opening"
153 case StreamOpen:
154 return "open"
155 case StreamClosing:
156 return "closing"
157 case StreamTerminal:
158 return "terminal"
159 default:
160 return "unknown"
161 }
162}
163
164// StreamDisposition classifies one stream frame: applied, or a late frame for
165// a retired stream that is ignored without effect.
166type StreamDisposition uint8
167
168const (
169 // StreamAccepted reports the frame was applied.
170 StreamAccepted StreamDisposition = iota + 1
171 // StreamDiscarded reports a late frame for a retired stream. It is
172 // ignored and never revives, reopens, or terminalizes anything.
173 StreamDiscarded
174)
175
176// StreamStatus is one immutable read of a tracked stream.
177type StreamStatus struct {
178 // Ref is the full reference recorded at admission.
179 Ref StreamRef
180 // State is the stream lifecycle state.
181 State StreamState
182 // Done closes exactly once when the stream reaches its terminal state.
183 Done <-chan struct{}
184 // Err is the stream's terminal cause: nil while live and for an orderly
185 // Close, and the failure cause for a Reset, a queue overflow, or a
186 // physical loss.
187 Err error
188 // FailureKind classifies Err: RemoteFailureNone while live, for an
189 // orderly Close, and for a refusal.
190 FailureKind domain.RemoteFailureKind
191 // Refused reports that the opening side refused the stream.
192 Refused bool
193 // Refusal is the typed refusal carried by Refused when Refused is true.
194 Refusal ErrorDetail
195 // QueuedChunks and QueuedBytes are the currently queued inbound chunks
196 // and bytes.
197 QueuedChunks int
198 QueuedBytes int
199 // SendCredit is the remaining outbound credit in credit bytes: payload
200 // plus MuxChunkCreditOverhead per Data frame (see chunkCredit).
201 SendCredit uint64
202}
203
204// muxStream is one tracked logical stream. Every field is guarded by the
205// owning StreamEngine's lock; terminal is its own publish-once authority and
206// is safe to read independently once published.
207type muxStream struct {
208 ref StreamRef
209 state StreamState
210 terminal *terminalState
211 queue [][]byte
212 bytes int
213 refused bool
214 refusal ErrorDetail
215
216 // recvUsed is the inbound credit the peer spent that has not been
217 // returned yet: queued chunks plus consumed chunks awaiting a
218 // WindowUpdate. It never exceeds the window for a conforming peer.
219 recvUsed uint64
220 // recvReturn is consumed credit not yet granted back to the peer.
221 recvReturn uint64
222 // sendCredit is the outbound credit this side may still spend.
223 sendCredit uint64
224 // creditWake closes and is replaced whenever sendCredit grows or the
225 // stream settles, so a blocked writer wakes without polling.
226 creditWake chan struct{}
227}
228
229// StreamEngine is the in-memory stream table and admission authority of one
230// physical daemon connection. It is safe for concurrent use. It owns no I/O.
231// The admission, chunk, and aggregate budgets are copied from the negotiated
232// MuxCeilings at construction, and the per-stream window is derived from
233// them. They are fields so the ceilings are fixed per instance.
234type StreamEngine struct {
235 mu sync.Mutex
236 physical *terminalState
237 streams map[PhysicalStreamID]*muxStream
238 retired []PhysicalStreamID
239 highest PhysicalStreamID
240 live int
241 aggregate int
242
243 maxStreams int
244 maxChunkBytes uint64
245 window uint64
246 returnAt uint64
247 maxAggregateBytes int
248
249 // restricted, when set, binds every fresh inbound Open to the exact
250 // accepted physical policy in policy. It is configured once, before any
251 // stream is admitted, by a daemon-side consumer that owns the connection's
252 // authoritative binding (RestrictAdmissions).
253 restricted bool
254 policy ports.BrokerPolicy
255}
256
257// NewStreamEngineWithCeilings returns an open engine bound to one physical
258// connection whose admission, chunk, and aggregate budgets enforce the given
259// negotiated MuxCeilings. The value is validated and copied, so the ceilings
260// an engine enforces are fixed for its lifetime. An invalid advertisement is
261// refused with ErrInvalidCeilings.
262func NewStreamEngineWithCeilings(ceilings MuxCeilings) (*StreamEngine, error) {
263 if err := ceilings.Validate(); err != nil {
264 return nil, err
265 }
266 return newStreamEngine(ceilings), nil
267}
268
269// newStreamEngine builds an engine from already-validated ceilings. The
270// exported constructors validate first; NewStreamEngine delegates with the
271// package defaults, which are valid by construction.
272func newStreamEngine(ceilings MuxCeilings) *StreamEngine {
273 return &StreamEngine{
274 physical: newTerminalState(),
275 streams: make(map[PhysicalStreamID]*muxStream, int(ceilings.MaxStreams)),
276 maxStreams: int(ceilings.MaxStreams),
277 maxChunkBytes: ceilings.StreamChunkLimit,
278 window: ceilings.StreamWindow(),
279 returnAt: ceilings.creditReturnThreshold(),
280 maxAggregateBytes: int(ceilings.MaxAggregateBytes),
281 }
282}
283
284// Done returns the physical connection's terminal channel. It is closed
285// exactly once, before every live stream is terminalized.
286func (e *StreamEngine) Done() <-chan struct{} {
287 if e == nil || e.physical == nil {
288 return nil
289 }
290 return e.physical.Done()
291}
292
293// Err returns the physical connection's terminal cause: nil while open and
294// for an orderly Terminate.
295func (e *StreamEngine) Err() error {
296 if e == nil || e.physical == nil {
297 return nil
298 }
299 return e.physical.Err()
300}
301
302// FailureKind returns the physical connection's terminal classification:
303// RemoteFailureNone while open and for an orderly Terminate.
304func (e *StreamEngine) FailureKind() domain.RemoteFailureKind {
305 if e == nil || e.physical == nil {
306 return domain.RemoteFailureNone
307 }
308 return e.physical.FailureKind()
309}
310
311// Closed reports whether the physical connection already reached its terminal
312// outcome.
313func (e *StreamEngine) Closed() bool {
314 if e == nil {
315 return true
316 }
317 e.mu.Lock()
318 defer e.mu.Unlock()
319 return e.closedLocked()
320}
321
322// Live returns the number of admitted streams that have not yet reached their
323// terminal state: opening, open, or closing (a closing stream still holds the
324// admission slot and its queued bytes until Take drains them).
325func (e *StreamEngine) Live() int {
326 if e == nil {
327 return 0
328 }
329 e.mu.Lock()
330 defer e.mu.Unlock()
331 return e.live
332}
333
334// AggregateBytes returns the inbound bytes currently charged across every
335// stream. It never exceeds MaxMuxAggregateBytes.
336func (e *StreamEngine) AggregateBytes() int {
337 if e == nil {
338 return 0
339 }
340 e.mu.Lock()
341 defer e.mu.Unlock()
342 return e.aggregate
343}
344
345// RestrictAdmissions binds the engine to one exact accepted physical policy:
346// every fresh inbound Open must then carry a policy exactly compatible with
347// policy, and an Open that differs in any field is refused with
348// ErrOpenPolicyRefused before it is recorded or admitted. A replay or
349// duplicate Open is still classified by the identity fence first, so a
350// restriction never resets or disturbs the live stream a duplicate names.
351//
352// It must be configured before the first inbound Open - that is, before the
353// pump driving the engine starts - because changing the accepted policy after
354// admission would leave earlier streams bound to a policy the engine no longer
355// enforces. A second call, a call after any admission, an already-terminal
356// connection, or an invalid policy is refused without changing enforcement.
357func (e *StreamEngine) RestrictAdmissions(policy ports.BrokerPolicy) error {
358 if e == nil {
359 return ErrPhysicalClosed
360 }
361 if err := policy.Validate(); err != nil {
362 return err
363 }
364 e.mu.Lock()
365 defer e.mu.Unlock()
366 if e.closedLocked() {
367 return ErrPhysicalClosed
368 }
369 if e.restricted || e.highest != 0 {
370 return ErrStreamState
371 }
372 e.policy = policy
373 e.restricted = true
374 return nil
375}
376
377// Open admits one incoming logical stream in opening state. The reference
378// must be complete, its physical ID must be strictly above every admitted ID
379// (IDs are never reused), and the pending-plus-live ceiling must not be
380// reached. When the engine is restricted to an accepted physical policy
381// (RestrictAdmissions), a fresh Open whose requested policy does not exactly
382// match is refused before it takes an admission slot. A physical connection
383// that already reached its terminal outcome refuses all admission.
384func (e *StreamEngine) Open(msg Open) error {
385 if e == nil {
386 return ErrPhysicalClosed
387 }
388 e.mu.Lock()
389 defer e.mu.Unlock()
390 if e.closedLocked() {
391 return ErrPhysicalClosed
392 }
393 if err := msg.Ref.Validate(); err != nil {
394 return ErrInvalidStream
395 }
396 id := msg.Ref.Physical
397 if id <= e.highest {
398 return ErrStreamIDReused
399 }
400 if e.restricted && !e.policy.Compatible(msg.Policy) {
401 return ErrOpenPolicyRefused
402 }
403 if e.live >= e.maxStreams {
404 return ErrTooManyStreams
405 }
406 e.streams[id] = &muxStream{ref: msg.Ref, state: StreamOpening, terminal: newTerminalState(), sendCredit: e.window, creditWake: make(chan struct{})}
407 e.highest = id
408 e.live++
409 return nil
410}
411
412// Opened confirms one admitted stream and moves it to open. It is legal only
413// for an opening stream, and the full reference must match the one recorded
414// at admission.
415func (e *StreamEngine) Opened(msg Opened) error {
416 if e == nil {
417 return ErrPhysicalClosed
418 }
419 e.mu.Lock()
420 defer e.mu.Unlock()
421 stream, state, err := e.lookupLocked(msg.Ref.Physical)
422 if err != nil {
423 return err
424 }
425 if state != StreamOpening {
426 return ErrStreamState
427 }
428 if stream.ref != msg.Ref {
429 return ErrStreamRefMismatch
430 }
431 stream.state = StreamOpen
432 return nil
433}
434
435// Refused settles one admitted stream after the opening side refused it. It is
436// legal only for an opening stream, and the full reference must match. A
437// refusal is an orderly settlement: the stream's Done closes with a nil Err,
438// and the typed refusal is retained for the caller.
439func (e *StreamEngine) Refused(msg Refused) error {
440 if e == nil {
441 return ErrPhysicalClosed
442 }
443 e.mu.Lock()
444 defer e.mu.Unlock()
445 stream, state, err := e.lookupLocked(msg.Ref.Physical)
446 if err != nil {
447 return err
448 }
449 if state != StreamOpening {
450 return ErrStreamState
451 }
452 if stream.ref != msg.Ref {
453 return ErrStreamRefMismatch
454 }
455 stream.refused = true
456 stream.refusal = msg.Error
457 e.retireLocked(stream)
458 stream.terminal.Close()
459 return nil
460}
461
462// Data queues one already-validated chunk of an open stream. Data is legal
463// only while open: a retired stream discards it, and a stream that was never
464// opened is refused. A stream that already received an orderly Close
465// (closing) accepts no new data and refuses it with ErrStreamState, exactly as
466// an unconfirmed stream refuses data. A chunk above the negotiated chunk
467// ceiling is refused statelessly with ErrTooLarge and mutates nothing. A chunk
468// beyond the stream's receive window or the aggregate byte bound is a peer
469// flow-control violation: it resets exactly this stream and releases the
470// bytes it held, leaving every sibling untouched.
471func (e *StreamEngine) Data(msg Data) (StreamDisposition, error) {
472 if e == nil {
473 return 0, ErrPhysicalClosed
474 }
475 if uint64(len(msg.Data)) > e.maxChunkBytes {
476 return 0, ErrTooLarge
477 }
478 e.mu.Lock()
479 defer e.mu.Unlock()
480 stream, state, err := e.lookupLocked(msg.Physical)
481 if err != nil {
482 return 0, err
483 }
484 switch state {
485 case StreamTerminal:
486 return StreamDiscarded, nil
487 case StreamOpening, StreamClosing:
488 return 0, ErrStreamState
489 }
490 if e.queueOverflowsLocked(stream, len(msg.Data)) {
491 e.retireLocked(stream)
492 stream.terminal.Fail(domain.RemoteFailureInvalidResponse, ErrStreamQueueFull)
493 return 0, ErrStreamQueueFull
494 }
495 e.enqueueLocked(stream, msg.Data)
496 return StreamAccepted, nil
497}
498
499// WindowUpdate grows one stream's send credit by the credit the peer returned.
500// A late update for a retired stream is discarded. Credit that would exceed
501// the stream window is a peer flow-control violation (ErrStreamCredit).
502func (e *StreamEngine) WindowUpdate(msg WindowUpdate) (StreamDisposition, error) {
503 if e == nil {
504 return 0, ErrPhysicalClosed
505 }
506 e.mu.Lock()
507 defer e.mu.Unlock()
508 stream, state, err := e.lookupLocked(msg.Physical)
509 if err != nil {
510 return 0, err
511 }
512 if state == StreamTerminal {
513 return StreamDiscarded, nil
514 }
515 if msg.Credit > e.window-stream.sendCredit {
516 return 0, ErrStreamCredit
517 }
518 stream.sendCredit += msg.Credit
519 wakeCreditLocked(stream)
520 return StreamAccepted, nil
521}
522
523// reserveSend spends the credit of one outbound Data frame of n bytes. It
524// reports ok=false with a wake channel when credit is short; the caller waits
525// on it and retries. A closing, terminal, or unknown stream reports
526// errStreamSettled: the peer already retired it and grants no more credit.
527func (e *StreamEngine) reserveSend(physical PhysicalStreamID, n int) (bool, <-chan struct{}, error) {
528 if e == nil {
529 return false, nil, ErrPhysicalClosed
530 }
531 cost := chunkCredit(n)
532 e.mu.Lock()
533 defer e.mu.Unlock()
534 stream := e.streams[physical]
535 if stream == nil || stream.state == StreamTerminal || stream.state == StreamClosing {
536 return false, nil, errStreamSettled
537 }
538 if stream.sendCredit < cost {
539 return false, stream.creditWake, nil
540 }
541 stream.sendCredit -= cost
542 return true, nil, nil
543}
544
545// refundSend returns the credit of one outbound Data frame that was reserved
546// but never enqueued.
547func (e *StreamEngine) refundSend(physical PhysicalStreamID, n int) {
548 if e == nil {
549 return
550 }
551 e.mu.Lock()
552 defer e.mu.Unlock()
553 if stream := e.streams[physical]; stream != nil && stream.state != StreamTerminal {
554 stream.sendCredit = min(stream.sendCredit+chunkCredit(n), e.window)
555 wakeCreditLocked(stream)
556 }
557}
558
559// takeReturn claims the consumed inbound credit of one stream once it reaches
560// the return threshold, so the peer is granted credit in batches instead of
561// one WindowUpdate per chunk. It returns zero when nothing is due yet.
562func (e *StreamEngine) takeReturn(physical PhysicalStreamID) uint64 {
563 if e == nil {
564 return 0
565 }
566 e.mu.Lock()
567 defer e.mu.Unlock()
568 stream := e.streams[physical]
569 if stream == nil || stream.state != StreamOpen || stream.recvReturn < e.returnAt {
570 return 0
571 }
572 grant := stream.recvReturn
573 stream.recvReturn = 0
574 stream.recvUsed -= grant
575 return grant
576}
577
578// wakeCreditLocked wakes every writer waiting on this stream's credit.
579func wakeCreditLocked(stream *muxStream) {
580 if stream.creditWake != nil {
581 close(stream.creditWake)
582 }
583 stream.creditWake = make(chan struct{})
584}
585
586// Close orderly terminates one opening, open, or closing stream. A late close
587// for an already terminal or closing stream is discarded. It never
588// terminalizes the physical connection or a sibling stream.
589//
590// An orderly Close preserves data that was already accepted: when the inbound
591// queue holds unread chunks, the stream moves to closing, accepts no new data,
592// and keeps its queue readable until Take drains it; the stream only reaches
593// terminal, releasing its admission slot and closing Done, on the Take that
594// empties the queue. A Close with an empty queue terminalizes immediately.
595func (e *StreamEngine) Close(msg Close) (StreamDisposition, error) {
596 if e == nil {
597 return 0, ErrPhysicalClosed
598 }
599 e.mu.Lock()
600 defer e.mu.Unlock()
601 stream, state, err := e.lookupLocked(msg.Physical)
602 if err != nil {
603 return 0, err
604 }
605 if state == StreamTerminal || state == StreamClosing {
606 return StreamDiscarded, nil
607 }
608 if len(stream.queue) == 0 {
609 e.retireLocked(stream)
610 stream.terminal.Close()
611 return StreamAccepted, nil
612 }
613 stream.state = StreamClosing
614 // The peer retired the stream: a writer waiting on credit must observe it
615 // instead of waiting for a WindowUpdate that never comes.
616 wakeCreditLocked(stream)
617 return StreamAccepted, nil
618}
619
620// Reset abnormally terminates one opening or open stream. The termination is
621// stream-local: it never terminalizes the physical connection or a sibling
622// stream. A late reset for a retired stream is discarded, so cancellation is
623// idempotent.
624func (e *StreamEngine) Reset(msg Reset) (StreamDisposition, error) {
625 if e == nil {
626 return 0, ErrPhysicalClosed
627 }
628 e.mu.Lock()
629 defer e.mu.Unlock()
630 stream, state, err := e.lookupLocked(msg.Physical)
631 if err != nil {
632 return 0, err
633 }
634 if state == StreamTerminal {
635 return StreamDiscarded, nil
636 }
637 e.retireLocked(stream)
638 stream.terminal.Fail(msg.Error.FailureKind, resetCause(msg))
639 return StreamAccepted, nil
640}
641
642// Fail publishes the physical connection's terminal failure and then
643// terminalizes every live stream independently with the same cause. Done is
644// closed before any stream is terminalized; a stream that already settled
645// keeps its own outcome. All queued bytes are released.
646//
647// Physical publication is first-wins: a Fail after a Terminate is a no-op
648// that leaves the orderly nil Err and RemoteFailureNone, exactly as a
649// Terminate after a Fail leaves the failure cause. Streams that were live at
650// the winning publication all observe that one outcome.
651func (e *StreamEngine) Fail(kind domain.RemoteFailureKind, err error) {
652 if e == nil {
653 return
654 }
655 e.mu.Lock()
656 defer e.mu.Unlock()
657 e.physical.Fail(kind, err)
658 e.terminalizeAllLocked(func(stream *muxStream) { stream.terminal.Fail(kind, err) })
659}
660
661// Terminate orderly closes the physical connection and then terminalizes every
662// live stream independently with a nil error. Done is closed before any stream
663// is terminalized. All queued bytes are released.
664//
665// Physical publication is first-wins: a Terminate after a Fail is a no-op
666// that leaves the failure cause and its kind, exactly as a Fail after a
667// Terminate leaves the orderly nil Err and RemoteFailureNone.
668func (e *StreamEngine) Terminate() {
669 if e == nil {
670 return
671 }
672 e.mu.Lock()
673 defer e.mu.Unlock()
674 e.physical.Close()
675 e.terminalizeAllLocked(func(stream *muxStream) { stream.terminal.Close() })
676}
677
678// Status returns one immutable read of a tracked stream and whether it exists.
679func (e *StreamEngine) Status(physical PhysicalStreamID) (StreamStatus, bool) {
680 if e == nil {
681 return StreamStatus{}, false
682 }
683 e.mu.Lock()
684 defer e.mu.Unlock()
685 stream := e.streams[physical]
686 if stream == nil {
687 return StreamStatus{}, false
688 }
689 return StreamStatus{
690 Ref: stream.ref,
691 State: stream.state,
692 Done: stream.terminal.Done(),
693 Err: stream.terminal.Err(),
694 FailureKind: stream.terminal.FailureKind(),
695 Refused: stream.refused,
696 Refusal: stream.refusal,
697 QueuedChunks: len(stream.queue),
698 QueuedBytes: stream.bytes,
699 SendCredit: stream.sendCredit,
700 }, true
701}
702
703// cause returns the retained publish-once terminal authority of one admitted
704// stream. The pointer stays valid for the connection's lifetime even after the
705// engine evicts the bounded record, so a settled stream keeps its exact cause
706// independent of retention. It reports false for a physical identity the
707// engine never admitted.
708func (e *StreamEngine) cause(physical PhysicalStreamID) (*terminalState, bool) {
709 if e == nil {
710 return nil, false
711 }
712 e.mu.Lock()
713 defer e.mu.Unlock()
714 stream := e.streams[physical]
715 if stream == nil {
716 return nil, false
717 }
718 return stream.terminal, true
719}
720
721// Take removes and returns the oldest queued inbound chunk of one stream.
722// It reports false when the stream does not exist or its queue is empty. The
723// returned chunk is an engine-owned copy: it releases the chunk's bytes from
724// both budgets. Draining the last queued chunk of a closing stream terminalizes
725// it, releasing its admission slot and closing its Done.
726func (e *StreamEngine) Take(physical PhysicalStreamID) ([]byte, bool) {
727 chunk, ok, _ := e.take(physical)
728 return chunk, ok
729}
730
731// take is Take plus whether this pop drained a closing stream to its terminal
732// state, so the pump can rotate the stream's watch channel and wake a consumer
733// blocked on the queue transition. It holds the engine lock only for the pop
734// and the terminal publication.
735func (e *StreamEngine) take(physical PhysicalStreamID) ([]byte, bool, bool) {
736 if e == nil {
737 return nil, false, false
738 }
739 e.mu.Lock()
740 defer e.mu.Unlock()
741 stream := e.streams[physical]
742 if stream == nil || len(stream.queue) == 0 {
743 return nil, false, false
744 }
745 chunk := stream.queue[0]
746 stream.queue[0] = nil
747 stream.queue = stream.queue[1:]
748 stream.bytes -= len(chunk)
749 e.aggregate -= len(chunk)
750 stream.recvReturn += chunkCredit(len(chunk))
751 if stream.state == StreamClosing && len(stream.queue) == 0 {
752 e.retireLocked(stream)
753 stream.terminal.Close()
754 return chunk, true, true
755 }
756 return chunk, true, false
757}
758
759// lookupLocked classifies one frame: the tracked record of a live or settled
760// stream, a retired stream at or below the high-water mark, or a future
761// identity that was never opened.
762func (e *StreamEngine) lookupLocked(id PhysicalStreamID) (*muxStream, StreamState, error) {
763 if err := id.Validate(); err != nil {
764 return nil, 0, ErrInvalidStream
765 }
766 if stream, ok := e.streams[id]; ok {
767 return stream, stream.state, nil
768 }
769 if id > e.highest {
770 return nil, 0, ErrFutureStream
771 }
772 return nil, StreamTerminal, nil
773}
774
775// terminalizeAllLocked publishes one terminal outcome on every live stream.
776// The physical outcome is published by the caller before this runs; settled
777// streams are left untouched so they keep their own cause.
778func (e *StreamEngine) terminalizeAllLocked(publish func(*muxStream)) {
779 for _, stream := range e.streams {
780 if stream.state == StreamTerminal {
781 continue
782 }
783 e.retireLocked(stream)
784 publish(stream)
785 }
786}
787
788// retireLocked moves one stream to terminal, releases the bytes it holds, and
789// records it for bounded retention: the oldest retired record is evicted past
790// MaxRetiredStreamRecords. The high-water mark is never rolled back, so an
791// evicted identity still classifies as retired rather than future.
792func (e *StreamEngine) retireLocked(stream *muxStream) {
793 e.releaseQueueLocked(stream)
794 stream.state = StreamTerminal
795 stream.sendCredit = 0
796 wakeCreditLocked(stream)
797 e.live--
798 e.retired = append(e.retired, stream.ref.Physical)
799 for len(e.retired) > MaxRetiredStreamRecords {
800 evicted := e.retired[0]
801 e.retired = e.retired[1:]
802 if tracked := e.streams[evicted]; tracked != nil && tracked.state == StreamTerminal {
803 delete(e.streams, evicted)
804 }
805 }
806}
807
808// releaseQueueLocked drops every queued chunk and returns its bytes to the
809// aggregate budget.
810func (e *StreamEngine) releaseQueueLocked(stream *muxStream) {
811 for i := range stream.queue {
812 stream.queue[i] = nil
813 }
814 stream.queue = nil
815 if stream.bytes != 0 {
816 e.aggregate -= stream.bytes
817 stream.bytes = 0
818 }
819}
820
821// queueOverflowsLocked reports whether one inbound chunk would exceed the
822// credit granted to the peer or the negotiated aggregate byte bound.
823func (e *StreamEngine) queueOverflowsLocked(stream *muxStream, n int) bool {
824 return stream.recvUsed+chunkCredit(n) > e.window ||
825 e.aggregate+n > e.maxAggregateBytes
826}
827
828// enqueueLocked charges the chunk against both budgets and only then copies it
829// into the queue, so a refused chunk is never allocated a queue-owned copy.
830func (e *StreamEngine) enqueueLocked(stream *muxStream, chunk []byte) {
831 n := len(chunk)
832 e.aggregate += n
833 stream.bytes += n
834 stream.recvUsed += chunkCredit(n)
835 owned := make([]byte, n)
836 copy(owned, chunk)
837 stream.queue = append(stream.queue, owned)
838}
839
840// closedLocked reports whether the physical connection already reached its
841// terminal outcome. The caller holds the engine lock.
842func (e *StreamEngine) closedLocked() bool {
843 select {
844 case <-e.physical.Done():
845 return true
846 default:
847 return false
848 }
849}
850
851// resetCause builds the stream-local terminal cause of one reset. A reset
852// without a declared detail publishes no cause and lets terminalState
853// substitute its failure sentinel.
854func resetCause(msg Reset) error {
855 if !msg.HasError {
856 return nil
857 }
858 return ports.BrokerError{Code: msg.Error.Code, Text: msg.Error.Text}
859}