1package broker
2
3import (
4 "context"
5 "encoding/binary"
6 "errors"
7 "log/slog"
8 "sync"
9 "time"
10
11 "github.com/bnema/vev/internal/domain"
12 "github.com/bnema/vev/internal/ports"
13)
14
15// PoolLimits bound live physical keys, clients, reservations (including I/O in
16// flight), and logical streams. No unbounded queue or alias cache is retained.
17//
18// Warm and Idle bound warm reuse: a published physical transport whose last
19// logical stream closed stays warm so a returning client opens a new logical
20// stream without another dial, SSH bootstrap, or QUIC handshake. At most Warm
21// idle transports are retained; releasing one more retires the least recently
22// idled first. Zero Warm retires a transport as soon as it goes idle. A
23// positive Idle retires a transport that long after it went idle; zero Idle
24// sets no age bound, so retention then ends only through Warm eviction,
25// physical loss, or Close. Warm transports never pin the broker lifecycle.
26type PoolLimits struct {
27 Physical, Clients, Streams, StreamsPerClient, Warm int
28 Idle time.Duration
29}
30
31// poolKey is the exact authenticated (identity, policy) pair the broker pools
32// one physical transport by. The resolved endpoint's opaque address is
33// deliberately excluded: it selects the transport, not the pooling identity.
34//
35// A local stream request carries no endpoint and no registration, so its key is
36// the broker-owned local identity and policy the resolver returns. Every local
37// stream therefore shares one pooled physical connection to the local daemon,
38// exactly as every stream to one remote identity does; the local daemon is not
39// a configured host, so no registration participates in its key.
40type poolKey struct {
41 identity ports.BrokerDaemonIdentity
42 policy ports.BrokerPolicy
43}
44type pendingKey struct {
45 fence ports.BrokerEndpointFence
46 policy ports.BrokerPolicy
47 // startMode is part of the pending key on purpose: an acquisition that may
48 // start the target daemon must never be coalesced with an ExistingOnly
49 // acquisition that must not, even when they name the same unbound endpoint
50 // and policy. Once the physical connection is authenticated and published,
51 // the canonical key is identity plus exact policy alone: a compatible
52 // transport is shareable regardless of how it was acquired.
53 startMode ports.BrokerDaemonStartMode
54}
55type poolClient struct {
56 // window is this client's bounded anti-replay admission window. Stream IDs
57 // are allocated by the calling service and may arrive out of order when two
58 // opens race, so a plain monotone high-water mark would refuse a valid
59 // concurrent open; the window admits every unconsumed ID inside it and
60 // refuses duplicates and IDs evicted past its bound.
61 window ports.BrokerStreamWindow
62 streams map[ports.BrokerStreamID]*reservation
63}
64type reservation struct {
65 cancel context.CancelFunc
66 entry *poolEntry
67}
68
69// retire records why one entry ends and cancels it with that reason.
70func (e *poolEntry) retire(reason error) func() {
71 return func() { e.cancel(reason) }
72}
73
74// Pool entry retirement reasons. Each one is the cancellation cause of the
75// entry context, so the streams it ends and the broker log name the reason.
76var (
77 errEntryIdle = errors.New("broker: pooled connection idle timeout")
78 errEntryEvicted = errors.New("broker: pooled connection evicted for a newer warm connection")
79 errEntryAbandoned = errors.New("broker: pooled connection abandoned before it was ready")
80 errEntryRedirect = errors.New("broker: pooled connection merged into an existing connection")
81 errEntryLost = errors.New("broker: pooled physical connection lost")
82)
83
84type poolEntry struct {
85 key poolKey
86 pending pendingKey
87 ctx context.Context
88 cancel context.CancelCauseFunc
89 ready, wake chan struct{}
90 physical ports.BrokerPhysicalConnection
91 redirect *poolEntry
92 err error
93 refs int
94 retiring bool
95 // idleOrder is nonzero while the published entry is warm (no reservation)
96 // and orders warm entries for least-recently-idled eviction.
97 idleOrder uint64
98}
99
100// Pool is transport independent; the broker service composes it.
101// mu protects bookkeeping only; no port I/O or Close runs under it.
102type Pool struct {
103 mu sync.Mutex
104 epoch ports.BrokerEpoch
105 routes ports.BrokerRouteAuthority
106 binder ports.BrokerIdentityBinder
107 connector ports.BrokerEndpointConnector
108 clock ports.Clock
109 limits PoolLimits
110 ctx context.Context
111 cancel context.CancelFunc
112 closed bool
113 next uint64
114 idleOrder uint64
115 clients map[ports.BrokerConnectionID]*poolClient
116 entries map[poolKey]*poolEntry
117 pending map[pendingKey]*poolEntry
118 streams int
119 wg sync.WaitGroup
120}
121
122func NewPool(epoch ports.BrokerEpoch, routes ports.BrokerRouteAuthority, binder ports.BrokerIdentityBinder, connector ports.BrokerEndpointConnector, clock ports.Clock, limits PoolLimits) (*Pool, error) {
123 if epoch == 0 || routes == nil || binder == nil || connector == nil || clock == nil || limits.Physical <= 0 || limits.Clients <= 0 || limits.Streams <= 0 || limits.StreamsPerClient <= 0 || limits.Warm < 0 || limits.Idle < 0 {
124 return nil, ports.BrokerAdmissionInvalid
125 }
126 ctx, cancel := context.WithCancel(context.Background())
127 return &Pool{epoch: epoch, routes: routes, binder: binder, connector: connector, clock: clock, limits: limits, ctx: ctx, cancel: cancel, clients: make(map[ports.BrokerConnectionID]*poolClient), entries: make(map[poolKey]*poolEntry), pending: make(map[pendingKey]*poolEntry)}, nil
128}
129
130// RegisterClient issues epoch-scoped nonreusable identities. The bounded active
131// map needs no retired-ID tombstones. Stream IDs must increase within a client.
132func (p *Pool) RegisterClient() (ports.BrokerConnectionID, error) {
133 p.mu.Lock()
134 defer p.mu.Unlock()
135 if p.closed {
136 return ports.BrokerConnectionID{}, ports.BrokerAdmissionClosed
137 }
138 if len(p.clients) >= p.limits.Clients || p.next == ^uint64(0) {
139 return ports.BrokerConnectionID{}, ports.BrokerAdmissionLimit
140 }
141 p.next++
142 var id ports.BrokerConnectionID
143 binary.BigEndian.PutUint64(id[:8], uint64(p.epoch))
144 binary.BigEndian.PutUint64(id[8:], p.next)
145 p.clients[id] = &poolClient{streams: make(map[ports.BrokerStreamID]*reservation)}
146 return id, nil
147}
148
149func (p *Pool) CloseClient(id ports.BrokerConnectionID) {
150 p.mu.Lock()
151 c := p.clients[id]
152 delete(p.clients, id)
153 var cancels []context.CancelFunc
154 if c != nil {
155 for _, r := range c.streams {
156 cancels = append(cancels, r.cancel)
157 }
158 }
159 p.mu.Unlock()
160 for _, cancel := range cancels {
161 cancel()
162 }
163}
164
165func (p *Pool) CloseStream(id ports.BrokerConnectionID, stream ports.BrokerStreamID) error {
166 p.mu.Lock()
167 c := p.clients[id]
168 var r *reservation
169 if c != nil {
170 r = c.streams[stream]
171 }
172 p.mu.Unlock()
173 if r == nil {
174 return ports.BrokerAdmissionStale
175 }
176 r.cancel()
177 return nil
178}
179
180func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest) (ports.BrokerEnvelopeStream, error) {
181 if req.Epoch != p.epoch {
182 return nil, ports.BrokerError{Code: ports.BrokerErrorStaleEpoch}
183 }
184 ctx, cancel := context.WithCancel(ctx)
185 p.mu.Lock()
186 c := p.clients[req.Connection]
187 if p.closed {
188 p.mu.Unlock()
189 cancel()
190 return nil, ports.BrokerAdmissionClosed
191 }
192 if c == nil {
193 p.mu.Unlock()
194 cancel()
195 return nil, ports.BrokerAdmissionStale
196 }
197 if err := c.window.Admit(req.Stream); err != nil {
198 p.mu.Unlock()
199 cancel()
200 return nil, err
201 }
202 if p.streams >= p.limits.Streams || len(c.streams) >= p.limits.StreamsPerClient {
203 p.mu.Unlock()
204 cancel()
205 return nil, ports.BrokerAdmissionLimit
206 }
207 r := &reservation{cancel: cancel}
208 c.streams[req.Stream] = r
209 p.streams++
210 p.wg.Add(1)
211 p.mu.Unlock()
212 stop := context.AfterFunc(p.ctx, cancel)
213 var entry *poolEntry
214 release := func() {
215 stop()
216 cancel()
217 p.mu.Lock()
218 delete(c.streams, req.Stream)
219 p.streams--
220 var abandon func()
221 var evicted []func()
222 reservedEntry := r.entry
223 if reservedEntry != nil {
224 reservedEntry.refs--
225 if reservedEntry.refs == 0 {
226 select {
227 case <-reservedEntry.ready:
228 evicted = p.idleLocked(reservedEntry)
229 default:
230 reservedEntry.retiring = true
231 abandon = reservedEntry.retire(errEntryAbandoned)
232 }
233 }
234 select {
235 case reservedEntry.wake <- struct{}{}:
236 default:
237 }
238 }
239 p.mu.Unlock()
240 if abandon != nil {
241 abandon()
242 }
243 for _, evict := range evicted {
244 evict()
245 }
246 p.wg.Done()
247 }
248 success := false
249 defer func() {
250 if !success {
251 release()
252 }
253 }()
254 req.Env = append([]string(nil), req.Env...)
255 resolved, err := p.routes.ResolveDialTarget(ctx, req)
256 if err != nil {
257 return nil, normalizePoolError(err)
258 }
259 if err := resolved.Validate(); err != nil {
260 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: err}
261 }
262 if !resolved.Policy.Compatible(req.Policy) {
263 return nil, ports.BrokerError{Code: ports.BrokerErrorConflictingPolicy}
264 }
265 if err = ctx.Err(); err != nil {
266 return nil, normalizePoolError(err)
267 }
268 key := poolKey{resolved.ExpectedIdentity.Identity, resolved.Policy}
269 pending := pendingKey{fence: resolved.Fence, policy: resolved.Policy, startMode: resolved.StartMode}
270 p.mu.Lock()
271 if p.closed {
272 p.mu.Unlock()
273 return nil, ports.BrokerAdmissionClosed
274 }
275 if resolved.ExpectedIdentity.Bound {
276 entry = p.entries[key]
277 } else {
278 entry = p.pending[pending]
279 }
280 if entry != nil && entry.retiring {
281 p.mu.Unlock()
282 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "selected physical entry is retiring", Cause: errors.New("broker: selected physical entry is retiring")}
283 }
284 if entry == nil {
285 if len(p.entries)+len(p.pending) >= p.limits.Physical {
286 p.mu.Unlock()
287 return nil, ports.BrokerAdmissionLimit
288 }
289 ectx, ecancel := context.WithCancelCause(p.ctx)
290 entry = &poolEntry{key: key, pending: pending, ctx: ectx, cancel: ecancel, ready: make(chan struct{}), wake: make(chan struct{}, 1)}
291 if resolved.ExpectedIdentity.Bound {
292 p.entries[key] = entry
293 } else {
294 p.pending[pending] = entry
295 }
296 p.wg.Add(1)
297 go p.runEntry(entry, resolved)
298 }
299 entry.refs++
300 entry.idleOrder = 0
301 r.entry = entry
302 p.mu.Unlock()
303 select {
304 case <-ctx.Done():
305 return nil, normalizePoolError(ctx.Err())
306 case <-entry.ready:
307 }
308 winner, err := p.redirectReservation(entry, r)
309 if err != nil {
310 return nil, err
311 }
312 if winner != nil {
313 entry = winner
314 select {
315 case <-ctx.Done():
316 return nil, normalizePoolError(ctx.Err())
317 case <-entry.ready:
318 }
319 }
320 if entry.err != nil {
321 return nil, entry.err
322 }
323 openCtx, openCancel := context.WithCancel(ctx)
324 stopEntry := context.AfterFunc(entry.ctx, openCancel)
325 raw, err := entry.physical.OpenStream(openCtx, req)
326 stopEntry()
327 openCancel()
328 if err != nil {
329 // A failed open may still carry a typed-nil connection, which compares
330 // unequal to nil while holding no receiver: closing it would dereference
331 // a nil pointer.
332 if !nilDependency(raw) {
333 _ = raw.Close()
334 }
335 return nil, p.streamTerminal(ctx, entry, req, normalizePoolError(err))
336 }
337 if nilDependency(raw) {
338 return nil, ports.BrokerError{Code: ports.BrokerErrorIncompatible, Cause: ports.BrokerAdmissionInvalid}
339 }
340 if terminal := p.streamTerminal(ctx, entry, req, nil); terminal != nil {
341 _ = raw.Close()
342 return nil, terminal
343 }
344 stream := &pooledStream{BrokerEnvelopeStream: raw, done: make(chan struct{}), release: release}
345 success = true
346 go func() {
347 var terminal error
348 select {
349 case <-ctx.Done():
350 terminal = ports.BrokerError{Code: ports.BrokerErrorCancelled}
351 case <-entry.ctx.Done():
352 case <-entry.physical.Done():
353 case <-raw.Done():
354 terminal = raw.Err()
355 case <-stream.done:
356 return
357 }
358 stream.finish(p.streamTerminal(ctx, entry, req, terminal))
359 }()
360 return stream, nil
361}
362
363func (p *Pool) redirectReservation(entry *poolEntry, reservation *reservation) (*poolEntry, error) {
364 p.mu.Lock()
365 defer p.mu.Unlock()
366 winner := entry.redirect
367 if winner == nil {
368 return nil, nil
369 }
370 if current := p.entries[winner.key]; current != winner || winner.retiring {
371 return nil, ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "physical identity redirect unavailable"}
372 }
373 entry.refs--
374 winner.refs++
375 winner.idleOrder = 0
376 reservation.entry = winner
377 return winner, nil
378}
379
380func (p *Pool) runEntry(e *poolEntry, endpoint ports.BrokerDialTarget) {
381 defer p.wg.Done()
382 physical, err := p.connector.Connect(e.ctx, endpoint)
383 if err == nil && (nilDependency(physical) || (endpoint.ExpectedIdentity.Bound && physical.Identity() != endpoint.ExpectedIdentity.Identity) || physical.Identity().Validate() != nil || !physical.Policy().Compatible(endpoint.Policy) || physical.Incarnation().Validate() != nil) {
384 err = ports.BrokerError{Code: ports.BrokerErrorIncompatible}
385 }
386 if err == nil {
387 identity, bindErr := p.binder.BindAuthenticatedIdentity(e.ctx, ports.BrokerIdentityBindingRequest{Fence: endpoint.Fence, Policy: endpoint.Policy, Identity: physical.Identity()})
388 if bindErr != nil {
389 err = bindErr
390 } else if identity != physical.Identity() {
391 err = ports.BrokerError{Code: ports.BrokerErrorIncompatible}
392 }
393 }
394 p.mu.Lock()
395 if !endpoint.ExpectedIdentity.Bound {
396 delete(p.pending, e.pending)
397 }
398 if err == nil {
399 e.key = poolKey{identity: physical.Identity(), policy: endpoint.Policy}
400 if winner := p.entries[e.key]; winner != nil && winner != e && !winner.retiring {
401 e.redirect = winner
402 } else {
403 p.entries[e.key] = e
404 }
405 }
406 e.physical = physical
407 e.err = normalizePoolError(err)
408 close(e.ready)
409 redirect := e.redirect
410 p.mu.Unlock()
411 if redirect != nil {
412 _ = physical.Close()
413 e.cancel(errEntryRedirect)
414 return
415 }
416 if err == nil {
417 // Zero Idle sets no age bound: no timer exists and a nil expiry channel
418 // never fires, so only Warm eviction, physical loss, or Close retire.
419 var timer ports.Timer
420 if p.limits.Idle > 0 {
421 timer = p.clock.NewTimer(p.limits.Idle)
422 defer stopPoolTimer(timer)
423 }
424 for {
425 p.mu.Lock()
426 idle := e.refs == 0
427 p.mu.Unlock()
428 var expiry <-chan time.Time
429 if timer != nil {
430 stopPoolTimer(timer)
431 if idle {
432 timer.Reset(p.limits.Idle)
433 expiry = timer.C()
434 }
435 }
436 select {
437 case <-e.ctx.Done():
438 goto retire
439 case <-physical.Done():
440 e.cancel(errEntryLost)
441 goto retire
442 case <-e.wake:
443 continue
444 case <-expiry:
445 p.mu.Lock()
446 if e.refs == 0 {
447 e.retiring = true
448 p.mu.Unlock()
449 e.cancel(errEntryIdle)
450 goto retire
451 }
452 p.mu.Unlock()
453 }
454 }
455 }
456retire:
457 p.mu.Lock()
458 e.retiring = true
459 refs := e.refs
460 p.mu.Unlock()
461 if err != nil {
462 e.cancel(err)
463 } else {
464 e.cancel(errEntryLost)
465 }
466 if physical != nil {
467 p.logRetire(e, physical, refs)
468 _ = physical.Close()
469 }
470 // Keep the key occupied until Close completes: never overlap physicals.
471 p.mu.Lock()
472 if p.entries[e.key] == e {
473 delete(p.entries, e.key)
474 }
475 if p.pending[e.pending] == e {
476 delete(p.pending, e.pending)
477 }
478 p.mu.Unlock()
479}
480
481// idleLocked marks a published entry whose last reservation just ended as warm
482// and returns the cancellations that retire warm entries beyond limits.Warm,
483// least recently idled first. The caller runs them after unlocking. An entry
484// that failed, lost a redirect race, or is already retiring is never warm.
485func (p *Pool) idleLocked(e *poolEntry) []func() {
486 if e.err != nil || e.redirect != nil || e.retiring || p.entries[e.key] != e {
487 return nil
488 }
489 p.idleOrder++
490 e.idleOrder = p.idleOrder
491 var warm []*poolEntry
492 for _, candidate := range p.entries {
493 if candidate.idleOrder != 0 && candidate.refs == 0 && !candidate.retiring {
494 warm = append(warm, candidate)
495 }
496 }
497 var evicted []func()
498 for len(warm) > p.limits.Warm {
499 oldest := 0
500 for i := range warm {
501 if warm[i].idleOrder < warm[oldest].idleOrder {
502 oldest = i
503 }
504 }
505 victim := warm[oldest]
506 victim.retiring = true
507 evicted = append(evicted, victim.retire(errEntryEvicted))
508 warm = append(warm[:oldest], warm[oldest+1:]...)
509 }
510 return evicted
511}
512
513// AdoptPhysical publishes a probe-authenticated transport as a warm entry.
514// The caller retains ownership when adoption fails and must close it. A
515// concurrently opened pooled transport wins: probes must not replace it.
516func (p *Pool) AdoptPhysical(physical ports.BrokerPhysicalConnection, policy ports.BrokerPolicy) bool {
517 if nilDependency(physical) || physical.Identity().Validate() != nil ||
518 physical.Incarnation().Validate() != nil || !physical.Policy().Compatible(policy) || p.limits.Warm == 0 {
519 return false
520 }
521 select {
522 case <-physical.Done():
523 return false
524 default:
525 }
526 key := poolKey{identity: physical.Identity(), policy: policy}
527 p.mu.Lock()
528 if p.closed || len(p.entries)+len(p.pending) >= p.limits.Physical || p.entries[key] != nil {
529 p.mu.Unlock()
530 return false
531 }
532 ctx, cancel := context.WithCancelCause(p.ctx)
533 e := &poolEntry{key: key, ctx: ctx, cancel: cancel, ready: make(chan struct{}), wake: make(chan struct{}, 1), physical: physical}
534 close(e.ready)
535 p.entries[key] = e
536 evicted := p.idleLocked(e)
537 p.wg.Add(1)
538 p.mu.Unlock()
539 for _, evict := range evicted {
540 evict()
541 }
542 go p.watchAdopted(e)
543 return true
544}
545
546func (p *Pool) watchAdopted(e *poolEntry) {
547 defer p.wg.Done()
548 physical := e.physical
549 var timer ports.Timer
550 if p.limits.Idle > 0 {
551 timer = p.clock.NewTimer(p.limits.Idle)
552 defer stopPoolTimer(timer)
553 }
554 for {
555 p.mu.Lock()
556 idle := e.refs == 0
557 p.mu.Unlock()
558 var expiry <-chan time.Time
559 if timer != nil {
560 stopPoolTimer(timer)
561 if idle {
562 timer.Reset(p.limits.Idle)
563 expiry = timer.C()
564 }
565 }
566 select {
567 case <-e.ctx.Done():
568 goto retire
569 case <-physical.Done():
570 e.cancel(errEntryLost)
571 goto retire
572 case <-e.wake:
573 continue
574 case <-expiry:
575 p.mu.Lock()
576 if e.refs == 0 && e.idleOrder != 0 {
577 e.retiring = true
578 p.mu.Unlock()
579 e.cancel(errEntryIdle)
580 goto retire
581 }
582 p.mu.Unlock()
583 }
584 }
585retire:
586 p.mu.Lock()
587 e.retiring = true
588 refs := e.refs
589 p.mu.Unlock()
590 e.cancel(errEntryLost)
591 p.logRetire(e, physical, refs)
592 _ = physical.Close()
593 p.mu.Lock()
594 if p.entries[e.key] == e {
595 delete(p.entries, e.key)
596 }
597 p.mu.Unlock()
598}
599
600// SharedPhysical returns the published physical transport the pool holds for
601// one authenticated identity and exact policy, active or warm, without
602// reserving it. Observation borrows it to avoid a second bootstrap; the
603// borrower must not close it and must tolerate its retirement at any time.
604// Borrowing never refreshes warm order or the idle deadline.
605func (p *Pool) SharedPhysical(identity ports.BrokerDaemonIdentity, policy ports.BrokerPolicy) (ports.BrokerPhysicalConnection, bool) {
606 p.mu.Lock()
607 defer p.mu.Unlock()
608 e := p.entries[poolKey{identity: identity, policy: policy}]
609 if p.closed || e == nil || e.retiring || e.redirect != nil {
610 return nil, false
611 }
612 select {
613 case <-e.ready:
614 default:
615 return nil, false
616 }
617 if e.err != nil || nilDependency(e.physical) {
618 return nil, false
619 }
620 select {
621 case <-e.physical.Done():
622 return nil, false
623 default:
624 }
625 return e.physical, true
626}
627
628// logRetire records one physical connection leaving the pool: why the entry
629// ended, the transport's own failure when it died, and how many streams were
630// still using it. Pool shutdown is not logged.
631func (p *Pool) logRetire(e *poolEntry, physical ports.BrokerPhysicalConnection, refs int) {
632 if p.ctx.Err() != nil {
633 return
634 }
635 level := slog.LevelInfo
636 if refs > 0 {
637 level = slog.LevelWarn
638 }
639 args := []any{"reason", context.Cause(e.ctx), "active_streams", refs}
640 select {
641 case <-physical.Done():
642 args = append(args, "failure", physical.FailureKind().String(), "cause", physical.Err())
643 default:
644 }
645 slog.Log(context.Background(), level, "broker_pool_connection_retired", args...)
646}
647
648// Close cancels pending operations, closes transports, and joins all owned
649// workers. Bounded completion depends on the explicit cancellation/Close port
650// contract; callers must not supply adapters that ignore it.
651func (p *Pool) Close() error {
652 p.mu.Lock()
653 p.closed = true
654 p.mu.Unlock()
655 p.cancel()
656 p.wg.Wait()
657 return nil
658}
659
660type pooledStream struct {
661 ports.BrokerEnvelopeStream
662 once sync.Once
663 done chan struct{}
664 err error
665 closeErr error
666 release func()
667}
668
669func (s *pooledStream) Done() <-chan struct{} { return s.done }
670func (s *pooledStream) Err() error {
671 select {
672 case <-s.done:
673 return s.err
674 default:
675 return nil
676 }
677}
678func (s *pooledStream) Close() error { s.finish(nil); return s.closeErr }
679func (s *pooledStream) finish(err error) {
680 s.once.Do(func() { s.err = err; s.closeErr = s.BrokerEnvelopeStream.Close(); s.release(); close(s.done) })
681}
682
683// One goroutine owns this timer. Drain a buffered expiry before every reset,
684// including clocks implementing pre-Go-1.23 timer semantics.
685func stopPoolTimer(timer ports.Timer) {
686 if !timer.Stop() {
687 select {
688 case <-timer.C():
689 default:
690 }
691 }
692}
693
694func normalizePoolError(err error) error {
695 if err == nil {
696 return nil
697 }
698 code := ports.BrokerErrorUnavailable
699 var typed ports.BrokerError
700 var typedPtr *ports.BrokerError
701 switch {
702 case errors.Is(err, context.DeadlineExceeded):
703 code = ports.BrokerErrorTimeout
704 case errors.Is(err, context.Canceled):
705 code = ports.BrokerErrorCancelled
706 case errors.As(err, &typed) && typed.Validate() == nil:
707 code = typed.Code
708 case errors.As(err, &typedPtr) && typedPtr != nil && typedPtr.Validate() == nil:
709 code = typedPtr.Code
710 }
711 return ports.BrokerError{Code: code, Cause: err}
712}
713
714// Precedence at terminal observation: pool shutdown, request/client cancellation,
715// physical Done, entry retirement, logical result. A completed stream is immutable.
716func (p *Pool) streamTerminal(ctx context.Context, e *poolEntry, req ports.BrokerOpenStreamRequest, fallback error) error {
717 if p.ctx.Err() != nil {
718 return ports.BrokerError{Code: ports.BrokerErrorCancelled, Cause: p.ctx.Err()}
719 }
720 if ctx.Err() != nil {
721 return normalizePoolError(ctx.Err())
722 }
723 select {
724 case <-e.physical.Done():
725 kind := e.physical.FailureKind()
726 if kind < domain.RemoteFailureTransport || kind > domain.RemoteFailureInvalidResponse {
727 kind = domain.RemoteFailureTransport
728 }
729 return ports.BrokerStreamLost{Epoch: req.Epoch, Connection: req.Connection, Stream: req.Stream, Cause: kind, Err: e.physical.Err()}
730 default:
731 }
732 if e.ctx.Err() != nil {
733 reason := context.Cause(e.ctx)
734 return ports.BrokerError{Code: ports.BrokerErrorAttachmentLost, Text: reason.Error(), Cause: reason}
735 }
736 return normalizePoolError(fallback)
737}