1package client
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "sync"
8 "sync/atomic"
9 "time"
10
11 "github.com/bnema/vev/internal/ports"
12 "github.com/bnema/vev/internal/protocol"
13)
14
15// Committed attachment orchestration.
16//
17// After the picker resolves a committed selection to one exact
18// ports.BrokerOpenStreamRequest, the supervisor opens that logical stream on
19// the broker and hands the typed connection to the real attachment worker
20// through its existing foreground grant. The supervisor still owns the
21// terminal, raw mode, the single input lifetime, and the close authority over
22// the stream; the picker never opens, writes, or retargets a session.
23//
24// One absolute deadline bounds the whole attachment: stream open, the session
25// preamble, Hello/Welcome, and the committed initial publication. It is never
26// restarted and never extended. When the admitted stream exposes an accepted
27// absolute deadline (ports.HandshakeDeadlineProvider) the earlier of the two is
28// kept, so queue time spent before the stream reached this client is consumed
29// rather than repaid with a second budget; an already-elapsed deadline fails the
30// attachment immediately. The deadline is driven by ports.Clock/ports.Timer,
31// never the wall clock and never a sleep.
32
33// attachmentTimeoutError is the one typed failure a deadline ends an attachment
34// with. The cause stays local for errors.Is/As and is never display text.
35func attachmentTimeoutError(cause error) error {
36 return ports.BrokerError{Code: ports.BrokerErrorTimeout, Text: "attachment timed out", Cause: cause}
37}
38
39// errResumeForegroundBusy ends a resume whose reattach could not claim the
40// terminal foreground.
41var errResumeForegroundBusy = ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "resume could not claim the terminal"}
42
43// attachmentDeadline owns the one absolute deadline of a whole attachment. It
44// is safe for concurrent use: its timer goroutine expires it while the
45// supervisor's run goroutine may adopt a later deadline after the stream opens.
46type attachmentDeadline struct {
47 clock ports.Clock
48 ctx context.Context
49 cancel context.CancelFunc
50 timedOut chan struct{}
51
52 mu sync.Mutex
53 deadline time.Time
54 timer ports.Timer
55 stop chan struct{}
56 done chan struct{}
57 expired atomic.Bool
58 expireOnce sync.Once
59 // finishOnce makes finish idempotent across the deferred teardown and any
60 // early failure path.
61 finishOnce sync.Once
62}
63
64// startAttachmentDeadline arms the pre-open budget of one attachment. The
65// returned deadline already bounds the OpenStream call; adopt may fold the
66// stream's own accepted absolute deadline into it, but the earlier deadline
67// always stays in force.
68func startAttachmentDeadline(parent context.Context, clock ports.Clock) *attachmentDeadline {
69 if parent == nil {
70 parent = context.Background()
71 }
72 if supervisorNil(clock) {
73 clock = systemClock{}
74 }
75 ctx, cancel := context.WithCancel(parent)
76 deadline := &attachmentDeadline{
77 clock: clock,
78 ctx: ctx,
79 cancel: cancel,
80 timedOut: make(chan struct{}),
81 }
82 deadline.mu.Lock()
83 deadline.deadline = clock.Now().Add(protocol.HandshakeTimeout)
84 deadline.armLocked(protocol.HandshakeTimeout)
85 deadline.mu.Unlock()
86 return deadline
87}
88
89// armLocked starts a fresh timer for exactly duration. Callers hold d.mu and
90// must have stopped any previous timer. It never derives duration from
91// protocol.HandshakeTimeout itself, so folding in an absolute deadline can only
92// shrink the remaining budget, never restart it.
93func (d *attachmentDeadline) armLocked(duration time.Duration) {
94 if duration <= 0 {
95 d.expire()
96 return
97 }
98 timer := d.clock.NewTimer(duration)
99 if supervisorNil(timer) {
100 timer = systemClock{}.NewTimer(duration)
101 }
102 stop := make(chan struct{})
103 done := make(chan struct{})
104 d.timer = timer
105 d.stop = stop
106 d.done = done
107 go func() {
108 defer close(done)
109 select {
110 case <-timer.C():
111 d.expire()
112 case <-stop:
113 }
114 }()
115}
116
117// stopTimerLocked stops and joins the current timer goroutine. Callers hold
118// d.mu. It is a no-op when no timer is armed.
119func (d *attachmentDeadline) stopTimerLocked() {
120 if d.timer == nil {
121 return
122 }
123 close(d.stop)
124 <-d.done
125 d.timer.Stop()
126 d.timer, d.stop, d.done = nil, nil, nil
127}
128
129// expire closes the timeout signal and cancels the attachment context exactly
130// once without taking d.mu. Timer retirement may join a goroutine that has
131// already selected the timer channel, so expiry must never need that mutex.
132func (d *attachmentDeadline) expire() {
133 d.expireOnce.Do(func() {
134 d.expired.Store(true)
135 close(d.timedOut)
136 d.cancel()
137 })
138}
139
140// adopt folds the admitted stream's accepted absolute deadline into the one
141// attachment deadline, keeping the EARLIER of the two. It reports
142// errAttachmentDeadline when the effective deadline already elapsed (including
143// during stream open), which fails the attachment instead of starting a second
144// budget. A stream without the optional seam keeps the pre-open deadline
145// unchanged.
146//
147// Adoption never extends the deadline. The accepted deadline is anchored
148// before this client's attempt began, so it is normally later than the client's
149// own budget once the stream is admitted; adopting it verbatim would repay the
150// open stage with a second full budget and let the whole attachment outlive the
151// one absolute deadline. Keeping the earlier value still consumes the time
152// already spent before this client saw the stream, because that anchor is
153// earlier.
154func (d *attachmentDeadline) adopt(stream ports.BrokerLogicalConnection) error {
155 provider, ok := stream.(ports.HandshakeDeadlineProvider)
156 if !ok {
157 return nil
158 }
159 accepted := provider.HandshakeDeadline()
160 if accepted.IsZero() {
161 return nil
162 }
163 d.mu.Lock()
164 defer d.mu.Unlock()
165 d.stopTimerLocked()
166 // Expiry is lock-free and may win while the old timer is being joined.
167 // The post-join flag is authoritative; never re-arm a cancelled context.
168 if d.expired.Load() {
169 return errAttachmentDeadline
170 }
171 if accepted.Before(d.deadline) {
172 d.deadline = accepted
173 }
174 remaining := d.deadline.Sub(d.clock.Now())
175 if remaining <= 0 {
176 d.expire()
177 return errAttachmentDeadline
178 }
179 if d.expired.Load() {
180 return errAttachmentDeadline
181 }
182 d.armLocked(remaining)
183 return nil
184}
185
186// Context bounds every stage of the attachment (stream open, preamble,
187// Hello/Welcome, initial publication) with the one absolute deadline.
188func (d *attachmentDeadline) Context() context.Context { return d.ctx }
189
190// TimedOut reports whether the absolute deadline fired.
191func (d *attachmentDeadline) TimedOut() bool {
192 select {
193 case <-d.timedOut:
194 return true
195 default:
196 return false
197 }
198}
199
200// cause maps an open-stream failure to the typed timeout when the deadline
201// fired, so an expired budget is never reported as an ordinary unavailability.
202func (d *attachmentDeadline) cause(err error) error {
203 if d.TimedOut() {
204 return attachmentTimeoutError(errors.Join(errAttachmentDeadline, err))
205 }
206 return normalizeUnavailable(err)
207}
208
209// disarm stops the deadline without cancelling the worker context. It is safe
210// to race with expiry and with the deferred finish.
211func (d *attachmentDeadline) disarm() {
212 if d == nil {
213 return
214 }
215 d.mu.Lock()
216 d.stopTimerLocked()
217 d.mu.Unlock()
218}
219
220// finish stops the deadline timer and cancels the attachment context. It is
221// idempotent and is the caller's deferred teardown for one attachment.
222func (d *attachmentDeadline) finish() {
223 if d == nil {
224 return
225 }
226 d.finishOnce.Do(func() {
227 d.mu.Lock()
228 d.stopTimerLocked()
229 d.mu.Unlock()
230 d.cancel()
231 })
232}
233
234// attachmentSettlement is one worker run's outcome as observed by the
235// supervisor: whether the supervisor adopted the event or retired the run.
236type attachmentSettlement struct {
237 event AttachmentEvent
238 adopted bool
239}
240
241// attachmentOutcomeKind is what one attachment attempt tells the resume loop
242// to do next.
243type attachmentOutcomeKind uint8
244
245const (
246 // attachmentEnded returns to the picker; any failure is already presented.
247 attachmentEnded attachmentOutcomeKind = iota
248 // attachmentTerminated ends the whole run (cancellation, terminal EOF, or
249 // detach-and-exit); cause is the run's termination error, possibly nil.
250 attachmentTerminated
251 // attachmentSwap attaches to target, which the picker overlay or a daemon
252 // navigation committed while this attachment was live.
253 attachmentSwap
254 // attachmentResume reconnects to target after a live attachment was lost
255 // with cause.
256 attachmentResume
257 // attachmentRetry repeats a resume attempt that hit the transport-class
258 // failure cause before attaching.
259 attachmentRetry
260)
261
262// attachmentOutcome is the complete result of one attachment attempt.
263type attachmentOutcome struct {
264 kind attachmentOutcomeKind
265 target pickerAttachmentTarget
266 cause error
267}
268
269func attachmentEndedOutcome() attachmentOutcome { return attachmentOutcome{kind: attachmentEnded} }
270
271func attachmentTerminatedOutcome(err error) attachmentOutcome {
272 return attachmentOutcome{kind: attachmentTerminated, cause: err}
273}
274
275// resolveCommittedStreamRequest turns exactly one committed catalogue key into
276// the exact broker stream request the user committed. The stream ID is
277// allocated by the broker service — the single allocator on its own connection
278// — and is consumed even when resolution refuses, matching the never-reused
279// contract.
280func (s *Supervisor) resolveInitialStreamRequest(service ports.BrokerNavigator, snapshot ports.BrokerSnapshot, navigation InitialNavigation) (ports.BrokerOpenStreamRequest, error) {
281 // The intent is resolved against the very snapshot the resolver saw, so the
282 // authority the resolver translated its target into and the authority the
283 // request is resolved from can never diverge.
284 request, err := navigation.Resolve(snapshot.Clone())
285 if err != nil {
286 return ports.BrokerOpenStreamRequest{}, classifyInitialNavigationFailure(err)
287 }
288 stream, err := service.NextStreamID()
289 if err != nil {
290 return ports.BrokerOpenStreamRequest{}, err
291 }
292 // The gabarit deliberately leaves the connection and stream identity zero:
293 // the supervisor, not the resolver and not the intention, owns them, and the
294 // environment is never read from the broker process.
295 request.Connection = service.ConnectionID()
296 request.Stream = stream
297 return request, nil
298}
299
300// classifyInitialNavigationFailure turns a resolver or validation failure into
301// the bounded selection-unavailable notice a composition can display. A typed
302// catalogue refusal keeps its own classification; anything else is reported as
303// a bounded no-selection refusal rather than as a destination failure, because
304// no destination was ever dialed.
305func classifyInitialNavigationFailure(err error) error {
306 var catalogueErr pickerCatalogueError
307 if errors.As(err, &catalogueErr) {
308 return err
309 }
310 return pickerCatalogueError{Code: pickerCatalogueNoSelection, Text: "initial navigation could not be resolved"}
311}
312
313func (s *Supervisor) resolveCommittedStreamRequest(service ports.BrokerNavigator, key string) (ports.BrokerOpenStreamRequest, attachmentTab, error) {
314 stream, err := service.NextStreamID()
315 if err != nil {
316 return ports.BrokerOpenStreamRequest{}, attachmentTab{}, err
317 }
318 return s.cfg.Picker.ResolveCommit(key, pickerResolveBase{
319 Connection: service.ConnectionID(),
320 Stream: stream,
321 })
322}
323
324// pickerAttachmentTarget is one resolved attachment: the exact broker stream
325// request and the tab its Hello selects.
326type pickerAttachmentTarget struct {
327 request ports.BrokerOpenStreamRequest
328 tab attachmentTab
329}
330
331// newAttachmentWorker builds the real typed-session worker for one admitted
332// request. The worker receives no endpoint, route, or raw-mode authority: it
333// only drives the typed session protocol on the stream the supervisor opened.
334func (s *Supervisor) newAttachmentWorker(request ports.BrokerOpenStreamRequest, tab attachmentTab, beforeAttached func() error, sessionEnv SessionEnvironment) (AttachmentWorker, error) {
335 geometry, ok := s.attachments.latestGeometry()
336 if !ok {
337 var err error
338 geometry, err = s.cfg.Terminal.Geometry()
339 if err != nil {
340 return nil, fmt.Errorf("vev: reading terminal geometry: %w", err)
341 }
342 }
343 environment := s.cfg.AttachmentEnvironment
344 worker, err := newSessionAttachmentWorker(sessionAttachmentConfig{
345 Request: request,
346 ClientID: s.clientID,
347 Geometry: geometry,
348 TermEnv: environment.TermEnv,
349 Cwd: environment.Cwd,
350 TrueColor: environment.TrueColor,
351 PID: environment.PID,
352 Capabilities: s.capabilities,
353 SessionEnvironment: sessionEnv,
354 BeforeAttached: beforeAttached,
355 Clock: s.cfg.Clock,
356 Theme: &s.theme,
357 Clipboard: s.cfg.Clipboard,
358 Tab: tab,
359 Tabs: &s.tabs,
360 })
361 if err != nil {
362 return nil, err
363 }
364 return worker, nil
365}
366
367// takeInitialNavigation consumes the one-shot intent exactly once. Consumption
368// is recorded before the resolver runs and before any stream is opened, so a
369// refusal, a cancellation, a timeout, or a reconnection can never re-arm it: a
370// later ready cycle observes the intent already consumed. Before the first
371// committed publication this method is never reached, so a reconnection that
372// precedes readiness cannot consume the intent either.
373//
374// The resolver runs on a defensive copy of the committed publication and is
375// consulted at most once for the whole run. A resolved picker is a success
376// without an operation.
377func (s *Supervisor) takeInitialNavigation(snapshot ports.BrokerSnapshot) (InitialNavigation, bool, error) {
378 // The resolver is a translation of a CLI target into identity, so it is only
379 // consulted against a snapshot that satisfies its own contract. A malformed
380 // publication consumes the intent and reports a bounded refusal instead of
381 // handing an unvalidated authority to the resolver or the request.
382 if err := snapshot.Validate(); err != nil {
383 return InitialNavigation{}, true, fmt.Errorf("vev: initial navigation snapshot: %w", err)
384 }
385 s.mu.Lock()
386 if s.navigationConsumed {
387 s.mu.Unlock()
388 return InitialNavigation{}, false, nil
389 }
390 s.navigationConsumed = true
391 navigation := s.navigation
392 resolver := s.cfg.ResolveInitialNavigation
393 s.mu.Unlock()
394
395 if resolver != nil {
396 resolved, err := resolver(snapshot.Clone())
397 if errors.Is(err, ErrInitialNavigationNotObserved) {
398 // Not a refusal: the target's daemon has not been observed yet.
399 // The intent stays armed for a later publication.
400 s.mu.Lock()
401 s.navigationConsumed = false
402 s.mu.Unlock()
403 return InitialNavigation{}, false, err
404 }
405 if err != nil {
406 return InitialNavigation{}, true, err
407 }
408 navigation = resolved
409 }
410 if navigation.Kind == InitialNavigationPicker {
411 return InitialNavigation{}, true, nil
412 }
413 if err := navigation.Validate(); err != nil {
414 return InitialNavigation{}, true, err
415 }
416 return navigation, true, nil
417}
418
419// ErrInitialNavigationNotObserved is returned by an InitialNavigationResolver
420// whose decision depends on a daemon inventory the committed publication does
421// not carry yet (for example attach-or-create before the host's first
422// observation). The supervisor keeps the intent armed and resolves it again on
423// later publications, bounded by initialNavigationObservationBudget.
424var ErrInitialNavigationNotObserved = errors.New("vev: initial navigation target is not observed yet")
425
426// InitialNavigationNotObserved is the resolver's ErrInitialNavigationNotObserved
427// naming the configured endpoint whose fresh observation it needs; the
428// supervisor asks the broker to reconcile that endpoint once per wait.
429// Fallback, when it is not the picker, is the navigation to take if the budget
430// expires before the endpoint is observed (a remote host whose observation
431// fails can still serve a stream); the destination daemon stays the final
432// authority over it.
433type InitialNavigationNotObserved struct {
434 Endpoint string
435 Fallback InitialNavigation
436}
437
438func (e InitialNavigationNotObserved) Error() string {
439 return ErrInitialNavigationNotObserved.Error() + ": " + e.Endpoint
440}
441
442func (e InitialNavigationNotObserved) Is(target error) bool {
443 return target == ErrInitialNavigationNotObserved
444}
445
446// initialNavigationObservationBudget bounds how long an initial navigation
447// waits for its target's first observation; it stays inside the 15-second
448// handshake budget.
449const initialNavigationObservationBudget = 10 * time.Second
450
451// initialNavigationTake is one consumption attempt of the armed initial
452// navigation against the committed publication that decided it.
453type initialNavigationTake struct {
454 snapshot ports.BrokerSnapshot
455 navigation InitialNavigation
456 consumed bool
457 err error
458 terminated bool
459 termErr error
460}
461
462// awaitInitialNavigationObservation re-resolves the armed initial navigation
463// on each committed publication until the resolver can decide, the budget
464// expires (the resolver's fallback, or a bounded refusal without one), or the
465// run ends. The returned snapshot is the publication the decision was made on,
466// so the stream request resolves against the same authority.
467func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, pending InitialNavigationNotObserved) initialNavigationTake {
468 timer := s.cfg.Clock.NewTimer(initialNavigationObservationBudget)
469 defer timer.Stop()
470 for {
471 select {
472 case <-s.brokerChanged():
473 snapshot := service.Snapshot()
474 navigation, consumed, err := s.takeInitialNavigation(snapshot)
475 if errors.Is(err, ErrInitialNavigationNotObserved) {
476 _ = errors.As(err, &pending)
477 continue
478 }
479 return initialNavigationTake{snapshot: snapshot, navigation: navigation, consumed: consumed, err: err}
480 case <-timer.C():
481 s.mu.Lock()
482 s.navigationConsumed = true
483 s.mu.Unlock()
484 if pending.Fallback.Kind != InitialNavigationPicker {
485 return initialNavigationTake{snapshot: service.Snapshot(), navigation: pending.Fallback, consumed: true}
486 }
487 return initialNavigationTake{consumed: true, err: ErrInitialNavigationNotObserved}
488 case <-service.Done():
489 // The ready loop observes the loss; the intent stays armed for the
490 // next adopted connection.
491 return initialNavigationTake{}
492 case <-ctx.Done():
493 return initialNavigationTake{terminated: true, termErr: ctx.Err()}
494 case err := <-input.EOF():
495 return initialNavigationTake{terminated: true, termErr: terminalReadCause(err)}
496 }
497 }
498}
499
500// runCommittedAttachment admits one committed selection end to end: resolve,
501// open the exact stream under the one absolute deadline, run the real
502// attachment worker through the supervisor's foreground grant, and return to
503// the picker. It reports terminated=true only when the run itself must end
504// (process cancellation or terminal EOF); every attachment outcome returns to
505// the picker in the same process.
506func (s *Supervisor) runInitialNavigation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator) (bool, error) {
507 // The intent is consumed against the first committed publication of the
508 // adopted connection; the resolver and the request resolution share that one
509 // read.
510 snapshot := service.Snapshot()
511 navigation, consumed, err := s.takeInitialNavigation(snapshot)
512 if errors.Is(err, ErrInitialNavigationNotObserved) {
513 var pending InitialNavigationNotObserved
514 if errors.As(err, &pending) && pending.Endpoint != "" {
515 service.RequestReconcile(pending.Endpoint)
516 }
517 take := s.awaitInitialNavigationObservation(ctx, input, service, pending)
518 if take.terminated {
519 return true, take.termErr
520 }
521 snapshot, navigation, consumed, err = take.snapshot, take.navigation, take.consumed, take.err
522 }
523 if !consumed {
524 return false, nil
525 }
526 if err != nil {
527 s.reportAttachmentFailure(classifyInitialNavigationFailure(err))
528 return false, nil
529 }
530 if navigation.Kind == InitialNavigationPicker {
531 // Picker is a success without an operation: no stream is opened, and the
532 // intent stays consumed so a later reconnect cannot replay it.
533 return false, nil
534 }
535 request, err := s.resolveInitialStreamRequest(service, snapshot, navigation)
536 if err != nil {
537 s.reportAttachmentFailure(err)
538 return false, nil
539 }
540 // The picker was hidden behind Connecting while this navigation was
541 // pending: keys typed then must not replay as a selection or a close
542 // after the attachment returns to it.
543 if s.cfg.Picker != nil {
544 s.cfg.Picker.TakeOp()
545 }
546 s.pendingPickerKey = ""
547 return s.runResolvedAttachment(ctx, input, service, pickerAttachmentTarget{request: request}, SessionEnvironmentLocalCLI)
548}
549
550func (s *Supervisor) runCommittedAttachment(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, key string) (bool, error) {
551 if s.cfg.Picker == nil {
552 return false, nil
553 }
554 request, tab, err := s.resolveCommittedStreamRequest(service, key)
555 if err != nil {
556 // A refused selection is a typed presentation notice; it never opens a
557 // stream or becomes a session write.
558 s.reportAttachmentFailure(err)
559 return false, nil
560 }
561 return s.runResolvedAttachment(ctx, input, service, pickerAttachmentTarget{request: request, tab: tab}, SessionEnvironmentLocalPicker)
562}
563
564func (s *Supervisor) runResolvedAttachment(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance) (terminated bool, termErr error) {
565 picker := s.cfg.Picker
566 if picker == nil {
567 return false, nil
568 }
569 picker.SetOwnsInput(false)
570 input.releasePicker()
571 defer func() {
572 // A terminated run never returns to the picker: handing it input back
573 // during teardown would let a retired picker consume keys.
574 if terminated {
575 return
576 }
577 // Whoever owned the terminal since the last picker frame, the next
578 // frame must draw the whole box again.
579 s.invalidatePickerPresentation()
580 // Back on the plain picker there is no attachment to start on.
581 s.setPickerCurrent(pickerCurrent{})
582 // Publications folded into routes while attached never reached the
583 // picker catalogue.
584 picker.ApplySnapshot(service.Snapshot())
585 picker.SetOwnsInput(true)
586 input.acquirePicker()
587 }()
588 // budget.resuming is true while the loop reconnects a lost attachment:
589 // attempts then return transport-class failures as attachmentRetry instead
590 // of presenting them, so a retry stays on Connecting.
591 var budget attachmentResumeBudget
592 for {
593 attemptStart := s.cfg.Clock.Now()
594 outcome := s.attachResolved(ctx, input, service, target, localProvenance, budget.resuming)
595 switch outcome.kind {
596 case attachmentTerminated:
597 return true, outcome.cause
598 case attachmentEnded:
599 // The attempt already presented a final outcome (session gone,
600 // ended, refused).
601 return false, nil
602 case attachmentSwap:
603 // A commit from the picker overlay to another target: the previous
604 // attachment detached cleanly, and the swap goes through Connecting
605 // exactly like any committed selection.
606 budget = attachmentResumeBudget{}
607 next, ok := s.withFreshStream(service, outcome.target)
608 if !ok {
609 return false, nil
610 }
611 target, localProvenance = next, SessionEnvironmentLocalPicker
612 continue
613 case attachmentRetry:
614 // A resume attempt hit a transport-class failure before
615 // attaching (the route or destination is still down): try the
616 // same target again.
617 outcome.target = target
618 }
619 attempt, ok := budget.spend(outcome.kind, s.cfg.Clock.Now().Sub(attemptStart))
620 if !ok {
621 s.logger.Warn("client_attachment_resume_exhausted", "endpoint", outcome.target.request.Endpoint, "session", outcome.target.request.Target.SessionName, "attempts", attempt-1, "error", outcome.cause)
622 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: outcome.cause})
623 s.notifyLifecycle(LifecycleNoticeDestinationFailed)
624 s.notifyAttachment(outcome.cause)
625 return false, nil
626 }
627 if terminated, termErr, brokerLost := s.waitResume(ctx, input, service, attempt); terminated || brokerLost {
628 return terminated, termErr
629 }
630 next, ok := s.withFreshStream(service, outcome.target)
631 if !ok {
632 return false, nil
633 }
634 s.logger.Info("client_attachment_resume", "endpoint", next.request.Endpoint, "session", next.request.Target.SessionName, "attempt", attempt)
635 target = next
636 }
637}
638
639// attachmentResumeBudget bounds how often one lost attachment reconnects.
640type attachmentResumeBudget struct {
641 used int
642 resuming bool
643}
644
645// spend records one resume after an attempt of the given kind that ran for
646// up. A live attachment that stayed up for resumeStableAttachment starts a
647// fresh budget; one that drops right after attaching keeps counting, so a
648// flapping session gives up.
649// It returns the 1-based resume attempt, and false once the budget is
650// exhausted.
651func (b *attachmentResumeBudget) spend(kind attachmentOutcomeKind, up time.Duration) (int, bool) {
652 if kind == attachmentResume && up >= resumeStableAttachment {
653 b.used = 0
654 }
655 b.used++
656 b.resuming = true
657 return b.used, b.used <= maxAttachmentResumes
658}
659
660// withFreshStream gives target a newly allocated stream identity. It is
661// allocated right before the open, never earlier: streams opened meanwhile
662// (route publications) may have moved the connection's anti-replay window past
663// an identity reserved at commit. A refusal is presented.
664func (s *Supervisor) withFreshStream(service ports.BrokerNavigator, target pickerAttachmentTarget) (pickerAttachmentTarget, bool) {
665 stream, err := service.NextStreamID()
666 if err != nil {
667 s.reportAttachmentFailure(err)
668 return pickerAttachmentTarget{}, false
669 }
670 target.request.Stream = stream
671 return target, true
672}
673
674// attachResolved runs one resolved attachment attempt end to end: open the
675// exact stream under one absolute deadline, admit the worker through the
676// foreground grant, settle it, and classify the result. resuming marks an
677// attempt of the resume loop, whose transport-class failures are returned as
678// attachmentRetry instead of presented.
679func (s *Supervisor) attachResolved(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance, resuming bool) attachmentOutcome {
680 request := target.request
681 s.transition(supervisorEvent{kind: supervisorAttachBegin})
682 deadline := startAttachmentDeadline(ctx, s.cfg.Clock)
683 defer deadline.finish()
684
685 sessionEnv := s.cfg.SessionEnvironment.Clone()
686 if request.Local {
687 sessionEnv.Provenance = localProvenance
688 } else {
689 sessionEnv.Provenance = SessionEnvironmentRemote
690 sessionEnv.Env = nil
691 sessionEnv.Cwd = ""
692 }
693 if err := sessionEnv.Validate(); err != nil {
694 return s.attemptFailed(err, resuming)
695 }
696 request.Env = append([]string(nil), sessionEnv.Env...)
697
698 s.logger.Debug("client_stream_open_begin", "local", request.Local, "endpoint", request.Endpoint, "admission", request.Admission, "generation", s.State().Generation)
699 stream, err := service.OpenStream(deadline.Context(), request)
700 if err != nil {
701 s.logger.Warn("client_stream_open_failed", "local", request.Local, "endpoint", request.Endpoint, "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
702 return s.attemptFailed(deadline.cause(err), resuming)
703 }
704 if supervisorNil(stream) {
705 return s.attemptFailed(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker returned no stream"}, resuming)
706 }
707 if err := deadline.adopt(stream); err != nil {
708 // The accepted absolute deadline already elapsed before this client
709 // saw the stream. Fail the attachment instead of restarting a budget.
710 _ = stream.Close()
711 return s.attemptFailed(attachmentTimeoutError(err), resuming)
712 }
713
714 worker, err := s.newAttachmentWorker(request, target.tab, func() error {
715 deadline.disarm()
716 if deadline.expired.Load() {
717 return errAttachmentDeadline
718 }
719 return nil
720 }, sessionEnv)
721 if err != nil {
722 _ = stream.Close()
723 return s.attemptFailed(err, resuming)
724 }
725
726 token := AttachmentToken{Generation: s.State().Generation, Attempt: s.nextAttachment.Add(1)}
727 s.logger.Debug("client_stream_opened", "local", request.Local, "endpoint", request.Endpoint, "generation", token.Generation, "attempt", token.Attempt)
728 run, ok := s.attachments.Begin(deadline.Context(), token, worker, stream)
729 if !ok {
730 // Another foreground owns the terminal, or the shared reader is
731 // claimed. Begin never closes the stream, so the supervisor does.
732 _ = stream.Close()
733 if resuming {
734 // A resume must say why it stopped instead of dropping silently
735 // to the picker, so the busy error is presented, never retried.
736 s.reportAttachmentFailure(errResumeForegroundBusy)
737 return attachmentEndedOutcome()
738 }
739 s.transition(supervisorEvent{kind: supervisorAttachEnded})
740 return attachmentEndedOutcome()
741 }
742 return s.settleAttachment(ctx, input, service, deadline, run, request, resuming)
743}
744
745// settleAttachment joins one admitted attachment while watching the parent
746// context, terminal EOF, and broker-connection loss. The deadline is the only
747// other stop: when it fires before the worker settled, the attachment ends as
748// a typed timeout without claiming attachment.
749func (s *Supervisor) settleAttachment(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, deadline *attachmentDeadline, run *attachmentRun, request ports.BrokerOpenStreamRequest, resuming bool) attachmentOutcome {
750 settled := make(chan attachmentSettlement, 1)
751 go func() {
752 event, adopted := run.Wait(deadline.Context())
753 settled <- attachmentSettlement{event: event, adopted: adopted}
754 }()
755
756 overlay := &attachmentPickerOverlay{sup: s, run: run, service: service, request: request}
757 defer overlay.drop()
758
759 var result attachmentSettlement
760 var brokerLost bool
761 var terminated bool
762 var termErr error
763 notices := noticeWake{clock: s.cfg.Clock}
764 defer notices.stop()
765settlement:
766 for {
767 // Begin has admitted the foreground, which can write before MarkAttached
768 // and before the initial publication. Keep presentation invalidation
769 // buffered for the next picker state unless the picker overlay owns the
770 // terminal; consuming it otherwise could overlay session output with a
771 // second terminal writer.
772 select {
773 case result = <-settled:
774 break settlement
775 case <-s.attachments.NavigationRequests():
776 overlay.enter()
777 case <-s.brokerChanged():
778 if overlay.active {
779 overlay.applyPublication()
780 }
781 s.publishRoutes(service, run, request)
782 case <-s.attachments.RouteDemands():
783 s.publishRoutes(service, run, request)
784 case message := <-s.attachments.Navigations():
785 if !overlay.swapping {
786 s.settleDaemonNavigation(service, overlay, message)
787 }
788 case <-overlay.previewChanged():
789 overlay.publishPreview()
790 case <-overlay.ops():
791 overlay.takeOp()
792 case consumed := <-s.attachments.OverlayActions():
793 overlay.settleAction(consumed)
794 case <-overlay.invalidation():
795 overlay.resize()
796 case <-notices.arm(s.cfg.Picker):
797 s.renderCurrent()
798 case outcome := <-s.kills.results():
799 s.finishPickerKill(service, outcome)
800 case <-service.Done():
801 brokerLost = true
802 // The broker connection is gone; the supervisor cancels the run and
803 // lets the ready loop observe the loss and retire the connection.
804 run.Cancel()
805 result = <-settled
806 break settlement
807 case <-ctx.Done():
808 run.Cancel()
809 result = <-settled
810 terminated, termErr = true, ctx.Err()
811 break settlement
812 case err := <-input.EOF():
813 run.Cancel()
814 result = <-settled
815 terminated, termErr = true, terminalReadCause(err)
816 break settlement
817 }
818 }
819
820 // Parent cancellation wins over a settlement that raced it: the attachment
821 // context derives from ctx, so cancellation can settle the run before the
822 // ctx.Done case is selected.
823 if !terminated {
824 if cause := ctx.Err(); cause != nil {
825 terminated, termErr = true, cause
826 }
827 }
828
829 // The attachment settled, so its foreground (and overlay slot) is gone.
830 overlay.drop()
831 if !terminated && !brokerLost {
832 s.takeSettledNavigation(service, run.token, request)
833 }
834 // The overlay or a daemon navigation may have committed another target
835 // while this attachment was live; this attempt consumes it either way.
836 swap := s.pendingSwap
837 s.pendingSwap = nil
838 if terminated {
839 return attachmentTerminatedOutcome(termErr)
840 }
841 if swap != nil && !brokerLost {
842 // The overlay committed another target and this attachment detached
843 // for it: go straight to Connecting for the swap, never through the
844 // plain picker.
845 return attachmentOutcome{kind: attachmentSwap, target: *swap}
846 }
847 if deadline.TimedOut() && !result.adopted {
848 timeout := attachmentTimeoutError(errAttachmentDeadline)
849 if outcome, ok := s.resumeRetry(timeout, resuming); ok {
850 return outcome
851 }
852 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: timeout})
853 s.notifyAttachment(timeout)
854 return attachmentEndedOutcome()
855 }
856 if errors.Is(result.event.Err, errDetachAndExit) {
857 s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", "explicit_detach")
858 s.transition(supervisorEvent{kind: supervisorAttachEnded})
859 s.notifyLifecycle(LifecycleNoticeDetachAndExit)
860 return attachmentTerminatedOutcome(nil)
861 }
862 if errors.Is(result.event.Err, errDetachToPicker) {
863 s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", "detach_to_picker")
864 s.transition(supervisorEvent{kind: supervisorAttachEnded})
865 s.notifyLifecycle(LifecycleNoticeDetachToPicker)
866 return attachmentEndedOutcome()
867 }
868 if result.adopted && result.event.Kind == AttachmentEventLost && run.fg.Attached() {
869 // The stream carrying a live attachment was lost while the broker stayed
870 // up. The daemon parks the session for resume, so reconnect to the exact
871 // session and tab that were showing instead of dropping to the picker.
872 if resume, ok := resumeTarget(request, run.fg); ok {
873 s.logger.Warn("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", result.event.Kind.String(), "error", result.event.Err, "cause", errors.Unwrap(result.event.Err), "resume", true)
874 return attachmentOutcome{kind: attachmentResume, target: resume, cause: result.event.Err}
875 }
876 }
877 if result.adopted && (result.event.Kind == AttachmentEventLost || result.event.Kind == AttachmentEventFailed) {
878 s.logger.Warn("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", result.event.Kind.String(), "error", result.event.Err, "cause", errors.Unwrap(result.event.Err))
879 // A resume attempt whose stream was lost before attaching is still a
880 // transport failure: retry instead of presenting it.
881 if result.event.Kind == AttachmentEventLost && resuming {
882 return attachmentOutcome{kind: attachmentRetry, cause: result.event.Err}
883 }
884 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: result.event.Err})
885 s.notifyLifecycle(LifecycleNoticeDestinationFailed)
886 s.notifyAttachment(result.event.Err)
887 return attachmentEndedOutcome()
888 }
889 // An orderly end, or a run the supervisor retired for a broker loss,
890 // returns to the picker with no attachment failure reported. A stream that
891 // failed after attachment was lost/failed above, so it can never leave the
892 // attached presentation showing.
893 s.transition(supervisorEvent{kind: supervisorAttachEnded})
894 if !brokerLost {
895 s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", result.event.Kind.String())
896 s.notifyLifecycle(LifecycleNoticeSessionEnded)
897 }
898 return attachmentEndedOutcome()
899}
900
901// maxAttachmentResumes bounds how many times one lost attachment reconnects
902// before the client gives up and returns to the picker. With the supervisor
903// backoff this spans a few seconds, enough to ride out a transient drop.
904const maxAttachmentResumes = 5
905
906// resumeStableAttachment is how long a resumed attachment must stay up before
907// its next loss gets a fresh resume budget.
908const resumeStableAttachment = 30 * time.Second
909
910// resumeTarget builds the exact reattach request for a lost attachment: the
911// session and tab it last committed, over the same local or remote route. A
912// foreground that never committed a session has nothing to resume.
913func resumeTarget(request ports.BrokerOpenStreamRequest, fg *attachmentForeground) (pickerAttachmentTarget, bool) {
914 target, tab, known := fg.committedSnapshot()
915 if !known || target.Validate() != nil {
916 return pickerAttachmentTarget{}, false
917 }
918 resume := request
919 resume.Admission = ports.BrokerAdmissionExact
920 resume.Target = target
921 resume.Name = ""
922 return pickerAttachmentTarget{request: resume, tab: attachmentTab{preferred: tab}}, true
923}
924
925// waitResume presents Connecting and waits one backoff interval before the
926// next resume attempt. It settles early on cancellation or terminal EOF
927// (terminated), or on broker loss, which ends the resume and leaves the ready
928// loop to observe and retire the connection.
929func (s *Supervisor) waitResume(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, attempt int) (terminated bool, termErr error, brokerLost bool) {
930 s.transition(supervisorEvent{kind: supervisorAttachBegin})
931 delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
932 timer := s.cfg.Clock.NewTimer(delay)
933 defer stopSupervisorTimer(timer)
934 for {
935 select {
936 case <-timer.C():
937 return false, nil, false
938 case <-ctx.Done():
939 return true, ctx.Err(), false
940 case err := <-input.EOF():
941 return true, terminalReadCause(err), false
942 case <-service.Done():
943 s.transition(supervisorEvent{kind: supervisorAttachEnded})
944 return false, nil, true
945 case <-s.presentationInvalidation():
946 s.renderResizeInvalidation()
947 case <-s.spinnerTick():
948 s.cfg.Spinner.AdvanceSpinner(s.State())
949 }
950 }
951}
952
953// resumeRetry turns a transport-class failure of a resume attempt into an
954// attachmentRetry outcome, so runResolvedAttachment retries while Connecting
955// stays up. Any other failure, or a failure outside a resume, is not retried.
956func (s *Supervisor) resumeRetry(err error, resuming bool) (attachmentOutcome, bool) {
957 if !resuming || !resumableFailure(err) {
958 return attachmentOutcome{}, false
959 }
960 s.logger.Info("client_attachment_resume_failed", "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
961 return attachmentOutcome{kind: attachmentRetry, cause: err}, true
962}
963
964// attemptFailed ends one attachment attempt that failed before settlement:
965// a resume retries a transport-class failure, anything else is presented.
966func (s *Supervisor) attemptFailed(err error, resuming bool) attachmentOutcome {
967 if outcome, ok := s.resumeRetry(err, resuming); ok {
968 return outcome
969 }
970 s.reportAttachmentFailure(err)
971 return attachmentEndedOutcome()
972}
973
974// resumableFailure reports whether a resume attempt failed for a transport
975// reason worth retrying. Refusals, protocol failures, and an ambiguous outcome
976// are final.
977func resumableFailure(err error) bool {
978 var lost ports.BrokerStreamLost
979 if errors.As(err, &lost) {
980 return true
981 }
982 var typed ports.BrokerError
983 if !errors.As(err, &typed) {
984 return false
985 }
986 switch typed.Code {
987 case ports.BrokerErrorUnavailable, ports.BrokerErrorTimeout, ports.BrokerErrorAttachmentLost:
988 return true
989 default:
990 return false
991 }
992}
993
994// reportAttachmentFailure returns to the picker with a typed, visible failure.
995func (s *Supervisor) reportAttachmentFailure(err error) {
996 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: err})
997 // Classification order matters: a local refusal never dialed a destination,
998 // so it must not be reported as one, and only a genuinely unknown outcome
999 // gets its own notice.
1000 var catalogueErr pickerCatalogueError
1001 var brokerErr ports.BrokerError
1002 switch {
1003 case errors.As(err, &catalogueErr):
1004 s.notifyLifecycle(LifecycleNoticeSelectionUnavailable)
1005 case errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorOutcomeUnknown:
1006 s.notifyLifecycle(LifecycleNoticeOutcomeUnknown)
1007 default:
1008 s.notifyLifecycle(LifecycleNoticeDestinationFailed)
1009 }
1010 s.notifyAttachment(err)
1011}
1012
1013// notifyAttachment publishes the typed failure to the optional observer.
1014func (s *Supervisor) notifyAttachment(err error) {
1015 if s.cfg.Notify == nil || err == nil {
1016 return
1017 }
1018 s.cfg.Notify(s.State(), err)
1019}