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// pendingInPlace is one in-place switch the supervisor is waiting on: its
332// choice number and the target to reconnect to if the daemon refuses.
333type pendingInPlace struct {
334 seq uint64
335 target pickerAttachmentTarget
336}
337
338// newAttachmentWorker builds the real typed-session worker for one admitted
339// request. The worker receives no endpoint, route, or raw-mode authority: it
340// only drives the typed session protocol on the stream the supervisor opened.
341func (s *Supervisor) newAttachmentWorker(request ports.BrokerOpenStreamRequest, tab attachmentTab, beforeAttached func() error, sessionEnv SessionEnvironment) (AttachmentWorker, error) {
342 geometry, ok := s.attachments.latestGeometry()
343 if !ok {
344 var err error
345 geometry, err = s.cfg.Terminal.Geometry()
346 if err != nil {
347 return nil, fmt.Errorf("reading terminal geometry: %w", err)
348 }
349 }
350 environment := s.cfg.AttachmentEnvironment
351 worker, err := newSessionAttachmentWorker(sessionAttachmentConfig{
352 Request: request,
353 ClientID: s.clientID,
354 Geometry: geometry,
355 TermEnv: environment.TermEnv,
356 Cwd: environment.Cwd,
357 Color: environment.Color,
358 PID: environment.PID,
359 Capabilities: s.capabilities,
360 SessionEnvironment: sessionEnv,
361 BeforeAttached: beforeAttached,
362 Clock: s.cfg.Clock,
363 Theme: &s.theme,
364 Clipboard: s.cfg.Clipboard,
365 Tab: tab,
366 Tabs: &s.tabs,
367 EchoPredict: environment.EchoPredict,
368 })
369 if err != nil {
370 return nil, err
371 }
372 return worker, nil
373}
374
375// takeInitialNavigation consumes the one-shot intent exactly once. Consumption
376// is recorded before the resolver runs and before any stream is opened, so a
377// refusal, a cancellation, a timeout, or a reconnection can never re-arm it: a
378// later ready cycle observes the intent already consumed. Before the first
379// committed publication this method is never reached, so a reconnection that
380// precedes readiness cannot consume the intent either.
381//
382// The resolver runs on a defensive copy of the committed publication and is
383// consulted at most once for the whole run. A resolved picker is a success
384// without an operation.
385func (s *Supervisor) takeInitialNavigation(snapshot ports.BrokerSnapshot) (InitialNavigation, bool, error) {
386 // The resolver is a translation of a CLI target into identity, so it is only
387 // consulted against a snapshot that satisfies its own contract. A malformed
388 // publication consumes the intent and reports a bounded refusal instead of
389 // handing an unvalidated authority to the resolver or the request.
390 if err := snapshot.Validate(); err != nil {
391 return InitialNavigation{}, true, fmt.Errorf("initial navigation snapshot: %w", err)
392 }
393 s.mu.Lock()
394 if s.navigationConsumed {
395 s.mu.Unlock()
396 return InitialNavigation{}, false, nil
397 }
398 s.navigationConsumed = true
399 navigation := s.navigation
400 resolver := s.cfg.ResolveInitialNavigation
401 s.mu.Unlock()
402
403 if resolver != nil {
404 resolved, err := resolver(snapshot.Clone())
405 if errors.Is(err, ErrInitialNavigationNotObserved) {
406 // Not a refusal: the target's daemon has not been observed yet.
407 // The intent stays armed for a later publication.
408 s.mu.Lock()
409 s.navigationConsumed = false
410 s.mu.Unlock()
411 return InitialNavigation{}, false, err
412 }
413 if err != nil {
414 return InitialNavigation{}, true, err
415 }
416 navigation = resolved
417 }
418 if navigation.Kind == InitialNavigationPicker {
419 return InitialNavigation{}, true, nil
420 }
421 if err := navigation.Validate(); err != nil {
422 return InitialNavigation{}, true, err
423 }
424 return navigation, true, nil
425}
426
427// ErrInitialNavigationNotObserved is returned by an InitialNavigationResolver
428// whose decision depends on a daemon inventory the committed publication does
429// not carry yet (for example attach-or-create before the host's first
430// observation). The supervisor keeps the intent armed and resolves it again on
431// later publications, bounded by initialNavigationObservationBudget.
432var ErrInitialNavigationNotObserved = errors.New("initial navigation target is not observed yet")
433
434// InitialNavigationNotObserved is the resolver's ErrInitialNavigationNotObserved
435// naming the configured endpoint whose fresh observation it needs; the
436// supervisor asks the broker to reconcile that endpoint once per wait.
437// Fallback, when it is not the picker, is the navigation to take if the budget
438// expires before the endpoint is observed (a remote host whose observation
439// fails can still serve a stream); the destination daemon stays the final
440// authority over it.
441type InitialNavigationNotObserved struct {
442 Endpoint string
443 Fallback InitialNavigation
444}
445
446func (e InitialNavigationNotObserved) Error() string {
447 return ErrInitialNavigationNotObserved.Error() + ": " + e.Endpoint
448}
449
450func (e InitialNavigationNotObserved) Is(target error) bool {
451 return target == ErrInitialNavigationNotObserved
452}
453
454// initialNavigationObservationBudget bounds how long an initial navigation
455// waits for its target's first observation; it stays inside the 15-second
456// handshake budget.
457const initialNavigationObservationBudget = 10 * time.Second
458
459// initialNavigationTake is one consumption attempt of the armed initial
460// navigation against the committed publication that decided it.
461type initialNavigationTake struct {
462 snapshot ports.BrokerSnapshot
463 navigation InitialNavigation
464 consumed bool
465 err error
466 terminated bool
467 termErr error
468}
469
470// awaitInitialNavigationObservation re-resolves the armed initial navigation
471// on each committed publication until the resolver can decide, the budget
472// expires (the resolver's fallback, or a bounded refusal without one), or the
473// run ends. The returned snapshot is the publication the decision was made on,
474// so the stream request resolves against the same authority.
475func (s *Supervisor) awaitInitialNavigationObservation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, pending InitialNavigationNotObserved) initialNavigationTake {
476 timer := s.cfg.Clock.NewTimer(initialNavigationObservationBudget)
477 defer timer.Stop()
478 for {
479 select {
480 case <-s.nav.brokerChanged():
481 snapshot := service.Snapshot()
482 navigation, consumed, err := s.takeInitialNavigation(snapshot)
483 if errors.Is(err, ErrInitialNavigationNotObserved) {
484 _ = errors.As(err, &pending)
485 continue
486 }
487 return initialNavigationTake{snapshot: snapshot, navigation: navigation, consumed: consumed, err: err}
488 case <-timer.C():
489 s.mu.Lock()
490 s.navigationConsumed = true
491 s.mu.Unlock()
492 if pending.Fallback.Kind != InitialNavigationPicker {
493 return initialNavigationTake{snapshot: service.Snapshot(), navigation: pending.Fallback, consumed: true}
494 }
495 return initialNavigationTake{consumed: true, err: ErrInitialNavigationNotObserved}
496 case <-service.Done():
497 // The ready loop observes the loss; the intent stays armed for the
498 // next adopted connection.
499 return initialNavigationTake{}
500 case <-ctx.Done():
501 return initialNavigationTake{terminated: true, termErr: ctx.Err()}
502 case err := <-input.EOF():
503 return initialNavigationTake{terminated: true, termErr: terminalReadCause(err)}
504 }
505 }
506}
507
508// runCommittedAttachment admits one committed selection end to end: resolve,
509// open the exact stream under the one absolute deadline, run the real
510// attachment worker through the supervisor's foreground grant, and return to
511// the picker. It reports terminated=true only when the run itself must end
512// (process cancellation or terminal EOF); every attachment outcome returns to
513// the picker in the same process.
514func (s *Supervisor) runInitialNavigation(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator) (bool, error) {
515 // The intent is consumed against the first committed publication of the
516 // adopted connection; the resolver and the request resolution share that one
517 // read.
518 snapshot := service.Snapshot()
519 navigation, consumed, err := s.takeInitialNavigation(snapshot)
520 if errors.Is(err, ErrInitialNavigationNotObserved) {
521 var pending InitialNavigationNotObserved
522 if errors.As(err, &pending) && pending.Endpoint != "" {
523 service.RequestReconcile(pending.Endpoint)
524 }
525 take := s.awaitInitialNavigationObservation(ctx, input, service, pending)
526 if take.terminated {
527 return true, take.termErr
528 }
529 snapshot, navigation, consumed, err = take.snapshot, take.navigation, take.consumed, take.err
530 }
531 if !consumed {
532 return false, nil
533 }
534 if err != nil {
535 s.reportAttachmentFailure(classifyInitialNavigationFailure(err))
536 return false, nil
537 }
538 if navigation.Kind == InitialNavigationPicker {
539 // Picker is a success without an operation: no stream is opened, and the
540 // intent stays consumed so a later reconnect cannot replay it.
541 return false, nil
542 }
543 request, err := s.resolveInitialStreamRequest(service, snapshot, navigation)
544 if err != nil {
545 s.reportAttachmentFailure(err)
546 return false, nil
547 }
548 // The picker was hidden behind Connecting while this navigation was
549 // pending: keys typed then must not replay as a selection or a close
550 // after the attachment returns to it.
551 if s.cfg.Picker != nil {
552 s.cfg.Picker.TakeOp()
553 }
554 s.picker.pendingPickerKey = ""
555 return s.runResolvedAttachment(ctx, input, service, pickerAttachmentTarget{request: request}, SessionEnvironmentLocalCLI)
556}
557
558func (s *Supervisor) runCommittedAttachment(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, key string) (bool, error) {
559 if s.cfg.Picker == nil {
560 return false, nil
561 }
562 request, tab, err := s.resolveCommittedStreamRequest(service, key)
563 if err != nil {
564 // A refused selection is a typed presentation notice; it never opens a
565 // stream or becomes a session write.
566 s.reportAttachmentFailure(err)
567 return false, nil
568 }
569 return s.runResolvedAttachment(ctx, input, service, pickerAttachmentTarget{request: request, tab: tab}, SessionEnvironmentLocalPicker)
570}
571
572func (s *Supervisor) runResolvedAttachment(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance) (terminated bool, termErr error) {
573 picker := s.cfg.Picker
574 if picker == nil {
575 return false, nil
576 }
577 picker.SetOwnsInput(false)
578 input.releasePicker()
579 defer func() {
580 // A terminated run never returns to the picker: handing it input back
581 // during teardown would let a retired picker consume keys.
582 if terminated {
583 return
584 }
585 // Whoever owned the terminal since the last picker frame, the next
586 // frame must draw the whole box again.
587 s.invalidatePickerPresentation()
588 // Back on the plain picker there is no attachment to start on.
589 s.setPickerCurrent(pickerCurrent{})
590 // Publications folded into routes while attached never reached the
591 // picker catalogue.
592 picker.ApplySnapshot(service.Snapshot())
593 picker.SetOwnsInput(true)
594 if input != nil && input.pump != nil {
595 input.pump.dropUnclaimed()
596 }
597 input.acquirePicker()
598 }()
599 // budget.resuming is true while the loop reconnects a lost attachment:
600 // attempts then return transport-class failures as attachmentRetry instead
601 // of presenting them, so a retry stays on Connecting.
602 var budget attachmentResumeBudget
603 for {
604 attemptStart := s.cfg.Clock.Now()
605 outcome := s.attachResolved(ctx, input, service, target, localProvenance, budget.resuming)
606 switch outcome.kind {
607 case attachmentTerminated:
608 return true, outcome.cause
609 case attachmentEnded:
610 // The attempt already presented a final outcome (session gone,
611 // ended, refused).
612 return false, nil
613 case attachmentSwap:
614 // A commit from the picker overlay to another target: the previous
615 // attachment detached cleanly, and the swap goes through Connecting
616 // exactly like any committed selection.
617 budget = attachmentResumeBudget{}
618 // Keys kept for the previous session must not run in the next one;
619 // only a resume of the same target replays them.
620 if input != nil && input.pump != nil {
621 input.pump.dropUnclaimed()
622 }
623 next, ok := s.withFreshStream(service, outcome.target)
624 if !ok {
625 return false, nil
626 }
627 target, localProvenance = next, SessionEnvironmentLocalPicker
628 continue
629 case attachmentRetry:
630 // A resume attempt hit a transport-class failure before
631 // attaching (the route or destination is still down): try the
632 // same target again.
633 outcome.target = target
634 }
635 now := s.cfg.Clock.Now()
636 attempt, ok := budget.spend(outcome.kind, now.Sub(attemptStart), now)
637 if !ok {
638 s.logger.Warn("client_attachment_resume_exhausted", "endpoint", outcome.target.request.Endpoint, "session", outcome.target.request.Target.SessionName, "attempts", attempt-1, "error", outcome.cause)
639 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: outcome.cause})
640 s.notifyLifecycle(LifecycleNoticeDestinationFailed)
641 s.notifyAttachment(outcome.cause)
642 return false, nil
643 }
644 if terminated, termErr, stop := s.waitResume(ctx, input, service, attempt); terminated || stop {
645 return terminated, termErr
646 }
647 next, ok := s.withFreshStream(service, outcome.target)
648 if !ok {
649 return false, nil
650 }
651 s.logger.Info("client_attachment_resume", "endpoint", next.request.Endpoint, "session", next.request.Target.SessionName, "attempt", attempt)
652 target = next
653 }
654}
655
656// attachmentResumeBudget bounds how often one lost attachment reconnects.
657type attachmentResumeBudget struct {
658 used int
659 resuming bool
660 // flaps counts attachments lost before they became stable. They prove
661 // the route works but the session drops, so they are capped by count
662 // regardless of the outage window.
663 flaps int
664 // since is when the current outage began; zero before the first spend.
665 since time.Time
666}
667
668// spend records one resume after an attempt of the given kind that ran for
669// up. A live attachment that stayed up for resumeStableAttachment starts a
670// fresh budget; one that drops right after attaching keeps counting, so a
671// flapping session gives up.
672// It returns the 1-based resume attempt, and false once the budget is
673// exhausted: more than maxAttachmentResumes flaps, or more than
674// maxAttachmentResumes attempts once resumeOutageWindow has passed since the
675// outage began. A route that stays down therefore keeps resuming for the
676// whole window, while a session that keeps dropping right after it attaches
677// gives up quickly.
678func (b *attachmentResumeBudget) spend(kind attachmentOutcomeKind, up time.Duration, now time.Time) (int, bool) {
679 if kind == attachmentResume {
680 if up >= resumeStableAttachment {
681 b.used = 0
682 b.flaps = 0
683 } else {
684 b.flaps++
685 }
686 }
687 if b.used == 0 {
688 b.since = now
689 }
690 b.used++
691 b.resuming = true
692 if b.flaps > maxAttachmentResumes {
693 return b.used, false
694 }
695 return b.used, b.used <= maxAttachmentResumes || now.Sub(b.since) < resumeOutageWindow
696}
697
698// withFreshStream gives target a newly allocated stream identity. It is
699// allocated right before the open, never earlier: streams opened meanwhile
700// (route publications) may have moved the connection's anti-replay window past
701// an identity reserved at commit. A refusal is presented.
702func (s *Supervisor) withFreshStream(service ports.BrokerNavigator, target pickerAttachmentTarget) (pickerAttachmentTarget, bool) {
703 stream, err := service.NextStreamID()
704 if err != nil {
705 s.reportAttachmentFailure(err)
706 return pickerAttachmentTarget{}, false
707 }
708 target.request.Stream = stream
709 return target, true
710}
711
712// attachResolved runs one resolved attachment attempt end to end: open the
713// exact stream under one absolute deadline, admit the worker through the
714// foreground grant, settle it, and classify the result. resuming marks an
715// attempt of the resume loop, whose transport-class failures are returned as
716// attachmentRetry instead of presented.
717func (s *Supervisor) attachResolved(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, target pickerAttachmentTarget, localProvenance SessionEnvironmentProvenance, resuming bool) attachmentOutcome {
718 request := target.request
719 // A live local session usually commits within the notice delay; a
720 // remote handshake, a resume, or a restore may not, and nothing can paint
721 // once the foreground owns the terminal, so their notice shows at once.
722 slow := resuming || !request.Local || target.tab.stopped != nil ||
723 (request.Admission == ports.BrokerAdmissionExact && !catalogueSessionLive(service.Snapshot(), request))
724 s.transition(supervisorEvent{kind: supervisorAttachBegin, slow: slow})
725 deadline := startAttachmentDeadline(ctx, s.cfg.Clock)
726 defer deadline.finish()
727
728 sessionEnv := s.cfg.SessionEnvironment.Clone()
729 if request.Local {
730 sessionEnv.Provenance = localProvenance
731 } else {
732 sessionEnv.Provenance = SessionEnvironmentRemote
733 sessionEnv.Env = nil
734 sessionEnv.Cwd = ""
735 }
736 if err := sessionEnv.Validate(); err != nil {
737 return s.attemptFailed(err, resuming)
738 }
739 request.Env = append([]string(nil), sessionEnv.Env...)
740
741 s.logger.Debug("client_stream_open_begin", "local", request.Local, "endpoint", request.Endpoint, "admission", request.Admission, "generation", s.State().Generation)
742 openCtx := deadline.Context()
743 var keys *resumeWatch
744 var cancelOpen context.CancelFunc
745 var watchDone chan struct{}
746 if resuming {
747 keys = input.watchResume(s.cfg.Clock, s.logger)
748 openCtx, cancelOpen = context.WithCancel(openCtx)
749 watchDone = make(chan struct{})
750 go func() {
751 defer close(watchDone)
752 select {
753 case <-keys.cancelled:
754 cancelOpen()
755 case <-openCtx.Done():
756 }
757 }()
758 }
759 stream, err := s.openStreamTicking(openCtx, service, request)
760 if keys != nil {
761 keys.release()
762 cancelOpen()
763 <-watchDone
764 select {
765 case <-keys.cancelled:
766 s.logger.Info("client_attachment_resume_cancelled", "phase", "stream_open")
767 if !supervisorNil(stream) {
768 _ = stream.Close()
769 }
770 s.transition(supervisorEvent{kind: supervisorAttachEnded})
771 return attachmentEndedOutcome()
772 default:
773 }
774 }
775 if err != nil {
776 s.logger.Warn("client_stream_open_failed", "local", request.Local, "endpoint", request.Endpoint, "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
777 return s.attemptFailed(deadline.cause(err), resuming)
778 }
779 if supervisorNil(stream) {
780 return s.attemptFailed(ports.BrokerError{Code: ports.BrokerErrorUnavailable, Text: "broker returned no stream"}, resuming)
781 }
782 if err := deadline.adopt(stream); err != nil {
783 // The accepted absolute deadline already elapsed before this client
784 // saw the stream. Fail the attachment instead of restarting a budget.
785 _ = stream.Close()
786 return s.attemptFailed(attachmentTimeoutError(err), resuming)
787 }
788
789 worker, err := s.newAttachmentWorker(request, target.tab, func() error {
790 deadline.disarm()
791 if deadline.expired.Load() {
792 return errAttachmentDeadline
793 }
794 return nil
795 }, sessionEnv)
796 if err != nil {
797 _ = stream.Close()
798 return s.attemptFailed(err, resuming)
799 }
800
801 token := AttachmentToken{Generation: s.State().Generation, Attempt: s.nextAttachment.Add(1)}
802 s.logger.Debug("client_stream_opened", "local", request.Local, "endpoint", request.Endpoint, "generation", token.Generation, "attempt", token.Attempt)
803 run, ok := s.attachments.Begin(deadline.Context(), token, worker, stream)
804 if !ok {
805 // Another foreground owns the terminal, or the shared reader is
806 // claimed. Begin never closes the stream, so the supervisor does.
807 _ = stream.Close()
808 if resuming {
809 // A resume must say why it stopped instead of dropping silently
810 // to the picker, so the busy error is presented, never retried.
811 s.reportAttachmentFailure(errResumeForegroundBusy)
812 return attachmentEndedOutcome()
813 }
814 s.transition(supervisorEvent{kind: supervisorAttachEnded})
815 return attachmentEndedOutcome()
816 }
817 if input != nil {
818 // The attachment owns input again: keys typed from now on reach the
819 // session, so the outage's overflow ends here, not at stream open.
820 input.resumeOverflowed.Store(false)
821 }
822 return s.settleAttachment(ctx, input, service, deadline, run, request, resuming)
823}
824
825// openStreamTicking opens one stream while advancing the Connecting spinner,
826// so a slow open still shows its delayed notice. No foreground owns the
827// terminal before Begin, so painting here never overwrites session output.
828// OpenStream honors ctx, so the wait always ends.
829func (s *Supervisor) openStreamTicking(ctx context.Context, service ports.BrokerNavigator, request ports.BrokerOpenStreamRequest) (ports.BrokerLogicalConnection, error) {
830 type opened struct {
831 stream ports.BrokerLogicalConnection
832 err error
833 }
834 result := make(chan opened, 1)
835 go func() {
836 stream, err := service.OpenStream(ctx, request)
837 result <- opened{stream: stream, err: err}
838 }()
839 for {
840 select {
841 case r := <-result:
842 return r.stream, r.err
843 case <-s.spinnerTick():
844 s.cfg.Spinner.AdvanceSpinner(s.State())
845 }
846 }
847}
848
849// settleAttachment joins one admitted attachment while watching the parent
850// context, terminal EOF, and broker-connection loss. The deadline is the only
851// other stop: when it fires before the worker settled, the attachment ends as
852// a typed timeout without claiming attachment.
853func (s *Supervisor) settleAttachment(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, deadline *attachmentDeadline, run *attachmentRun, request ports.BrokerOpenStreamRequest, resuming bool) attachmentOutcome {
854 settled := make(chan attachmentSettlement, 1)
855 go func() {
856 event, adopted := run.Wait(deadline.Context())
857 settled <- attachmentSettlement{event: event, adopted: adopted}
858 }()
859
860 overlay := &attachmentPickerOverlay{sup: s, run: run, service: service, request: request}
861 defer overlay.drop()
862
863 var result attachmentSettlement
864 var brokerLost bool
865 var terminated bool
866 var termErr error
867 notices := noticeWake{clock: s.cfg.Clock}
868 defer notices.stop()
869settlement:
870 for {
871 // Begin has admitted the foreground, which can write before MarkAttached
872 // and before the initial publication. Keep presentation invalidation
873 // buffered for the next picker state unless the picker overlay owns the
874 // terminal; consuming it otherwise could overlay session output with a
875 // second terminal writer.
876 select {
877 case result = <-settled:
878 break settlement
879 case <-s.attachments.NavigationRequests():
880 overlay.enter()
881 case <-s.nav.brokerChanged():
882 if overlay.active {
883 overlay.applyPublication()
884 }
885 s.publishRoutes(service, run, request)
886 case <-s.attachments.RouteDemands():
887 s.publishRoutes(service, run, request)
888 case message := <-s.attachments.Navigations():
889 if !overlay.swapping {
890 s.settleDaemonNavigation(service, overlay, message)
891 }
892 case <-overlay.previewChanged():
893 overlay.publishPreview()
894 case <-overlay.ops():
895 overlay.takeOp()
896 case consumed := <-s.attachments.OverlayActions():
897 overlay.settleAction(consumed)
898 case result := <-s.attachments.InPlaceResults():
899 overlay.inPlaceSettled(result)
900 case <-overlay.invalidation():
901 overlay.resize()
902 case <-notices.arm(s.cfg.Picker):
903 s.renderCurrent()
904 case outcome := <-s.picker.kills.results():
905 s.finishPickerKill(service, outcome)
906 case <-service.Done():
907 brokerLost = true
908 // The broker connection is gone; the supervisor cancels the run and
909 // lets the ready loop observe the loss and retire the connection.
910 run.Cancel()
911 result = <-settled
912 break settlement
913 case <-ctx.Done():
914 run.Cancel()
915 result = <-settled
916 terminated, termErr = true, ctx.Err()
917 break settlement
918 case err := <-input.EOF():
919 run.Cancel()
920 result = <-settled
921 terminated, termErr = true, terminalReadCause(err)
922 break settlement
923 }
924 }
925
926 // Parent cancellation wins over a settlement that raced it: the attachment
927 // context derives from ctx, so cancellation can settle the run before the
928 // ctx.Done case is selected.
929 if !terminated {
930 if cause := ctx.Err(); cause != nil {
931 terminated, termErr = true, cause
932 }
933 }
934
935 // The attachment settled, so its foreground (and overlay slot) is gone.
936 overlay.drop()
937 if !terminated && !brokerLost {
938 s.takeSettledNavigation(service, run.token, request)
939 }
940 // The overlay or a daemon navigation may have committed another target
941 // while this attachment was live; this attempt consumes it either way.
942 swap := s.nav.takeSwap()
943 // An in-place switch still in flight ended with its attachment; the
944 // resume or picker path below decides what comes next.
945 s.nav.clearInPlace()
946 if terminated {
947 return attachmentTerminatedOutcome(termErr)
948 }
949 if swap != nil && !brokerLost {
950 // The overlay committed another target and this attachment detached
951 // for it: go straight to Connecting for the swap, never through the
952 // plain picker.
953 return attachmentOutcome{kind: attachmentSwap, target: *swap}
954 }
955 if deadline.TimedOut() && !result.adopted {
956 timeout := attachmentTimeoutError(errAttachmentDeadline)
957 if outcome, ok := s.resumeRetry(timeout, resuming); ok {
958 return outcome
959 }
960 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: timeout})
961 s.notifyAttachment(timeout)
962 return attachmentEndedOutcome()
963 }
964 if errors.Is(result.event.Err, errDetachAndExit) {
965 s.logger.Info("client_stream_closed", "local", request.Local, "endpoint", request.Endpoint, "lifecycle", request.Target.LifecycleID.String(), "session", request.Target.SessionName, "reason", "explicit_detach")
966 s.transition(supervisorEvent{kind: supervisorAttachEnded})
967 s.notifyLifecycle(LifecycleNoticeDetachAndExit)
968 return attachmentTerminatedOutcome(nil)
969 }
970 if errors.Is(result.event.Err, errDetachToPicker) {
971 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")
972 s.transition(supervisorEvent{kind: supervisorAttachEnded})
973 s.notifyLifecycle(LifecycleNoticeDetachToPicker)
974 return attachmentEndedOutcome()
975 }
976 if result.adopted && result.event.Kind == AttachmentEventLost && run.fg.Attached() {
977 // The stream carrying a live attachment was lost while the broker stayed
978 // up. The daemon parks the session for resume, so reconnect to the exact
979 // session and tab that were showing instead of dropping to the picker.
980 if resume, ok := resumeTarget(request, run.fg); ok {
981 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)
982 return attachmentOutcome{kind: attachmentResume, target: resume, cause: result.event.Err}
983 }
984 }
985 if result.adopted && (result.event.Kind == AttachmentEventLost || result.event.Kind == AttachmentEventFailed) {
986 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))
987 // A resume attempt whose stream was lost before attaching is still a
988 // transport failure: retry instead of presenting it.
989 if result.event.Kind == AttachmentEventLost && resuming {
990 return attachmentOutcome{kind: attachmentRetry, cause: result.event.Err}
991 }
992 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: result.event.Err})
993 s.notifyLifecycle(LifecycleNoticeDestinationFailed)
994 s.notifyAttachment(result.event.Err)
995 return attachmentEndedOutcome()
996 }
997 // An orderly end, or a run the supervisor retired for a broker loss,
998 // returns to the picker with no attachment failure reported. A stream that
999 // failed after attachment was lost/failed above, so it can never leave the
1000 // attached presentation showing.
1001 s.transition(supervisorEvent{kind: supervisorAttachEnded})
1002 if !brokerLost {
1003 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())
1004 s.notifyLifecycle(LifecycleNoticeSessionEnded)
1005 }
1006 return attachmentEndedOutcome()
1007}
1008
1009// maxAttachmentResumes is the minimum number of resume attempts before the
1010// client may give up on a lost attachment. Giving up also requires the outage
1011// to have lasted resumeOutageWindow, so a flaky mobile link keeps resuming
1012// instead of dropping to the picker after a few seconds.
1013const maxAttachmentResumes = 5
1014
1015// resumeOutageWindow is how long one outage keeps resuming. It matches the
1016// daemon's park window for a detached remote attachment: past it the session
1017// view is gone anyway.
1018const resumeOutageWindow = 15 * time.Minute
1019
1020// resumeStableAttachment is how long a resumed attachment must stay up before
1021// its next loss gets a fresh resume budget.
1022const resumeStableAttachment = 30 * time.Second
1023
1024// resumeTarget builds the exact reattach request for a lost attachment: the
1025// session and tab it last committed, over the same local or remote route. A
1026// foreground that never committed a session has nothing to resume.
1027func resumeTarget(request ports.BrokerOpenStreamRequest, fg *attachmentForeground) (pickerAttachmentTarget, bool) {
1028 target, tab, known := fg.committedSnapshot()
1029 if !known || target.Validate() != nil {
1030 return pickerAttachmentTarget{}, false
1031 }
1032 resume := request
1033 resume.Admission = ports.BrokerAdmissionExact
1034 resume.Target = target
1035 resume.Name = ""
1036 return pickerAttachmentTarget{request: resume, tab: attachmentTab{preferred: tab}}, true
1037}
1038
1039// waitResume presents Connecting and waits one backoff interval before the
1040// next resume attempt. It settles early on cancellation or terminal EOF
1041// (terminated), or with stop on broker loss, which leaves the ready loop to
1042// observe and retire the connection, or when the user cancels the resume
1043// with Ctrl-C or Esc, which returns to the picker. Other keys typed during
1044// the outage are kept and replayed into the session once it resumes.
1045func (s *Supervisor) waitResume(ctx context.Context, input *terminalInputLifetime, service ports.BrokerNavigator, attempt int) (terminated bool, termErr error, stop bool) {
1046 s.transition(supervisorEvent{kind: supervisorAttachBegin, slow: true})
1047 delay := supervisorBackoffDelay(uint64(attempt), s.cfg.Jitter)
1048 timer := s.cfg.Clock.NewTimer(delay)
1049 defer stopSupervisorTimer(timer)
1050 keys := input.watchResume(s.cfg.Clock, s.logger)
1051 defer keys.release()
1052 for {
1053 select {
1054 case <-keys.cancelled:
1055 // The user gave up on the outage: back to the picker, quietly.
1056 s.logger.Info("client_attachment_resume_cancelled", "attempt", attempt)
1057 s.transition(supervisorEvent{kind: supervisorAttachEnded})
1058 return false, nil, true
1059 case <-timer.C():
1060 // Release resolves an Esc typed just before the backoff ended,
1061 // which may still cancel.
1062 keys.release()
1063 select {
1064 case <-keys.cancelled:
1065 s.logger.Info("client_attachment_resume_cancelled", "attempt", attempt)
1066 s.transition(supervisorEvent{kind: supervisorAttachEnded})
1067 return false, nil, true
1068 default:
1069 }
1070 return false, nil, false
1071 case <-ctx.Done():
1072 return true, ctx.Err(), false
1073 case err := <-input.EOF():
1074 return true, terminalReadCause(err), false
1075 case <-service.Done():
1076 s.transition(supervisorEvent{kind: supervisorAttachEnded})
1077 return false, nil, true
1078 case <-s.presentationInvalidation():
1079 s.renderResizeInvalidation()
1080 case <-s.spinnerTick():
1081 s.cfg.Spinner.AdvanceSpinner(s.State())
1082 }
1083 }
1084}
1085
1086// resumeRetry turns a transport-class failure of a resume attempt into an
1087// attachmentRetry outcome, so runResolvedAttachment retries while Connecting
1088// stays up. Any other failure, or a failure outside a resume, is not retried.
1089func (s *Supervisor) resumeRetry(err error, resuming bool) (attachmentOutcome, bool) {
1090 if !resuming || !resumableFailure(err) {
1091 return attachmentOutcome{}, false
1092 }
1093 s.logger.Info("client_attachment_resume_failed", "code", brokerErrorCode(err), "error", err, "cause", errors.Unwrap(err))
1094 return attachmentOutcome{kind: attachmentRetry, cause: err}, true
1095}
1096
1097// attemptFailed ends one attachment attempt that failed before settlement:
1098// a resume retries a transport-class failure, anything else is presented.
1099func (s *Supervisor) attemptFailed(err error, resuming bool) attachmentOutcome {
1100 if outcome, ok := s.resumeRetry(err, resuming); ok {
1101 return outcome
1102 }
1103 s.reportAttachmentFailure(err)
1104 return attachmentEndedOutcome()
1105}
1106
1107// resumableFailure reports whether a resume attempt failed for a transport
1108// reason worth retrying. Refusals, protocol failures, and an ambiguous outcome
1109// are final.
1110func resumableFailure(err error) bool {
1111 var lost ports.BrokerStreamLost
1112 if errors.As(err, &lost) {
1113 return true
1114 }
1115 var typed ports.BrokerError
1116 if !errors.As(err, &typed) {
1117 return false
1118 }
1119 switch typed.Code {
1120 case ports.BrokerErrorUnavailable, ports.BrokerErrorTimeout, ports.BrokerErrorAttachmentLost:
1121 return true
1122 default:
1123 return false
1124 }
1125}
1126
1127// reportAttachmentFailure returns to the picker with a typed, visible failure.
1128func (s *Supervisor) reportAttachmentFailure(err error) {
1129 s.transition(supervisorEvent{kind: supervisorAttachEnded, err: err})
1130 // Classification order matters: a local refusal never dialed a destination,
1131 // so it must not be reported as one, and only a genuinely unknown outcome
1132 // gets its own notice.
1133 var catalogueErr pickerCatalogueError
1134 var brokerErr ports.BrokerError
1135 switch {
1136 case errors.As(err, &catalogueErr):
1137 s.notifyLifecycle(LifecycleNoticeSelectionUnavailable)
1138 case errors.As(err, &brokerErr) && brokerErr.Code == ports.BrokerErrorOutcomeUnknown:
1139 s.notifyLifecycle(LifecycleNoticeOutcomeUnknown)
1140 default:
1141 s.notifyLifecycle(LifecycleNoticeDestinationFailed)
1142 }
1143 s.notifyAttachment(err)
1144}
1145
1146// notifyAttachment publishes the typed failure to the optional observer.
1147func (s *Supervisor) notifyAttachment(err error) {
1148 if s.cfg.Notify == nil || err == nil {
1149 return
1150 }
1151 s.cfg.Notify(s.State(), err)
1152}