1// Per-connection broker state.
2//
3// Connection walks one exact lifecycle: New is entered when the listener
4// assigns the connection scope, one locally validated preamble moves it to
5// AwaitRegister, the first Register moves it to Ready, and shutdown runs
6// Closing then Closed. Every later frame must carry the scope assigned at
7// accept: a mismatched epoch or connection is rejected before it can touch
8// connection state.
9//
10// The connection owns one subscription generation series, one snapshot
11// assembler per active generation, one bounded operation tracker, and one
12// bounded stream tracker. Subscription generations advance strictly: a
13// resubscribe must carry a newer generation, while a resync or unsubscribe
14// must carry the active one. A generation advance discards in-flight
15// snapshot staging; the committed snapshot survives until a newer transfer
16// commits. Operation and stream state is scoped to exactly this connection
17// and epoch.
18//
19// There is no I/O, socket, pump, or transport here.
20package brokerwire
21
22import (
23 "errors"
24 "fmt"
25 "sync"
26
27 "github.com/bnema/vev/internal/ports"
28)
29
30// Per-connection bounds.
31const (
32 // MaxPendingOperations bounds admitted, unfinished mutating operations
33 // per connection.
34 MaxPendingOperations = 64
35 // MaxTrackedOperations bounds the completed-operation dedup set.
36 MaxTrackedOperations = 4096
37 // operationIDBytes is the encoded size of one operation identity.
38 operationIDBytes = 16
39 // MaxOperationDedupBytes bounds the completed-operation dedup memory:
40 // MaxTrackedOperations identities of 16 bytes each (64 KiB).
41 MaxOperationDedupBytes = MaxTrackedOperations * operationIDBytes
42 // MaxStreams bounds concurrently opening or open logical streams per
43 // connection.
44 MaxStreams = 128
45 // MaxRetiredStreams bounds retired stream records retained per
46 // connection so a late frame can be classified and discarded instead of
47 // mistaken for a future stream.
48 MaxRetiredStreams = 128
49)
50
51var (
52 // ErrInvalidScope reports an incomplete broker scope: a zero epoch or a
53 // zero connection identity.
54 ErrInvalidScope = errors.New("brokerwire: invalid broker scope")
55 // ErrConnectionState reports an operation that is not legal in the
56 // connection's current lifecycle state.
57 ErrConnectionState = errors.New("brokerwire: invalid connection state")
58 // ErrConnectionClosed reports work presented to a closing or closed
59 // connection.
60 ErrConnectionClosed = errors.New("brokerwire: connection is closed")
61 // ErrDuplicateRegister reports a second Register on one connection.
62 ErrDuplicateRegister = errors.New("brokerwire: duplicate register")
63 // ErrNotSubscribed reports a snapshot part or resync without an active
64 // subscription.
65 ErrNotSubscribed = errors.New("brokerwire: connection has no active subscription")
66 // ErrInvalidOperation reports a zero operation identity.
67 ErrInvalidOperation = errors.New("brokerwire: invalid broker operation")
68 // ErrInvalidOutcome reports an outcome outside the closed mutation
69 // taxonomy.
70 ErrInvalidOutcome = errors.New("brokerwire: invalid broker mutation outcome")
71 // ErrOperationPending reports a duplicate admission of an in-flight
72 // operation.
73 ErrOperationPending = errors.New("brokerwire: operation is already pending")
74 // ErrOperationCompleted reports an operation that already has a recorded
75 // outcome. It is deduped and never replayed.
76 ErrOperationCompleted = errors.New("brokerwire: operation already has an outcome")
77 // ErrTooManyPendingOperations reports a full pending set.
78 ErrTooManyPendingOperations = errors.New("brokerwire: too many pending operations")
79 // ErrUnknownOperation reports a completion for an operation that was
80 // never admitted, including one whose dedup record was evicted.
81 ErrUnknownOperation = errors.New("brokerwire: unknown operation")
82 // ErrInvalidStream reports a zero stream identity.
83 ErrInvalidStream = errors.New("brokerwire: invalid broker stream")
84 // ErrStreamIDReused reports a stream identity that is already consumed or
85 // evicted outside the bounded anti-replay window. Stream identities are
86 // allocated by the calling service and are never reused.
87 ErrStreamIDReused = errors.New("brokerwire: stream ID is not fresh")
88 // ErrFutureStream reports a frame for a stream identity that was never
89 // allocated.
90 ErrFutureStream = errors.New("brokerwire: stream was never opened")
91 // ErrTooManyStreams reports the per-connection concurrent stream
92 // ceiling.
93 ErrTooManyStreams = errors.New("brokerwire: too many open streams")
94 // ErrStreamState reports an event that is illegal in the stream's
95 // current state.
96 ErrStreamState = errors.New("brokerwire: invalid stream state")
97)
98
99// Scope is one exact broker connection scope: the broker epoch plus the
100// connection identity assigned at accept. Every frame carries it, and a
101// mismatched scope is rejected rather than applied.
102type Scope struct {
103 Epoch ports.BrokerEpoch
104 Connection ports.BrokerConnectionID
105}
106
107// Validate rejects an incomplete scope.
108func (s Scope) Validate() error {
109 if s.Epoch == 0 {
110 return fmt.Errorf("%w: missing epoch", ErrInvalidScope)
111 }
112 if err := s.Connection.Validate(); err != nil {
113 return fmt.Errorf("%w: %v", ErrInvalidScope, err)
114 }
115 return nil
116}
117
118// ConnState is the closed broker connection lifecycle.
119type ConnState uint8
120
121const (
122 // ConnNew is the accepted connection state: only a preamble is legal.
123 ConnNew ConnState = iota
124 // ConnAwaitRegister waits for exactly one Register after the preamble.
125 ConnAwaitRegister
126 // ConnReady accepts subscription, snapshot, operation, and stream work.
127 ConnReady
128 // ConnClosing has stopped accepting new work and discarded in-flight
129 // snapshot staging.
130 ConnClosing
131 // ConnClosed has released all connection state.
132 ConnClosed
133)
134
135func (s ConnState) String() string {
136 switch s {
137 case ConnNew:
138 return "new"
139 case ConnAwaitRegister:
140 return "await_register"
141 case ConnReady:
142 return "ready"
143 case ConnClosing:
144 return "closing"
145 case ConnClosed:
146 return "closed"
147 default:
148 return "unknown"
149 }
150}
151
152// Connection is the per-connection broker state machine: handshake
153// progression, one subscription generation series, one snapshot assembler
154// per active generation, and the bounded operation and stream trackers for
155// this exact scope. It is safe for concurrent use.
156type Connection struct {
157 mu sync.Mutex
158 scope Scope
159 state ConnState
160 operations *OperationTracker
161 streams *StreamTracker
162
163 subscribed bool
164 generation SubscriptionGeneration
165 highest SubscriptionGeneration
166 assembler *SnapshotAssembler
167}
168
169// NewConnection binds one connection to the exact scope assigned at accept.
170func NewConnection(scope Scope) (*Connection, error) {
171 if err := scope.Validate(); err != nil {
172 return nil, err
173 }
174 operations, err := newOperationTracker(scope)
175 if err != nil {
176 return nil, err
177 }
178 streams, err := newStreamTracker(scope)
179 if err != nil {
180 return nil, err
181 }
182 return &Connection{scope: scope, state: ConnNew, operations: operations, streams: streams}, nil
183}
184
185// State returns the current lifecycle state.
186func (c *Connection) State() ConnState {
187 if c == nil {
188 return ConnClosed
189 }
190 c.mu.Lock()
191 defer c.mu.Unlock()
192 return c.state
193}
194
195// Scope returns the exact scope assigned at accept.
196func (c *Connection) Scope() Scope {
197 if c == nil {
198 return Scope{}
199 }
200 c.mu.Lock()
201 defer c.mu.Unlock()
202 return c.scope
203}
204
205// Ready reports whether the handshake completed.
206func (c *Connection) Ready() bool { return c.State() == ConnReady }
207
208// SignalPreamble records one locally validated preamble. It is legal exactly
209// once, in New, and moves the connection to AwaitRegister.
210func (c *Connection) SignalPreamble() error {
211 if c == nil {
212 return ErrConnectionClosed
213 }
214 c.mu.Lock()
215 defer c.mu.Unlock()
216 if c.state != ConnNew {
217 return c.stateErrorLocked()
218 }
219 c.state = ConnAwaitRegister
220 return nil
221}
222
223// Register accepts the first Register and moves to Ready. A second Register,
224// or one before the preamble, is refused.
225func (c *Connection) Register() error {
226 if c == nil {
227 return ErrConnectionClosed
228 }
229 c.mu.Lock()
230 defer c.mu.Unlock()
231 switch c.state {
232 case ConnAwaitRegister:
233 c.state = ConnReady
234 return nil
235 case ConnReady:
236 return ErrDuplicateRegister
237 default:
238 return c.stateErrorLocked()
239 }
240}
241
242// BeginClose stops accepting new work and discards in-flight snapshot
243// staging. The committed snapshot survives until the connection is finally
244// closed. It is idempotent.
245func (c *Connection) BeginClose() error {
246 if c == nil {
247 return nil
248 }
249 c.mu.Lock()
250 defer c.mu.Unlock()
251 switch c.state {
252 case ConnClosing, ConnClosed:
253 return nil
254 }
255 c.state = ConnClosing
256 c.subscribed = false
257 c.generation = 0
258 if c.assembler != nil {
259 c.assembler.DiscardStaging()
260 }
261 return nil
262}
263
264// FinishClose completes shutdown and releases operation and stream state.
265// Only a closing connection may close; a second call is a no-op.
266func (c *Connection) FinishClose() error {
267 if c == nil {
268 return nil
269 }
270 c.mu.Lock()
271 defer c.mu.Unlock()
272 switch c.state {
273 case ConnClosed:
274 return nil
275 case ConnClosing:
276 c.state = ConnClosed
277 c.assembler = nil
278 c.operations.reset()
279 c.streams.reset()
280 return nil
281 default:
282 return ErrConnectionState
283 }
284}
285
286// Subscribe starts one subscription generation series. The generation must be
287// nonzero and strictly above every generation this connection has used, so a
288// superseded series can never be revived. A new generation discards any
289// in-flight staging and keeps publishing the committed snapshot until a newer
290// transfer commits.
291func (c *Connection) Subscribe(generation uint64) error {
292 if c == nil {
293 return ErrConnectionClosed
294 }
295 c.mu.Lock()
296 defer c.mu.Unlock()
297 if err := c.requireReadyLocked(); err != nil {
298 return err
299 }
300 if generation == 0 {
301 return ErrInvalidGeneration
302 }
303 if generation <= c.highest {
304 return ErrStaleGeneration
305 }
306 var committed ports.BrokerSnapshot
307 var hasCommitted bool
308 if c.assembler != nil {
309 committed, hasCommitted = c.assembler.Snapshot()
310 }
311 next := NewSnapshotAssembler(c.scope.Epoch, c.scope.Connection, WithGeneration(generation))
312 if hasCommitted {
313 next.adoptCommitted(committed)
314 }
315 c.subscribed = true
316 c.generation = generation
317 c.highest = generation
318 c.assembler = next
319 return nil
320}
321
322// Resync restarts the active generation's publication: it must carry the
323// active generation exactly, and it discards any in-flight staging so a full
324// transfer can restart at index 0.
325func (c *Connection) Resync(generation uint64) error {
326 if c == nil {
327 return ErrConnectionClosed
328 }
329 c.mu.Lock()
330 defer c.mu.Unlock()
331 if err := c.requireReadyLocked(); err != nil {
332 return err
333 }
334 if !c.subscribed {
335 return ErrNotSubscribed
336 }
337 if generation != c.generation {
338 return generationFence(generation, c.generation)
339 }
340 c.assembler.DiscardStaging()
341 return nil
342}
343
344// Unsubscribe ends the active generation. It must carry the active generation
345// exactly; the generation remains consumed and is never reused. The committed
346// snapshot survives until the connection is finally closed.
347func (c *Connection) Unsubscribe(generation uint64) error {
348 if c == nil {
349 return ErrConnectionClosed
350 }
351 c.mu.Lock()
352 defer c.mu.Unlock()
353 if err := c.requireReadyLocked(); err != nil {
354 return err
355 }
356 if !c.subscribed {
357 return ErrNotSubscribed
358 }
359 if generation != c.generation {
360 return generationFence(generation, c.generation)
361 }
362 c.subscribed = false
363 c.generation = 0
364 if c.assembler != nil {
365 c.assembler.DiscardStaging()
366 }
367 return nil
368}
369
370// AddSnapshotPart stages one part of the active generation's publication. The
371// committed snapshot changes only when End validates. The ready check, the
372// subscription check, and the assembler mutation share one connection lock, so
373// a concurrent close can never commit a part after it discarded staging.
374func (c *Connection) AddSnapshotPart(part SnapshotPart) (bool, ports.BrokerSnapshot, error) {
375 if c == nil {
376 return false, ports.BrokerSnapshot{}, ErrConnectionClosed
377 }
378 c.mu.Lock()
379 defer c.mu.Unlock()
380 if err := c.requireReadyLocked(); err != nil {
381 return false, ports.BrokerSnapshot{}, err
382 }
383 if !c.subscribed || c.assembler == nil {
384 return false, ports.BrokerSnapshot{}, ErrNotSubscribed
385 }
386 return c.assembler.Add(part)
387}
388
389// Snapshot returns the last committed snapshot and whether one exists.
390func (c *Connection) Snapshot() (ports.BrokerSnapshot, bool) {
391 if c == nil {
392 return ports.BrokerSnapshot{}, false
393 }
394 c.mu.Lock()
395 assembler := c.assembler
396 c.mu.Unlock()
397 if assembler == nil {
398 return ports.BrokerSnapshot{}, false
399 }
400 return assembler.Snapshot()
401}
402
403// StagingActive reports whether a snapshot transfer is in flight for the
404// active subscription generation.
405func (c *Connection) StagingActive() bool {
406 if c == nil {
407 return false
408 }
409 c.mu.Lock()
410 assembler := c.assembler
411 c.mu.Unlock()
412 if assembler == nil {
413 return false
414 }
415 return assembler.StagingActive()
416}
417
418// AdmitOperation records one in-flight mutating operation for this scope.
419// The ready check and the tracker mutation share one connection lock, so a
420// concurrent close can never admit work after shutdown.
421func (c *Connection) AdmitOperation(operation ports.BrokerOperationID) error {
422 if c == nil {
423 return ErrConnectionClosed
424 }
425 c.mu.Lock()
426 defer c.mu.Unlock()
427 if err := c.requireReadyLocked(); err != nil {
428 return err
429 }
430 return c.operations.Admit(c.scope, operation)
431}
432
433// CompleteOperation records exactly one outcome for an in-flight operation.
434// The ready check and the tracker mutation share one connection lock.
435func (c *Connection) CompleteOperation(operation ports.BrokerOperationID, outcome ports.BrokerMutationOutcome) error {
436 if c == nil {
437 return ErrConnectionClosed
438 }
439 c.mu.Lock()
440 defer c.mu.Unlock()
441 if err := c.requireReadyLocked(); err != nil {
442 return err
443 }
444 return c.operations.Complete(c.scope, operation, outcome)
445}
446
447// OpenStream allocates one fresh logical stream in this scope. The ready
448// check and the tracker mutation share one connection lock.
449func (c *Connection) OpenStream(stream ports.BrokerStreamID) error {
450 if c == nil {
451 return ErrConnectionClosed
452 }
453 c.mu.Lock()
454 defer c.mu.Unlock()
455 if err := c.requireReadyLocked(); err != nil {
456 return err
457 }
458 return c.streams.Open(c.scope, stream)
459}
460
461// StreamOpened confirms one logical stream the broker established. The
462// ready check and the tracker mutation share one connection lock.
463func (c *Connection) StreamOpened(stream ports.BrokerStreamID) error {
464 if c == nil {
465 return ErrConnectionClosed
466 }
467 c.mu.Lock()
468 defer c.mu.Unlock()
469 if err := c.requireReadyLocked(); err != nil {
470 return err
471 }
472 return c.streams.Opened(c.scope, stream)
473}
474
475// StreamData delivers one opaque daemon frame to an open stream. The ready
476// check and the tracker mutation share one connection lock.
477func (c *Connection) StreamData(stream ports.BrokerStreamID) (StreamDisposition, error) {
478 if c == nil {
479 return 0, ErrConnectionClosed
480 }
481 c.mu.Lock()
482 defer c.mu.Unlock()
483 if err := c.requireReadyLocked(); err != nil {
484 return 0, err
485 }
486 return c.streams.Data(c.scope, stream)
487}
488
489// StreamProgress reports one phase of an in-flight stream operation. The
490// ready check and the tracker mutation share one connection lock.
491func (c *Connection) StreamProgress(stream ports.BrokerStreamID) (StreamDisposition, error) {
492 if c == nil {
493 return 0, ErrConnectionClosed
494 }
495 c.mu.Lock()
496 defer c.mu.Unlock()
497 if err := c.requireReadyLocked(); err != nil {
498 return 0, err
499 }
500 return c.streams.Progress(c.scope, stream)
501}
502
503// CloseStream ends one logical stream. The ready check and the tracker
504// mutation share one connection lock.
505func (c *Connection) CloseStream(stream ports.BrokerStreamID) (StreamDisposition, error) {
506 if c == nil {
507 return 0, ErrConnectionClosed
508 }
509 c.mu.Lock()
510 defer c.mu.Unlock()
511 if err := c.requireReadyLocked(); err != nil {
512 return 0, err
513 }
514 return c.streams.Close(c.scope, stream)
515}
516
517func (c *Connection) requireReadyLocked() error {
518 switch c.state {
519 case ConnReady:
520 return nil
521 case ConnClosing, ConnClosed:
522 return ErrConnectionClosed
523 default:
524 return ErrConnectionState
525 }
526}
527
528func (c *Connection) stateErrorLocked() error {
529 if c.state == ConnClosing || c.state == ConnClosed {
530 return ErrConnectionClosed
531 }
532 return ErrConnectionState
533}
534
535// generationFence classifies a generation that is not the active one.
536func generationFence(got, active uint64) error {
537 if got < active {
538 return ErrStaleGeneration
539 }
540 return ErrFutureGeneration
541}
542
543// OperationTracker bounds the mutating operations admitted on one
544// connection scope. At most MaxPendingOperations operations are in flight at
545// once. Completed operations keep only their identity and outcome, bounded
546// to MaxTrackedOperations identities (MaxOperationDedupBytes), so a repeated
547// operation is deduped rather than re-executed inside that window.
548// Outcome-unknown is recorded like any other outcome: the client refreshes
549// authoritative state and never blindly replays a non-idempotent operation.
550// It is safe for concurrent use.
551type OperationTracker struct {
552 mu sync.Mutex
553 scope Scope
554 pending map[ports.BrokerOperationID]struct{}
555 results map[ports.BrokerOperationID]ports.BrokerMutationOutcome
556 order []ports.BrokerOperationID
557}
558
559// newOperationTracker binds a tracker to one exact connection scope. It is
560// unexported on purpose: the trackers are connection-owned state, and every
561// production mutation goes through the Connection methods that hold the
562// connection lock across the ready check and the tracker mutation.
563func newOperationTracker(scope Scope) (*OperationTracker, error) {
564 if err := scope.Validate(); err != nil {
565 return nil, err
566 }
567 return &OperationTracker{
568 scope: scope,
569 pending: make(map[ports.BrokerOperationID]struct{}, MaxPendingOperations),
570 results: make(map[ports.BrokerOperationID]ports.BrokerMutationOutcome, MaxTrackedOperations),
571 }, nil
572}
573
574// Admit registers one mutating operation as in flight. A mismatched scope, a
575// zero identity, a duplicate in-flight operation, an operation that already
576// has a recorded outcome, and a full pending set are all refused.
577func (t *OperationTracker) Admit(scope Scope, operation ports.BrokerOperationID) error {
578 if t == nil {
579 return ErrConnectionClosed
580 }
581 t.mu.Lock()
582 defer t.mu.Unlock()
583 if scope != t.scope {
584 return ErrScopeMismatch
585 }
586 if operation.IsZero() {
587 return ErrInvalidOperation
588 }
589 if _, done := t.results[operation]; done {
590 return ErrOperationCompleted
591 }
592 if _, pending := t.pending[operation]; pending {
593 return ErrOperationPending
594 }
595 if len(t.pending) >= MaxPendingOperations {
596 return ErrTooManyPendingOperations
597 }
598 t.pending[operation] = struct{}{}
599 return nil
600}
601
602// Complete records exactly one outcome for an in-flight operation. The
603// identity moves into the bounded dedup set, evicting the oldest record when
604// full, so a repeat of a recently completed operation is deduped instead of
605// replayed.
606func (t *OperationTracker) Complete(scope Scope, operation ports.BrokerOperationID, outcome ports.BrokerMutationOutcome) error {
607 if t == nil {
608 return ErrConnectionClosed
609 }
610 t.mu.Lock()
611 defer t.mu.Unlock()
612 if scope != t.scope {
613 return ErrScopeMismatch
614 }
615 if err := outcome.Validate(); err != nil {
616 return ErrInvalidOutcome
617 }
618 if _, pending := t.pending[operation]; !pending {
619 if _, done := t.results[operation]; done {
620 return ErrOperationCompleted
621 }
622 return ErrUnknownOperation
623 }
624 delete(t.pending, operation)
625 t.results[operation] = outcome
626 t.order = append(t.order, operation)
627 for len(t.order) > MaxTrackedOperations {
628 evicted := t.order[0]
629 t.order = t.order[1:]
630 delete(t.results, evicted)
631 }
632 return nil
633}
634
635// Pending returns the number of in-flight operations.
636func (t *OperationTracker) Pending() int {
637 if t == nil {
638 return 0
639 }
640 t.mu.Lock()
641 defer t.mu.Unlock()
642 return len(t.pending)
643}
644
645// Tracked returns the number of dedup records retained.
646func (t *OperationTracker) Tracked() int {
647 if t == nil {
648 return 0
649 }
650 t.mu.Lock()
651 defer t.mu.Unlock()
652 return len(t.results)
653}
654
655// DedupBytes returns the dedup memory in bytes: one fixed-size identity per
656// retained record. It never exceeds MaxOperationDedupBytes.
657func (t *OperationTracker) DedupBytes() int {
658 return t.Tracked() * operationIDBytes
659}
660
661// Outcome returns the recorded outcome for one operation.
662func (t *OperationTracker) Outcome(operation ports.BrokerOperationID) (ports.BrokerMutationOutcome, bool) {
663 if t == nil {
664 return 0, false
665 }
666 t.mu.Lock()
667 defer t.mu.Unlock()
668 outcome, ok := t.results[operation]
669 return outcome, ok
670}
671
672func (t *OperationTracker) reset() {
673 if t == nil {
674 return
675 }
676 t.mu.Lock()
677 defer t.mu.Unlock()
678 t.pending = make(map[ports.BrokerOperationID]struct{}, MaxPendingOperations)
679 t.results = make(map[ports.BrokerOperationID]ports.BrokerMutationOutcome, MaxTrackedOperations)
680 t.order = nil
681}
682
683// StreamState is the closed lifecycle of one logical stream.
684type StreamState uint8
685
686const (
687 // StreamOpening is an allocated stream the broker has not established.
688 StreamOpening StreamState = iota + 1
689 // StreamOpen carries data and progress.
690 StreamOpen
691 // StreamRetired is terminal: no further frame is applied.
692 StreamRetired
693)
694
695func (s StreamState) String() string {
696 switch s {
697 case StreamOpening:
698 return "opening"
699 case StreamOpen:
700 return "open"
701 case StreamRetired:
702 return "retired"
703 default:
704 return "unknown"
705 }
706}
707
708// StreamDisposition classifies one stream frame: applied, or a late frame for
709// a retired stream that is discarded without effect.
710type StreamDisposition uint8
711
712const (
713 // StreamAccepted reports the frame was applied.
714 StreamAccepted StreamDisposition = iota + 1
715 // StreamDiscarded reports a late frame for a retired stream. It is
716 // ignored and never revives, reopens, or closes anything.
717 StreamDiscarded
718)
719
720// StreamTracker bounds the logical streams of one connection scope. Stream
721// identities are allocated by the calling service and may arrive slightly out
722// of order when two opens race, so admission uses one bounded anti-replay
723// window instead of a bare high-water mark: every unconsumed identity inside
724// the window is admitted exactly once, a duplicate is refused, and an identity
725// that trails the newest admitted identity by the window size is stale. At most
726// MaxStreams streams are opening or open at once; at most MaxRetiredStreams
727// retired records are retained, so an older late frame is still classified as
728// retired rather than mistaken for a future stream. It is safe for concurrent
729// use.
730type StreamTracker struct {
731 mu sync.Mutex
732 scope Scope
733 states map[ports.BrokerStreamID]StreamState
734 retired []ports.BrokerStreamID
735 highest ports.BrokerStreamID
736 window ports.BrokerStreamWindow
737 active int
738}
739
740// newStreamTracker binds a tracker to one exact connection scope. It is
741// unexported on purpose: the trackers are connection-owned state, and every
742// production mutation goes through the Connection methods that hold the
743// connection lock across the ready check and the tracker mutation.
744func newStreamTracker(scope Scope) (*StreamTracker, error) {
745 if err := scope.Validate(); err != nil {
746 return nil, err
747 }
748 return &StreamTracker{scope: scope, states: make(map[ports.BrokerStreamID]StreamState, MaxStreams)}, nil
749}
750
751// Open allocates one fresh logical stream in Opening state. The identity is
752// admitted on this connection's bounded anti-replay window: it must be
753// unconsumed and inside the window, and it must not already be in use. An
754// identity allocated by a concurrent open is therefore admitted even when it
755// arrives just after a higher one, while a replay and an identity evicted past
756// the window are both refused.
757func (t *StreamTracker) Open(scope Scope, stream ports.BrokerStreamID) error {
758 if t == nil {
759 return ErrConnectionClosed
760 }
761 t.mu.Lock()
762 defer t.mu.Unlock()
763 if scope != t.scope {
764 return ErrScopeMismatch
765 }
766 if err := stream.Validate(); err != nil {
767 return ErrInvalidStream
768 }
769 if err := t.window.Admit(stream); err != nil {
770 // A duplicate or an identity evicted from the window is stale rather
771 // than malformed, so the caller can classify it as a never-reused
772 // identity instead of a protocol violation.
773 return ErrStreamIDReused
774 }
775 if t.active >= MaxStreams {
776 return ErrTooManyStreams
777 }
778 t.states[stream] = StreamOpening
779 if stream > t.highest {
780 t.highest = stream
781 }
782 t.active++
783 return nil
784}
785
786// Opened moves one opening stream to Open. Every other state is illegal:
787// opening is only ever confirmed once.
788func (t *StreamTracker) Opened(scope Scope, stream ports.BrokerStreamID) error {
789 if t == nil {
790 return ErrConnectionClosed
791 }
792 t.mu.Lock()
793 defer t.mu.Unlock()
794 state, err := t.lookupLocked(scope, stream)
795 if err != nil {
796 return err
797 }
798 if state != StreamOpening {
799 return ErrStreamState
800 }
801 t.states[stream] = StreamOpen
802 return nil
803}
804
805// Data delivers one opaque daemon frame. Only an open stream accepts data; a
806// retired stream discards it, and a stream that was never allocated is
807// refused.
808func (t *StreamTracker) Data(scope Scope, stream ports.BrokerStreamID) (StreamDisposition, error) {
809 if t == nil {
810 return 0, ErrConnectionClosed
811 }
812 t.mu.Lock()
813 defer t.mu.Unlock()
814 state, err := t.lookupLocked(scope, stream)
815 if err != nil {
816 return 0, err
817 }
818 switch state {
819 case StreamOpen:
820 return StreamAccepted, nil
821 case StreamRetired:
822 return StreamDiscarded, nil
823 default:
824 return 0, ErrStreamState
825 }
826}
827
828// Progress reports one phase of an in-flight stream operation. It is legal
829// while opening and while open; a retired stream discards it.
830func (t *StreamTracker) Progress(scope Scope, stream ports.BrokerStreamID) (StreamDisposition, error) {
831 if t == nil {
832 return 0, ErrConnectionClosed
833 }
834 t.mu.Lock()
835 defer t.mu.Unlock()
836 state, err := t.lookupLocked(scope, stream)
837 if err != nil {
838 return 0, err
839 }
840 if state == StreamRetired {
841 return StreamDiscarded, nil
842 }
843 return StreamAccepted, nil
844}
845
846// Close moves one opening or open stream to its terminal retired state. A
847// late close for a retired stream is discarded.
848func (t *StreamTracker) Close(scope Scope, stream ports.BrokerStreamID) (StreamDisposition, error) {
849 if t == nil {
850 return 0, ErrConnectionClosed
851 }
852 t.mu.Lock()
853 defer t.mu.Unlock()
854 state, err := t.lookupLocked(scope, stream)
855 if err != nil {
856 return 0, err
857 }
858 if state == StreamRetired {
859 return StreamDiscarded, nil
860 }
861 t.retireLocked(stream)
862 return StreamAccepted, nil
863}
864
865// Active returns the number of opening or open streams.
866func (t *StreamTracker) Active() int {
867 if t == nil {
868 return 0
869 }
870 t.mu.Lock()
871 defer t.mu.Unlock()
872 return t.active
873}
874
875// State returns the tracked state of one stream.
876func (t *StreamTracker) State(stream ports.BrokerStreamID) (StreamState, bool) {
877 if t == nil {
878 return 0, false
879 }
880 t.mu.Lock()
881 defer t.mu.Unlock()
882 state, ok := t.states[stream]
883 return state, ok
884}
885
886// lookupLocked fences the frame scope and classifies the stream: a tracked
887// opening or open state, a retired stream that was allocated earlier (a
888// tracked retired record or one evicted past the retention bound), or a
889// future identity that was never allocated.
890func (t *StreamTracker) lookupLocked(scope Scope, stream ports.BrokerStreamID) (StreamState, error) {
891 if scope != t.scope {
892 return 0, ErrScopeMismatch
893 }
894 if err := stream.Validate(); err != nil {
895 return 0, ErrInvalidStream
896 }
897 if state, ok := t.states[stream]; ok {
898 return state, nil
899 }
900 if stream > t.highest {
901 return 0, ErrFutureStream
902 }
903 return StreamRetired, nil
904}
905
906func (t *StreamTracker) retireLocked(stream ports.BrokerStreamID) {
907 t.states[stream] = StreamRetired
908 t.retired = append(t.retired, stream)
909 t.active--
910 for len(t.retired) > MaxRetiredStreams {
911 evicted := t.retired[0]
912 t.retired = t.retired[1:]
913 delete(t.states, evicted)
914 }
915}
916
917func (t *StreamTracker) reset() {
918 if t == nil {
919 return
920 }
921 t.mu.Lock()
922 defer t.mu.Unlock()
923 t.states = make(map[ports.BrokerStreamID]StreamState, MaxStreams)
924 t.retired = nil
925 t.highest = 0
926 t.window = ports.BrokerStreamWindow{}
927 t.active = 0
928}