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("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 Color: environment.Color,
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("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("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 if input != nil && input.pump != nil {
587 input.pump.dropUnclaimed()
588 }
589 input.acquirePicker()
590 }()
591 // budget.resuming is true while the loop reconnects a lost attachment:
592 // attempts then return transport-class failures as attachmentRetry instead
593 // of presenting them, so a retry stays on Connecting.
594 var budget attachmentResumeBudget
595 for {
596 attemptStart := s.cfg.Clock.Now()
597 outcome := s.attachResolved(ctx, input, service, target, localProvenance, budget.resuming)
598 switch outcome.kind {
599 case attachmentTerminated:
600 return true, outcome.cause
601 case attachmentEnded:
602 // The attempt already presented a final outcome (session gone,
603 // ended, refused).
604 return false, nil
605 case attachmentSwap:
606 // A commit from the picker overlay to another target: the previous
607 // attachment detached cleanly, and the swap goes through Connecting
608 // exactly like any committed selection.
609 budget = attachmentResumeBudget{}
610 // Keys kept for the previous session must not run in the next one;
611 // only a resume of the same target replays them.
612 if input != nil && input.pump != nil {
613 input.pump.dropUnclaimed()
614 }
615 next, ok := s.withFreshStream(service, outcome.target)
616 if !ok {
617 return false, nil
618 }
619 target, localProvenance = next, SessionEnvironmentLocalPicker
620 continue
621 case attachmentRetry:
622 // A resume attempt hit a transport-class failure before
623 // attaching (the route or destination is still down): try the
624 // same target again.
625 outcome.target = target
626 }
627 now := s.cfg.Clock.Now()
628 attempt, ok := budget.spend(outcome.kind, now.Sub(attemptStart), now)
629 if !ok {
630 s.logger.Warn("client_attachment_resume_exhausted", "endpoint", outcome.target.request.Endpoint, "session", outcome.target.request.Target.SessionName, "attempts", attempt-1, "error", outcome.cause)
631 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: outcome.cause})
632 s.notifyLifecycle(LifecycleNoticeDestinationFailed)
633 s.notifyAttachment(outcome.cause)
634 return false, nil
635 }
636 if terminated, termErr, stop := s.waitResume(ctx, input, service, attempt); terminated || stop {
637 return terminated, termErr
638 }
639 next, ok := s.withFreshStream(service, outcome.target)
640 if !ok {
641 return false, nil
642 }
643 s.logger.Info("client_attachment_resume", "endpoint", next.request.Endpoint, "session", next.request.Target.SessionName, "attempt", attempt)
644 target = next
645 }
646}
647
648// attachmentResumeBudget bounds how often one lost attachment reconnects.
649type attachmentResumeBudget struct {
650 used int
651 resuming bool
652 // flaps counts attachments lost before they became stable. They prove
653 // the route works but the session drops, so they are capped by count
654 // regardless of the outage window.
655 flaps int
656 // since is when the current outage began; zero before the first spend.
657 since time.Time
658}
659
660// spend records one resume after an attempt of the given kind that ran for
661// up. A live attachment that stayed up for resumeStableAttachment starts a
662// fresh budget; one that drops right after attaching keeps counting, so a
663// flapping session gives up.
664// It returns the 1-based resume attempt, and false once the budget is
665// exhausted: more than maxAttachmentResumes flaps, or more than
666// maxAttachmentResumes attempts once resumeOutageWindow has passed since the
667// outage began. A route that stays down therefore keeps resuming for the
668// whole window, while a session that keeps dropping right after it attaches
669// gives up quickly.
670func (b *attachmentResumeBudget) spend(kind attachmentOutcomeKind, up time.Duration, now time.Time) (int, bool) {
671 if kind == attachmentResume {
672 if up >= resumeStableAttachment {
673 b.used = 0
674 b.flaps = 0
675 } else {
676 b.flaps++
677 }
678 }
679 if b.used == 0 {
680 b.since = now
681 }
682 b.used++
683 b.resuming = true
684 if b.flaps > maxAttachmentResumes {
685 return b.used, false
686 }
687 return b.used, b.used <= maxAttachmentResumes || now.Sub(b.since) < resumeOutageWindow
688}
689
690// withFreshStream gives target a newly allocated stream identity. It is
691// allocated right before the open, never earlier: streams opened meanwhile
692// (route publications) may have moved the connection's anti-replay window past
693// an identity reserved at commit. A refusal is presented.
694func (s *Supervisor) withFreshStream(service ports.BrokerNavigator, target pickerAttachmentTarget) (pickerAttachmentTarget, bool) {
695 stream, err := service.NextStreamID()
696 if err != nil {
697 s.reportAttachmentFailure(err)
698 return pickerAttachmentTarget{}, false
699 }
700 target.request.Stream = stream
701 return target, true
702}
703
704// attachResolved runs one resolved attachment attempt end to end: open the
705// exact stream under one absolute deadline, admit the worker through the
706// foreground grant, settle it, and classify the result. resuming marks an
707// attempt of the resume loop, whose transport-class failures are returned as
708// attachmentRetry instead of presented.
709func (s *Supervisor) attachResolved(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance, resuming bool) attachmentOutcome {
710 request := target.request
711 s.transition(supervisorEvent{kind: supervisorAttachBegin})
712 deadline := startAttachmentDeadline(ctx, s.cfg.Clock)
713 defer deadline.finish()
714
715 sessionEnv := s.cfg.SessionEnvironment.Clone()
716 if request.Local {
717 sessionEnv.Provenance = localProvenance
718 } else {
719 sessionEnv.Provenance = SessionEnvironmentRemote
720 sessionEnv.Env = nil
721 sessionEnv.Cwd = ""
722 }
723 if err := sessionEnv.Validate(); err != nil {
724 return s.attemptFailed(err, resuming)
725 }
726 request.Env = append([]string(nil), sessionEnv.Env...)
727
728 s.logger.Debug("client_stream_open_begin", "local", request.Local, "endpoint", request.Endpoint, "admission", request.Admission, "generation", s.State().Generation)
729 openCtx := deadline.Context()
730 var keys *resumeWatch
731 var cancelOpen context.CancelFunc
732 var watchDone chan struct{}
733 if resuming {
734 keys = input.watchResume(s.cfg.Clock)
735 openCtx, cancelOpen = context.WithCancel(openCtx)
736 watchDone = make(chan struct{})
737 go func() {
738 defer close(watchDone)
739 select {
740 case <-keys.cancelled:
741 cancelOpen()
742 case <-openCtx.Done():
743 }
744 }()
745 }
746 stream, err := service.OpenStream(openCtx, request)
747 if keys != nil {
748 keys.release()
749 cancelOpen()
750 <-watchDone
751 select {
752 case <-keys.cancelled:
753 s.logger.Info("client_attachment_resume_cancelled", "phase", "stream_open")
754 if !supervisorNil(stream) {
755 _ = stream.Close()
756 }
757 s.transition(supervisorEvent{kind: supervisorAttachEnded})
758 return attachmentEndedOutcome()
759 default:
760 }
761 }
762 if err != nil {
763 s.logger.Warn("client_stream_open_failed", "local", request.Local, "endpoint", request.Endpoint, "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
764 return s.attemptFailed(deadline.cause(err), resuming)
765 }
766 if supervisorNil(stream) {
767 return s.attemptFailed(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker returned no stream"}, resuming)
768 }
769 if err := deadline.adopt(stream); err != nil {
770 // The accepted absolute deadline already elapsed before this client
771 // saw the stream. Fail the attachment instead of restarting a budget.
772 _ = stream.Close()
773 return s.attemptFailed(attachmentTimeoutError(err), resuming)
774 }
775
776 worker, err := s.newAttachmentWorker(request, target.tab, func() error {
777 deadline.disarm()
778 if deadline.expired.Load() {
779 return errAttachmentDeadline
780 }
781 return nil
782 }, sessionEnv)
783 if err != nil {
784 _ = stream.Close()
785 return s.attemptFailed(err, resuming)
786 }
787
788 token := AttachmentToken{Generation: s.State().Generation, Attempt: s.nextAttachment.Add(1)}
789 s.logger.Debug("client_stream_opened", "local", request.Local, "endpoint", request.Endpoint, "generation", token.Generation, "attempt", token.Attempt)
790 run, ok := s.attachments.Begin(deadline.Context(), token, worker, stream)
791 if !ok {
792 // Another foreground owns the terminal, or the shared reader is
793 // claimed. Begin never closes the stream, so the supervisor does.
794 _ = stream.Close()
795 if resuming {
796 // A resume must say why it stopped instead of dropping silently
797 // to the picker, so the busy error is presented, never retried.
798 s.reportAttachmentFailure(errResumeForegroundBusy)
799 return attachmentEndedOutcome()
800 }
801 s.transition(supervisorEvent{kind: supervisorAttachEnded})
802 return attachmentEndedOutcome()
803 }
804 return s.settleAttachment(ctx, input, service, deadline, run, request, resuming)
805}
806
807// settleAttachment joins one admitted attachment while watching the parent
808// context, terminal EOF, and broker-connection loss. The deadline is the only
809// other stop: when it fires before the worker settled, the attachment ends as
810// a typed timeout without claiming attachment.
811func (s *Supervisor) settleAttachment(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, deadline *attachmentDeadline, run *attachmentRun, request ports.BrokerOpenStreamRequest, resuming bool) attachmentOutcome {
812 settled := make(chan attachmentSettlement, 1)
813 go func() {
814 event, adopted := run.Wait(deadline.Context())
815 settled <- attachmentSettlement{event: event, adopted: adopted}
816 }()
817
818 overlay := &attachmentPickerOverlay{sup: s, run: run, service: service, request: request}
819 defer overlay.drop()
820
821 var result attachmentSettlement
822 var brokerLost bool
823 var terminated bool
824 var termErr error
825 notices := noticeWake{clock: s.cfg.Clock}
826 defer notices.stop()
827settlement:
828 for {
829 // Begin has admitted the foreground, which can write before MarkAttached
830 // and before the initial publication. Keep presentation invalidation
831 // buffered for the next picker state unless the picker overlay owns the
832 // terminal; consuming it otherwise could overlay session output with a
833 // second terminal writer.
834 select {
835 case result = <-settled:
836 break settlement
837 case <-s.attachments.NavigationRequests():
838 overlay.enter()
839 case <-s.brokerChanged():
840 if overlay.active {
841 overlay.applyPublication()
842 }
843 s.publishRoutes(service, run, request)
844 case <-s.attachments.RouteDemands():
845 s.publishRoutes(service, run, request)
846 case message := <-s.attachments.Navigations():
847 if !overlay.swapping {
848 s.settleDaemonNavigation(service, overlay, message)
849 }
850 case <-overlay.previewChanged():
851 overlay.publishPreview()
852 case <-overlay.ops():
853 overlay.takeOp()
854 case consumed := <-s.attachments.OverlayActions():
855 overlay.settleAction(consumed)
856 case <-overlay.invalidation():
857 overlay.resize()
858 case <-notices.arm(s.cfg.Picker):
859 s.renderCurrent()
860 case outcome := <-s.kills.results():
861 s.finishPickerKill(service, outcome)
862 case <-service.Done():
863 brokerLost = true
864 // The broker connection is gone; the supervisor cancels the run and
865 // lets the ready loop observe the loss and retire the connection.
866 run.Cancel()
867 result = <-settled
868 break settlement
869 case <-ctx.Done():
870 run.Cancel()
871 result = <-settled
872 terminated, termErr = true, ctx.Err()
873 break settlement
874 case err := <-input.EOF():
875 run.Cancel()
876 result = <-settled
877 terminated, termErr = true, terminalReadCause(err)
878 break settlement
879 }
880 }
881
882 // Parent cancellation wins over a settlement that raced it: the attachment
883 // context derives from ctx, so cancellation can settle the run before the
884 // ctx.Done case is selected.
885 if !terminated {
886 if cause := ctx.Err(); cause != nil {
887 terminated, termErr = true, cause
888 }
889 }
890
891 // The attachment settled, so its foreground (and overlay slot) is gone.
892 overlay.drop()
893 if !terminated && !brokerLost {
894 s.takeSettledNavigation(service, run.token, request)
895 }
896 // The overlay or a daemon navigation may have committed another target
897 // while this attachment was live; this attempt consumes it either way.
898 swap := s.pendingSwap
899 s.pendingSwap = nil
900 if terminated {
901 return attachmentTerminatedOutcome(termErr)
902 }
903 if swap != nil && !brokerLost {
904 // The overlay committed another target and this attachment detached
905 // for it: go straight to Connecting for the swap, never through the
906 // plain picker.
907 return attachmentOutcome{kind: attachmentSwap, target: *swap}
908 }
909 if deadline.TimedOut() && !result.adopted {
910 timeout := attachmentTimeoutError(errAttachmentDeadline)
911 if outcome, ok := s.resumeRetry(timeout, resuming); ok {
912 return outcome
913 }
914 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: timeout})
915 s.notifyAttachment(timeout)
916 return attachmentEndedOutcome()
917 }
918 if errors.Is(result.event.Err, errDetachAndExit) {
919 s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", "explicit_detach")
920 s.transition(supervisorEvent{kind: supervisorAttachEnded})
921 s.notifyLifecycle(LifecycleNoticeDetachAndExit)
922 return attachmentTerminatedOutcome(nil)
923 }
924 if errors.Is(result.event.Err, errDetachToPicker) {
925 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")
926 s.transition(supervisorEvent{kind: supervisorAttachEnded})
927 s.notifyLifecycle(LifecycleNoticeDetachToPicker)
928 return attachmentEndedOutcome()
929 }
930 if result.adopted && result.event.Kind == AttachmentEventLost && run.fg.Attached() {
931 // The stream carrying a live attachment was lost while the broker stayed
932 // up. The daemon parks the session for resume, so reconnect to the exact
933 // session and tab that were showing instead of dropping to the picker.
934 if resume, ok := resumeTarget(request, run.fg); ok {
935 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)
936 return attachmentOutcome{kind: attachmentResume, target: resume, cause: result.event.Err}
937 }
938 }
939 if result.adopted && (result.event.Kind == AttachmentEventLost || result.event.Kind == AttachmentEventFailed) {
940 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))
941 // A resume attempt whose stream was lost before attaching is still a
942 // transport failure: retry instead of presenting it.
943 if result.event.Kind == AttachmentEventLost && resuming {
944 return attachmentOutcome{kind: attachmentRetry, cause: result.event.Err}
945 }
946 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: result.event.Err})
947 s.notifyLifecycle(LifecycleNoticeDestinationFailed)
948 s.notifyAttachment(result.event.Err)
949 return attachmentEndedOutcome()
950 }
951 // An orderly end, or a run the supervisor retired for a broker loss,
952 // returns to the picker with no attachment failure reported. A stream that
953 // failed after attachment was lost/failed above, so it can never leave the
954 // attached presentation showing.
955 s.transition(supervisorEvent{kind: supervisorAttachEnded})
956 if !brokerLost {
957 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())
958 s.notifyLifecycle(LifecycleNoticeSessionEnded)
959 }
960 return attachmentEndedOutcome()
961}
962
963// maxAttachmentResumes is the minimum number of resume attempts before the
964// client may give up on a lost attachment. Giving up also requires the outage
965// to have lasted resumeOutageWindow, so a flaky mobile link keeps resuming
966// instead of dropping to the picker after a few seconds.
967const maxAttachmentResumes = 5
968
969// resumeOutageWindow is how long one outage keeps resuming. It matches the
970// daemon's park window for a detached remote attachment: past it the session
971// view is gone anyway.
972const resumeOutageWindow = 15 * time.Minute
973
974// resumeStableAttachment is how long a resumed attachment must stay up before
975// its next loss gets a fresh resume budget.
976const resumeStableAttachment = 30 * time.Second
977
978// resumeTarget builds the exact reattach request for a lost attachment: the
979// session and tab it last committed, over the same local or remote route. A
980// foreground that never committed a session has nothing to resume.
981func resumeTarget(request ports.BrokerOpenStreamRequest, fg *attachmentForeground) (pickerAttachmentTarget, bool) {
982 target, tab, known := fg.committedSnapshot()
983 if !known || target.Validate() != nil {
984 return pickerAttachmentTarget{}, false
985 }
986 resume := request
987 resume.Admission = ports.BrokerAdmissionExact
988 resume.Target = target
989 resume.Name = ""
990 return pickerAttachmentTarget{request: resume, tab: attachmentTab{preferred: tab}}, true
991}
992
993// waitResume presents Connecting and waits one backoff interval before the
994// next resume attempt. It settles early on cancellation or terminal EOF
995// (terminated), or with stop on broker loss, which leaves the ready loop to
996// observe and retire the connection, or when the user cancels the resume
997// with Ctrl-C or Esc, which returns to the picker. Other keys typed during
998// the outage are kept and replayed into the session once it resumes.
999func (s *Supervisor) waitResume(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, attempt int) (terminated bool, termErr error, stop bool) {
1000 s.transition(supervisorEvent{kind: supervisorAttachBegin})
1001 delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
1002 timer := s.cfg.Clock.NewTimer(delay)
1003 defer stopSupervisorTimer(timer)
1004 keys := input.watchResume(s.cfg.Clock)
1005 defer keys.release()
1006 for {
1007 select {
1008 case <-keys.cancelled:
1009 // The user gave up on the outage: back to the picker, quietly.
1010 s.logger.Info("client_attachment_resume_cancelled", "attempt", attempt)
1011 s.transition(supervisorEvent{kind: supervisorAttachEnded})
1012 return false, nil, true
1013 case <-timer.C():
1014 // Release resolves an Esc typed just before the backoff ended,
1015 // which may still cancel.
1016 keys.release()
1017 select {
1018 case <-keys.cancelled:
1019 s.logger.Info("client_attachment_resume_cancelled", "attempt", attempt)
1020 s.transition(supervisorEvent{kind: supervisorAttachEnded})
1021 return false, nil, true
1022 default:
1023 }
1024 return false, nil, false
1025 case <-ctx.Done():
1026 return true, ctx.Err(), false
1027 case err := <-input.EOF():
1028 return true, terminalReadCause(err), false
1029 case <-service.Done():
1030 s.transition(supervisorEvent{kind: supervisorAttachEnded})
1031 return false, nil, true
1032 case <-s.presentationInvalidation():
1033 s.renderResizeInvalidation()
1034 case <-s.spinnerTick():
1035 s.cfg.Spinner.AdvanceSpinner(s.State())
1036 }
1037 }
1038}
1039
1040// resumeRetry turns a transport-class failure of a resume attempt into an
1041// attachmentRetry outcome, so runResolvedAttachment retries while Connecting
1042// stays up. Any other failure, or a failure outside a resume, is not retried.
1043func (s *Supervisor) resumeRetry(err error, resuming bool) (attachmentOutcome, bool) {
1044 if !resuming || !resumableFailure(err) {
1045 return attachmentOutcome{}, false
1046 }
1047 s.logger.Info("client_attachment_resume_failed", "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
1048 return attachmentOutcome{kind: attachmentRetry, cause: err}, true
1049}
1050
1051// attemptFailed ends one attachment attempt that failed before settlement:
1052// a resume retries a transport-class failure, anything else is presented.
1053func (s *Supervisor) attemptFailed(err error, resuming bool) attachmentOutcome {
1054 if outcome, ok := s.resumeRetry(err, resuming); ok {
1055 return outcome
1056 }
1057 s.reportAttachmentFailure(err)
1058 return attachmentEndedOutcome()
1059}
1060
1061// resumableFailure reports whether a resume attempt failed for a transport
1062// reason worth retrying. Refusals, protocol failures, and an ambiguous outcome
1063// are final.
1064func resumableFailure(err error) bool {
1065 var lost ports.BrokerStreamLost
1066 if errors.As(err, &lost) {
1067 return true
1068 }
1069 var typed ports.BrokerError
1070 if !errors.As(err, &typed) {
1071 return false
1072 }
1073 switch typed.Code {
1074 case ports.BrokerErrorUnavailable, ports.BrokerErrorTimeout, ports.BrokerErrorAttachmentLost:
1075 return true
1076 default:
1077 return false
1078 }
1079}
1080
1081// reportAttachmentFailure returns to the picker with a typed, visible failure.
1082func (s *Supervisor) reportAttachmentFailure(err error) {
1083 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: err})
1084 // Classification order matters: a local refusal never dialed a destination,
1085 // so it must not be reported as one, and only a genuinely unknown outcome
1086 // gets its own notice.
1087 var catalogueErr pickerCatalogueError
1088 var brokerErr ports.BrokerError
1089 switch {
1090 case errors.As(err, &catalogueErr):
1091 s.notifyLifecycle(LifecycleNoticeSelectionUnavailable)
1092 case errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorOutcomeUnknown:
1093 s.notifyLifecycle(LifecycleNoticeOutcomeUnknown)
1094 default:
1095 s.notifyLifecycle(LifecycleNoticeDestinationFailed)
1096 }
1097 s.notifyAttachment(err)
1098}
1099
1100// notifyAttachment publishes the typed failure to the optional observer.
1101func (s *Supervisor) notifyAttachment(err error) {
1102 if s.cfg.Notify == nil || err == nil {
1103 return
1104 }
1105 s.cfg.Notify(s.State(), err)
1106}