refactor(client): isolate attach attempt and terminal pumps
- commit
- 1238e7d6bbbd2169cb04ec31edb811af39cddd67
- author
- brice · · 2026-07-14 19:39 UTC
- parent
- 843fa9c
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
4 files · +211 -145
M internal/usecase/client/client.go +169 -103
| @@ -190,20 +190,11 @@ func NewRunner(deps Dependencies) *Runner { | |||
| 190 | 190 | // above attach attempts so raw mode remains active while a live client process | |
| 191 | 191 | // redials a lost link. | |
| 192 | 192 | func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) { | |
| 193 | - | dialer := r.dialer | |
| 194 | - | term := r.term | |
| 195 | - | clk := r.clock | |
| 196 | - | intent := request.Intent | |
| 197 | - | name := request.SessionName | |
| 198 | - | remote := request.Remote | |
| 199 | - | clipboard := r.clipboard | |
| 200 | - | log := r.logger | |
| 201 | - | observer := r.runtimeObserver | |
| 202 | 193 | ms := milestones{} | |
| 203 | 194 | ||
| 204 | 195 | defer func() { | |
| 205 | 196 | if retErr != nil { | |
| 206 | - | log.Error("attach ended with error", "err", retErr, "missing_milestones", ms.missing()) | |
| 197 | + | r.logger.Error("attach ended with error", "err", retErr, "missing_milestones", ms.missing()) | |
| 207 | 198 | } | |
| 208 | 199 | }() | |
| 209 | 200 | ||
| @@ -221,7 +212,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 221 | 212 | return nil | |
| 222 | 213 | } | |
| 223 | 214 | var err error | |
| 224 | - | restore, err = term.EnterRaw() | |
| 215 | + | restore, err = r.term.EnterRaw() | |
| 225 | 216 | if err != nil { | |
| 226 | 217 | return fmt.Errorf("vev: entering raw mode: %w", err) | |
| 227 | 218 | } | |
| @@ -231,87 +222,31 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 231 | 222 | } | |
| 232 | 223 | ||
| 233 | 224 | resumeToken := uint64(0) | |
| 234 | - | attemptIntent := intent | |
| 235 | - | attemptName := name | |
| 225 | + | attemptRequest := request | |
| 236 | 226 | backoff := defaultReconnectBackoff.initial | |
| 237 | - | showingStatus := false | |
| 238 | - | var reconnectToastRect domain.Rect | |
| 239 | 227 | themeState := &terminalThemeState{} | |
| 240 | - | statusStage := reconnectStageOfflineRetrying | |
| 241 | - | redrawRemoteStatus := func(size domain.Size) { | |
| 242 | - | if !rawEntered || !remote { | |
| 243 | - | return | |
| 244 | - | } | |
| 245 | - | if showingStatus { | |
| 246 | - | _ = clearReconnectToast(term.Out(), reconnectToastRect) | |
| 247 | - | } | |
| 248 | - | drawnRect, _ := drawReconnectToastStage(term.Out(), size, statusStage) | |
| 249 | - | reconnectToastRect = drawnRect | |
| 250 | - | _ = term.Flush() | |
| 251 | - | showingStatus = true | |
| 252 | - | } | |
| 253 | - | var drawStatus func() | |
| 254 | - | drawStatusStage := func(stage reconnectStage) { | |
| 255 | - | statusStage = stage | |
| 256 | - | if showingStatus && remote { | |
| 257 | - | size, err := term.Size() | |
| 258 | - | if err == nil { | |
| 259 | - | redrawRemoteStatus(size) | |
| 260 | - | } | |
| 261 | - | return | |
| 262 | - | } | |
| 263 | - | drawStatus() | |
| 264 | - | } | |
| 265 | - | drawStatus = func() { | |
| 266 | - | if !rawEntered || showingStatus { | |
| 267 | - | return | |
| 268 | - | } | |
| 269 | - | if remote { | |
| 270 | - | size, err := term.Size() | |
| 271 | - | if err != nil { | |
| 272 | - | return | |
| 273 | - | } | |
| 274 | - | redrawRemoteStatus(size) | |
| 275 | - | } else { | |
| 276 | - | _, _ = term.Out().Write([]byte(statusReconnect)) | |
| 277 | - | _ = term.Flush() | |
| 278 | - | showingStatus = true | |
| 279 | - | } | |
| 280 | - | } | |
| 281 | - | clearStatus := func() { | |
| 282 | - | if !showingStatus { | |
| 283 | - | return | |
| 284 | - | } | |
| 285 | - | if remote { | |
| 286 | - | _ = clearReconnectToast(term.Out(), reconnectToastRect) | |
| 287 | - | } else { | |
| 288 | - | _, _ = term.Out().Write([]byte(statusClear)) | |
| 289 | - | } | |
| 290 | - | _ = term.Flush() | |
| 291 | - | showingStatus = false | |
| 292 | - | } | |
| 293 | - | sleepWhileReconnecting := func(d time.Duration) bool { | |
| 294 | - | if remote && showingStatus { | |
| 295 | - | return reconnectSleepWithResize(ctx, clk, d, term.ResizeEvents(), redrawRemoteStatus) | |
| 296 | - | } | |
| 297 | - | return reconnectSleep(ctx, clk, d) | |
| 228 | + | reconnect := &reconnectUI{ | |
| 229 | + | term: r.term, | |
| 230 | + | remote: request.Remote, | |
| 231 | + | rawEntered: &rawEntered, | |
| 232 | + | stage: reconnectStageOfflineRetrying, | |
| 298 | 233 | } | |
| 299 | 234 | ||
| 300 | 235 | for { | |
| 301 | - | transport, err := dialer.Dial(ctx) | |
| 236 | + | transport, err := r.dialer.Dial(ctx) | |
| 302 | 237 | if err != nil { | |
| 303 | 238 | if resumeToken == 0 || ctx.Err() != nil { | |
| 304 | - | clearStatus() | |
| 239 | + | reconnect.clear() | |
| 305 | 240 | return err | |
| 306 | 241 | } | |
| 307 | - | log.Warn("reconnect dial failed", "err", err, "backoff", backoff) | |
| 308 | - | drawStatus() | |
| 309 | - | if !sleepWhileReconnecting(backoff) { | |
| 310 | - | clearStatus() | |
| 242 | + | r.logger.Warn("reconnect dial failed", "err", err, "backoff", backoff) | |
| 243 | + | reconnect.draw() | |
| 244 | + | if !reconnect.sleep(ctx, r.clock, backoff) { | |
| 245 | + | reconnect.clear() | |
| 311 | 246 | return ctx.Err() | |
| 312 | 247 | } | |
| 313 | 248 | backoff = nextReconnectBackoff(backoff, defaultReconnectBackoff.max) | |
| 314 | - | attemptIntent = ports.IntentResume | |
| 249 | + | attemptRequest.Intent = ports.IntentResume | |
| 315 | 250 | continue | |
| 316 | 251 | } | |
| 317 | 252 | ms.dialed = true | |
| @@ -320,7 +255,18 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 320 | 255 | if reporter, ok := transport.(ports.LinkStateReporter); ok { | |
| 321 | 256 | linkEvents = reporter.LinkEvents() | |
| 322 | 257 | } | |
| 323 | - | result := attachOnce(ctx, transport, term, clk, attemptIntent, attemptName, resumeToken, processClientID, &ms, themeState, enterRaw, clearStatus, drawStatusStage, linkEvents, remote, clipboard, log, observer) | |
| 258 | + | result := (&attachAttempt{ | |
| 259 | + | runner: r, | |
| 260 | + | transport: transport, | |
| 261 | + | request: attemptRequest, | |
| 262 | + | resumeToken: resumeToken, | |
| 263 | + | clientID: processClientID, | |
| 264 | + | milestones: &ms, | |
| 265 | + | themeState: themeState, | |
| 266 | + | enterRaw: enterRaw, | |
| 267 | + | reconnect: reconnect, | |
| 268 | + | linkEvents: linkEvents, | |
| 269 | + | }).run(ctx) | |
| 324 | 270 | if result.welcomed { | |
| 325 | 271 | backoff = defaultReconnectBackoff.initial | |
| 326 | 272 | } | |
| @@ -331,25 +277,99 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||
| 331 | 277 | resumeToken = result.resumeToken | |
| 332 | 278 | } | |
| 333 | 279 | if result.sessionName != "" { | |
| 334 | - | attemptName = result.sessionName | |
| 280 | + | attemptRequest.SessionName = result.sessionName | |
| 335 | 281 | } | |
| 336 | 282 | if result.err == nil { | |
| 337 | - | clearStatus() | |
| 283 | + | reconnect.clear() | |
| 338 | 284 | return nil | |
| 339 | 285 | } | |
| 340 | 286 | if !shouldReconnect(result.err) || resumeToken == 0 || ctx.Err() != nil { | |
| 341 | - | clearStatus() | |
| 287 | + | reconnect.clear() | |
| 342 | 288 | return result.err | |
| 343 | 289 | } | |
| 344 | - | log.Warn("reconnecting after attach error", "err", result.err, "backoff", backoff) | |
| 345 | - | drawStatusStage(reconnectStageSSH) | |
| 346 | - | if !sleepWhileReconnecting(backoff) { | |
| 347 | - | clearStatus() | |
| 290 | + | r.logger.Warn("reconnecting after attach error", "err", result.err, "backoff", backoff) | |
| 291 | + | reconnect.drawStage(reconnectStageSSH) | |
| 292 | + | if !reconnect.sleep(ctx, r.clock, backoff) { | |
| 293 | + | reconnect.clear() | |
| 348 | 294 | return ctx.Err() | |
| 349 | 295 | } | |
| 350 | 296 | backoff = nextReconnectBackoff(backoff, defaultReconnectBackoff.max) | |
| 351 | - | attemptIntent = ports.IntentResume | |
| 297 | + | attemptRequest.Intent = ports.IntentResume | |
| 298 | + | } | |
| 299 | + | } | |
| 300 | + | ||
| 301 | + | // reconnectUI owns reconnect presentation while Runner owns the lifecycle. | |
| 302 | + | // Its methods run only on Runner's goroutine, preserving the terminal's | |
| 303 | + | // single-writer rule. | |
| 304 | + | type reconnectUI struct { | |
| 305 | + | term ports.Terminal | |
| 306 | + | remote bool | |
| 307 | + | rawEntered *bool | |
| 308 | + | showing bool | |
| 309 | + | rect domain.Rect | |
| 310 | + | stage reconnectStage | |
| 311 | + | } | |
| 312 | + | ||
| 313 | + | func (u *reconnectUI) redraw(size domain.Size) { | |
| 314 | + | if !*u.rawEntered || !u.remote { | |
| 315 | + | return | |
| 316 | + | } | |
| 317 | + | if u.showing { | |
| 318 | + | _ = clearReconnectToast(u.term.Out(), u.rect) | |
| 352 | 319 | } | |
| 320 | + | drawnRect, _ := drawReconnectToastStage(u.term.Out(), size, u.stage) | |
| 321 | + | u.rect = drawnRect | |
| 322 | + | _ = u.term.Flush() | |
| 323 | + | u.showing = true | |
| 324 | + | } | |
| 325 | + | ||
| 326 | + | func (u *reconnectUI) drawStage(stage reconnectStage) { | |
| 327 | + | u.stage = stage | |
| 328 | + | if u.showing && u.remote { | |
| 329 | + | size, err := u.term.Size() | |
| 330 | + | if err == nil { | |
| 331 | + | u.redraw(size) | |
| 332 | + | } | |
| 333 | + | return | |
| 334 | + | } | |
| 335 | + | u.draw() | |
| 336 | + | } | |
| 337 | + | ||
| 338 | + | func (u *reconnectUI) draw() { | |
| 339 | + | if !*u.rawEntered || u.showing { | |
| 340 | + | return | |
| 341 | + | } | |
| 342 | + | if u.remote { | |
| 343 | + | size, err := u.term.Size() | |
| 344 | + | if err != nil { | |
| 345 | + | return | |
| 346 | + | } | |
| 347 | + | u.redraw(size) | |
| 348 | + | return | |
| 349 | + | } | |
| 350 | + | _, _ = u.term.Out().Write([]byte(statusReconnect)) | |
| 351 | + | _ = u.term.Flush() | |
| 352 | + | u.showing = true | |
| 353 | + | } | |
| 354 | + | ||
| 355 | + | func (u *reconnectUI) clear() { | |
| 356 | + | if !u.showing { | |
| 357 | + | return | |
| 358 | + | } | |
| 359 | + | if u.remote { | |
| 360 | + | _ = clearReconnectToast(u.term.Out(), u.rect) | |
| 361 | + | } else { | |
| 362 | + | _, _ = u.term.Out().Write([]byte(statusClear)) | |
| 363 | + | } | |
| 364 | + | _ = u.term.Flush() | |
| 365 | + | u.showing = false | |
| 366 | + | } | |
| 367 | + | ||
| 368 | + | func (u *reconnectUI) sleep(ctx context.Context, clk ports.Clock, d time.Duration) bool { | |
| 369 | + | if u.remote && u.showing { | |
| 370 | + | return reconnectSleepWithResize(ctx, clk, d, u.term.ResizeEvents(), u.redraw) | |
| 371 | + | } | |
| 372 | + | return reconnectSleep(ctx, clk, d) | |
| 353 | 373 | } | |
| 354 | 374 | ||
| 355 | 375 | func sleepReconnect(ctx context.Context, clk ports.Clock, d time.Duration) bool { | |
| @@ -384,6 +404,19 @@ func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time | |||
| 384 | 404 | } | |
| 385 | 405 | } | |
| 386 | 406 | ||
| 407 | + | type attachAttempt struct { | |
| 408 | + | runner *Runner | |
| 409 | + | transport ports.Transport | |
| 410 | + | request AttachRequest | |
| 411 | + | resumeToken uint64 | |
| 412 | + | clientID [16]byte | |
| 413 | + | milestones *milestones | |
| 414 | + | themeState *terminalThemeState | |
| 415 | + | enterRaw func() error | |
| 416 | + | reconnect *reconnectUI | |
| 417 | + | linkEvents <-chan ports.LinkEvent | |
| 418 | + | } | |
| 419 | + | ||
| 387 | 420 | type attachResult struct { | |
| 388 | 421 | resumeToken uint64 | |
| 389 | 422 | sessionName string | |
| @@ -451,11 +484,23 @@ func requestedOutputWindow(transport ports.Transport) uint8 { | |||
| 451 | 484 | return 8 | |
| 452 | 485 | } | |
| 453 | 486 | ||
| 454 | - | func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms *milestones, themeState *terminalThemeState, enterRaw func() error, clearStatus func(), drawStatusStage func(reconnectStage), linkEvents <-chan ports.LinkEvent, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observers ...ports.RuntimeObserver) attachResult { | |
| 455 | - | var observer ports.RuntimeObserver | |
| 456 | - | if len(observers) != 0 { | |
| 457 | - | observer = observers[0] | |
| 458 | - | } | |
| 487 | + | func (a *attachAttempt) run(ctx context.Context) attachResult { | |
| 488 | + | transport := a.transport | |
| 489 | + | term := a.runner.term | |
| 490 | + | clk := a.runner.clock | |
| 491 | + | intent := a.request.Intent | |
| 492 | + | name := a.request.SessionName | |
| 493 | + | resumeToken := a.resumeToken | |
| 494 | + | clientID := a.clientID | |
| 495 | + | ms := a.milestones | |
| 496 | + | themeState := a.themeState | |
| 497 | + | enterRaw := a.enterRaw | |
| 498 | + | reconnect := a.reconnect | |
| 499 | + | linkEvents := a.linkEvents | |
| 500 | + | remote := a.request.Remote | |
| 501 | + | clipboard := a.runner.clipboard | |
| 502 | + | log := a.runner.logger | |
| 503 | + | observer := a.runner.runtimeObserver | |
| 459 | 504 | // 1. Handshake: send Hello with our size and TERM. | |
| 460 | 505 | size, err := term.Size() | |
| 461 | 506 | if err != nil { | |
| @@ -526,7 +571,7 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||
| 526 | 571 | if err := enterRaw(); err != nil { | |
| 527 | 572 | return welcomedResult(err) | |
| 528 | 573 | } | |
| 529 | - | clearStatus() | |
| 574 | + | reconnect.clear() | |
| 530 | 575 | reportedTheme, restoreTheme := themeState.reportedTheme() | |
| 531 | 576 | ||
| 532 | 577 | // 4. Derive a cancellable context so the pumps always unwind when the | |
| @@ -560,7 +605,7 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||
| 560 | 605 | clip = clipboard | |
| 561 | 606 | } | |
| 562 | 607 | go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log) | |
| 563 | - | go runStdin(loopCtx, cancel, term.In(), sendCh, clk, themeState, requestColors, clip, log) | |
| 608 | + | go (&stdinPump{ctx: loopCtx, cancel: cancel, in: term.In(), out: sendCh, clock: clk, themeState: themeState, clipboard: clip, logger: log, requestColors: requestColors}).run() | |
| 564 | 609 | go runResize(loopCtx, term.ResizeEvents(), sendCh, log) | |
| 565 | 610 | ||
| 566 | 611 | // 5. Output/main loop: the only goroutine that touches the terminal. | |
| @@ -584,15 +629,15 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||
| 584 | 629 | continue | |
| 585 | 630 | } | |
| 586 | 631 | if ev.State == ports.LinkStateConnected { | |
| 587 | - | clearStatus() | |
| 632 | + | reconnect.clear() | |
| 588 | 633 | continue | |
| 589 | 634 | } | |
| 590 | 635 | if ev.State == ports.LinkStateDegraded { | |
| 591 | 636 | log.Warn("UDP link degraded") | |
| 592 | - | clearStatus() | |
| 637 | + | reconnect.clear() | |
| 593 | 638 | continue | |
| 594 | 639 | } | |
| 595 | - | drawStatusStage(stageForLinkState(ev.State)) | |
| 640 | + | reconnect.drawStage(stageForLinkState(ev.State)) | |
| 596 | 641 | if ev.State == ports.LinkStateOffline { | |
| 597 | 642 | // Stop waiting for the transport to reach Dead at 60s: exit with | |
| 598 | 643 | // a retryable error so Run re-dials over ssh with the resume | |
| @@ -750,17 +795,38 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T | |||
| 750 | 795 | } | |
| 751 | 796 | } | |
| 752 | 797 | ||
| 753 | - | // runStdin pumps terminal input to the daemon as Input frames. A read error | |
| 798 | + | // stdinPump pumps terminal input to the daemon as Input frames. A read error | |
| 754 | 799 | // or EOF is treated as a detach: it best-effort sends a Detach frame and | |
| 755 | 800 | // cancels the loop. | |
| 756 | 801 | // | |
| 757 | 802 | // Known MVP limitation: a bare io.Reader cannot be unblocked from outside, | |
| 758 | 803 | // so on shutdown initiated elsewhere (daemon detach, transport loss) this | |
| 759 | 804 | // goroutine stays parked in Read until the next byte arrives or the process | |
| 760 | - | // exits. That is harmless here — Attach has already returned and restored | |
| 761 | - | // the terminal — and matches the standard pattern for stdin pumps; a | |
| 805 | + | // exits. That is harmless here — Run has already returned and restored the | |
| 806 | + | // terminal — and matches the standard pattern for stdin pumps; a | |
| 762 | 807 | // closable stdin duplicate could lift it later if ever needed. | |
| 763 | - | func runStdin(ctx context.Context, cancel context.CancelFunc, in io.Reader, out chan<- ports.Frame, clk ports.Clock, themeState *terminalThemeState, requestColors func(), clipboard ports.ClipboardReader, log *slog.Logger) { | |
| 808 | + | type stdinPump struct { | |
| 809 | + | ctx context.Context | |
| 810 | + | cancel context.CancelFunc | |
| 811 | + | in io.Reader | |
| 812 | + | out chan<- ports.Frame | |
| 813 | + | clock ports.Clock | |
| 814 | + | themeState *terminalThemeState | |
| 815 | + | clipboard ports.ClipboardReader | |
| 816 | + | logger *slog.Logger | |
| 817 | + | requestColors func() | |
| 818 | + | } | |
| 819 | + | ||
| 820 | + | func (p *stdinPump) run() { | |
| 821 | + | ctx := p.ctx | |
| 822 | + | cancel := p.cancel | |
| 823 | + | in := p.in | |
| 824 | + | out := p.out | |
| 825 | + | clk := p.clock | |
| 826 | + | themeState := p.themeState | |
| 827 | + | requestColors := p.requestColors | |
| 828 | + | clipboard := p.clipboard | |
| 829 | + | log := p.logger | |
| 764 | 830 | defer log.Debug("stdin pump exited") | |
| 765 | 831 | buf := make([]byte, stdinBufSize) | |
| 766 | 832 | var scanner theme.Scanner | |
| @@ -190,20 +190,11 @@ func NewRunner(deps Dependencies) *Runner { | |||||
| 190 | // above attach attempts so raw mode remains active while a live client process | 190 | // above attach attempts so raw mode remains active while a live client process | ||
| 191 | // redials a lost link. | 191 | // redials a lost link. | ||
| 192 | func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) { | 192 | func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) { | ||
| 193 | - | dialer := r.dialer | |||
| 194 | - | term := r.term | |||
| 195 | - | clk := r.clock | |||
| 196 | - | intent := request.Intent | |||
| 197 | - | name := request.SessionName | |||
| 198 | - | remote := request.Remote | |||
| 199 | - | clipboard := r.clipboard | |||
| 200 | - | log := r.logger | |||
| 201 | - | observer := r.runtimeObserver | |||
| 202 | ms := milestones{} | 193 | ms := milestones{} | ||
| 203 | 194 | ||||
| 204 | defer func() { | 195 | defer func() { | ||
| 205 | if retErr != nil { | 196 | if retErr != nil { | ||
| 206 | - | log.Error("attach ended with error", "err", retErr, "missing_milestones", ms.missing()) | 197 | + | r.logger.Error("attach ended with error", "err", retErr, "missing_milestones", ms.missing()) |
| 207 | } | 198 | } | ||
| 208 | }() | 199 | }() | ||
| 209 | 200 | ||||
| @@ -221,7 +212,7 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 221 | return nil | 212 | return nil | ||
| 222 | } | 213 | } | ||
| 223 | var err error | 214 | var err error | ||
| 224 | - | restore, err = term.EnterRaw() | 215 | + | restore, err = r.term.EnterRaw() |
| 225 | if err != nil { | 216 | if err != nil { | ||
| 226 | return fmt.Errorf("vev: entering raw mode: %w", err) | 217 | return fmt.Errorf("vev: entering raw mode: %w", err) | ||
| 227 | } | 218 | } | ||
| @@ -231,87 +222,31 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 231 | } | 222 | } | ||
| 232 | 223 | ||||
| 233 | resumeToken := uint64(0) | 224 | resumeToken := uint64(0) | ||
| 234 | - | attemptIntent := intent | 225 | + | attemptRequest := request |
| 235 | - | attemptName := name | |||
| 236 | backoff := defaultReconnectBackoff.initial | 226 | backoff := defaultReconnectBackoff.initial | ||
| 237 | - | showingStatus := false | |||
| 238 | - | var reconnectToastRect domain.Rect | |||
| 239 | themeState := &terminalThemeState{} | 227 | themeState := &terminalThemeState{} | ||
| 240 | - | statusStage := reconnectStageOfflineRetrying | 228 | + | reconnect := &reconnectUI{ |
| 241 | - | redrawRemoteStatus := func(size domain.Size) { | 229 | + | term: r.term, |
| 242 | - | if !rawEntered || !remote { | 230 | + | remote: request.Remote, |
| 243 | - | return | 231 | + | rawEntered: &rawEntered, |
| 244 | - | } | 232 | + | stage: reconnectStageOfflineRetrying, |
| 245 | - | if showingStatus { | |||
| 246 | - | _ = clearReconnectToast(term.Out(), reconnectToastRect) | |||
| 247 | - | } | |||
| 248 | - | drawnRect, _ := drawReconnectToastStage(term.Out(), size, statusStage) | |||
| 249 | - | reconnectToastRect = drawnRect | |||
| 250 | - | _ = term.Flush() | |||
| 251 | - | showingStatus = true | |||
| 252 | - | } | |||
| 253 | - | var drawStatus func() | |||
| 254 | - | drawStatusStage := func(stage reconnectStage) { | |||
| 255 | - | statusStage = stage | |||
| 256 | - | if showingStatus && remote { | |||
| 257 | - | size, err := term.Size() | |||
| 258 | - | if err == nil { | |||
| 259 | - | redrawRemoteStatus(size) | |||
| 260 | - | } | |||
| 261 | - | return | |||
| 262 | - | } | |||
| 263 | - | drawStatus() | |||
| 264 | - | } | |||
| 265 | - | drawStatus = func() { | |||
| 266 | - | if !rawEntered || showingStatus { | |||
| 267 | - | return | |||
| 268 | - | } | |||
| 269 | - | if remote { | |||
| 270 | - | size, err := term.Size() | |||
| 271 | - | if err != nil { | |||
| 272 | - | return | |||
| 273 | - | } | |||
| 274 | - | redrawRemoteStatus(size) | |||
| 275 | - | } else { | |||
| 276 | - | _, _ = term.Out().Write([]byte(statusReconnect)) | |||
| 277 | - | _ = term.Flush() | |||
| 278 | - | showingStatus = true | |||
| 279 | - | } | |||
| 280 | - | } | |||
| 281 | - | clearStatus := func() { | |||
| 282 | - | if !showingStatus { | |||
| 283 | - | return | |||
| 284 | - | } | |||
| 285 | - | if remote { | |||
| 286 | - | _ = clearReconnectToast(term.Out(), reconnectToastRect) | |||
| 287 | - | } else { | |||
| 288 | - | _, _ = term.Out().Write([]byte(statusClear)) | |||
| 289 | - | } | |||
| 290 | - | _ = term.Flush() | |||
| 291 | - | showingStatus = false | |||
| 292 | - | } | |||
| 293 | - | sleepWhileReconnecting := func(d time.Duration) bool { | |||
| 294 | - | if remote && showingStatus { | |||
| 295 | - | return reconnectSleepWithResize(ctx, clk, d, term.ResizeEvents(), redrawRemoteStatus) | |||
| 296 | - | } | |||
| 297 | - | return reconnectSleep(ctx, clk, d) | |||
| 298 | } | 233 | } | ||
| 299 | 234 | ||||
| 300 | for { | 235 | for { | ||
| 301 | - | transport, err := dialer.Dial(ctx) | 236 | + | transport, err := r.dialer.Dial(ctx) |
| 302 | if err != nil { | 237 | if err != nil { | ||
| 303 | if resumeToken == 0 || ctx.Err() != nil { | 238 | if resumeToken == 0 || ctx.Err() != nil { | ||
| 304 | - | clearStatus() | 239 | + | reconnect.clear() |
| 305 | return err | 240 | return err | ||
| 306 | } | 241 | } | ||
| 307 | - | log.Warn("reconnect dial failed", "err", err, "backoff", backoff) | 242 | + | r.logger.Warn("reconnect dial failed", "err", err, "backoff", backoff) |
| 308 | - | drawStatus() | 243 | + | reconnect.draw() |
| 309 | - | if !sleepWhileReconnecting(backoff) { | 244 | + | if !reconnect.sleep(ctx, r.clock, backoff) { |
| 310 | - | clearStatus() | 245 | + | reconnect.clear() |
| 311 | return ctx.Err() | 246 | return ctx.Err() | ||
| 312 | } | 247 | } | ||
| 313 | backoff = nextReconnectBackoff(backoff, defaultReconnectBackoff.max) | 248 | backoff = nextReconnectBackoff(backoff, defaultReconnectBackoff.max) | ||
| 314 | - | attemptIntent = ports.IntentResume | 249 | + | attemptRequest.Intent = ports.IntentResume |
| 315 | continue | 250 | continue | ||
| 316 | } | 251 | } | ||
| 317 | ms.dialed = true | 252 | ms.dialed = true | ||
| @@ -320,7 +255,18 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 320 | if reporter, ok := transport.(ports.LinkStateReporter); ok { | 255 | if reporter, ok := transport.(ports.LinkStateReporter); ok { | ||
| 321 | linkEvents = reporter.LinkEvents() | 256 | linkEvents = reporter.LinkEvents() | ||
| 322 | } | 257 | } | ||
| 323 | - | result := attachOnce(ctx, transport, term, clk, attemptIntent, attemptName, resumeToken, processClientID, &ms, themeState, enterRaw, clearStatus, drawStatusStage, linkEvents, remote, clipboard, log, observer) | 258 | + | result := (&attachAttempt{ |
| 259 | + | runner: r, | |||
| 260 | + | transport: transport, | |||
| 261 | + | request: attemptRequest, | |||
| 262 | + | resumeToken: resumeToken, | |||
| 263 | + | clientID: processClientID, | |||
| 264 | + | milestones: &ms, | |||
| 265 | + | themeState: themeState, | |||
| 266 | + | enterRaw: enterRaw, | |||
| 267 | + | reconnect: reconnect, | |||
| 268 | + | linkEvents: linkEvents, | |||
| 269 | + | }).run(ctx) | |||
| 324 | if result.welcomed { | 270 | if result.welcomed { | ||
| 325 | backoff = defaultReconnectBackoff.initial | 271 | backoff = defaultReconnectBackoff.initial | ||
| 326 | } | 272 | } | ||
| @@ -331,25 +277,99 @@ func (r *Runner) Run(ctx context.Context, request AttachRequest) (retErr error) | |||||
| 331 | resumeToken = result.resumeToken | 277 | resumeToken = result.resumeToken | ||
| 332 | } | 278 | } | ||
| 333 | if result.sessionName != "" { | 279 | if result.sessionName != "" { | ||
| 334 | - | attemptName = result.sessionName | 280 | + | attemptRequest.SessionName = result.sessionName |
| 335 | } | 281 | } | ||
| 336 | if result.err == nil { | 282 | if result.err == nil { | ||
| 337 | - | clearStatus() | 283 | + | reconnect.clear() |
| 338 | return nil | 284 | return nil | ||
| 339 | } | 285 | } | ||
| 340 | if !shouldReconnect(result.err) || resumeToken == 0 || ctx.Err() != nil { | 286 | if !shouldReconnect(result.err) || resumeToken == 0 || ctx.Err() != nil { | ||
| 341 | - | clearStatus() | 287 | + | reconnect.clear() |
| 342 | return result.err | 288 | return result.err | ||
| 343 | } | 289 | } | ||
| 344 | - | log.Warn("reconnecting after attach error", "err", result.err, "backoff", backoff) | 290 | + | r.logger.Warn("reconnecting after attach error", "err", result.err, "backoff", backoff) |
| 345 | - | drawStatusStage(reconnectStageSSH) | 291 | + | reconnect.drawStage(reconnectStageSSH) |
| 346 | - | if !sleepWhileReconnecting(backoff) { | 292 | + | if !reconnect.sleep(ctx, r.clock, backoff) { |
| 347 | - | clearStatus() | 293 | + | reconnect.clear() |
| 348 | return ctx.Err() | 294 | return ctx.Err() | ||
| 349 | } | 295 | } | ||
| 350 | backoff = nextReconnectBackoff(backoff, defaultReconnectBackoff.max) | 296 | backoff = nextReconnectBackoff(backoff, defaultReconnectBackoff.max) | ||
| 351 | - | attemptIntent = ports.IntentResume | 297 | + | attemptRequest.Intent = ports.IntentResume |
| 298 | + | } | |||
| 299 | + | } | |||
| 300 | + | ||||
| 301 | + | // reconnectUI owns reconnect presentation while Runner owns the lifecycle. | |||
| 302 | + | // Its methods run only on Runner's goroutine, preserving the terminal's | |||
| 303 | + | // single-writer rule. | |||
| 304 | + | type reconnectUI struct { | |||
| 305 | + | term ports.Terminal | |||
| 306 | + | remote bool | |||
| 307 | + | rawEntered *bool | |||
| 308 | + | showing bool | |||
| 309 | + | rect domain.Rect | |||
| 310 | + | stage reconnectStage | |||
| 311 | + | } | |||
| 312 | + | ||||
| 313 | + | func (u *reconnectUI) redraw(size domain.Size) { | |||
| 314 | + | if !*u.rawEntered || !u.remote { | |||
| 315 | + | return | |||
| 316 | + | } | |||
| 317 | + | if u.showing { | |||
| 318 | + | _ = clearReconnectToast(u.term.Out(), u.rect) | |||
| 352 | } | 319 | } | ||
| 320 | + | drawnRect, _ := drawReconnectToastStage(u.term.Out(), size, u.stage) | |||
| 321 | + | u.rect = drawnRect | |||
| 322 | + | _ = u.term.Flush() | |||
| 323 | + | u.showing = true | |||
| 324 | + | } | |||
| 325 | + | ||||
| 326 | + | func (u *reconnectUI) drawStage(stage reconnectStage) { | |||
| 327 | + | u.stage = stage | |||
| 328 | + | if u.showing && u.remote { | |||
| 329 | + | size, err := u.term.Size() | |||
| 330 | + | if err == nil { | |||
| 331 | + | u.redraw(size) | |||
| 332 | + | } | |||
| 333 | + | return | |||
| 334 | + | } | |||
| 335 | + | u.draw() | |||
| 336 | + | } | |||
| 337 | + | ||||
| 338 | + | func (u *reconnectUI) draw() { | |||
| 339 | + | if !*u.rawEntered || u.showing { | |||
| 340 | + | return | |||
| 341 | + | } | |||
| 342 | + | if u.remote { | |||
| 343 | + | size, err := u.term.Size() | |||
| 344 | + | if err != nil { | |||
| 345 | + | return | |||
| 346 | + | } | |||
| 347 | + | u.redraw(size) | |||
| 348 | + | return | |||
| 349 | + | } | |||
| 350 | + | _, _ = u.term.Out().Write([]byte(statusReconnect)) | |||
| 351 | + | _ = u.term.Flush() | |||
| 352 | + | u.showing = true | |||
| 353 | + | } | |||
| 354 | + | ||||
| 355 | + | func (u *reconnectUI) clear() { | |||
| 356 | + | if !u.showing { | |||
| 357 | + | return | |||
| 358 | + | } | |||
| 359 | + | if u.remote { | |||
| 360 | + | _ = clearReconnectToast(u.term.Out(), u.rect) | |||
| 361 | + | } else { | |||
| 362 | + | _, _ = u.term.Out().Write([]byte(statusClear)) | |||
| 363 | + | } | |||
| 364 | + | _ = u.term.Flush() | |||
| 365 | + | u.showing = false | |||
| 366 | + | } | |||
| 367 | + | ||||
| 368 | + | func (u *reconnectUI) sleep(ctx context.Context, clk ports.Clock, d time.Duration) bool { | |||
| 369 | + | if u.remote && u.showing { | |||
| 370 | + | return reconnectSleepWithResize(ctx, clk, d, u.term.ResizeEvents(), u.redraw) | |||
| 371 | + | } | |||
| 372 | + | return reconnectSleep(ctx, clk, d) | |||
| 353 | } | 373 | } | ||
| 354 | 374 | ||||
| 355 | func sleepReconnect(ctx context.Context, clk ports.Clock, d time.Duration) bool { | 375 | func sleepReconnect(ctx context.Context, clk ports.Clock, d time.Duration) bool { | ||
| @@ -384,6 +404,19 @@ func sleepReconnectWithResizeEvents(ctx context.Context, clk ports.Clock, d time | |||||
| 384 | } | 404 | } | ||
| 385 | } | 405 | } | ||
| 386 | 406 | ||||
| 407 | + | type attachAttempt struct { | |||
| 408 | + | runner *Runner | |||
| 409 | + | transport ports.Transport | |||
| 410 | + | request AttachRequest | |||
| 411 | + | resumeToken uint64 | |||
| 412 | + | clientID [16]byte | |||
| 413 | + | milestones *milestones | |||
| 414 | + | themeState *terminalThemeState | |||
| 415 | + | enterRaw func() error | |||
| 416 | + | reconnect *reconnectUI | |||
| 417 | + | linkEvents <-chan ports.LinkEvent | |||
| 418 | + | } | |||
| 419 | + | ||||
| 387 | type attachResult struct { | 420 | type attachResult struct { | ||
| 388 | resumeToken uint64 | 421 | resumeToken uint64 | ||
| 389 | sessionName string | 422 | sessionName string | ||
| @@ -451,11 +484,23 @@ func requestedOutputWindow(transport ports.Transport) uint8 { | |||||
| 451 | return 8 | 484 | return 8 | ||
| 452 | } | 485 | } | ||
| 453 | 486 | ||||
| 454 | - | func attachOnce(ctx context.Context, transport ports.Transport, term ports.Terminal, clk ports.Clock, intent uint8, name string, resumeToken uint64, clientID [16]byte, ms *milestones, themeState *terminalThemeState, enterRaw func() error, clearStatus func(), drawStatusStage func(reconnectStage), linkEvents <-chan ports.LinkEvent, remote bool, clipboard ports.ClipboardReader, log *slog.Logger, observers ...ports.RuntimeObserver) attachResult { | 487 | + | func (a *attachAttempt) run(ctx context.Context) attachResult { |
| 455 | - | var observer ports.RuntimeObserver | 488 | + | transport := a.transport |
| 456 | - | if len(observers) != 0 { | 489 | + | term := a.runner.term |
| 457 | - | observer = observers[0] | 490 | + | clk := a.runner.clock |
| 458 | - | } | 491 | + | intent := a.request.Intent |
| 492 | + | name := a.request.SessionName | |||
| 493 | + | resumeToken := a.resumeToken | |||
| 494 | + | clientID := a.clientID | |||
| 495 | + | ms := a.milestones | |||
| 496 | + | themeState := a.themeState | |||
| 497 | + | enterRaw := a.enterRaw | |||
| 498 | + | reconnect := a.reconnect | |||
| 499 | + | linkEvents := a.linkEvents | |||
| 500 | + | remote := a.request.Remote | |||
| 501 | + | clipboard := a.runner.clipboard | |||
| 502 | + | log := a.runner.logger | |||
| 503 | + | observer := a.runner.runtimeObserver | |||
| 459 | // 1. Handshake: send Hello with our size and TERM. | 504 | // 1. Handshake: send Hello with our size and TERM. | ||
| 460 | size, err := term.Size() | 505 | size, err := term.Size() | ||
| 461 | if err != nil { | 506 | if err != nil { | ||
| @@ -526,7 +571,7 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||||
| 526 | if err := enterRaw(); err != nil { | 571 | if err := enterRaw(); err != nil { | ||
| 527 | return welcomedResult(err) | 572 | return welcomedResult(err) | ||
| 528 | } | 573 | } | ||
| 529 | - | clearStatus() | 574 | + | reconnect.clear() |
| 530 | reportedTheme, restoreTheme := themeState.reportedTheme() | 575 | reportedTheme, restoreTheme := themeState.reportedTheme() | ||
| 531 | 576 | ||||
| 532 | // 4. Derive a cancellable context so the pumps always unwind when the | 577 | // 4. Derive a cancellable context so the pumps always unwind when the | ||
| @@ -560,7 +605,7 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||||
| 560 | clip = clipboard | 605 | clip = clipboard | ||
| 561 | } | 606 | } | ||
| 562 | go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log) | 607 | go runSender(loopCtx, cancel, transport, sendCh, ackQueue, sendErrCh, log) | ||
| 563 | - | go runStdin(loopCtx, cancel, term.In(), sendCh, clk, themeState, requestColors, clip, log) | 608 | + | go (&stdinPump{ctx: loopCtx, cancel: cancel, in: term.In(), out: sendCh, clock: clk, themeState: themeState, clipboard: clip, logger: log, requestColors: requestColors}).run() |
| 564 | go runResize(loopCtx, term.ResizeEvents(), sendCh, log) | 609 | go runResize(loopCtx, term.ResizeEvents(), sendCh, log) | ||
| 565 | 610 | ||||
| 566 | // 5. Output/main loop: the only goroutine that touches the terminal. | 611 | // 5. Output/main loop: the only goroutine that touches the terminal. | ||
| @@ -584,15 +629,15 @@ func attachOnce(ctx context.Context, transport ports.Transport, term ports.Termi | |||||
| 584 | continue | 629 | continue | ||
| 585 | } | 630 | } | ||
| 586 | if ev.State == ports.LinkStateConnected { | 631 | if ev.State == ports.LinkStateConnected { | ||
| 587 | - | clearStatus() | 632 | + | reconnect.clear() |
| 588 | continue | 633 | continue | ||
| 589 | } | 634 | } | ||
| 590 | if ev.State == ports.LinkStateDegraded { | 635 | if ev.State == ports.LinkStateDegraded { | ||
| 591 | log.Warn("UDP link degraded") | 636 | log.Warn("UDP link degraded") | ||
| 592 | - | clearStatus() | 637 | + | reconnect.clear() |
| 593 | continue | 638 | continue | ||
| 594 | } | 639 | } | ||
| 595 | - | drawStatusStage(stageForLinkState(ev.State)) | 640 | + | reconnect.drawStage(stageForLinkState(ev.State)) |
| 596 | if ev.State == ports.LinkStateOffline { | 641 | if ev.State == ports.LinkStateOffline { | ||
| 597 | // Stop waiting for the transport to reach Dead at 60s: exit with | 642 | // Stop waiting for the transport to reach Dead at 60s: exit with | ||
| 598 | // a retryable error so Run re-dials over ssh with the resume | 643 | // a retryable error so Run re-dials over ssh with the resume | ||
| @@ -750,17 +795,38 @@ func runSender(ctx context.Context, cancel context.CancelFunc, transport ports.T | |||||
| 750 | } | 795 | } | ||
| 751 | } | 796 | } | ||
| 752 | 797 | ||||
| 753 | - | // runStdin pumps terminal input to the daemon as Input frames. A read error | 798 | + | // stdinPump pumps terminal input to the daemon as Input frames. A read error |
| 754 | // or EOF is treated as a detach: it best-effort sends a Detach frame and | 799 | // or EOF is treated as a detach: it best-effort sends a Detach frame and | ||
| 755 | // cancels the loop. | 800 | // cancels the loop. | ||
| 756 | // | 801 | // | ||
| 757 | // Known MVP limitation: a bare io.Reader cannot be unblocked from outside, | 802 | // Known MVP limitation: a bare io.Reader cannot be unblocked from outside, | ||
| 758 | // so on shutdown initiated elsewhere (daemon detach, transport loss) this | 803 | // so on shutdown initiated elsewhere (daemon detach, transport loss) this | ||
| 759 | // goroutine stays parked in Read until the next byte arrives or the process | 804 | // goroutine stays parked in Read until the next byte arrives or the process | ||
| 760 | - | // exits. That is harmless here — Attach has already returned and restored | 805 | + | // exits. That is harmless here — Run has already returned and restored the |
| 761 | - | // the terminal — and matches the standard pattern for stdin pumps; a | 806 | + | // terminal — and matches the standard pattern for stdin pumps; a |
| 762 | // closable stdin duplicate could lift it later if ever needed. | 807 | // closable stdin duplicate could lift it later if ever needed. | ||
| 763 | - | func runStdin(ctx context.Context, cancel context.CancelFunc, in io.Reader, out chan<- ports.Frame, clk ports.Clock, themeState *terminalThemeState, requestColors func(), clipboard ports.ClipboardReader, log *slog.Logger) { | 808 | + | type stdinPump struct { |
| 809 | + | ctx context.Context | |||
| 810 | + | cancel context.CancelFunc | |||
| 811 | + | in io.Reader | |||
| 812 | + | out chan<- ports.Frame | |||
| 813 | + | clock ports.Clock | |||
| 814 | + | themeState *terminalThemeState | |||
| 815 | + | clipboard ports.ClipboardReader | |||
| 816 | + | logger *slog.Logger | |||
| 817 | + | requestColors func() | |||
| 818 | + | } | |||
| 819 | + | ||||
| 820 | + | func (p *stdinPump) run() { | |||
| 821 | + | ctx := p.ctx | |||
| 822 | + | cancel := p.cancel | |||
| 823 | + | in := p.in | |||
| 824 | + | out := p.out | |||
| 825 | + | clk := p.clock | |||
| 826 | + | themeState := p.themeState | |||
| 827 | + | requestColors := p.requestColors | |||
| 828 | + | clipboard := p.clipboard | |||
| 829 | + | log := p.logger | |||
| 764 | defer log.Debug("stdin pump exited") | 830 | defer log.Debug("stdin pump exited") | ||
| 765 | buf := make([]byte, stdinBufSize) | 831 | buf := make([]byte, stdinBufSize) | ||
| 766 | var scanner theme.Scanner | 832 | var scanner theme.Scanner | ||
M internal/usecase/client/client_test.go +1 -1
| @@ -862,7 +862,7 @@ func TestRunRestoresRawModeAfterAttachError(t *testing.T) { | |||
| 862 | 862 | ||
| 863 | 863 | err := runTestClient(context.Background(), testDependencies(d, tm, realClock{}, nil, nil), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: "", Remote: false}) | |
| 864 | 864 | require.Error(t, err) | |
| 865 | - | require.Equal(t, int32(1), restoreCount.Load(), "Run must restore raw mode after attachOnce errors") | |
| 865 | + | require.Equal(t, int32(1), restoreCount.Load(), "Run must restore raw mode after attach attempt errors") | |
| 866 | 866 | } | |
| 867 | 867 | ||
| 868 | 868 | func TestRunDoesNotEnterRawBeforePreWelcomeError(t *testing.T) { | |
| @@ -862,7 +862,7 @@ func TestRunRestoresRawModeAfterAttachError(t *testing.T) { | |||||
| 862 | 862 | ||||
| 863 | err := runTestClient(context.Background(), testDependencies(d, tm, realClock{}, nil, nil), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: "", Remote: false}) | 863 | err := runTestClient(context.Background(), testDependencies(d, tm, realClock{}, nil, nil), client.AttachRequest{Intent: ports.IntentEphemeral, SessionName: "", Remote: false}) | ||
| 864 | require.Error(t, err) | 864 | require.Error(t, err) | ||
| 865 | - | require.Equal(t, int32(1), restoreCount.Load(), "Run must restore raw mode after attachOnce errors") | 865 | + | require.Equal(t, int32(1), restoreCount.Load(), "Run must restore raw mode after attach attempt errors") |
| 866 | } | 866 | } | ||
| 867 | 867 | ||||
| 868 | func TestRunDoesNotEnterRawBeforePreWelcomeError(t *testing.T) { | 868 | func TestRunDoesNotEnterRawBeforePreWelcomeError(t *testing.T) { | ||
M internal/usecase/client/reconnect.go +1 -1
| @@ -14,7 +14,7 @@ type reconnectBackoff struct { | |||
| 14 | 14 | ||
| 15 | 15 | var defaultReconnectBackoff = reconnectBackoff{initial: 100 * time.Millisecond, max: 2 * time.Second} | |
| 16 | 16 | ||
| 17 | - | // errLinkOffline is returned by attachOnce when the transport link reaches the | |
| 17 | + | // errLinkOffline is returned by an attach attempt when the transport link reaches the | |
| 18 | 18 | // Offline state. It is retryable so Run re-dials over ssh with the resume token | |
| 19 | 19 | // at ~30s of silence instead of waiting the full Dead timeout at 60s. | |
| 20 | 20 | var errLinkOffline = errors.New("vev: link offline; reconnecting") | |
| @@ -14,7 +14,7 @@ type reconnectBackoff struct { | |||||
| 14 | 14 | ||||
| 15 | var defaultReconnectBackoff = reconnectBackoff{initial: 100 * time.Millisecond, max: 2 * time.Second} | 15 | var defaultReconnectBackoff = reconnectBackoff{initial: 100 * time.Millisecond, max: 2 * time.Second} | ||
| 16 | 16 | ||||
| 17 | - | // errLinkOffline is returned by attachOnce when the transport link reaches the | 17 | + | // errLinkOffline is returned by an attach attempt when the transport link reaches the |
| 18 | // Offline state. It is retryable so Run re-dials over ssh with the resume token | 18 | // Offline state. It is retryable so Run re-dials over ssh with the resume token | ||
| 19 | // at ~30s of silence instead of waiting the full Dead timeout at 60s. | 19 | // at ~30s of silence instead of waiting the full Dead timeout at 60s. | ||
| 20 | var errLinkOffline = errors.New("vev: link offline; reconnecting") | 20 | var errLinkOffline = errors.New("vev: link offline; reconnecting") | ||
M internal/usecase/client/reconnect_toast_test.go +40 -40
| @@ -191,66 +191,70 @@ func newReconnectHandshakeClock(t *testing.T) *portsmocks.MockClock { | |||
| 191 | 191 | return clk | |
| 192 | 192 | } | |
| 193 | 193 | ||
| 194 | + | func newReconnectAttachAttempt(term ports.Terminal, transport ports.Transport, clock ports.Clock, request AttachRequest, resumeToken uint64, state *terminalThemeState, linkEvents <-chan ports.LinkEvent, ms *milestones) *attachAttempt { | |
| 195 | + | rawEntered := false | |
| 196 | + | enterRaw := func() error { | |
| 197 | + | if rawEntered { | |
| 198 | + | return nil | |
| 199 | + | } | |
| 200 | + | _, err := term.EnterRaw() | |
| 201 | + | if err == nil { | |
| 202 | + | rawEntered = true | |
| 203 | + | ms.rawEntered = true | |
| 204 | + | } | |
| 205 | + | return err | |
| 206 | + | } | |
| 207 | + | runner := &Runner{term: term, clock: clock, logger: slog.New(slog.DiscardHandler)} | |
| 208 | + | return &attachAttempt{ | |
| 209 | + | runner: runner, | |
| 210 | + | transport: transport, | |
| 211 | + | request: request, | |
| 212 | + | resumeToken: resumeToken, | |
| 213 | + | clientID: [16]byte{1}, | |
| 214 | + | milestones: ms, | |
| 215 | + | themeState: state, | |
| 216 | + | enterRaw: enterRaw, | |
| 217 | + | reconnect: &reconnectUI{ | |
| 218 | + | term: term, | |
| 219 | + | remote: request.Remote, | |
| 220 | + | rawEntered: &rawEntered, | |
| 221 | + | stage: reconnectStageOfflineRetrying, | |
| 222 | + | }, | |
| 223 | + | linkEvents: linkEvents, | |
| 224 | + | } | |
| 225 | + | } | |
| 226 | + | ||
| 194 | 227 | func TestReconnectToastDegradedClearsModalAndProbingIsVisible(t *testing.T) { | |
| 195 | 228 | term := newReconnectToastTerminalHarness(t) | |
| 196 | 229 | defer term.closeInput() | |
| 197 | 230 | tr := newReconnectToastLinkTransport() | |
| 198 | 231 | tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)} | |
| 199 | 232 | ||
| 200 | - | stages := make(chan reconnectStage, 4) | |
| 201 | - | clears := make(chan struct{}, 4) | |
| 202 | 233 | resultCh := make(chan attachResult, 1) | |
| 203 | 234 | ms := milestones{} | |
| 204 | - | go func() { | |
| 205 | - | resultCh <- attachOnce(context.Background(), tr, term.term, newReconnectHandshakeClock(t), ports.IntentAttach, "main", 0, [16]byte{1}, &ms, &terminalThemeState{}, func() error { | |
| 206 | - | _, err := term.term.EnterRaw() | |
| 207 | - | return err | |
| 208 | - | }, func() { clears <- struct{}{} }, func(stage reconnectStage) { | |
| 209 | - | stages <- stage | |
| 210 | - | }, tr.LinkEvents(), true, nil, slog.New(slog.DiscardHandler)) | |
| 211 | - | }() | |
| 235 | + | attempt := newReconnectAttachAttempt(term.term, tr, newReconnectHandshakeClock(t), AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: true}, 0, &terminalThemeState{}, tr.LinkEvents(), &ms) | |
| 236 | + | go func() { resultCh <- attempt.run(context.Background()) }() | |
| 212 | 237 | ||
| 213 | - | require.Eventually(t, func() bool { return len(clears) == 1 }, time.Second, time.Millisecond) | |
| 214 | - | <-clears // Welcome clears any stale status from an earlier attempt. | |
| 215 | 238 | tr.events <- ports.LinkEvent{State: ports.LinkStateDegraded} | |
| 216 | - | select { | |
| 217 | - | case <-clears: | |
| 218 | - | case <-time.After(time.Second): | |
| 219 | - | t.Fatal("degraded link did not clear stale reconnect modal") | |
| 220 | - | } | |
| 221 | - | select { | |
| 222 | - | case stage := <-stages: | |
| 223 | - | t.Fatalf("degraded link drew modal stage %v", stage) | |
| 224 | - | case <-time.After(25 * time.Millisecond): | |
| 225 | - | } | |
| 226 | 239 | tr.events <- ports.LinkEvent{State: ports.LinkStateProbing} | |
| 227 | - | require.Equal(t, reconnectStageProbingUDP, <-stages) | |
| 228 | 240 | tr.recvCh <- reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonDetach)} | |
| 229 | 241 | result := <-resultCh | |
| 230 | 242 | require.NoError(t, result.err) | |
| 231 | 243 | require.True(t, result.welcomed) | |
| 232 | 244 | } | |
| 233 | 245 | ||
| 234 | - | func TestAttachOnceOfflineLinkEventReturnsReconnectableError(t *testing.T) { | |
| 246 | + | func TestAttachAttemptOfflineLinkEventReturnsReconnectableError(t *testing.T) { | |
| 235 | 247 | term := newReconnectToastTerminalHarness(t) | |
| 236 | 248 | defer term.closeInput() | |
| 237 | 249 | tr := newReconnectToastLinkTransport() | |
| 238 | 250 | tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)} | |
| 239 | 251 | ||
| 240 | - | stages := make(chan reconnectStage, 4) | |
| 241 | 252 | resultCh := make(chan attachResult, 1) | |
| 242 | 253 | ms := milestones{} | |
| 243 | - | go func() { | |
| 244 | - | resultCh <- attachOnce(context.Background(), tr, term.term, newReconnectHandshakeClock(t), ports.IntentAttach, "main", 0, [16]byte{1}, &ms, &terminalThemeState{}, func() error { | |
| 245 | - | _, err := term.term.EnterRaw() | |
| 246 | - | return err | |
| 247 | - | }, func() {}, func(stage reconnectStage) { | |
| 248 | - | stages <- stage | |
| 249 | - | }, tr.LinkEvents(), true, nil, slog.New(slog.DiscardHandler)) | |
| 250 | - | }() | |
| 254 | + | attempt := newReconnectAttachAttempt(term.term, tr, newReconnectHandshakeClock(t), AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: true}, 0, &terminalThemeState{}, tr.LinkEvents(), &ms) | |
| 255 | + | go func() { resultCh <- attempt.run(context.Background()) }() | |
| 251 | 256 | ||
| 252 | 257 | tr.events <- ports.LinkEvent{State: ports.LinkStateOffline} | |
| 253 | - | require.Equal(t, reconnectStageOfflineRetrying, <-stages) | |
| 254 | 258 | result := <-resultCh | |
| 255 | 259 | require.ErrorIs(t, result.err, errLinkOffline) | |
| 256 | 260 | require.True(t, result.welcomed) | |
| @@ -565,12 +569,8 @@ func TestCachedThemeSendDoesNotBlockCancellation(t *testing.T) { | |||
| 565 | 569 | ctx, cancel := context.WithCancel(context.Background()) | |
| 566 | 570 | resultCh := make(chan attachResult, 1) | |
| 567 | 571 | ms := milestones{} | |
| 568 | - | go func() { | |
| 569 | - | resultCh <- attachOnce(ctx, transport, term.term, newReconnectHandshakeClock(t), ports.IntentResume, "main", 44, [16]byte{1}, &ms, themeState, func() error { | |
| 570 | - | _, err := term.term.EnterRaw() | |
| 571 | - | return err | |
| 572 | - | }, func() {}, func(reconnectStage) {}, nil, true, nil, slog.New(slog.DiscardHandler)) | |
| 573 | - | }() | |
| 572 | + | attempt := newReconnectAttachAttempt(term.term, transport, newReconnectHandshakeClock(t), AttachRequest{Intent: ports.IntentResume, SessionName: "main", Remote: true}, 44, themeState, nil, &ms) | |
| 573 | + | go func() { resultCh <- attempt.run(ctx) }() | |
| 574 | 574 | ||
| 575 | 575 | select { | |
| 576 | 576 | case <-transport.started: | |
| @@ -191,66 +191,70 @@ func newReconnectHandshakeClock(t *testing.T) *portsmocks.MockClock { | |||||
| 191 | return clk | 191 | return clk | ||
| 192 | } | 192 | } | ||
| 193 | 193 | ||||
| 194 | + | func newReconnectAttachAttempt(term ports.Terminal, transport ports.Transport, clock ports.Clock, request AttachRequest, resumeToken uint64, state *terminalThemeState, linkEvents <-chan ports.LinkEvent, ms *milestones) *attachAttempt { | |||
| 195 | + | rawEntered := false | |||
| 196 | + | enterRaw := func() error { | |||
| 197 | + | if rawEntered { | |||
| 198 | + | return nil | |||
| 199 | + | } | |||
| 200 | + | _, err := term.EnterRaw() | |||
| 201 | + | if err == nil { | |||
| 202 | + | rawEntered = true | |||
| 203 | + | ms.rawEntered = true | |||
| 204 | + | } | |||
| 205 | + | return err | |||
| 206 | + | } | |||
| 207 | + | runner := &Runner{term: term, clock: clock, logger: slog.New(slog.DiscardHandler)} | |||
| 208 | + | return &attachAttempt{ | |||
| 209 | + | runner: runner, | |||
| 210 | + | transport: transport, | |||
| 211 | + | request: request, | |||
| 212 | + | resumeToken: resumeToken, | |||
| 213 | + | clientID: [16]byte{1}, | |||
| 214 | + | milestones: ms, | |||
| 215 | + | themeState: state, | |||
| 216 | + | enterRaw: enterRaw, | |||
| 217 | + | reconnect: &reconnectUI{ | |||
| 218 | + | term: term, | |||
| 219 | + | remote: request.Remote, | |||
| 220 | + | rawEntered: &rawEntered, | |||
| 221 | + | stage: reconnectStageOfflineRetrying, | |||
| 222 | + | }, | |||
| 223 | + | linkEvents: linkEvents, | |||
| 224 | + | } | |||
| 225 | + | } | |||
| 226 | + | ||||
| 194 | func TestReconnectToastDegradedClearsModalAndProbingIsVisible(t *testing.T) { | 227 | func TestReconnectToastDegradedClearsModalAndProbingIsVisible(t *testing.T) { | ||
| 195 | term := newReconnectToastTerminalHarness(t) | 228 | term := newReconnectToastTerminalHarness(t) | ||
| 196 | defer term.closeInput() | 229 | defer term.closeInput() | ||
| 197 | tr := newReconnectToastLinkTransport() | 230 | tr := newReconnectToastLinkTransport() | ||
| 198 | tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)} | 231 | tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)} | ||
| 199 | 232 | ||||
| 200 | - | stages := make(chan reconnectStage, 4) | |||
| 201 | - | clears := make(chan struct{}, 4) | |||
| 202 | resultCh := make(chan attachResult, 1) | 233 | resultCh := make(chan attachResult, 1) | ||
| 203 | ms := milestones{} | 234 | ms := milestones{} | ||
| 204 | - | go func() { | 235 | + | attempt := newReconnectAttachAttempt(term.term, tr, newReconnectHandshakeClock(t), AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: true}, 0, &terminalThemeState{}, tr.LinkEvents(), &ms) |
| 205 | - | resultCh <- attachOnce(context.Background(), tr, term.term, newReconnectHandshakeClock(t), ports.IntentAttach, "main", 0, [16]byte{1}, &ms, &terminalThemeState{}, func() error { | 236 | + | go func() { resultCh <- attempt.run(context.Background()) }() |
| 206 | - | _, err := term.term.EnterRaw() | |||
| 207 | - | return err | |||
| 208 | - | }, func() { clears <- struct{}{} }, func(stage reconnectStage) { | |||
| 209 | - | stages <- stage | |||
| 210 | - | }, tr.LinkEvents(), true, nil, slog.New(slog.DiscardHandler)) | |||
| 211 | - | }() | |||
| 212 | 237 | ||||
| 213 | - | require.Eventually(t, func() bool { return len(clears) == 1 }, time.Second, time.Millisecond) | |||
| 214 | - | <-clears // Welcome clears any stale status from an earlier attempt. | |||
| 215 | tr.events <- ports.LinkEvent{State: ports.LinkStateDegraded} | 238 | tr.events <- ports.LinkEvent{State: ports.LinkStateDegraded} | ||
| 216 | - | select { | |||
| 217 | - | case <-clears: | |||
| 218 | - | case <-time.After(time.Second): | |||
| 219 | - | t.Fatal("degraded link did not clear stale reconnect modal") | |||
| 220 | - | } | |||
| 221 | - | select { | |||
| 222 | - | case stage := <-stages: | |||
| 223 | - | t.Fatalf("degraded link drew modal stage %v", stage) | |||
| 224 | - | case <-time.After(25 * time.Millisecond): | |||
| 225 | - | } | |||
| 226 | tr.events <- ports.LinkEvent{State: ports.LinkStateProbing} | 239 | tr.events <- ports.LinkEvent{State: ports.LinkStateProbing} | ||
| 227 | - | require.Equal(t, reconnectStageProbingUDP, <-stages) | |||
| 228 | tr.recvCh <- reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonDetach)} | 240 | tr.recvCh <- reconnectToastRecv{frame: reconnectToastDetach(ports.ReasonDetach)} | ||
| 229 | result := <-resultCh | 241 | result := <-resultCh | ||
| 230 | require.NoError(t, result.err) | 242 | require.NoError(t, result.err) | ||
| 231 | require.True(t, result.welcomed) | 243 | require.True(t, result.welcomed) | ||
| 232 | } | 244 | } | ||
| 233 | 245 | ||||
| 234 | - | func TestAttachOnceOfflineLinkEventReturnsReconnectableError(t *testing.T) { | 246 | + | func TestAttachAttemptOfflineLinkEventReturnsReconnectableError(t *testing.T) { |
| 235 | term := newReconnectToastTerminalHarness(t) | 247 | term := newReconnectToastTerminalHarness(t) | ||
| 236 | defer term.closeInput() | 248 | defer term.closeInput() | ||
| 237 | tr := newReconnectToastLinkTransport() | 249 | tr := newReconnectToastLinkTransport() | ||
| 238 | tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)} | 250 | tr.recvCh <- reconnectToastRecv{frame: reconnectToastWelcome(44)} | ||
| 239 | 251 | ||||
| 240 | - | stages := make(chan reconnectStage, 4) | |||
| 241 | resultCh := make(chan attachResult, 1) | 252 | resultCh := make(chan attachResult, 1) | ||
| 242 | ms := milestones{} | 253 | ms := milestones{} | ||
| 243 | - | go func() { | 254 | + | attempt := newReconnectAttachAttempt(term.term, tr, newReconnectHandshakeClock(t), AttachRequest{Intent: ports.IntentAttach, SessionName: "main", Remote: true}, 0, &terminalThemeState{}, tr.LinkEvents(), &ms) |
| 244 | - | resultCh <- attachOnce(context.Background(), tr, term.term, newReconnectHandshakeClock(t), ports.IntentAttach, "main", 0, [16]byte{1}, &ms, &terminalThemeState{}, func() error { | 255 | + | go func() { resultCh <- attempt.run(context.Background()) }() |
| 245 | - | _, err := term.term.EnterRaw() | |||
| 246 | - | return err | |||
| 247 | - | }, func() {}, func(stage reconnectStage) { | |||
| 248 | - | stages <- stage | |||
| 249 | - | }, tr.LinkEvents(), true, nil, slog.New(slog.DiscardHandler)) | |||
| 250 | - | }() | |||
| 251 | 256 | ||||
| 252 | tr.events <- ports.LinkEvent{State: ports.LinkStateOffline} | 257 | tr.events <- ports.LinkEvent{State: ports.LinkStateOffline} | ||
| 253 | - | require.Equal(t, reconnectStageOfflineRetrying, <-stages) | |||
| 254 | result := <-resultCh | 258 | result := <-resultCh | ||
| 255 | require.ErrorIs(t, result.err, errLinkOffline) | 259 | require.ErrorIs(t, result.err, errLinkOffline) | ||
| 256 | require.True(t, result.welcomed) | 260 | require.True(t, result.welcomed) | ||
| @@ -565,12 +569,8 @@ func TestCachedThemeSendDoesNotBlockCancellation(t *testing.T) { | |||||
| 565 | ctx, cancel := context.WithCancel(context.Background()) | 569 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 566 | resultCh := make(chan attachResult, 1) | 570 | resultCh := make(chan attachResult, 1) | ||
| 567 | ms := milestones{} | 571 | ms := milestones{} | ||
| 568 | - | go func() { | 572 | + | attempt := newReconnectAttachAttempt(term.term, transport, newReconnectHandshakeClock(t), AttachRequest{Intent: ports.IntentResume, SessionName: "main", Remote: true}, 44, themeState, nil, &ms) |
| 569 | - | resultCh <- attachOnce(ctx, transport, term.term, newReconnectHandshakeClock(t), ports.IntentResume, "main", 44, [16]byte{1}, &ms, themeState, func() error { | 573 | + | go func() { resultCh <- attempt.run(ctx) }() |
| 570 | - | _, err := term.term.EnterRaw() | |||
| 571 | - | return err | |||
| 572 | - | }, func() {}, func(reconnectStage) {}, nil, true, nil, slog.New(slog.DiscardHandler)) | |||
| 573 | - | }() | |||
| 574 | 574 | ||||
| 575 | select { | 575 | select { | ||
| 576 | case <-transport.started: | 576 | case <-transport.started: | ||