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