fix(transport): apply session-neutral connection review fixes
- commit
- bb0d730c838e160ad5291d864bcabfb76a596adb
- author
- brice · · 2026-08-04 09:23 UTC
- parent
- 68ba351
- signature
- GPG Verified
CD88F9E564E0102646FB6077A1A07C7C7C3E349D· arch-gpg Verified for the committer email
28 files · +325 -147
M AGENTS.md +1 -1
| @@ -75,5 +75,5 @@ Wire payload types/codecs live in `internal/ports/frame.go` and `internal/ports/ | |||
| 75 | 75 | - Named sessions survive headless and persist across daemon restarts. | |
| 76 | 76 | - The daemon starts on first use and exits when the last session ends. | |
| 77 | 77 | - Local attach sends `Hello`, receives `Welcome`, then pumps `Input`/`Resize` and `Output` frames. | |
| 78 | - | - Remote attach runs `ssh -- host vev _stdio [session]` and proxies the same protocol over stdio. | |
| 78 | + | - Remote attach runs `ssh -- host vev _stdio` and proxies the same protocol over stdio; session selection is carried in `Hello`. | |
| 79 | 79 | - Only one client may attach to a session; a new attach replaces the old client. | |
| @@ -75,5 +75,5 @@ Wire payload types/codecs live in `internal/ports/frame.go` and `internal/ports/ | |||||
| 75 | - Named sessions survive headless and persist across daemon restarts. | 75 | - Named sessions survive headless and persist across daemon restarts. | ||
| 76 | - The daemon starts on first use and exits when the last session ends. | 76 | - The daemon starts on first use and exits when the last session ends. | ||
| 77 | - Local attach sends `Hello`, receives `Welcome`, then pumps `Input`/`Resize` and `Output` frames. | 77 | - Local attach sends `Hello`, receives `Welcome`, then pumps `Input`/`Resize` and `Output` frames. | ||
| 78 | - | - Remote attach runs `ssh -- host vev _stdio [session]` and proxies the same protocol over stdio. | 78 | + | - Remote attach runs `ssh -- host vev _stdio` and proxies the same protocol over stdio; session selection is carried in `Hello`. |
| 79 | - Only one client may attach to a session; a new attach replaces the old client. | 79 | - Only one client may attach to a session; a new attach replaces the old client. | ||
M cmd/vev-perf-harness/evidence_test.go +1 -1
| @@ -44,7 +44,7 @@ func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) { | |||
| 44 | 44 | if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil { | |
| 45 | 45 | t.Fatal(err) | |
| 46 | 46 | } | |
| 47 | - | p, err := l.preparePeer(m, roleCommand{Args: []string{"_udp-proxy", "work"}, Transport: tc.tr}) | |
| 47 | + | p, err := l.preparePeer(m, roleCommand{Args: []string{"_udp-proxy"}, Transport: tc.tr}) | |
| 48 | 48 | if err != nil { | |
| 49 | 49 | t.Fatal(err) | |
| 50 | 50 | } | |
| @@ -44,7 +44,7 @@ func TestCLITransportSeamOwnsExclusivePeerTraceAndCleanup(t *testing.T) { | |||||
| 44 | if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil { | 44 | if err := os.WriteFile(m.TracePath, nil, 0o600); err != nil { | ||
| 45 | t.Fatal(err) | 45 | t.Fatal(err) | ||
| 46 | } | 46 | } | ||
| 47 | - | p, err := l.preparePeer(m, roleCommand{Args: []string{"_udp-proxy", "work"}, Transport: tc.tr}) | 47 | + | p, err := l.preparePeer(m, roleCommand{Args: []string{"_udp-proxy"}, Transport: tc.tr}) |
| 48 | if err != nil { | 48 | if err != nil { | ||
| 49 | t.Fatal(err) | 49 | t.Fatal(err) | ||
| 50 | } | 50 | } | ||
M cmd/vev-perf-harness/hangup_test.go +2 -3
| @@ -185,9 +185,8 @@ func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) { | |||
| 185 | 185 | case "daemon": | |
| 186 | 186 | command = roleCommand{Args: []string{"--daemon"}} | |
| 187 | 187 | case "ssh_stdio_peer": | |
| 188 | - | // An empty remote session requests an ephemeral session, avoiding a | |
| 189 | - | // separate bootstrap client while still executing the traced _stdio mode. | |
| 190 | - | command = roleCommand{Args: []string{"_stdio", ""}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}} | |
| 188 | + | // The transport-neutral _stdio mode requests an ephemeral session. | |
| 189 | + | command = roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}} | |
| 191 | 190 | case "client": | |
| 192 | 191 | command = roleCommand{Args: []string{"attach", "harness@127.0.0.1"}} | |
| 193 | 192 | } | |
| @@ -185,9 +185,8 @@ func TestClientGracefulShutdownDrainsTracedStdioDescendant(t *testing.T) { | |||||
| 185 | case "daemon": | 185 | case "daemon": | ||
| 186 | command = roleCommand{Args: []string{"--daemon"}} | 186 | command = roleCommand{Args: []string{"--daemon"}} | ||
| 187 | case "ssh_stdio_peer": | 187 | case "ssh_stdio_peer": | ||
| 188 | - | // An empty remote session requests an ephemeral session, avoiding a | 188 | + | // The transport-neutral _stdio mode requests an ephemeral session. |
| 189 | - | // separate bootstrap client while still executing the traced _stdio mode. | 189 | + | command = roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}} |
| 190 | - | command = roleCommand{Args: []string{"_stdio", ""}, Transport: transport{ID: "ssh_stdio", Kind: "ssh_stdio"}} | |||
| 191 | case "client": | 190 | case "client": | ||
| 192 | command = roleCommand{Args: []string{"attach", "harness@127.0.0.1"}} | 191 | command = roleCommand{Args: []string{"attach", "harness@127.0.0.1"}} | ||
| 193 | } | 192 | } | ||
M cmd/vev-perf-harness/launcher.go +11 -11
| @@ -332,7 +332,7 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP | |||
| 332 | 332 | if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil { | |
| 333 | 333 | return nil, fail(err) | |
| 334 | 334 | } | |
| 335 | - | if len(role.Args) != 2 || (role.Args[0] != "_stdio" && role.Args[0] != "_udp-proxy") { | |
| 335 | + | if len(role.Args) != 1 || (role.Args[0] != "_stdio" && role.Args[0] != "_udp-proxy") { | |
| 336 | 336 | return nil, fail(fmt.Errorf("unsupported public peer command %q", role.Args)) | |
| 337 | 337 | } | |
| 338 | 338 | route := peerRoute{mapping: m, command: role} | |
| @@ -373,27 +373,27 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP | |||
| 373 | 373 | set -eu | |
| 374 | 374 | case "$*" in | |
| 375 | 375 | *"_stdio"*) | |
| 376 | - | exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[12]q VEV_PERF_RUN=%[13]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio %[6]q | |
| 376 | + | exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[11]q VEV_PERF_RUN=%[12]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio | |
| 377 | 377 | ;; | |
| 378 | 378 | *"_udp-bootstrap"*) | |
| 379 | - | rm -f %[7]q %[8]q %[9]q | |
| 380 | - | env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[12]q VEV_PERF_RUN=%[13]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q & | |
| 379 | + | rm -f %[6]q %[7]q %[8]q | |
| 380 | + | env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[11]q VEV_PERF_RUN=%[12]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy >%[6]q 2>%[7]q & | |
| 381 | 381 | peer=$! | |
| 382 | - | printf '%%s\n' "$peer" > %[11]q | |
| 382 | + | printf '%%s\n' "$peer" > %[10]q | |
| 383 | 383 | i=0 | |
| 384 | - | while [ ! -s %[7]q ] && [ "$i" -lt 1000 ]; do sleep 0.01; i=$((i+1)); done | |
| 385 | - | test -s %[7]q | |
| 386 | - | line=$(head -n 1 %[7]q) | |
| 387 | - | printf '%%s\n' "$line" > %[9]q | |
| 384 | + | while [ ! -s %[6]q ] && [ "$i" -lt 1000 ]; do sleep 0.01; i=$((i+1)); done | |
| 385 | + | test -s %[6]q | |
| 386 | + | line=$(head -n 1 %[6]q) | |
| 387 | + | printf '%%s\n' "$line" > %[8]q | |
| 388 | 388 | set -- $line | |
| 389 | 389 | test "$1" = VEV-UDP | |
| 390 | 390 | test -n "$3" | |
| 391 | - | printf 'VEV-UDP %%s %%s\n' %[10]d "$3" | |
| 391 | + | printf 'VEV-UDP %%s %%s\n' %[9]d "$3" | |
| 392 | 392 | exit 0 | |
| 393 | 393 | ;; | |
| 394 | 394 | *) echo 'vev harness ssh seam rejected non-vev command' >&2; exit 64 ;; | |
| 395 | 395 | esac | |
| 396 | - | `, m.TracePath, m.ProcessID, runtimeDir, filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, target, netemPort, route.pidPath, m.Scenario, m.Run) | |
| 396 | + | `, m.TracePath, m.ProcessID, runtimeDir, filepath.Join(runDir, "state"), l.bin, ready, stderr, target, netemPort, route.pidPath, m.Scenario, m.Run) | |
| 397 | 397 | if err := os.WriteFile(shim, []byte(body), 0o700); err != nil { | |
| 398 | 398 | return nil, closeNetem(err) | |
| 399 | 399 | } | |
| @@ -332,7 +332,7 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP | |||||
| 332 | if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil { | 332 | if err := safedir.EnsurePrivate(filepath.Join(runDir, "state")); err != nil { | ||
| 333 | return nil, fail(err) | 333 | return nil, fail(err) | ||
| 334 | } | 334 | } | ||
| 335 | - | if len(role.Args) != 2 || (role.Args[0] != "_stdio" && role.Args[0] != "_udp-proxy") { | 335 | + | if len(role.Args) != 1 || (role.Args[0] != "_stdio" && role.Args[0] != "_udp-proxy") { |
| 336 | return nil, fail(fmt.Errorf("unsupported public peer command %q", role.Args)) | 336 | return nil, fail(fmt.Errorf("unsupported public peer command %q", role.Args)) | ||
| 337 | } | 337 | } | ||
| 338 | route := peerRoute{mapping: m, command: role} | 338 | route := peerRoute{mapping: m, command: role} | ||
| @@ -373,27 +373,27 @@ func (l *cliLauncher) preparePeer(m processMapping, role roleCommand) (launchedP | |||||
| 373 | set -eu | 373 | set -eu | ||
| 374 | case "$*" in | 374 | case "$*" in | ||
| 375 | *"_stdio"*) | 375 | *"_stdio"*) | ||
| 376 | - | exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[12]q VEV_PERF_RUN=%[13]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio %[6]q | 376 | + | exec env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[11]q VEV_PERF_RUN=%[12]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _stdio |
| 377 | ;; | 377 | ;; | ||
| 378 | *"_udp-bootstrap"*) | 378 | *"_udp-bootstrap"*) | ||
| 379 | - | rm -f %[7]q %[8]q %[9]q | 379 | + | rm -f %[6]q %[7]q %[8]q |
| 380 | - | env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[12]q VEV_PERF_RUN=%[13]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy %[6]q >%[7]q 2>%[8]q & | 380 | + | env VEV_PERF_TRACE=%[1]q VEV_PERF_PROCESS_ID=%[2]q VEV_PERF_SCENARIO=%[11]q VEV_PERF_RUN=%[12]d XDG_RUNTIME_DIR=%[3]q XDG_STATE_HOME=%[4]q TERM=xterm-256color %[5]q _udp-proxy >%[6]q 2>%[7]q & |
| 381 | peer=$! | 381 | peer=$! | ||
| 382 | - | printf '%%s\n' "$peer" > %[11]q | 382 | + | printf '%%s\n' "$peer" > %[10]q |
| 383 | i=0 | 383 | i=0 | ||
| 384 | - | while [ ! -s %[7]q ] && [ "$i" -lt 1000 ]; do sleep 0.01; i=$((i+1)); done | 384 | + | while [ ! -s %[6]q ] && [ "$i" -lt 1000 ]; do sleep 0.01; i=$((i+1)); done |
| 385 | - | test -s %[7]q | 385 | + | test -s %[6]q |
| 386 | - | line=$(head -n 1 %[7]q) | 386 | + | line=$(head -n 1 %[6]q) |
| 387 | - | printf '%%s\n' "$line" > %[9]q | 387 | + | printf '%%s\n' "$line" > %[8]q |
| 388 | set -- $line | 388 | set -- $line | ||
| 389 | test "$1" = VEV-UDP | 389 | test "$1" = VEV-UDP | ||
| 390 | test -n "$3" | 390 | test -n "$3" | ||
| 391 | - | printf 'VEV-UDP %%s %%s\n' %[10]d "$3" | 391 | + | printf 'VEV-UDP %%s %%s\n' %[9]d "$3" |
| 392 | exit 0 | 392 | exit 0 | ||
| 393 | ;; | 393 | ;; | ||
| 394 | *) echo 'vev harness ssh seam rejected non-vev command' >&2; exit 64 ;; | 394 | *) echo 'vev harness ssh seam rejected non-vev command' >&2; exit 64 ;; | ||
| 395 | esac | 395 | esac | ||
| 396 | - | `, m.TracePath, m.ProcessID, runtimeDir, filepath.Join(runDir, "state"), l.bin, role.Args[1], ready, stderr, target, netemPort, route.pidPath, m.Scenario, m.Run) | 396 | + | `, m.TracePath, m.ProcessID, runtimeDir, filepath.Join(runDir, "state"), l.bin, ready, stderr, target, netemPort, route.pidPath, m.Scenario, m.Run) |
| 397 | if err := os.WriteFile(shim, []byte(body), 0o700); err != nil { | 397 | if err := os.WriteFile(shim, []byte(body), 0o700); err != nil { | ||
| 398 | return nil, closeNetem(err) | 398 | return nil, closeNetem(err) | ||
| 399 | } | 399 | } | ||
M cmd/vev-perf-harness/launcher_test.go +2 -2
| @@ -614,8 +614,8 @@ func TestHarnessUsesPublicRoleCommandsAndPTYWorkloads(t *testing.T) { | |||
| 614 | 614 | }{ | |
| 615 | 615 | {"daemon", []string{"--daemon"}}, | |
| 616 | 616 | {"client", []string{"new", "perf-s-001"}}, | |
| 617 | - | {"ssh_stdio_peer", []string{"_stdio", "perf-s-001"}}, | |
| 618 | - | {"udp_peer", []string{"_udp-proxy", "perf-s-001"}}, | |
| 617 | + | {"ssh_stdio_peer", []string{"_stdio"}}, | |
| 618 | + | {"udp_peer", []string{"_udp-proxy"}}, | |
| 619 | 619 | } { | |
| 620 | 620 | t.Run(tc.role, func(t *testing.T) { | |
| 621 | 621 | got := roleArgs(scenario{ID: "s", Transport: "local"}, processMapping{Role: tc.role, Run: 1}) | |
| @@ -614,8 +614,8 @@ func TestHarnessUsesPublicRoleCommandsAndPTYWorkloads(t *testing.T) { | |||||
| 614 | }{ | 614 | }{ | ||
| 615 | {"daemon", []string{"--daemon"}}, | 615 | {"daemon", []string{"--daemon"}}, | ||
| 616 | {"client", []string{"new", "perf-s-001"}}, | 616 | {"client", []string{"new", "perf-s-001"}}, | ||
| 617 | - | {"ssh_stdio_peer", []string{"_stdio", "perf-s-001"}}, | 617 | + | {"ssh_stdio_peer", []string{"_stdio"}}, |
| 618 | - | {"udp_peer", []string{"_udp-proxy", "perf-s-001"}}, | 618 | + | {"udp_peer", []string{"_udp-proxy"}}, |
| 619 | } { | 619 | } { | ||
| 620 | t.Run(tc.role, func(t *testing.T) { | 620 | t.Run(tc.role, func(t *testing.T) { | ||
| 621 | got := roleArgs(scenario{ID: "s", Transport: "local"}, processMapping{Role: tc.role, Run: 1}) | 621 | got := roleArgs(scenario{ID: "s", Transport: "local"}, processMapping{Role: tc.role, Run: 1}) | ||
M cmd/vev-perf-harness/runner.go +2 -2
| @@ -396,9 +396,9 @@ func routeRoleArgs(s scenario, m processMapping, selected transport) roleCommand | |||
| 396 | 396 | return roleCommand{Args: remote} | |
| 397 | 397 | } | |
| 398 | 398 | case "ssh_stdio_peer": | |
| 399 | - | return roleCommand{Args: []string{"_stdio", session}, Transport: transport{ID: s.Transport, Kind: "ssh_stdio"}} | |
| 399 | + | return roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: s.Transport, Kind: "ssh_stdio"}} | |
| 400 | 400 | case "udp_peer": | |
| 401 | - | return roleCommand{Args: []string{"_udp-proxy", session}, Transport: selected} | |
| 401 | + | return roleCommand{Args: []string{"_udp-proxy"}, Transport: selected} | |
| 402 | 402 | } | |
| 403 | 403 | return roleCommand{} | |
| 404 | 404 | } | |
| @@ -396,9 +396,9 @@ func routeRoleArgs(s scenario, m processMapping, selected transport) roleCommand | |||||
| 396 | return roleCommand{Args: remote} | 396 | return roleCommand{Args: remote} | ||
| 397 | } | 397 | } | ||
| 398 | case "ssh_stdio_peer": | 398 | case "ssh_stdio_peer": | ||
| 399 | - | return roleCommand{Args: []string{"_stdio", session}, Transport: transport{ID: s.Transport, Kind: "ssh_stdio"}} | 399 | + | return roleCommand{Args: []string{"_stdio"}, Transport: transport{ID: s.Transport, Kind: "ssh_stdio"}} |
| 400 | case "udp_peer": | 400 | case "udp_peer": | ||
| 401 | - | return roleCommand{Args: []string{"_udp-proxy", session}, Transport: selected} | 401 | + | return roleCommand{Args: []string{"_udp-proxy"}, Transport: selected} |
| 402 | } | 402 | } | ||
| 403 | return roleCommand{} | 403 | return roleCommand{} | ||
| 404 | } | 404 | } | ||
M cmd/vev-perf-harness/runner_test.go +2 -2
| @@ -256,10 +256,10 @@ func TestHarnessRoutesEveryRemoteFixtureThroughItsDeclaredPeer(t *testing.T) { | |||
| 256 | 256 | if !equalStrings(client.Args, []string{"attach", "harness@127.0.0.1"}) { | |
| 257 | 257 | t.Fatalf("client did not use ephemeral public remote attach: %q", client.Args) | |
| 258 | 258 | } | |
| 259 | - | if tc.peer == "ssh_stdio_peer" && !equalStrings(peer.Args, []string{"_stdio", "perf-route-002"}) { | |
| 259 | + | if tc.peer == "ssh_stdio_peer" && !equalStrings(peer.Args, []string{"_stdio"}) { | |
| 260 | 260 | t.Fatalf("ssh peer command=%q", peer.Args) | |
| 261 | 261 | } | |
| 262 | - | if tc.peer == "udp_peer" && !equalStrings(peer.Args, []string{"_udp-proxy", "perf-route-002"}) { | |
| 262 | + | if tc.peer == "udp_peer" && !equalStrings(peer.Args, []string{"_udp-proxy"}) { | |
| 263 | 263 | t.Fatalf("udp peer command=%q", peer.Args) | |
| 264 | 264 | } | |
| 265 | 265 | if peer.Transport.RTTMS != tc.wantRTT || peer.Transport.LossPercent != tc.wantLoss { | |
| @@ -256,10 +256,10 @@ func TestHarnessRoutesEveryRemoteFixtureThroughItsDeclaredPeer(t *testing.T) { | |||||
| 256 | if !equalStrings(client.Args, []string{"attach", "harness@127.0.0.1"}) { | 256 | if !equalStrings(client.Args, []string{"attach", "harness@127.0.0.1"}) { | ||
| 257 | t.Fatalf("client did not use ephemeral public remote attach: %q", client.Args) | 257 | t.Fatalf("client did not use ephemeral public remote attach: %q", client.Args) | ||
| 258 | } | 258 | } | ||
| 259 | - | if tc.peer == "ssh_stdio_peer" && !equalStrings(peer.Args, []string{"_stdio", "perf-route-002"}) { | 259 | + | if tc.peer == "ssh_stdio_peer" && !equalStrings(peer.Args, []string{"_stdio"}) { |
| 260 | t.Fatalf("ssh peer command=%q", peer.Args) | 260 | t.Fatalf("ssh peer command=%q", peer.Args) | ||
| 261 | } | 261 | } | ||
| 262 | - | if tc.peer == "udp_peer" && !equalStrings(peer.Args, []string{"_udp-proxy", "perf-route-002"}) { | 262 | + | if tc.peer == "udp_peer" && !equalStrings(peer.Args, []string{"_udp-proxy"}) { |
| 263 | t.Fatalf("udp peer command=%q", peer.Args) | 263 | t.Fatalf("udp peer command=%q", peer.Args) | ||
| 264 | } | 264 | } | ||
| 265 | if peer.Transport.RTTMS != tc.wantRTT || peer.Transport.LossPercent != tc.wantLoss { | 265 | if peer.Transport.RTTMS != tc.wantRTT || peer.Transport.LossPercent != tc.wantLoss { | ||
M internal/adapters/dgram/ssh_dialer.go +6 -7
| @@ -39,7 +39,7 @@ func (p execBootstrapProcess) Kill() error { return p.cmd | |||
| 39 | 39 | func (p execBootstrapProcess) Wait() error { return p.cmd.Wait() } | |
| 40 | 40 | ||
| 41 | 41 | var ( | |
| 42 | - | startUDPBootstrap = func(ctx context.Context, target, session string, stderr io.Writer) bootstrapProcess { | |
| 42 | + | startUDPBootstrap = func(ctx context.Context, target string, stderr io.Writer) bootstrapProcess { | |
| 43 | 43 | spec := sshstdio.BuildCommandForMode(target, "_udp-bootstrap", "") | |
| 44 | 44 | cmd := exec.CommandContext(ctx, spec.Path, spec.Args...) | |
| 45 | 45 | cmd.Stderr = stderr | |
| @@ -98,19 +98,18 @@ func (b *limitedBuffer) String() string { return string(b.buf) } | |||
| 98 | 98 | // attach. SSH stdio is selected only through the explicit stdio transport mode. | |
| 99 | 99 | type RemoteDialer struct { | |
| 100 | 100 | Target string | |
| 101 | - | Session string | |
| 102 | 101 | BootstrapTimeout time.Duration | |
| 103 | 102 | ProbeTimeout time.Duration | |
| 104 | 103 | Log *slog.Logger | |
| 105 | 104 | RuntimeObserver ports.SerializedRuntimeObserver | |
| 106 | 105 | } | |
| 107 | 106 | ||
| 108 | - | func NewRemoteDialer(target, session string) RemoteDialer { | |
| 109 | - | return RemoteDialer{Target: target, Session: session, ProbeTimeout: defaultProbeTimeout} | |
| 107 | + | func NewRemoteDialer(target, _ string) RemoteDialer { | |
| 108 | + | return RemoteDialer{Target: target, ProbeTimeout: defaultProbeTimeout} | |
| 110 | 109 | } | |
| 111 | 110 | ||
| 112 | - | func NewRemoteDialerWithLogger(target, session string, log *slog.Logger) RemoteDialer { | |
| 113 | - | return RemoteDialer{Target: target, Session: session, ProbeTimeout: defaultProbeTimeout, Log: log} | |
| 111 | + | func NewRemoteDialerWithLogger(target, _ string, log *slog.Logger) RemoteDialer { | |
| 112 | + | return RemoteDialer{Target: target, ProbeTimeout: defaultProbeTimeout, Log: log} | |
| 114 | 113 | } | |
| 115 | 114 | ||
| 116 | 115 | func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) { | |
| @@ -129,7 +128,7 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) { | |||
| 129 | 128 | defer bootstrapCancel() | |
| 130 | 129 | ||
| 131 | 130 | var stderr limitedBuffer | |
| 132 | - | proc := startUDPBootstrap(bootstrapCtx, d.Target, "", &stderr) | |
| 131 | + | proc := startUDPBootstrap(bootstrapCtx, d.Target, &stderr) | |
| 133 | 132 | stdout, err := proc.StdoutPipe() | |
| 134 | 133 | if err != nil { | |
| 135 | 134 | return nil, udpUnavailable("bootstrap stdout", err, &stderr) | |
| @@ -39,7 +39,7 @@ func (p execBootstrapProcess) Kill() error { return p.cmd | |||||
| 39 | func (p execBootstrapProcess) Wait() error { return p.cmd.Wait() } | 39 | func (p execBootstrapProcess) Wait() error { return p.cmd.Wait() } | ||
| 40 | 40 | ||||
| 41 | var ( | 41 | var ( | ||
| 42 | - | startUDPBootstrap = func(ctx context.Context, target, session string, stderr io.Writer) bootstrapProcess { | 42 | + | startUDPBootstrap = func(ctx context.Context, target string, stderr io.Writer) bootstrapProcess { |
| 43 | spec := sshstdio.BuildCommandForMode(target, "_udp-bootstrap", "") | 43 | spec := sshstdio.BuildCommandForMode(target, "_udp-bootstrap", "") | ||
| 44 | cmd := exec.CommandContext(ctx, spec.Path, spec.Args...) | 44 | cmd := exec.CommandContext(ctx, spec.Path, spec.Args...) | ||
| 45 | cmd.Stderr = stderr | 45 | cmd.Stderr = stderr | ||
| @@ -98,19 +98,18 @@ func (b *limitedBuffer) String() string { return string(b.buf) } | |||||
| 98 | // attach. SSH stdio is selected only through the explicit stdio transport mode. | 98 | // attach. SSH stdio is selected only through the explicit stdio transport mode. | ||
| 99 | type RemoteDialer struct { | 99 | type RemoteDialer struct { | ||
| 100 | Target string | 100 | Target string | ||
| 101 | - | Session string | |||
| 102 | BootstrapTimeout time.Duration | 101 | BootstrapTimeout time.Duration | ||
| 103 | ProbeTimeout time.Duration | 102 | ProbeTimeout time.Duration | ||
| 104 | Log *slog.Logger | 103 | Log *slog.Logger | ||
| 105 | RuntimeObserver ports.SerializedRuntimeObserver | 104 | RuntimeObserver ports.SerializedRuntimeObserver | ||
| 106 | } | 105 | } | ||
| 107 | 106 | ||||
| 108 | - | func NewRemoteDialer(target, session string) RemoteDialer { | 107 | + | func NewRemoteDialer(target, _ string) RemoteDialer { |
| 109 | - | return RemoteDialer{Target: target, Session: session, ProbeTimeout: defaultProbeTimeout} | 108 | + | return RemoteDialer{Target: target, ProbeTimeout: defaultProbeTimeout} |
| 110 | } | 109 | } | ||
| 111 | 110 | ||||
| 112 | - | func NewRemoteDialerWithLogger(target, session string, log *slog.Logger) RemoteDialer { | 111 | + | func NewRemoteDialerWithLogger(target, _ string, log *slog.Logger) RemoteDialer { |
| 113 | - | return RemoteDialer{Target: target, Session: session, ProbeTimeout: defaultProbeTimeout, Log: log} | 112 | + | return RemoteDialer{Target: target, ProbeTimeout: defaultProbeTimeout, Log: log} |
| 114 | } | 113 | } | ||
| 115 | 114 | ||||
| 116 | func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) { | 115 | func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) { | ||
| @@ -129,7 +128,7 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) { | |||||
| 129 | defer bootstrapCancel() | 128 | defer bootstrapCancel() | ||
| 130 | 129 | ||||
| 131 | var stderr limitedBuffer | 130 | var stderr limitedBuffer | ||
| 132 | - | proc := startUDPBootstrap(bootstrapCtx, d.Target, "", &stderr) | 131 | + | proc := startUDPBootstrap(bootstrapCtx, d.Target, &stderr) |
| 133 | stdout, err := proc.StdoutPipe() | 132 | stdout, err := proc.StdoutPipe() | ||
| 134 | if err != nil { | 133 | if err != nil { | ||
| 135 | return nil, udpUnavailable("bootstrap stdout", err, &stderr) | 134 | return nil, udpUnavailable("bootstrap stdout", err, &stderr) | ||
M internal/adapters/dgram/ssh_dialer_test.go +5 -5
| @@ -33,7 +33,7 @@ func (p *fakeBootstrapProcess) Start() error { return p.st | |||
| 33 | 33 | func (p *fakeBootstrapProcess) Kill() error { p.killed = true; return nil } | |
| 34 | 34 | func (p *fakeBootstrapProcess) Wait() error { p.waited = true; return p.waitErr } | |
| 35 | 35 | ||
| 36 | - | func withBootstrapStarter(t *testing.T, fn func(context.Context, string, string, io.Writer) bootstrapProcess) { | |
| 36 | + | func withBootstrapStarter(t *testing.T, fn func(context.Context, string, io.Writer) bootstrapProcess) { | |
| 37 | 37 | t.Helper() | |
| 38 | 38 | old := startUDPBootstrap | |
| 39 | 39 | startUDPBootstrap = fn | |
| @@ -71,7 +71,7 @@ func TestRemoteDialerUDPBootstrapFailures(t *testing.T) { | |||
| 71 | 71 | if tt.name == "bootstrap wait failure" { | |
| 72 | 72 | proc.waitErr = errors.New("exit status 255") | |
| 73 | 73 | } | |
| 74 | - | withBootstrapStarter(t, func(context.Context, string, string, io.Writer) bootstrapProcess { return proc }) | |
| 74 | + | withBootstrapStarter(t, func(context.Context, string, io.Writer) bootstrapProcess { return proc }) | |
| 75 | 75 | if tt.listen != nil { | |
| 76 | 76 | withListenUDP(t, func(context.Context) (net.PacketConn, error) { return nil, tt.listen }) | |
| 77 | 77 | } | |
| @@ -120,9 +120,9 @@ func TestListenUDPPacketUsesDualStackWildcard(t *testing.T) { | |||
| 120 | 120 | func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) { | |
| 121 | 121 | key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize)) | |
| 122 | 122 | proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))} | |
| 123 | - | withBootstrapStarter(t, func(_ context.Context, target, session string, _ io.Writer) bootstrapProcess { | |
| 124 | - | if target != "127.0.0.1" || session != "" { | |
| 125 | - | t.Fatalf("bootstrap target/session = %q/%q", target, session) | |
| 123 | + | withBootstrapStarter(t, func(_ context.Context, target string, _ io.Writer) bootstrapProcess { | |
| 124 | + | if target != "127.0.0.1" { | |
| 125 | + | t.Fatalf("bootstrap target = %q", target) | |
| 126 | 126 | } | |
| 127 | 127 | return proc | |
| 128 | 128 | }) | |
| @@ -33,7 +33,7 @@ func (p *fakeBootstrapProcess) Start() error { return p.st | |||||
| 33 | func (p *fakeBootstrapProcess) Kill() error { p.killed = true; return nil } | 33 | func (p *fakeBootstrapProcess) Kill() error { p.killed = true; return nil } | ||
| 34 | func (p *fakeBootstrapProcess) Wait() error { p.waited = true; return p.waitErr } | 34 | func (p *fakeBootstrapProcess) Wait() error { p.waited = true; return p.waitErr } | ||
| 35 | 35 | ||||
| 36 | - | func withBootstrapStarter(t *testing.T, fn func(context.Context, string, string, io.Writer) bootstrapProcess) { | 36 | + | func withBootstrapStarter(t *testing.T, fn func(context.Context, string, io.Writer) bootstrapProcess) { |
| 37 | t.Helper() | 37 | t.Helper() | ||
| 38 | old := startUDPBootstrap | 38 | old := startUDPBootstrap | ||
| 39 | startUDPBootstrap = fn | 39 | startUDPBootstrap = fn | ||
| @@ -71,7 +71,7 @@ func TestRemoteDialerUDPBootstrapFailures(t *testing.T) { | |||||
| 71 | if tt.name == "bootstrap wait failure" { | 71 | if tt.name == "bootstrap wait failure" { | ||
| 72 | proc.waitErr = errors.New("exit status 255") | 72 | proc.waitErr = errors.New("exit status 255") | ||
| 73 | } | 73 | } | ||
| 74 | - | withBootstrapStarter(t, func(context.Context, string, string, io.Writer) bootstrapProcess { return proc }) | 74 | + | withBootstrapStarter(t, func(context.Context, string, io.Writer) bootstrapProcess { return proc }) |
| 75 | if tt.listen != nil { | 75 | if tt.listen != nil { | ||
| 76 | withListenUDP(t, func(context.Context) (net.PacketConn, error) { return nil, tt.listen }) | 76 | withListenUDP(t, func(context.Context) (net.PacketConn, error) { return nil, tt.listen }) | ||
| 77 | } | 77 | } | ||
| @@ -120,9 +120,9 @@ func TestListenUDPPacketUsesDualStackWildcard(t *testing.T) { | |||||
| 120 | func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) { | 120 | func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) { | ||
| 121 | key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize)) | 121 | key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize)) | ||
| 122 | proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))} | 122 | proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))} | ||
| 123 | - | withBootstrapStarter(t, func(_ context.Context, target, session string, _ io.Writer) bootstrapProcess { | 123 | + | withBootstrapStarter(t, func(_ context.Context, target string, _ io.Writer) bootstrapProcess { |
| 124 | - | if target != "127.0.0.1" || session != "" { | 124 | + | if target != "127.0.0.1" { |
| 125 | - | t.Fatalf("bootstrap target/session = %q/%q", target, session) | 125 | + | t.Fatalf("bootstrap target = %q", target) |
| 126 | } | 126 | } | ||
| 127 | return proc | 127 | return proc | ||
| 128 | }) | 128 | }) | ||
M internal/adapters/remote/remote_factory.go +0 -1
| @@ -46,7 +46,6 @@ func (f DialerFactory) DialerForRemote(target string, session string, mode ports | |||
| 46 | 46 | ||
| 47 | 47 | type stdioDialer struct { | |
| 48 | 48 | target string | |
| 49 | - | session string | |
| 50 | 49 | log *slog.Logger | |
| 51 | 50 | observer ports.SerializedRuntimeObserver | |
| 52 | 51 | } | |
| @@ -46,7 +46,6 @@ func (f DialerFactory) DialerForRemote(target string, session string, mode ports | |||||
| 46 | 46 | ||||
| 47 | type stdioDialer struct { | 47 | type stdioDialer struct { | ||
| 48 | target string | 48 | target string | ||
| 49 | - | session string | |||
| 50 | log *slog.Logger | 49 | log *slog.Logger | ||
| 51 | observer ports.SerializedRuntimeObserver | 50 | observer ports.SerializedRuntimeObserver | ||
| 52 | } | 51 | } | ||
M internal/adapters/remote/remote_factory_test.go +5 -5
| @@ -32,16 +32,16 @@ func TestDialerFactorySelectsExplicitModes(t *testing.T) { | |||
| 32 | 32 | if !ok { | |
| 33 | 33 | t.Fatalf("dialer type = %T, want %T", dialer, dgram.RemoteDialer{}) | |
| 34 | 34 | } | |
| 35 | - | if got.Target != "remote.example" || got.Session != "" { | |
| 36 | - | t.Fatalf("dialer = %+v, want target and no session metadata", got) | |
| 35 | + | if got.Target != "remote.example" { | |
| 36 | + | t.Fatalf("dialer target = %q, want %q", got.Target, "remote.example") | |
| 37 | 37 | } | |
| 38 | 38 | case stdioDialer: | |
| 39 | 39 | got, ok := dialer.(stdioDialer) | |
| 40 | 40 | if !ok { | |
| 41 | 41 | t.Fatalf("dialer type = %T, want %T", dialer, stdioDialer{}) | |
| 42 | 42 | } | |
| 43 | - | if got.target != "remote.example" || got.session != "" { | |
| 44 | - | t.Fatalf("dialer = %+v, want target and no session metadata", got) | |
| 43 | + | if got.target != "remote.example" { | |
| 44 | + | t.Fatalf("dialer target = %q, want %q", got.target, "remote.example") | |
| 45 | 45 | } | |
| 46 | 46 | } | |
| 47 | 47 | }) | |
| @@ -64,7 +64,7 @@ func TestDialerFactoryRejectsUnsupportedMode(t *testing.T) { | |||
| 64 | 64 | } | |
| 65 | 65 | ||
| 66 | 66 | func TestStdioDialerUsesContextBeforeStartingSSH(t *testing.T) { | |
| 67 | - | dialer := stdioDialer{target: "remote.example", session: "work"} | |
| 67 | + | dialer := stdioDialer{target: "remote.example"} | |
| 68 | 68 | ctx, cancel := context.WithCancel(context.Background()) | |
| 69 | 69 | cancel() | |
| 70 | 70 | ||
| @@ -32,16 +32,16 @@ func TestDialerFactorySelectsExplicitModes(t *testing.T) { | |||||
| 32 | if !ok { | 32 | if !ok { | ||
| 33 | t.Fatalf("dialer type = %T, want %T", dialer, dgram.RemoteDialer{}) | 33 | t.Fatalf("dialer type = %T, want %T", dialer, dgram.RemoteDialer{}) | ||
| 34 | } | 34 | } | ||
| 35 | - | if got.Target != "remote.example" || got.Session != "" { | 35 | + | if got.Target != "remote.example" { |
| 36 | - | t.Fatalf("dialer = %+v, want target and no session metadata", got) | 36 | + | t.Fatalf("dialer target = %q, want %q", got.Target, "remote.example") |
| 37 | } | 37 | } | ||
| 38 | case stdioDialer: | 38 | case stdioDialer: | ||
| 39 | got, ok := dialer.(stdioDialer) | 39 | got, ok := dialer.(stdioDialer) | ||
| 40 | if !ok { | 40 | if !ok { | ||
| 41 | t.Fatalf("dialer type = %T, want %T", dialer, stdioDialer{}) | 41 | t.Fatalf("dialer type = %T, want %T", dialer, stdioDialer{}) | ||
| 42 | } | 42 | } | ||
| 43 | - | if got.target != "remote.example" || got.session != "" { | 43 | + | if got.target != "remote.example" { |
| 44 | - | t.Fatalf("dialer = %+v, want target and no session metadata", got) | 44 | + | t.Fatalf("dialer target = %q, want %q", got.target, "remote.example") |
| 45 | } | 45 | } | ||
| 46 | } | 46 | } | ||
| 47 | }) | 47 | }) | ||
| @@ -64,7 +64,7 @@ func TestDialerFactoryRejectsUnsupportedMode(t *testing.T) { | |||||
| 64 | } | 64 | } | ||
| 65 | 65 | ||||
| 66 | func TestStdioDialerUsesContextBeforeStartingSSH(t *testing.T) { | 66 | func TestStdioDialerUsesContextBeforeStartingSSH(t *testing.T) { | ||
| 67 | - | dialer := stdioDialer{target: "remote.example", session: "work"} | 67 | + | dialer := stdioDialer{target: "remote.example"} |
| 68 | ctx, cancel := context.WithCancel(context.Background()) | 68 | ctx, cancel := context.WithCancel(context.Background()) | ||
| 69 | cancel() | 69 | cancel() | ||
| 70 | 70 | ||||
M internal/adapters/sshstdio/transport.go +3 -3
| @@ -332,9 +332,9 @@ func shellQuote(s string) string { | |||
| 332 | 332 | return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" | |
| 333 | 333 | } | |
| 334 | 334 | ||
| 335 | - | // Dial starts ssh target vev _stdio [session] and returns a Transport over the | |
| 336 | - | // child process' stdio. The subprocess is started with exec.Command argv, never | |
| 337 | - | // through a shell. | |
| 335 | + | // Dial starts ssh target vev _stdio and returns a Transport over the child | |
| 336 | + | // process' stdio. Session selection is carried in Hello. The subprocess is | |
| 337 | + | // started with exec.Command argv, never through a shell. | |
| 338 | 338 | func Dial(target, session string) (ports.Transport, error) { | |
| 339 | 339 | return DialContext(context.Background(), target, session) | |
| 340 | 340 | } | |
| @@ -332,9 +332,9 @@ func shellQuote(s string) string { | |||||
| 332 | return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" | 332 | return "'" + strings.ReplaceAll(s, "'", "'\\''") + "'" | ||
| 333 | } | 333 | } | ||
| 334 | 334 | ||||
| 335 | - | // Dial starts ssh target vev _stdio [session] and returns a Transport over the | 335 | + | // Dial starts ssh target vev _stdio and returns a Transport over the child |
| 336 | - | // child process' stdio. The subprocess is started with exec.Command argv, never | 336 | + | // process' stdio. Session selection is carried in Hello. The subprocess is |
| 337 | - | // through a shell. | 337 | + | // started with exec.Command argv, never through a shell. |
| 338 | func Dial(target, session string) (ports.Transport, error) { | 338 | func Dial(target, session string) (ports.Transport, error) { | ||
| 339 | return DialContext(context.Background(), target, session) | 339 | return DialContext(context.Background(), target, session) | ||
| 340 | } | 340 | } | ||
M internal/adapters/sshstdio/transport_test.go +3 -2
| @@ -18,8 +18,9 @@ import ( | |||
| 18 | 18 | ) | |
| 19 | 19 | ||
| 20 | 20 | func TestBuildCommandUsesExecArgs(t *testing.T) { | |
| 21 | - | // BuildCommand deliberately starts `ssh -- target 'vev' '_stdio' [session]`. | |
| 22 | - | // Terminal color capability is carried in vev's Hello message, not in ssh env. | |
| 21 | + | // BuildCommand deliberately starts `ssh -- target 'vev' '_stdio'`. | |
| 22 | + | // Session selection and terminal color capability are carried in vev's Hello | |
| 23 | + | // message, not in ssh env. | |
| 23 | 24 | tests := []struct { | |
| 24 | 25 | name string | |
| 25 | 26 | target string | |
| @@ -18,8 +18,9 @@ import ( | |||||
| 18 | ) | 18 | ) | ||
| 19 | 19 | ||||
| 20 | func TestBuildCommandUsesExecArgs(t *testing.T) { | 20 | func TestBuildCommandUsesExecArgs(t *testing.T) { | ||
| 21 | - | // BuildCommand deliberately starts `ssh -- target 'vev' '_stdio' [session]`. | 21 | + | // BuildCommand deliberately starts `ssh -- target 'vev' '_stdio'`. |
| 22 | - | // Terminal color capability is carried in vev's Hello message, not in ssh env. | 22 | + | // Session selection and terminal color capability are carried in vev's Hello |
| 23 | + | // message, not in ssh env. | |||
| 23 | tests := []struct { | 24 | tests := []struct { | ||
| 24 | name string | 25 | name string | ||
| 25 | target string | 26 | target string | ||
M internal/app/cmd.go +1 -0
| @@ -200,6 +200,7 @@ func runCmdWithDeps(ctx context.Context, invocation cmdInvocation, deps cmdDeps) | |||
| 200 | 200 | tracker.Fail(requestID, generation, fmt.Errorf("decoding command reply: %w", decodeErr)) | |
| 201 | 201 | return | |
| 202 | 202 | } | |
| 203 | + | result.RequestID = requestID | |
| 203 | 204 | tracker.Complete(generation, result) | |
| 204 | 205 | }() | |
| 205 | 206 | commandClock := deps.clock | |
| @@ -200,6 +200,7 @@ func runCmdWithDeps(ctx context.Context, invocation cmdInvocation, deps cmdDeps) | |||||
| 200 | tracker.Fail(requestID, generation, fmt.Errorf("decoding command reply: %w", decodeErr)) | 200 | tracker.Fail(requestID, generation, fmt.Errorf("decoding command reply: %w", decodeErr)) | ||
| 201 | return | 201 | return | ||
| 202 | } | 202 | } | ||
| 203 | + | result.RequestID = requestID | |||
| 203 | tracker.Complete(generation, result) | 204 | tracker.Complete(generation, result) | ||
| 204 | }() | 205 | }() | ||
| 205 | commandClock := deps.clock | 206 | commandClock := deps.clock | ||
M internal/app/cmd_test.go +30 -26
| @@ -12,7 +12,9 @@ import ( | |||
| 12 | 12 | "time" | |
| 13 | 13 | ||
| 14 | 14 | "github.com/bnema/vev/internal/ports" | |
| 15 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |
| 15 | 16 | "github.com/bnema/vev/internal/usecase/daemon" | |
| 17 | + | "github.com/stretchr/testify/mock" | |
| 16 | 18 | "github.com/stretchr/testify/require" | |
| 17 | 19 | ) | |
| 18 | 20 | ||
| @@ -290,8 +292,15 @@ func TestRunCmdRejectsTooManyArgumentsBeforeDialing(t *testing.T) { | |||
| 290 | 292 | } | |
| 291 | 293 | ||
| 292 | 294 | func TestRunCmdTimesOutPendingRequest(t *testing.T) { | |
| 293 | - | timer := &cmdTestTimer{ch: make(chan time.Time, 1)} | |
| 294 | - | clock := &cmdTestClock{timer: timer} | |
| 295 | + | timerCh := make(chan time.Time, 1) | |
| 296 | + | timer := portsmocks.NewMockTimer(t) | |
| 297 | + | timer.EXPECT().C().Return(timerCh).Once() | |
| 298 | + | timer.EXPECT().Stop().Return(true).Once() | |
| 299 | + | delayCh := make(chan time.Duration, 1) | |
| 300 | + | clock := portsmocks.NewMockClock(t) | |
| 301 | + | clock.EXPECT().NewTimer(mock.Anything).Run(func(delay time.Duration) { | |
| 302 | + | delayCh <- delay | |
| 303 | + | }).Return(timer).Once() | |
| 295 | 304 | recvStarted := make(chan struct{}) | |
| 296 | 305 | recvGate := make(chan struct{}) | |
| 297 | 306 | transport := &cmdTestTransport{recvStarted: recvStarted, recvGate: recvGate} | |
| @@ -305,7 +314,10 @@ func TestRunCmdTimesOutPendingRequest(t *testing.T) { | |||
| 305 | 314 | }) | |
| 306 | 315 | }() | |
| 307 | 316 | <-recvStarted | |
| 308 | - | timer.ch <- time.Time{} | |
| 317 | + | if delay := <-delayCh; delay != daemon.CommandRequestTimeout { | |
| 318 | + | t.Fatalf("command timer delay = %s, want %s", delay, daemon.CommandRequestTimeout) | |
| 319 | + | } | |
| 320 | + | timerCh <- time.Time{} | |
| 309 | 321 | err := <-done | |
| 310 | 322 | close(recvGate) | |
| 311 | 323 | if ExitCode(err) != 3 || !errors.Is(err, daemon.ErrCommandRequestTimeout) { | |
| @@ -317,20 +329,22 @@ func TestRunCmdTimesOutPendingRequest(t *testing.T) { | |||
| 317 | 329 | } | |
| 318 | 330 | } | |
| 319 | 331 | ||
| 320 | - | type cmdTestTimer struct{ ch chan time.Time } | |
| 321 | - | ||
| 322 | - | func (t *cmdTestTimer) C() <-chan time.Time { return t.ch } | |
| 323 | - | func (t *cmdTestTimer) Reset(time.Duration) bool { return false } | |
| 324 | - | func (t *cmdTestTimer) Stop() bool { return true } | |
| 325 | - | ||
| 326 | - | type cmdTestClock struct{ timer *cmdTestTimer } | |
| 327 | - | ||
| 328 | - | func (c *cmdTestClock) Now() time.Time { return time.Time{} } | |
| 329 | - | func (c *cmdTestClock) NewTimer(delay time.Duration) ports.Timer { | |
| 330 | - | if delay != daemon.CommandRequestTimeout { | |
| 331 | - | panic("unexpected command timeout") | |
| 332 | + | func TestRunCmdZeroRequestIDReplyReturnsDaemonError(t *testing.T) { | |
| 333 | + | const want = "daemon error" | |
| 334 | + | transport := &cmdTestTransport{recv: ports.Frame{Type: ports.MsgCommandResult, Payload: ports.MarshalCommandResult(ports.CommandResult{ | |
| 335 | + | Code: ports.ErrInternal, Text: want, | |
| 336 | + | })}} | |
| 337 | + | err := runCmdWithDeps(context.Background(), cmdInvocation{slug: "split-right"}, cmdDeps{ | |
| 338 | + | stdout: io.Discard, | |
| 339 | + | getenv: func(string) string { return "" }, | |
| 340 | + | dial: func(context.Context, string) (ports.Transport, error) { return transport, nil }, | |
| 341 | + | }) | |
| 342 | + | if err == nil || err.Error() != want { | |
| 343 | + | t.Fatalf("error = %v, want daemon error %q", err, want) | |
| 344 | + | } | |
| 345 | + | if errors.Is(err, daemon.ErrCommandRequestTimeout) { | |
| 346 | + | t.Fatalf("error = %v, want daemon error instead of timeout", err) | |
| 332 | 347 | } | |
| 333 | - | return c.timer | |
| 334 | 348 | } | |
| 335 | 349 | ||
| 336 | 350 | func TestRunCmdDoesNotAutostartAndClassifiesDialFailure(t *testing.T) { | |
| @@ -417,16 +431,6 @@ func (t *cmdTestTransport) Recv() (ports.Frame, error) { | |||
| 417 | 431 | if t.recvGate != nil { | |
| 418 | 432 | <-t.recvGate | |
| 419 | 433 | } | |
| 420 | - | if t.recv.Type == ports.MsgCommandResult && len(t.sent) != 0 { | |
| 421 | - | result, err := ports.UnmarshalCommandResult(t.recv.Payload) | |
| 422 | - | if err == nil && result.RequestID == 0 { | |
| 423 | - | request, err := ports.UnmarshalCommandRequest(t.sent[0].Payload) | |
| 424 | - | if err == nil { | |
| 425 | - | result.RequestID = request.RequestID | |
| 426 | - | return ports.Frame{Type: ports.MsgCommandResult, Payload: ports.MarshalCommandResult(result)}, t.recvErr | |
| 427 | - | } | |
| 428 | - | } | |
| 429 | - | } | |
| 430 | 434 | return t.recv, t.recvErr | |
| 431 | 435 | } | |
| 432 | 436 | ||
| @@ -12,7 +12,9 @@ import ( | |||||
| 12 | "time" | 12 | "time" | ||
| 13 | 13 | ||||
| 14 | "github.com/bnema/vev/internal/ports" | 14 | "github.com/bnema/vev/internal/ports" | ||
| 15 | + | portsmocks "github.com/bnema/vev/internal/ports/mocks" | |||
| 15 | "github.com/bnema/vev/internal/usecase/daemon" | 16 | "github.com/bnema/vev/internal/usecase/daemon" | ||
| 17 | + | "github.com/stretchr/testify/mock" | |||
| 16 | "github.com/stretchr/testify/require" | 18 | "github.com/stretchr/testify/require" | ||
| 17 | ) | 19 | ) | ||
| 18 | 20 | ||||
| @@ -290,8 +292,15 @@ func TestRunCmdRejectsTooManyArgumentsBeforeDialing(t *testing.T) { | |||||
| 290 | } | 292 | } | ||
| 291 | 293 | ||||
| 292 | func TestRunCmdTimesOutPendingRequest(t *testing.T) { | 294 | func TestRunCmdTimesOutPendingRequest(t *testing.T) { | ||
| 293 | - | timer := &cmdTestTimer{ch: make(chan time.Time, 1)} | 295 | + | timerCh := make(chan time.Time, 1) |
| 294 | - | clock := &cmdTestClock{timer: timer} | 296 | + | timer := portsmocks.NewMockTimer(t) |
| 297 | + | timer.EXPECT().C().Return(timerCh).Once() | |||
| 298 | + | timer.EXPECT().Stop().Return(true).Once() | |||
| 299 | + | delayCh := make(chan time.Duration, 1) | |||
| 300 | + | clock := portsmocks.NewMockClock(t) | |||
| 301 | + | clock.EXPECT().NewTimer(mock.Anything).Run(func(delay time.Duration) { | |||
| 302 | + | delayCh <- delay | |||
| 303 | + | }).Return(timer).Once() | |||
| 295 | recvStarted := make(chan struct{}) | 304 | recvStarted := make(chan struct{}) | ||
| 296 | recvGate := make(chan struct{}) | 305 | recvGate := make(chan struct{}) | ||
| 297 | transport := &cmdTestTransport{recvStarted: recvStarted, recvGate: recvGate} | 306 | transport := &cmdTestTransport{recvStarted: recvStarted, recvGate: recvGate} | ||
| @@ -305,7 +314,10 @@ func TestRunCmdTimesOutPendingRequest(t *testing.T) { | |||||
| 305 | }) | 314 | }) | ||
| 306 | }() | 315 | }() | ||
| 307 | <-recvStarted | 316 | <-recvStarted | ||
| 308 | - | timer.ch <- time.Time{} | 317 | + | if delay := <-delayCh; delay != daemon.CommandRequestTimeout { |
| 318 | + | t.Fatalf("command timer delay = %s, want %s", delay, daemon.CommandRequestTimeout) | |||
| 319 | + | } | |||
| 320 | + | timerCh <- time.Time{} | |||
| 309 | err := <-done | 321 | err := <-done | ||
| 310 | close(recvGate) | 322 | close(recvGate) | ||
| 311 | if ExitCode(err) != 3 || !errors.Is(err, daemon.ErrCommandRequestTimeout) { | 323 | if ExitCode(err) != 3 || !errors.Is(err, daemon.ErrCommandRequestTimeout) { | ||
| @@ -317,20 +329,22 @@ func TestRunCmdTimesOutPendingRequest(t *testing.T) { | |||||
| 317 | } | 329 | } | ||
| 318 | } | 330 | } | ||
| 319 | 331 | ||||
| 320 | - | type cmdTestTimer struct{ ch chan time.Time } | 332 | + | func TestRunCmdZeroRequestIDReplyReturnsDaemonError(t *testing.T) { |
| 321 | - | 333 | + | const want = "daemon error" | |
| 322 | - | func (t *cmdTestTimer) C() <-chan time.Time { return t.ch } | 334 | + | transport := &cmdTestTransport{recv: ports.Frame{Type: ports.MsgCommandResult, Payload: ports.MarshalCommandResult(ports.CommandResult{ |
| 323 | - | func (t *cmdTestTimer) Reset(time.Duration) bool { return false } | 335 | + | Code: ports.ErrInternal, Text: want, |
| 324 | - | func (t *cmdTestTimer) Stop() bool { return true } | 336 | + | })}} |
| 325 | - | 337 | + | err := runCmdWithDeps(context.Background(), cmdInvocation{slug: "split-right"}, cmdDeps{ | |
| 326 | - | type cmdTestClock struct{ timer *cmdTestTimer } | 338 | + | stdout: io.Discard, |
| 327 | - | 339 | + | getenv: func(string) string { return "" }, | |
| 328 | - | func (c *cmdTestClock) Now() time.Time { return time.Time{} } | 340 | + | dial: func(context.Context, string) (ports.Transport, error) { return transport, nil }, |
| 329 | - | func (c *cmdTestClock) NewTimer(delay time.Duration) ports.Timer { | 341 | + | }) |
| 330 | - | if delay != daemon.CommandRequestTimeout { | 342 | + | if err == nil || err.Error() != want { |
| 331 | - | panic("unexpected command timeout") | 343 | + | t.Fatalf("error = %v, want daemon error %q", err, want) |
| 344 | + | } | |||
| 345 | + | if errors.Is(err, daemon.ErrCommandRequestTimeout) { | |||
| 346 | + | t.Fatalf("error = %v, want daemon error instead of timeout", err) | |||
| 332 | } | 347 | } | ||
| 333 | - | return c.timer | |||
| 334 | } | 348 | } | ||
| 335 | 349 | ||||
| 336 | func TestRunCmdDoesNotAutostartAndClassifiesDialFailure(t *testing.T) { | 350 | func TestRunCmdDoesNotAutostartAndClassifiesDialFailure(t *testing.T) { | ||
| @@ -417,16 +431,6 @@ func (t *cmdTestTransport) Recv() (ports.Frame, error) { | |||||
| 417 | if t.recvGate != nil { | 431 | if t.recvGate != nil { | ||
| 418 | <-t.recvGate | 432 | <-t.recvGate | ||
| 419 | } | 433 | } | ||
| 420 | - | if t.recv.Type == ports.MsgCommandResult && len(t.sent) != 0 { | |||
| 421 | - | result, err := ports.UnmarshalCommandResult(t.recv.Payload) | |||
| 422 | - | if err == nil && result.RequestID == 0 { | |||
| 423 | - | request, err := ports.UnmarshalCommandRequest(t.sent[0].Payload) | |||
| 424 | - | if err == nil { | |||
| 425 | - | result.RequestID = request.RequestID | |||
| 426 | - | return ports.Frame{Type: ports.MsgCommandResult, Payload: ports.MarshalCommandResult(result)}, t.recvErr | |||
| 427 | - | } | |||
| 428 | - | } | |||
| 429 | - | } | |||
| 430 | return t.recv, t.recvErr | 434 | return t.recv, t.recvErr | ||
| 431 | } | 435 | } | ||
| 432 | 436 | ||||
M internal/app/run.go +17 -28
| @@ -161,32 +161,20 @@ func parseArgs(args []string) (command, error) { | |||
| 161 | 161 | case "--daemon-launcher": | |
| 162 | 162 | return command{kind: kindDaemonLauncher}, nil | |
| 163 | 163 | case "_stdio": | |
| 164 | - | if len(args) > 2 { | |
| 165 | - | return command{}, usagef("`_stdio` accepts at most one session name") | |
| 166 | - | } | |
| 167 | - | cmd := command{kind: kindStdio} | |
| 168 | - | if len(args) == 2 { | |
| 169 | - | cmd.name = args[1] | |
| 164 | + | if len(args) != 1 { | |
| 165 | + | return command{}, usagef("`_stdio` does not accept a session name") | |
| 170 | 166 | } | |
| 171 | - | return cmd, nil | |
| 167 | + | return command{kind: kindStdio}, nil | |
| 172 | 168 | case "_udp-bootstrap": | |
| 173 | - | if len(args) > 2 { | |
| 174 | - | return command{}, usagef("`_udp-bootstrap` accepts at most one session name") | |
| 175 | - | } | |
| 176 | - | cmd := command{kind: kindUDPBootstrap} | |
| 177 | - | if len(args) == 2 { | |
| 178 | - | cmd.name = args[1] | |
| 169 | + | if len(args) != 1 { | |
| 170 | + | return command{}, usagef("`_udp-bootstrap` does not accept a session name") | |
| 179 | 171 | } | |
| 180 | - | return cmd, nil | |
| 172 | + | return command{kind: kindUDPBootstrap}, nil | |
| 181 | 173 | case "_udp-proxy": | |
| 182 | - | if len(args) > 2 { | |
| 183 | - | return command{}, usagef("`_udp-proxy` accepts at most one session name") | |
| 184 | - | } | |
| 185 | - | cmd := command{kind: kindUDPProxy} | |
| 186 | - | if len(args) == 2 { | |
| 187 | - | cmd.name = args[1] | |
| 174 | + | if len(args) != 1 { | |
| 175 | + | return command{}, usagef("`_udp-proxy` does not accept a session name") | |
| 188 | 176 | } | |
| 189 | - | return cmd, nil | |
| 177 | + | return command{kind: kindUDPProxy}, nil | |
| 190 | 178 | case "new": | |
| 191 | 179 | if len(args) < 2 || args[1] == "" { | |
| 192 | 180 | return command{}, usagef("`new` requires a session name") | |
| @@ -278,9 +266,9 @@ func dispatch(ctx context.Context, cmd command) error { | |||
| 278 | 266 | case kindStdio: | |
| 279 | 267 | return runStdio(ctx) | |
| 280 | 268 | case kindUDPBootstrap: | |
| 281 | - | return runUDPBootstrap(ctx, cmd.name) | |
| 269 | + | return runUDPBootstrap(ctx) | |
| 282 | 270 | case kindUDPProxy: | |
| 283 | - | return runUDPProxy(ctx, cmd.name, os.Stdout) | |
| 271 | + | return runUDPProxy(ctx, os.Stdout) | |
| 284 | 272 | case kindList: | |
| 285 | 273 | return runList(ctx, cmd) | |
| 286 | 274 | case kindHost: | |
| @@ -1056,21 +1044,22 @@ func runStdio(ctx context.Context) (retErr error) { | |||
| 1056 | 1044 | // pipe drained by a small forwarding goroutine. | |
| 1057 | 1045 | stdin, stdout := os.Stdin, os.Stdout | |
| 1058 | 1046 | stdoutReader, stdoutWriter := io.Pipe() | |
| 1047 | + | forwardDone := make(chan struct{}) | |
| 1059 | 1048 | go func() { | |
| 1049 | + | defer close(forwardDone) | |
| 1060 | 1050 | if _, err := io.Copy(stdout, stdoutReader); err != nil { | |
| 1061 | 1051 | _ = stdoutReader.CloseWithError(err) | |
| 1062 | 1052 | } | |
| 1063 | 1053 | }() | |
| 1064 | 1054 | defer func() { | |
| 1065 | 1055 | _ = stdoutWriter.Close() | |
| 1056 | + | <-forwardDone | |
| 1066 | 1057 | _ = stdoutReader.Close() | |
| 1067 | 1058 | }() | |
| 1068 | - | stdio := sshstdio.NewTransport(nonClosingReader{Reader: stdin}, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer)) | |
| 1059 | + | stdio := sshstdio.NewTransport(stdin, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer)) | |
| 1069 | 1060 | return runStdioProxy(ctx, stdio, transport, log) | |
| 1070 | 1061 | } | |
| 1071 | 1062 | ||
| 1072 | - | type nonClosingReader struct{ io.Reader } | |
| 1073 | - | ||
| 1074 | 1063 | var ( | |
| 1075 | 1064 | udpBootstrapTimeout = 5 * time.Second | |
| 1076 | 1065 | udpProxyCommand = func(ctx context.Context, exe string, args ...string) *exec.Cmd { | |
| @@ -1080,7 +1069,7 @@ var ( | |||
| 1080 | 1069 | ||
| 1081 | 1070 | // runUDPBootstrap starts a detached _udp-proxy, forwards its single readiness | |
| 1082 | 1071 | // line, and exits so SSH stdio can close without owning the UDP proxy lifetime. | |
| 1083 | - | func runUDPBootstrap(ctx context.Context, _ string) error { | |
| 1072 | + | func runUDPBootstrap(ctx context.Context) error { | |
| 1084 | 1073 | // Always start a fresh proxy for a bootstrap request. A client attach begins | |
| 1085 | 1074 | // with MsgHello, which must reach the daemon handshake path; reusing an | |
| 1086 | 1075 | // already-running proxy would forward that Hello into the daemon connection's | |
| @@ -1148,7 +1137,7 @@ func runUDPBootstrap(ctx context.Context, _ string) error { | |||
| 1148 | 1137 | } | |
| 1149 | 1138 | ||
| 1150 | 1139 | // runUDPProxy is the detached long-lived remote-side datagram proxy. | |
| 1151 | - | func runUDPProxy(ctx context.Context, _ string, ready io.Writer) (retErr error) { | |
| 1140 | + | func runUDPProxy(ctx context.Context, ready io.Writer) (retErr error) { | |
| 1152 | 1141 | log, logCloser, err := configureLogging(logging.Stdio, false) | |
| 1153 | 1142 | if err != nil { | |
| 1154 | 1143 | return err | |
| @@ -161,32 +161,20 @@ func parseArgs(args []string) (command, error) { | |||||
| 161 | case "--daemon-launcher": | 161 | case "--daemon-launcher": | ||
| 162 | return command{kind: kindDaemonLauncher}, nil | 162 | return command{kind: kindDaemonLauncher}, nil | ||
| 163 | case "_stdio": | 163 | case "_stdio": | ||
| 164 | - | if len(args) > 2 { | 164 | + | if len(args) != 1 { |
| 165 | - | return command{}, usagef("`_stdio` accepts at most one session name") | 165 | + | return command{}, usagef("`_stdio` does not accept a session name") |
| 166 | - | } | |||
| 167 | - | cmd := command{kind: kindStdio} | |||
| 168 | - | if len(args) == 2 { | |||
| 169 | - | cmd.name = args[1] | |||
| 170 | } | 166 | } | ||
| 171 | - | return cmd, nil | 167 | + | return command{kind: kindStdio}, nil |
| 172 | case "_udp-bootstrap": | 168 | case "_udp-bootstrap": | ||
| 173 | - | if len(args) > 2 { | 169 | + | if len(args) != 1 { |
| 174 | - | return command{}, usagef("`_udp-bootstrap` accepts at most one session name") | 170 | + | return command{}, usagef("`_udp-bootstrap` does not accept a session name") |
| 175 | - | } | |||
| 176 | - | cmd := command{kind: kindUDPBootstrap} | |||
| 177 | - | if len(args) == 2 { | |||
| 178 | - | cmd.name = args[1] | |||
| 179 | } | 171 | } | ||
| 180 | - | return cmd, nil | 172 | + | return command{kind: kindUDPBootstrap}, nil |
| 181 | case "_udp-proxy": | 173 | case "_udp-proxy": | ||
| 182 | - | if len(args) > 2 { | 174 | + | if len(args) != 1 { |
| 183 | - | return command{}, usagef("`_udp-proxy` accepts at most one session name") | 175 | + | return command{}, usagef("`_udp-proxy` does not accept a session name") |
| 184 | - | } | |||
| 185 | - | cmd := command{kind: kindUDPProxy} | |||
| 186 | - | if len(args) == 2 { | |||
| 187 | - | cmd.name = args[1] | |||
| 188 | } | 176 | } | ||
| 189 | - | return cmd, nil | 177 | + | return command{kind: kindUDPProxy}, nil |
| 190 | case "new": | 178 | case "new": | ||
| 191 | if len(args) < 2 || args[1] == "" { | 179 | if len(args) < 2 || args[1] == "" { | ||
| 192 | return command{}, usagef("`new` requires a session name") | 180 | return command{}, usagef("`new` requires a session name") | ||
| @@ -278,9 +266,9 @@ func dispatch(ctx context.Context, cmd command) error { | |||||
| 278 | case kindStdio: | 266 | case kindStdio: | ||
| 279 | return runStdio(ctx) | 267 | return runStdio(ctx) | ||
| 280 | case kindUDPBootstrap: | 268 | case kindUDPBootstrap: | ||
| 281 | - | return runUDPBootstrap(ctx, cmd.name) | 269 | + | return runUDPBootstrap(ctx) |
| 282 | case kindUDPProxy: | 270 | case kindUDPProxy: | ||
| 283 | - | return runUDPProxy(ctx, cmd.name, os.Stdout) | 271 | + | return runUDPProxy(ctx, os.Stdout) |
| 284 | case kindList: | 272 | case kindList: | ||
| 285 | return runList(ctx, cmd) | 273 | return runList(ctx, cmd) | ||
| 286 | case kindHost: | 274 | case kindHost: | ||
| @@ -1056,21 +1044,22 @@ func runStdio(ctx context.Context) (retErr error) { | |||||
| 1056 | // pipe drained by a small forwarding goroutine. | 1044 | // pipe drained by a small forwarding goroutine. | ||
| 1057 | stdin, stdout := os.Stdin, os.Stdout | 1045 | stdin, stdout := os.Stdin, os.Stdout | ||
| 1058 | stdoutReader, stdoutWriter := io.Pipe() | 1046 | stdoutReader, stdoutWriter := io.Pipe() | ||
| 1047 | + | forwardDone := make(chan struct{}) | |||
| 1059 | go func() { | 1048 | go func() { | ||
| 1049 | + | defer close(forwardDone) | |||
| 1060 | if _, err := io.Copy(stdout, stdoutReader); err != nil { | 1050 | if _, err := io.Copy(stdout, stdoutReader); err != nil { | ||
| 1061 | _ = stdoutReader.CloseWithError(err) | 1051 | _ = stdoutReader.CloseWithError(err) | ||
| 1062 | } | 1052 | } | ||
| 1063 | }() | 1053 | }() | ||
| 1064 | defer func() { | 1054 | defer func() { | ||
| 1065 | _ = stdoutWriter.Close() | 1055 | _ = stdoutWriter.Close() | ||
| 1056 | + | <-forwardDone | |||
| 1066 | _ = stdoutReader.Close() | 1057 | _ = stdoutReader.Close() | ||
| 1067 | }() | 1058 | }() | ||
| 1068 | - | stdio := sshstdio.NewTransport(nonClosingReader{Reader: stdin}, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer)) | 1059 | + | stdio := sshstdio.NewTransport(stdin, stdoutWriter, nil, sshstdio.WithRuntimeObserver(observer)) |
| 1069 | return runStdioProxy(ctx, stdio, transport, log) | 1060 | return runStdioProxy(ctx, stdio, transport, log) | ||
| 1070 | } | 1061 | } | ||
| 1071 | 1062 | ||||
| 1072 | - | type nonClosingReader struct{ io.Reader } | |||
| 1073 | - | ||||
| 1074 | var ( | 1063 | var ( | ||
| 1075 | udpBootstrapTimeout = 5 * time.Second | 1064 | udpBootstrapTimeout = 5 * time.Second | ||
| 1076 | udpProxyCommand = func(ctx context.Context, exe string, args ...string) *exec.Cmd { | 1065 | udpProxyCommand = func(ctx context.Context, exe string, args ...string) *exec.Cmd { | ||
| @@ -1080,7 +1069,7 @@ var ( | |||||
| 1080 | 1069 | ||||
| 1081 | // runUDPBootstrap starts a detached _udp-proxy, forwards its single readiness | 1070 | // runUDPBootstrap starts a detached _udp-proxy, forwards its single readiness | ||
| 1082 | // line, and exits so SSH stdio can close without owning the UDP proxy lifetime. | 1071 | // line, and exits so SSH stdio can close without owning the UDP proxy lifetime. | ||
| 1083 | - | func runUDPBootstrap(ctx context.Context, _ string) error { | 1072 | + | func runUDPBootstrap(ctx context.Context) error { |
| 1084 | // Always start a fresh proxy for a bootstrap request. A client attach begins | 1073 | // Always start a fresh proxy for a bootstrap request. A client attach begins | ||
| 1085 | // with MsgHello, which must reach the daemon handshake path; reusing an | 1074 | // with MsgHello, which must reach the daemon handshake path; reusing an | ||
| 1086 | // already-running proxy would forward that Hello into the daemon connection's | 1075 | // already-running proxy would forward that Hello into the daemon connection's | ||
| @@ -1148,7 +1137,7 @@ func runUDPBootstrap(ctx context.Context, _ string) error { | |||||
| 1148 | } | 1137 | } | ||
| 1149 | 1138 | ||||
| 1150 | // runUDPProxy is the detached long-lived remote-side datagram proxy. | 1139 | // runUDPProxy is the detached long-lived remote-side datagram proxy. | ||
| 1151 | - | func runUDPProxy(ctx context.Context, _ string, ready io.Writer) (retErr error) { | 1140 | + | func runUDPProxy(ctx context.Context, ready io.Writer) (retErr error) { |
| 1152 | log, logCloser, err := configureLogging(logging.Stdio, false) | 1141 | log, logCloser, err := configureLogging(logging.Stdio, false) | ||
| 1153 | if err != nil { | 1142 | if err != nil { | ||
| 1154 | return err | 1143 | return err | ||
M internal/app/run_test.go +10 -9
| @@ -482,12 +482,13 @@ func TestParseArgs(t *testing.T) { | |||
| 482 | 482 | {name: "kill extra arg", args: []string{"kill", "work", "extra"}, wantErr: true}, | |
| 483 | 483 | {name: "daemon", args: []string{"--daemon"}, wantKind: kindDaemon}, | |
| 484 | 484 | {name: "stdio", args: []string{"_stdio"}, wantKind: kindStdio}, | |
| 485 | - | {name: "stdio with session", args: []string{"_stdio", "work"}, wantKind: kindStdio, wantName: "work"}, | |
| 486 | - | {name: "stdio preserves legacy unsafe name", args: []string{"_stdio", "my work"}, wantKind: kindStdio, wantName: "my work"}, | |
| 487 | - | {name: "stdio too many args", args: []string{"_stdio", "work", "extra"}, wantErr: true}, | |
| 488 | - | {name: "udp bootstrap", args: []string{"_udp-bootstrap", "work"}, wantKind: kindUDPBootstrap, wantName: "work"}, | |
| 489 | - | {name: "udp proxy", args: []string{"_udp-proxy", "work"}, wantKind: kindUDPProxy, wantName: "work"}, | |
| 490 | - | {name: "udp proxy too many args", args: []string{"_udp-proxy", "work", "extra"}, wantErr: true}, | |
| 485 | + | {name: "stdio rejects session", args: []string{"_stdio", "work"}, wantErr: true}, | |
| 486 | + | {name: "stdio rejects extra args", args: []string{"_stdio", "work", "extra"}, wantErr: true}, | |
| 487 | + | {name: "udp bootstrap", args: []string{"_udp-bootstrap"}, wantKind: kindUDPBootstrap}, | |
| 488 | + | {name: "udp bootstrap rejects session", args: []string{"_udp-bootstrap", "work"}, wantErr: true}, | |
| 489 | + | {name: "udp proxy", args: []string{"_udp-proxy"}, wantKind: kindUDPProxy}, | |
| 490 | + | {name: "udp proxy rejects session", args: []string{"_udp-proxy", "work"}, wantErr: true}, | |
| 491 | + | {name: "udp proxy rejects extra args", args: []string{"_udp-proxy", "work", "extra"}, wantErr: true}, | |
| 491 | 492 | {name: "help", args: []string{"--help"}, wantKind: kindHelp}, | |
| 492 | 493 | {name: "help subcommand", args: []string{"help"}, wantKind: kindHelp}, | |
| 493 | 494 | {name: "version", args: []string{"--version"}, wantKind: kindVersion}, | |
| @@ -1050,7 +1051,7 @@ func TestRunUDPBootstrapForwardsReadinessAndExits(t *testing.T) { | |||
| 1050 | 1051 | t.Cleanup(func() { udpProxyCommand = oldCommand }) | |
| 1051 | 1052 | ||
| 1052 | 1053 | got := captureStdout(t, func() { | |
| 1053 | - | if err := runUDPBootstrap(context.Background(), "work"); err != nil { | |
| 1054 | + | if err := runUDPBootstrap(context.Background()); err != nil { | |
| 1054 | 1055 | t.Fatalf("runUDPBootstrap() error = %v", err) | |
| 1055 | 1056 | } | |
| 1056 | 1057 | }) | |
| @@ -1075,7 +1076,7 @@ func TestRunUDPBootstrapReturnsReadinessEOF(t *testing.T) { | |||
| 1075 | 1076 | } | |
| 1076 | 1077 | t.Cleanup(func() { udpProxyCommand = oldCommand }) | |
| 1077 | 1078 | ||
| 1078 | - | err := runUDPBootstrap(context.Background(), "work") | |
| 1079 | + | err := runUDPBootstrap(context.Background()) | |
| 1079 | 1080 | if !errors.Is(err, io.EOF) { | |
| 1080 | 1081 | t.Fatalf("runUDPBootstrap() error = %v, want EOF", err) | |
| 1081 | 1082 | } | |
| @@ -1091,7 +1092,7 @@ func TestRunUDPBootstrapTimesOutWaitingForReadiness(t *testing.T) { | |||
| 1091 | 1092 | } | |
| 1092 | 1093 | t.Cleanup(func() { udpProxyCommand = oldCommand }) | |
| 1093 | 1094 | ||
| 1094 | - | err := runUDPBootstrap(context.Background(), "work") | |
| 1095 | + | err := runUDPBootstrap(context.Background()) | |
| 1095 | 1096 | if !errors.Is(err, context.DeadlineExceeded) { | |
| 1096 | 1097 | t.Fatalf("runUDPBootstrap() error = %v, want deadline exceeded", err) | |
| 1097 | 1098 | } | |
| @@ -482,12 +482,13 @@ func TestParseArgs(t *testing.T) { | |||||
| 482 | {name: "kill extra arg", args: []string{"kill", "work", "extra"}, wantErr: true}, | 482 | {name: "kill extra arg", args: []string{"kill", "work", "extra"}, wantErr: true}, | ||
| 483 | {name: "daemon", args: []string{"--daemon"}, wantKind: kindDaemon}, | 483 | {name: "daemon", args: []string{"--daemon"}, wantKind: kindDaemon}, | ||
| 484 | {name: "stdio", args: []string{"_stdio"}, wantKind: kindStdio}, | 484 | {name: "stdio", args: []string{"_stdio"}, wantKind: kindStdio}, | ||
| 485 | - | {name: "stdio with session", args: []string{"_stdio", "work"}, wantKind: kindStdio, wantName: "work"}, | 485 | + | {name: "stdio rejects session", args: []string{"_stdio", "work"}, wantErr: true}, |
| 486 | - | {name: "stdio preserves legacy unsafe name", args: []string{"_stdio", "my work"}, wantKind: kindStdio, wantName: "my work"}, | 486 | + | {name: "stdio rejects extra args", args: []string{"_stdio", "work", "extra"}, wantErr: true}, |
| 487 | - | {name: "stdio too many args", args: []string{"_stdio", "work", "extra"}, wantErr: true}, | 487 | + | {name: "udp bootstrap", args: []string{"_udp-bootstrap"}, wantKind: kindUDPBootstrap}, |
| 488 | - | {name: "udp bootstrap", args: []string{"_udp-bootstrap", "work"}, wantKind: kindUDPBootstrap, wantName: "work"}, | 488 | + | {name: "udp bootstrap rejects session", args: []string{"_udp-bootstrap", "work"}, wantErr: true}, |
| 489 | - | {name: "udp proxy", args: []string{"_udp-proxy", "work"}, wantKind: kindUDPProxy, wantName: "work"}, | 489 | + | {name: "udp proxy", args: []string{"_udp-proxy"}, wantKind: kindUDPProxy}, |
| 490 | - | {name: "udp proxy too many args", args: []string{"_udp-proxy", "work", "extra"}, wantErr: true}, | 490 | + | {name: "udp proxy rejects session", args: []string{"_udp-proxy", "work"}, wantErr: true}, |
| 491 | + | {name: "udp proxy rejects extra args", args: []string{"_udp-proxy", "work", "extra"}, wantErr: true}, | |||
| 491 | {name: "help", args: []string{"--help"}, wantKind: kindHelp}, | 492 | {name: "help", args: []string{"--help"}, wantKind: kindHelp}, | ||
| 492 | {name: "help subcommand", args: []string{"help"}, wantKind: kindHelp}, | 493 | {name: "help subcommand", args: []string{"help"}, wantKind: kindHelp}, | ||
| 493 | {name: "version", args: []string{"--version"}, wantKind: kindVersion}, | 494 | {name: "version", args: []string{"--version"}, wantKind: kindVersion}, | ||
| @@ -1050,7 +1051,7 @@ func TestRunUDPBootstrapForwardsReadinessAndExits(t *testing.T) { | |||||
| 1050 | t.Cleanup(func() { udpProxyCommand = oldCommand }) | 1051 | t.Cleanup(func() { udpProxyCommand = oldCommand }) | ||
| 1051 | 1052 | ||||
| 1052 | got := captureStdout(t, func() { | 1053 | got := captureStdout(t, func() { | ||
| 1053 | - | if err := runUDPBootstrap(context.Background(), "work"); err != nil { | 1054 | + | if err := runUDPBootstrap(context.Background()); err != nil { |
| 1054 | t.Fatalf("runUDPBootstrap() error = %v", err) | 1055 | t.Fatalf("runUDPBootstrap() error = %v", err) | ||
| 1055 | } | 1056 | } | ||
| 1056 | }) | 1057 | }) | ||
| @@ -1075,7 +1076,7 @@ func TestRunUDPBootstrapReturnsReadinessEOF(t *testing.T) { | |||||
| 1075 | } | 1076 | } | ||
| 1076 | t.Cleanup(func() { udpProxyCommand = oldCommand }) | 1077 | t.Cleanup(func() { udpProxyCommand = oldCommand }) | ||
| 1077 | 1078 | ||||
| 1078 | - | err := runUDPBootstrap(context.Background(), "work") | 1079 | + | err := runUDPBootstrap(context.Background()) |
| 1079 | if !errors.Is(err, io.EOF) { | 1080 | if !errors.Is(err, io.EOF) { | ||
| 1080 | t.Fatalf("runUDPBootstrap() error = %v, want EOF", err) | 1081 | t.Fatalf("runUDPBootstrap() error = %v, want EOF", err) | ||
| 1081 | } | 1082 | } | ||
| @@ -1091,7 +1092,7 @@ func TestRunUDPBootstrapTimesOutWaitingForReadiness(t *testing.T) { | |||||
| 1091 | } | 1092 | } | ||
| 1092 | t.Cleanup(func() { udpProxyCommand = oldCommand }) | 1093 | t.Cleanup(func() { udpProxyCommand = oldCommand }) | ||
| 1093 | 1094 | ||||
| 1094 | - | err := runUDPBootstrap(context.Background(), "work") | 1095 | + | err := runUDPBootstrap(context.Background()) |
| 1095 | if !errors.Is(err, context.DeadlineExceeded) { | 1096 | if !errors.Is(err, context.DeadlineExceeded) { | ||
| 1096 | t.Fatalf("runUDPBootstrap() error = %v, want deadline exceeded", err) | 1097 | t.Fatalf("runUDPBootstrap() error = %v, want deadline exceeded", err) | ||
| 1097 | } | 1098 | } | ||
M internal/usecase/client/handshake.go +8 -2
| @@ -105,13 +105,18 @@ func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, o | |||
| 105 | 105 | } | |
| 106 | 106 | ||
| 107 | 107 | func boundedDial(ctx context.Context, dialer ports.Dialer) (ports.Transport, error) { | |
| 108 | + | // The dial context only bounds startup. A successful carriage outlives the | |
| 109 | + | // handshake context; canceling that context after Welcome must not kill SSH. | |
| 110 | + | dialCtx, cancel := context.WithCancel(context.WithoutCancel(ctx)) | |
| 108 | 111 | result := make(chan struct { | |
| 109 | 112 | transport ports.Transport | |
| 110 | 113 | err error | |
| 111 | 114 | }, 1) | |
| 112 | 115 | go func() { | |
| 113 | - | transport, err := dialer.Dial(ctx) | |
| 114 | - | if ctx.Err() != nil && transport != nil { | |
| 116 | + | transport, err := dialer.Dial(dialCtx) | |
| 117 | + | if err != nil { | |
| 118 | + | cancel() | |
| 119 | + | } else if ctx.Err() != nil && transport != nil { | |
| 115 | 120 | _ = transport.Close() | |
| 116 | 121 | } | |
| 117 | 122 | result <- struct { | |
| @@ -123,6 +128,7 @@ func boundedDial(ctx context.Context, dialer ports.Dialer) (ports.Transport, err | |||
| 123 | 128 | case result := <-result: | |
| 124 | 129 | return result.transport, result.err | |
| 125 | 130 | case <-ctx.Done(): | |
| 131 | + | cancel() | |
| 126 | 132 | return nil, ctx.Err() | |
| 127 | 133 | } | |
| 128 | 134 | } | |
| @@ -105,13 +105,18 @@ func boundedHandshakeOperation(ctx context.Context, transport ports.Transport, o | |||||
| 105 | } | 105 | } | ||
| 106 | 106 | ||||
| 107 | func boundedDial(ctx context.Context, dialer ports.Dialer) (ports.Transport, error) { | 107 | func boundedDial(ctx context.Context, dialer ports.Dialer) (ports.Transport, error) { | ||
| 108 | + | // The dial context only bounds startup. A successful carriage outlives the | |||
| 109 | + | // handshake context; canceling that context after Welcome must not kill SSH. | |||
| 110 | + | dialCtx, cancel := context.WithCancel(context.WithoutCancel(ctx)) | |||
| 108 | result := make(chan struct { | 111 | result := make(chan struct { | ||
| 109 | transport ports.Transport | 112 | transport ports.Transport | ||
| 110 | err error | 113 | err error | ||
| 111 | }, 1) | 114 | }, 1) | ||
| 112 | go func() { | 115 | go func() { | ||
| 113 | - | transport, err := dialer.Dial(ctx) | 116 | + | transport, err := dialer.Dial(dialCtx) |
| 114 | - | if ctx.Err() != nil && transport != nil { | 117 | + | if err != nil { |
| 118 | + | cancel() | |||
| 119 | + | } else if ctx.Err() != nil && transport != nil { | |||
| 115 | _ = transport.Close() | 120 | _ = transport.Close() | ||
| 116 | } | 121 | } | ||
| 117 | result <- struct { | 122 | result <- struct { | ||
| @@ -123,6 +128,7 @@ func boundedDial(ctx context.Context, dialer ports.Dialer) (ports.Transport, err | |||||
| 123 | case result := <-result: | 128 | case result := <-result: | ||
| 124 | return result.transport, result.err | 129 | return result.transport, result.err | ||
| 125 | case <-ctx.Done(): | 130 | case <-ctx.Done(): | ||
| 131 | + | cancel() | |||
| 126 | return nil, ctx.Err() | 132 | return nil, ctx.Err() | ||
| 127 | } | 133 | } | ||
| 128 | } | 134 | } | ||
M internal/usecase/daemon/client.go +13 -5
| @@ -799,11 +799,19 @@ func (d *Daemon) resizeAttachmentForLease(token attachmentConnectionToken, size | |||
| 799 | 799 | ac.pipelineScratch = composeCacheInput{} | |
| 800 | 800 | ac.sendMu.Unlock() | |
| 801 | 801 | ||
| 802 | - | entry := token.sess | |
| 803 | - | // Rendering is attachment-scoped. Run it independently of the connection | |
| 804 | - | // reader so a blocked transport cannot hold session dispatch or peers. | |
| 805 | - | go d.paint(entry, ac, true, token.lease) | |
| 806 | - | return true | |
| 802 | + | sess, ok := localSession(token.sess) | |
| 803 | + | if !ok { | |
| 804 | + | return false | |
| 805 | + | } | |
| 806 | + | rc := sess.renderCoordinator() | |
| 807 | + | if rc == nil { | |
| 808 | + | return false | |
| 809 | + | } | |
| 810 | + | // Rendering is attachment-scoped. Schedule it through the coordinator so | |
| 811 | + | // rapid resize frames coalesce under the attachment lease and deadline. | |
| 812 | + | return rc.scheduleResizeForLease(size, ac, token.lease, func(uint64) { | |
| 813 | + | go d.paint(sess, ac, true, token.lease) | |
| 814 | + | }) != 0 | |
| 807 | 815 | } | |
| 808 | 816 | ||
| 809 | 817 | // handleClientNotice maps the closed client-event enum to daemon-owned notice | |
| @@ -799,11 +799,19 @@ func (d *Daemon) resizeAttachmentForLease(token attachmentConnectionToken, size | |||||
| 799 | ac.pipelineScratch = composeCacheInput{} | 799 | ac.pipelineScratch = composeCacheInput{} | ||
| 800 | ac.sendMu.Unlock() | 800 | ac.sendMu.Unlock() | ||
| 801 | 801 | ||||
| 802 | - | entry := token.sess | 802 | + | sess, ok := localSession(token.sess) |
| 803 | - | // Rendering is attachment-scoped. Run it independently of the connection | 803 | + | if !ok { |
| 804 | - | // reader so a blocked transport cannot hold session dispatch or peers. | 804 | + | return false |
| 805 | - | go d.paint(entry, ac, true, token.lease) | 805 | + | } |
| 806 | - | return true | 806 | + | rc := sess.renderCoordinator() |
| 807 | + | if rc == nil { | |||
| 808 | + | return false | |||
| 809 | + | } | |||
| 810 | + | // Rendering is attachment-scoped. Schedule it through the coordinator so | |||
| 811 | + | // rapid resize frames coalesce under the attachment lease and deadline. | |||
| 812 | + | return rc.scheduleResizeForLease(size, ac, token.lease, func(uint64) { | |||
| 813 | + | go d.paint(sess, ac, true, token.lease) | |||
| 814 | + | }) != 0 | |||
| 807 | } | 815 | } | ||
| 808 | 816 | ||||
| 809 | // handleClientNotice maps the closed client-event enum to daemon-owned notice | 817 | // handleClientNotice maps the closed client-event enum to daemon-owned notice | ||
M internal/usecase/daemon/command_tracker.go +3 -3
| @@ -127,15 +127,15 @@ func (t *CommandRequestTracker) Remove(requestID, generation uint64) { | |||
| 127 | 127 | // Complete accepts only the exact request generation. Unknown, late, and old | |
| 128 | 128 | // generation results are safe no-ops. | |
| 129 | 129 | func (t *CommandRequestTracker) Complete(generation uint64, result ports.CommandResult) { | |
| 130 | - | t.finish(generation, result.RequestID, CommandRequestOutcome{Result: result}) | |
| 130 | + | t.finish(result.RequestID, generation, CommandRequestOutcome{Result: result}) | |
| 131 | 131 | } | |
| 132 | 132 | ||
| 133 | 133 | // Fail completes the exact request with a transport or decode failure. | |
| 134 | 134 | func (t *CommandRequestTracker) Fail(requestID, generation uint64, err error) { | |
| 135 | - | t.finish(generation, requestID, CommandRequestOutcome{Err: err}) | |
| 135 | + | t.finish(requestID, generation, CommandRequestOutcome{Err: err}) | |
| 136 | 136 | } | |
| 137 | 137 | ||
| 138 | - | func (t *CommandRequestTracker) finish(generation, requestID uint64, outcome CommandRequestOutcome) { | |
| 138 | + | func (t *CommandRequestTracker) finish(requestID, generation uint64, outcome CommandRequestOutcome) { | |
| 139 | 139 | if t == nil { | |
| 140 | 140 | return | |
| 141 | 141 | } | |
| @@ -127,15 +127,15 @@ func (t *CommandRequestTracker) Remove(requestID, generation uint64) { | |||||
| 127 | // Complete accepts only the exact request generation. Unknown, late, and old | 127 | // Complete accepts only the exact request generation. Unknown, late, and old | ||
| 128 | // generation results are safe no-ops. | 128 | // generation results are safe no-ops. | ||
| 129 | func (t *CommandRequestTracker) Complete(generation uint64, result ports.CommandResult) { | 129 | func (t *CommandRequestTracker) Complete(generation uint64, result ports.CommandResult) { | ||
| 130 | - | t.finish(generation, result.RequestID, CommandRequestOutcome{Result: result}) | 130 | + | t.finish(result.RequestID, generation, CommandRequestOutcome{Result: result}) |
| 131 | } | 131 | } | ||
| 132 | 132 | ||||
| 133 | // Fail completes the exact request with a transport or decode failure. | 133 | // Fail completes the exact request with a transport or decode failure. | ||
| 134 | func (t *CommandRequestTracker) Fail(requestID, generation uint64, err error) { | 134 | func (t *CommandRequestTracker) Fail(requestID, generation uint64, err error) { | ||
| 135 | - | t.finish(generation, requestID, CommandRequestOutcome{Err: err}) | 135 | + | t.finish(requestID, generation, CommandRequestOutcome{Err: err}) |
| 136 | } | 136 | } | ||
| 137 | 137 | ||||
| 138 | - | func (t *CommandRequestTracker) finish(generation, requestID uint64, outcome CommandRequestOutcome) { | 138 | + | func (t *CommandRequestTracker) finish(requestID, generation uint64, outcome CommandRequestOutcome) { |
| 139 | if t == nil { | 139 | if t == nil { | ||
| 140 | return | 140 | return | ||
| 141 | } | 141 | } | ||
M internal/usecase/daemon/command_tracker_test.go +82 -0
| @@ -2,6 +2,7 @@ package daemon | |||
| 2 | 2 | ||
| 3 | 3 | import ( | |
| 4 | 4 | "context" | |
| 5 | + | "errors" | |
| 5 | 6 | "sync" | |
| 6 | 7 | "testing" | |
| 7 | 8 | "time" | |
| @@ -68,4 +69,85 @@ func TestCommandRequestTrackerTimeoutAndGenerationIsolation(t *testing.T) { | |||
| 68 | 69 | tracker.Complete(8, ports.CommandResult{RequestID: requestID, OK: true}) | |
| 69 | 70 | result := <-secondOutcome | |
| 70 | 71 | require.True(t, result.Result.OK) | |
| 72 | + | ||
| 73 | + | t.Run("wait outcomes", func(t *testing.T) { | |
| 74 | + | failure := errors.New("transport failed") | |
| 75 | + | tests := []struct { | |
| 76 | + | name string | |
| 77 | + | setup func(*testing.T, *CommandRequestTracker, uint64) (context.Context, ports.Clock, func()) | |
| 78 | + | wantErr error | |
| 79 | + | wantPending int | |
| 80 | + | }{ | |
| 81 | + | { | |
| 82 | + | name: "canceled context", | |
| 83 | + | setup: func(_ *testing.T, _ *CommandRequestTracker, _ uint64) (context.Context, ports.Clock, func()) { | |
| 84 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 85 | + | cancel() | |
| 86 | + | return ctx, newCommandTrackerTestClock(), nil | |
| 87 | + | }, | |
| 88 | + | wantErr: context.Canceled, | |
| 89 | + | wantPending: 0, | |
| 90 | + | }, | |
| 91 | + | { | |
| 92 | + | name: "nil clock", | |
| 93 | + | setup: func(_ *testing.T, _ *CommandRequestTracker, _ uint64) (context.Context, ports.Clock, func()) { | |
| 94 | + | return context.Background(), nil, nil | |
| 95 | + | }, | |
| 96 | + | wantErr: ErrCommandRequestUnavailable, | |
| 97 | + | wantPending: 1, | |
| 98 | + | }, | |
| 99 | + | { | |
| 100 | + | name: "fail", | |
| 101 | + | setup: func(_ *testing.T, tracker *CommandRequestTracker, requestID uint64) (context.Context, ports.Clock, func()) { | |
| 102 | + | tracker.Fail(requestID, 1, failure) | |
| 103 | + | return context.Background(), newCommandTrackerTestClock(), nil | |
| 104 | + | }, | |
| 105 | + | wantErr: failure, | |
| 106 | + | wantPending: 0, | |
| 107 | + | }, | |
| 108 | + | { | |
| 109 | + | name: "remove", | |
| 110 | + | setup: func(_ *testing.T, tracker *CommandRequestTracker, requestID uint64) (context.Context, ports.Clock, func()) { | |
| 111 | + | tracker.Remove(requestID, 1) | |
| 112 | + | clock := newCommandTrackerTestClock() | |
| 113 | + | return context.Background(), clock, func() { | |
| 114 | + | timer := <-clock.timers | |
| 115 | + | timer.fire() | |
| 116 | + | } | |
| 117 | + | }, | |
| 118 | + | wantErr: ErrCommandRequestTimeout, | |
| 119 | + | wantPending: 0, | |
| 120 | + | }, | |
| 121 | + | { | |
| 122 | + | name: "duplicate track", | |
| 123 | + | setup: func(t *testing.T, tracker *CommandRequestTracker, requestID uint64) (context.Context, ports.Clock, func()) { | |
| 124 | + | duplicate, ok := tracker.Track(requestID, 1) | |
| 125 | + | require.False(t, ok) | |
| 126 | + | require.Nil(t, duplicate) | |
| 127 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 128 | + | cancel() | |
| 129 | + | return ctx, newCommandTrackerTestClock(), nil | |
| 130 | + | }, | |
| 131 | + | wantErr: context.Canceled, | |
| 132 | + | wantPending: 0, | |
| 133 | + | }, | |
| 134 | + | } | |
| 135 | + | for _, tt := range tests { | |
| 136 | + | t.Run(tt.name, func(t *testing.T) { | |
| 137 | + | tracker := NewCommandRequestTracker() | |
| 138 | + | requestID, outcome := tracker.Publish(1) | |
| 139 | + | ctx, clock, drive := tt.setup(t, tracker, requestID) | |
| 140 | + | waitDone := make(chan error, 1) | |
| 141 | + | go func() { | |
| 142 | + | _, err := tracker.Wait(ctx, clock, requestID, 1, outcome) | |
| 143 | + | waitDone <- err | |
| 144 | + | }() | |
| 145 | + | if drive != nil { | |
| 146 | + | drive() | |
| 147 | + | } | |
| 148 | + | require.ErrorIs(t, <-waitDone, tt.wantErr) | |
| 149 | + | require.Equal(t, tt.wantPending, tracker.PendingCount()) | |
| 150 | + | }) | |
| 151 | + | } | |
| 152 | + | }) | |
| 71 | 153 | } | |
| @@ -2,6 +2,7 @@ package daemon | |||||
| 2 | 2 | ||||
| 3 | import ( | 3 | import ( | ||
| 4 | "context" | 4 | "context" | ||
| 5 | + | "errors" | |||
| 5 | "sync" | 6 | "sync" | ||
| 6 | "testing" | 7 | "testing" | ||
| 7 | "time" | 8 | "time" | ||
| @@ -68,4 +69,85 @@ func TestCommandRequestTrackerTimeoutAndGenerationIsolation(t *testing.T) { | |||||
| 68 | tracker.Complete(8, ports.CommandResult{RequestID: requestID, OK: true}) | 69 | tracker.Complete(8, ports.CommandResult{RequestID: requestID, OK: true}) | ||
| 69 | result := <-secondOutcome | 70 | result := <-secondOutcome | ||
| 70 | require.True(t, result.Result.OK) | 71 | require.True(t, result.Result.OK) | ||
| 72 | + | ||||
| 73 | + | t.Run("wait outcomes", func(t *testing.T) { | |||
| 74 | + | failure := errors.New("transport failed") | |||
| 75 | + | tests := []struct { | |||
| 76 | + | name string | |||
| 77 | + | setup func(*testing.T, *CommandRequestTracker, uint64) (context.Context, ports.Clock, func()) | |||
| 78 | + | wantErr error | |||
| 79 | + | wantPending int | |||
| 80 | + | }{ | |||
| 81 | + | { | |||
| 82 | + | name: "canceled context", | |||
| 83 | + | setup: func(_ *testing.T, _ *CommandRequestTracker, _ uint64) (context.Context, ports.Clock, func()) { | |||
| 84 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 85 | + | cancel() | |||
| 86 | + | return ctx, newCommandTrackerTestClock(), nil | |||
| 87 | + | }, | |||
| 88 | + | wantErr: context.Canceled, | |||
| 89 | + | wantPending: 0, | |||
| 90 | + | }, | |||
| 91 | + | { | |||
| 92 | + | name: "nil clock", | |||
| 93 | + | setup: func(_ *testing.T, _ *CommandRequestTracker, _ uint64) (context.Context, ports.Clock, func()) { | |||
| 94 | + | return context.Background(), nil, nil | |||
| 95 | + | }, | |||
| 96 | + | wantErr: ErrCommandRequestUnavailable, | |||
| 97 | + | wantPending: 1, | |||
| 98 | + | }, | |||
| 99 | + | { | |||
| 100 | + | name: "fail", | |||
| 101 | + | setup: func(_ *testing.T, tracker *CommandRequestTracker, requestID uint64) (context.Context, ports.Clock, func()) { | |||
| 102 | + | tracker.Fail(requestID, 1, failure) | |||
| 103 | + | return context.Background(), newCommandTrackerTestClock(), nil | |||
| 104 | + | }, | |||
| 105 | + | wantErr: failure, | |||
| 106 | + | wantPending: 0, | |||
| 107 | + | }, | |||
| 108 | + | { | |||
| 109 | + | name: "remove", | |||
| 110 | + | setup: func(_ *testing.T, tracker *CommandRequestTracker, requestID uint64) (context.Context, ports.Clock, func()) { | |||
| 111 | + | tracker.Remove(requestID, 1) | |||
| 112 | + | clock := newCommandTrackerTestClock() | |||
| 113 | + | return context.Background(), clock, func() { | |||
| 114 | + | timer := <-clock.timers | |||
| 115 | + | timer.fire() | |||
| 116 | + | } | |||
| 117 | + | }, | |||
| 118 | + | wantErr: ErrCommandRequestTimeout, | |||
| 119 | + | wantPending: 0, | |||
| 120 | + | }, | |||
| 121 | + | { | |||
| 122 | + | name: "duplicate track", | |||
| 123 | + | setup: func(t *testing.T, tracker *CommandRequestTracker, requestID uint64) (context.Context, ports.Clock, func()) { | |||
| 124 | + | duplicate, ok := tracker.Track(requestID, 1) | |||
| 125 | + | require.False(t, ok) | |||
| 126 | + | require.Nil(t, duplicate) | |||
| 127 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 128 | + | cancel() | |||
| 129 | + | return ctx, newCommandTrackerTestClock(), nil | |||
| 130 | + | }, | |||
| 131 | + | wantErr: context.Canceled, | |||
| 132 | + | wantPending: 0, | |||
| 133 | + | }, | |||
| 134 | + | } | |||
| 135 | + | for _, tt := range tests { | |||
| 136 | + | t.Run(tt.name, func(t *testing.T) { | |||
| 137 | + | tracker := NewCommandRequestTracker() | |||
| 138 | + | requestID, outcome := tracker.Publish(1) | |||
| 139 | + | ctx, clock, drive := tt.setup(t, tracker, requestID) | |||
| 140 | + | waitDone := make(chan error, 1) | |||
| 141 | + | go func() { | |||
| 142 | + | _, err := tracker.Wait(ctx, clock, requestID, 1, outcome) | |||
| 143 | + | waitDone <- err | |||
| 144 | + | }() | |||
| 145 | + | if drive != nil { | |||
| 146 | + | drive() | |||
| 147 | + | } | |||
| 148 | + | require.ErrorIs(t, <-waitDone, tt.wantErr) | |||
| 149 | + | require.Equal(t, tt.wantPending, tracker.PendingCount()) | |||
| 150 | + | }) | |||
| 151 | + | } | |||
| 152 | + | }) | |||
| 71 | } | 153 | } | ||
M internal/usecase/daemon/control.go +14 -17
| @@ -5,6 +5,7 @@ import ( | |||
| 5 | 5 | "encoding/json" | |
| 6 | 6 | "errors" | |
| 7 | 7 | "fmt" | |
| 8 | + | "runtime" | |
| 8 | 9 | "slices" | |
| 9 | 10 | "sort" | |
| 10 | 11 | "strings" | |
| @@ -37,14 +38,7 @@ func (d *Daemon) handleCommand(tr ports.Transport, f ports.Frame) error { | |||
| 37 | 38 | ||
| 38 | 39 | tracker := NewCommandRequestTracker() | |
| 39 | 40 | const generation = uint64(1) | |
| 40 | - | outcome, ok := tracker.Track(request.RequestID, generation) | |
| 41 | - | if !ok { | |
| 42 | - | return d.sendCommandResult(tr, ports.CommandResult{ | |
| 43 | - | RequestID: request.RequestID, | |
| 44 | - | Code: ports.ErrInternal, | |
| 45 | - | Text: "duplicate command request id", | |
| 46 | - | }) | |
| 47 | - | } | |
| 41 | + | outcome, _ := tracker.Track(request.RequestID, generation) | |
| 48 | 42 | commandCtx := d.serveCtx | |
| 49 | 43 | if commandCtx == nil { | |
| 50 | 44 | commandCtx = context.Background() | |
| @@ -60,15 +54,8 @@ func (d *Daemon) handleCommand(tr ports.Transport, f ports.Frame) error { | |||
| 60 | 54 | if commandClock == nil { | |
| 61 | 55 | commandClock = systemClock{} | |
| 62 | 56 | } | |
| 63 | - | result, waitErr := tracker.Wait(context.Background(), commandClock, request.RequestID, generation, outcome) | |
| 57 | + | result, waitErr := tracker.Wait(commandCtx, commandClock, request.RequestID, generation, outcome) | |
| 64 | 58 | if waitErr != nil { | |
| 65 | - | if errors.Is(waitErr, ErrCommandRequestTimeout) { | |
| 66 | - | return d.sendCommandResult(tr, ports.CommandResult{ | |
| 67 | - | RequestID: request.RequestID, | |
| 68 | - | Code: ports.ErrInternal, | |
| 69 | - | Text: waitErr.Error(), | |
| 70 | - | }) | |
| 71 | - | } | |
| 72 | 59 | return d.sendCommandResult(tr, ports.CommandResult{ | |
| 73 | 60 | RequestID: request.RequestID, | |
| 74 | 61 | Code: ports.ErrInternal, | |
| @@ -132,7 +119,17 @@ func (d *Daemon) dispatchCommand(ctx context.Context, request ports.CommandReque | |||
| 132 | 119 | return d.runControl(cmd, controlExec{ctx: ctx, d: d, sess: sess, tab: tb, target: daemonActionTarget{session: sess, tab: tb, pane: pane}}, request) | |
| 133 | 120 | } | |
| 134 | 121 | ||
| 135 | - | sess.dispatchMu.Lock() | |
| 122 | + | for { | |
| 123 | + | select { | |
| 124 | + | case <-ctx.Done(): | |
| 125 | + | return commandFailure(ports.ErrInternal, ctx.Err().Error()) | |
| 126 | + | default: | |
| 127 | + | } | |
| 128 | + | if sess.dispatchMu.TryLock() { | |
| 129 | + | break | |
| 130 | + | } | |
| 131 | + | runtime.Gosched() | |
| 132 | + | } | |
| 136 | 133 | defer sess.dispatchMu.Unlock() | |
| 137 | 134 | ||
| 138 | 135 | tb, pane, code, text := resolveControlTarget(sess, cmd.Target, request) | |
| @@ -5,6 +5,7 @@ import ( | |||||
| 5 | "encoding/json" | 5 | "encoding/json" | ||
| 6 | "errors" | 6 | "errors" | ||
| 7 | "fmt" | 7 | "fmt" | ||
| 8 | + | "runtime" | |||
| 8 | "slices" | 9 | "slices" | ||
| 9 | "sort" | 10 | "sort" | ||
| 10 | "strings" | 11 | "strings" | ||
| @@ -37,14 +38,7 @@ func (d *Daemon) handleCommand(tr ports.Transport, f ports.Frame) error { | |||||
| 37 | 38 | ||||
| 38 | tracker := NewCommandRequestTracker() | 39 | tracker := NewCommandRequestTracker() | ||
| 39 | const generation = uint64(1) | 40 | const generation = uint64(1) | ||
| 40 | - | outcome, ok := tracker.Track(request.RequestID, generation) | 41 | + | outcome, _ := tracker.Track(request.RequestID, generation) |
| 41 | - | if !ok { | |||
| 42 | - | return d.sendCommandResult(tr, ports.CommandResult{ | |||
| 43 | - | RequestID: request.RequestID, | |||
| 44 | - | Code: ports.ErrInternal, | |||
| 45 | - | Text: "duplicate command request id", | |||
| 46 | - | }) | |||
| 47 | - | } | |||
| 48 | commandCtx := d.serveCtx | 42 | commandCtx := d.serveCtx | ||
| 49 | if commandCtx == nil { | 43 | if commandCtx == nil { | ||
| 50 | commandCtx = context.Background() | 44 | commandCtx = context.Background() | ||
| @@ -60,15 +54,8 @@ func (d *Daemon) handleCommand(tr ports.Transport, f ports.Frame) error { | |||||
| 60 | if commandClock == nil { | 54 | if commandClock == nil { | ||
| 61 | commandClock = systemClock{} | 55 | commandClock = systemClock{} | ||
| 62 | } | 56 | } | ||
| 63 | - | result, waitErr := tracker.Wait(context.Background(), commandClock, request.RequestID, generation, outcome) | 57 | + | result, waitErr := tracker.Wait(commandCtx, commandClock, request.RequestID, generation, outcome) |
| 64 | if waitErr != nil { | 58 | if waitErr != nil { | ||
| 65 | - | if errors.Is(waitErr, ErrCommandRequestTimeout) { | |||
| 66 | - | return d.sendCommandResult(tr, ports.CommandResult{ | |||
| 67 | - | RequestID: request.RequestID, | |||
| 68 | - | Code: ports.ErrInternal, | |||
| 69 | - | Text: waitErr.Error(), | |||
| 70 | - | }) | |||
| 71 | - | } | |||
| 72 | return d.sendCommandResult(tr, ports.CommandResult{ | 59 | return d.sendCommandResult(tr, ports.CommandResult{ | ||
| 73 | RequestID: request.RequestID, | 60 | RequestID: request.RequestID, | ||
| 74 | Code: ports.ErrInternal, | 61 | Code: ports.ErrInternal, | ||
| @@ -132,7 +119,17 @@ func (d *Daemon) dispatchCommand(ctx context.Context, request ports.CommandReque | |||||
| 132 | return d.runControl(cmd, controlExec{ctx: ctx, d: d, sess: sess, tab: tb, target: daemonActionTarget{session: sess, tab: tb, pane: pane}}, request) | 119 | return d.runControl(cmd, controlExec{ctx: ctx, d: d, sess: sess, tab: tb, target: daemonActionTarget{session: sess, tab: tb, pane: pane}}, request) | ||
| 133 | } | 120 | } | ||
| 134 | 121 | ||||
| 135 | - | sess.dispatchMu.Lock() | 122 | + | for { |
| 123 | + | select { | |||
| 124 | + | case <-ctx.Done(): | |||
| 125 | + | return commandFailure(ports.ErrInternal, ctx.Err().Error()) | |||
| 126 | + | default: | |||
| 127 | + | } | |||
| 128 | + | if sess.dispatchMu.TryLock() { | |||
| 129 | + | break | |||
| 130 | + | } | |||
| 131 | + | runtime.Gosched() | |||
| 132 | + | } | |||
| 136 | defer sess.dispatchMu.Unlock() | 133 | defer sess.dispatchMu.Unlock() | ||
| 137 | 134 | ||||
| 138 | tb, pane, code, text := resolveControlTarget(sess, cmd.Target, request) | 135 | tb, pane, code, text := resolveControlTarget(sess, cmd.Target, request) | ||
M internal/usecase/daemon/control_test.go +4 -3
| @@ -129,13 +129,14 @@ func TestHandleCommandTimesOutWithRequestGeneration(t *testing.T) { | |||
| 129 | 129 | clock := &signalClock{timers: make(chan *signalTimer, 1)} | |
| 130 | 130 | d := newTestDaemon(t, factory, clock) | |
| 131 | 131 | addControlSession(d, "work", "t_work", "p_work") | |
| 132 | - | tr, sends, releaseConn := newConn(t, commandFrame(t, ports.CommandRequest{ | |
| 132 | + | frame := commandFrame(t, ports.CommandRequest{ | |
| 133 | 133 | RequestID: 17, Slug: "new-tab", TargetSession: "work", | |
| 134 | - | })) | |
| 134 | + | }) | |
| 135 | + | tr, sends, releaseConn := newConn(t, frame) | |
| 135 | 136 | defer releaseConn() | |
| 136 | 137 | done := make(chan error, 1) | |
| 137 | 138 | go func() { | |
| 138 | - | done <- d.handleCommand(tr, commandFrame(t, ports.CommandRequest{RequestID: 17, Slug: "new-tab", TargetSession: "work"})) | |
| 139 | + | done <- d.handleCommand(tr, frame) | |
| 139 | 140 | }() | |
| 140 | 141 | <-factory.entered | |
| 141 | 142 | timer := <-clock.timers | |
| @@ -129,13 +129,14 @@ func TestHandleCommandTimesOutWithRequestGeneration(t *testing.T) { | |||||
| 129 | clock := &signalClock{timers: make(chan *signalTimer, 1)} | 129 | clock := &signalClock{timers: make(chan *signalTimer, 1)} | ||
| 130 | d := newTestDaemon(t, factory, clock) | 130 | d := newTestDaemon(t, factory, clock) | ||
| 131 | addControlSession(d, "work", "t_work", "p_work") | 131 | addControlSession(d, "work", "t_work", "p_work") | ||
| 132 | - | tr, sends, releaseConn := newConn(t, commandFrame(t, ports.CommandRequest{ | 132 | + | frame := commandFrame(t, ports.CommandRequest{ |
| 133 | RequestID: 17, Slug: "new-tab", TargetSession: "work", | 133 | RequestID: 17, Slug: "new-tab", TargetSession: "work", | ||
| 134 | - | })) | 134 | + | }) |
| 135 | + | tr, sends, releaseConn := newConn(t, frame) | |||
| 135 | defer releaseConn() | 136 | defer releaseConn() | ||
| 136 | done := make(chan error, 1) | 137 | done := make(chan error, 1) | ||
| 137 | go func() { | 138 | go func() { | ||
| 138 | - | done <- d.handleCommand(tr, commandFrame(t, ports.CommandRequest{RequestID: 17, Slug: "new-tab", TargetSession: "work"})) | 139 | + | done <- d.handleCommand(tr, frame) |
| 139 | }() | 140 | }() | ||
| 140 | <-factory.entered | 141 | <-factory.entered | ||
| 141 | timer := <-clock.timers | 142 | timer := <-clock.timers | ||
M internal/usecase/daemon/daemon.go +7 -3
| @@ -1194,11 +1194,15 @@ func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut < | |||
| 1194 | 1194 | } | |
| 1195 | 1195 | postWelcomeTicket.End() | |
| 1196 | 1196 | paintToken := sess.attachmentToken(ac, tr) | |
| 1197 | - | painted := false | |
| 1197 | + | painted := make(chan bool, 1) | |
| 1198 | 1198 | if err := boundedHandshakeOperation(handshakeCtx, tr, func() error { | |
| 1199 | - | painted = d.firstPaintForTransition(paintToken) | |
| 1199 | + | painted <- d.firstPaintForTransition(paintToken) | |
| 1200 | 1200 | return nil | |
| 1201 | - | }); err != nil || !painted || handshakeCtx.Err() != nil { | |
| 1201 | + | }); err != nil { | |
| 1202 | + | failAttachment() | |
| 1203 | + | return | |
| 1204 | + | } | |
| 1205 | + | if !<-painted || handshakeCtx.Err() != nil { | |
| 1202 | 1206 | failAttachment() | |
| 1203 | 1207 | return | |
| 1204 | 1208 | } | |
| @@ -1194,11 +1194,15 @@ func (d *Daemon) handleHelloWithContext(handshakeCtx context.Context, timedOut < | |||||
| 1194 | } | 1194 | } | ||
| 1195 | postWelcomeTicket.End() | 1195 | postWelcomeTicket.End() | ||
| 1196 | paintToken := sess.attachmentToken(ac, tr) | 1196 | paintToken := sess.attachmentToken(ac, tr) | ||
| 1197 | - | painted := false | 1197 | + | painted := make(chan bool, 1) |
| 1198 | if err := boundedHandshakeOperation(handshakeCtx, tr, func() error { | 1198 | if err := boundedHandshakeOperation(handshakeCtx, tr, func() error { | ||
| 1199 | - | painted = d.firstPaintForTransition(paintToken) | 1199 | + | painted <- d.firstPaintForTransition(paintToken) |
| 1200 | return nil | 1200 | return nil | ||
| 1201 | - | }); err != nil || !painted || handshakeCtx.Err() != nil { | 1201 | + | }); err != nil { |
| 1202 | + | failAttachment() | |||
| 1203 | + | return | |||
| 1204 | + | } | |||
| 1205 | + | if !<-painted || handshakeCtx.Err() != nil { | |||
| 1202 | failAttachment() | 1206 | failAttachment() | ||
| 1203 | return | 1207 | return | ||
| 1204 | } | 1208 | } | ||
M internal/usecase/daemon/handshake.go +3 -0
| @@ -102,6 +102,9 @@ func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fal | |||
| 102 | 102 | return errors.Join(errHandshakeTimeout, context.DeadlineExceeded) | |
| 103 | 103 | default: | |
| 104 | 104 | } | |
| 105 | + | if parent == nil { | |
| 106 | + | return fallback | |
| 107 | + | } | |
| 105 | 108 | if err := parent.Err(); err != nil { | |
| 106 | 109 | return err | |
| 107 | 110 | } | |
| @@ -102,6 +102,9 @@ func handshakeContextError(parent context.Context, timedOut <-chan struct{}, fal | |||||
| 102 | return errors.Join(errHandshakeTimeout, context.DeadlineExceeded) | 102 | return errors.Join(errHandshakeTimeout, context.DeadlineExceeded) | ||
| 103 | default: | 103 | default: | ||
| 104 | } | 104 | } | ||
| 105 | + | if parent == nil { | |||
| 106 | + | return fallback | |||
| 107 | + | } | |||
| 105 | if err := parent.Err(); err != nil { | 108 | if err := parent.Err(); err != nil { | ||
| 106 | return err | 109 | return err | ||
| 107 | } | 110 | } | ||
M internal/usecase/daemon/handshake_test.go +38 -0
| @@ -107,6 +107,44 @@ func TestBoundedHandshakeCancellationDoesNotWaitForUninterruptibleOperation(t *t | |||
| 107 | 107 | awaitTestCompletion(t, operationDone, "handshake operation did not finish") | |
| 108 | 108 | } | |
| 109 | 109 | ||
| 110 | + | func TestHandshakeFirstPaintCancellationDoesNotRacePaintResult(t *testing.T) { | |
| 111 | + | pty, releasePTY := newBlockingPTY(t) | |
| 112 | + | defer releasePTY() | |
| 113 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |
| 114 | + | ctx, cancel := context.WithCancel(context.Background()) | |
| 115 | + | defer cancel() | |
| 116 | + | ||
| 117 | + | paintStarted := make(chan struct{}) | |
| 118 | + | releasePaint := make(chan struct{}) | |
| 119 | + | paintDone := make(chan struct{}) | |
| 120 | + | d.afterAttachmentEffectAdmitted = func(attachmentConnectionToken) { | |
| 121 | + | close(paintStarted) | |
| 122 | + | <-ctx.Done() | |
| 123 | + | <-releasePaint | |
| 124 | + | close(paintDone) | |
| 125 | + | } | |
| 126 | + | ||
| 127 | + | tr := newHandshakeBlockingTransport(false) | |
| 128 | + | handshakeDone := make(chan struct{}) | |
| 129 | + | go func() { | |
| 130 | + | d.handleHelloWithContext(ctx, make(chan struct{}), func() {}, func() {}, tr, | |
| 131 | + | mustHello(ports.IntentNew, "paint-cancel", domain.Size{Cols: 80, Rows: 24})) | |
| 132 | + | close(handshakeDone) | |
| 133 | + | }() | |
| 134 | + | ||
| 135 | + | awaitTestCompletion(t, paintStarted, "first paint did not start") | |
| 136 | + | cancel() | |
| 137 | + | awaitTestCompletion(t, tr.closed, "handshake cancellation did not close the transport") | |
| 138 | + | select { | |
| 139 | + | case <-paintDone: | |
| 140 | + | t.Fatal("first paint completed before cancellation returned") | |
| 141 | + | default: | |
| 142 | + | } | |
| 143 | + | close(releasePaint) | |
| 144 | + | awaitTestCompletion(t, paintDone, "delayed first paint did not finish") | |
| 145 | + | awaitTestCompletion(t, handshakeDone, "handshake cancellation did not finish") | |
| 146 | + | } | |
| 147 | + | ||
| 110 | 148 | func TestHandshakeTimeoutClosesBlockedReceive(t *testing.T) { | |
| 111 | 149 | clock := &signalClock{timers: make(chan *signalTimer, 1)} | |
| 112 | 150 | d := newTestDaemon(t, nil, clock) | |
| @@ -107,6 +107,44 @@ func TestBoundedHandshakeCancellationDoesNotWaitForUninterruptibleOperation(t *t | |||||
| 107 | awaitTestCompletion(t, operationDone, "handshake operation did not finish") | 107 | awaitTestCompletion(t, operationDone, "handshake operation did not finish") | ||
| 108 | } | 108 | } | ||
| 109 | 109 | ||||
| 110 | + | func TestHandshakeFirstPaintCancellationDoesNotRacePaintResult(t *testing.T) { | |||
| 111 | + | pty, releasePTY := newBlockingPTY(t) | |||
| 112 | + | defer releasePTY() | |||
| 113 | + | d := newTestDaemon(t, newFactory(t, pty), stubClock{}) | |||
| 114 | + | ctx, cancel := context.WithCancel(context.Background()) | |||
| 115 | + | defer cancel() | |||
| 116 | + | ||||
| 117 | + | paintStarted := make(chan struct{}) | |||
| 118 | + | releasePaint := make(chan struct{}) | |||
| 119 | + | paintDone := make(chan struct{}) | |||
| 120 | + | d.afterAttachmentEffectAdmitted = func(attachmentConnectionToken) { | |||
| 121 | + | close(paintStarted) | |||
| 122 | + | <-ctx.Done() | |||
| 123 | + | <-releasePaint | |||
| 124 | + | close(paintDone) | |||
| 125 | + | } | |||
| 126 | + | ||||
| 127 | + | tr := newHandshakeBlockingTransport(false) | |||
| 128 | + | handshakeDone := make(chan struct{}) | |||
| 129 | + | go func() { | |||
| 130 | + | d.handleHelloWithContext(ctx, make(chan struct{}), func() {}, func() {}, tr, | |||
| 131 | + | mustHello(ports.IntentNew, "paint-cancel", domain.Size{Cols: 80, Rows: 24})) | |||
| 132 | + | close(handshakeDone) | |||
| 133 | + | }() | |||
| 134 | + | ||||
| 135 | + | awaitTestCompletion(t, paintStarted, "first paint did not start") | |||
| 136 | + | cancel() | |||
| 137 | + | awaitTestCompletion(t, tr.closed, "handshake cancellation did not close the transport") | |||
| 138 | + | select { | |||
| 139 | + | case <-paintDone: | |||
| 140 | + | t.Fatal("first paint completed before cancellation returned") | |||
| 141 | + | default: | |||
| 142 | + | } | |||
| 143 | + | close(releasePaint) | |||
| 144 | + | awaitTestCompletion(t, paintDone, "delayed first paint did not finish") | |||
| 145 | + | awaitTestCompletion(t, handshakeDone, "handshake cancellation did not finish") | |||
| 146 | + | } | |||
| 147 | + | ||||
| 110 | func TestHandshakeTimeoutClosesBlockedReceive(t *testing.T) { | 148 | func TestHandshakeTimeoutClosesBlockedReceive(t *testing.T) { | ||
| 111 | clock := &signalClock{timers: make(chan *signalTimer, 1)} | 149 | clock := &signalClock{timers: make(chan *signalTimer, 1)} | ||
| 112 | d := newTestDaemon(t, nil, clock) | 150 | d := newTestDaemon(t, nil, clock) | ||
M internal/usecase/daemon/picker.go +5 -6
| @@ -682,9 +682,6 @@ func (d *Daemon) sendRemoteAttachTargetForAttachment(token attachmentConnectionT | |||
| 682 | 682 | if key.Validate() != nil || target.Session != key.ID() || target.Stopped || target.Name != "" || !d.remotePickerTargetReady(key) { | |
| 683 | 683 | return errAttachmentTransition | |
| 684 | 684 | } | |
| 685 | - | if guard.closePicker { | |
| 686 | - | d.closePicker(token.ac) | |
| 687 | - | } | |
| 688 | 685 | if err := token.sendControl(ports.Frame{Type: ports.MsgAttachTarget, Payload: ports.MarshalAttachTarget(ports.AttachTarget{ | |
| 689 | 686 | Endpoint: key.Host, | |
| 690 | 687 | Session: key.Name, | |
| @@ -692,6 +689,9 @@ func (d *Daemon) sendRemoteAttachTargetForAttachment(token attachmentConnectionT | |||
| 692 | 689 | })}); err != nil { | |
| 693 | 690 | return err | |
| 694 | 691 | } | |
| 692 | + | if guard.closePicker { | |
| 693 | + | d.closePicker(token.ac) | |
| 694 | + | } | |
| 695 | 695 | d.clientGoneForAttachment(token, false) | |
| 696 | 696 | return nil | |
| 697 | 697 | } | |
| @@ -981,15 +981,14 @@ func targetMatchesLifecycle(target picker.Target, name string, createdAt int64) | |||
| 981 | 981 | ||
| 982 | 982 | // stealClientForTarget is retained for direct-ID callers and tests. Named | |
| 983 | 983 | // targets must use switchToTarget so resolution and commit share d.mu. | |
| 984 | - | func (d *Daemon) stealClientForTarget(from *session, ac *attachedClient, targetSess *session, target picker.Target) *attachedClient { | |
| 984 | + | func (d *Daemon) stealClientForTarget(from *session, ac *attachedClient, targetSess *session, target picker.Target) { | |
| 985 | 985 | d.mu.Lock() | |
| 986 | 986 | _, transition, switched := d.switchToActiveTargetLocked(from, ac, targetSess, target, sessionHandoffGuard{}, nil, "") | |
| 987 | 987 | d.mu.Unlock() | |
| 988 | 988 | if !switched { | |
| 989 | - | return nil | |
| 989 | + | return | |
| 990 | 990 | } | |
| 991 | 991 | d.deferAttachmentTransitionCleanups(transition) | |
| 992 | - | return nil | |
| 993 | 992 | } | |
| 994 | 993 | ||
| 995 | 994 | // resumeStoppedAndSwitch is retained for direct callers and tests. It resolves | |
| @@ -682,9 +682,6 @@ func (d *Daemon) sendRemoteAttachTargetForAttachment(token attachmentConnectionT | |||||
| 682 | if key.Validate() != nil || target.Session != key.ID() || target.Stopped || target.Name != "" || !d.remotePickerTargetReady(key) { | 682 | if key.Validate() != nil || target.Session != key.ID() || target.Stopped || target.Name != "" || !d.remotePickerTargetReady(key) { | ||
| 683 | return errAttachmentTransition | 683 | return errAttachmentTransition | ||
| 684 | } | 684 | } | ||
| 685 | - | if guard.closePicker { | |||
| 686 | - | d.closePicker(token.ac) | |||
| 687 | - | } | |||
| 688 | if err := token.sendControl(ports.Frame{Type: ports.MsgAttachTarget, Payload: ports.MarshalAttachTarget(ports.AttachTarget{ | 685 | if err := token.sendControl(ports.Frame{Type: ports.MsgAttachTarget, Payload: ports.MarshalAttachTarget(ports.AttachTarget{ | ||
| 689 | Endpoint: key.Host, | 686 | Endpoint: key.Host, | ||
| 690 | Session: key.Name, | 687 | Session: key.Name, | ||
| @@ -692,6 +689,9 @@ func (d *Daemon) sendRemoteAttachTargetForAttachment(token attachmentConnectionT | |||||
| 692 | })}); err != nil { | 689 | })}); err != nil { | ||
| 693 | return err | 690 | return err | ||
| 694 | } | 691 | } | ||
| 692 | + | if guard.closePicker { | |||
| 693 | + | d.closePicker(token.ac) | |||
| 694 | + | } | |||
| 695 | d.clientGoneForAttachment(token, false) | 695 | d.clientGoneForAttachment(token, false) | ||
| 696 | return nil | 696 | return nil | ||
| 697 | } | 697 | } | ||
| @@ -981,15 +981,14 @@ func targetMatchesLifecycle(target picker.Target, name string, createdAt int64) | |||||
| 981 | 981 | ||||
| 982 | // stealClientForTarget is retained for direct-ID callers and tests. Named | 982 | // stealClientForTarget is retained for direct-ID callers and tests. Named | ||
| 983 | // targets must use switchToTarget so resolution and commit share d.mu. | 983 | // targets must use switchToTarget so resolution and commit share d.mu. | ||
| 984 | - | func (d *Daemon) stealClientForTarget(from *session, ac *attachedClient, targetSess *session, target picker.Target) *attachedClient { | 984 | + | func (d *Daemon) stealClientForTarget(from *session, ac *attachedClient, targetSess *session, target picker.Target) { |
| 985 | d.mu.Lock() | 985 | d.mu.Lock() | ||
| 986 | _, transition, switched := d.switchToActiveTargetLocked(from, ac, targetSess, target, sessionHandoffGuard{}, nil, "") | 986 | _, transition, switched := d.switchToActiveTargetLocked(from, ac, targetSess, target, sessionHandoffGuard{}, nil, "") | ||
| 987 | d.mu.Unlock() | 987 | d.mu.Unlock() | ||
| 988 | if !switched { | 988 | if !switched { | ||
| 989 | - | return nil | 989 | + | return |
| 990 | } | 990 | } | ||
| 991 | d.deferAttachmentTransitionCleanups(transition) | 991 | d.deferAttachmentTransitionCleanups(transition) | ||
| 992 | - | return nil | |||
| 993 | } | 992 | } | ||
| 994 | 993 | ||||
| 995 | // resumeStoppedAndSwitch is retained for direct callers and tests. It resolves | 994 | // resumeStoppedAndSwitch is retained for direct callers and tests. It resolves | ||
M internal/usecase/daemon/remote_picker_test.go +47 -0
| @@ -444,6 +444,53 @@ func TestRemotePickerHandoffSendsTargetAndLeavesNoShadowSession(t *testing.T) { | |||
| 444 | 444 | d.mu.Unlock() | |
| 445 | 445 | } | |
| 446 | 446 | ||
| 447 | + | func TestRemotePickerHandoffSendFailureKeepsPickerOpen(t *testing.T) { | |
| 448 | + | d := newRemotePickerDaemon(nil) | |
| 449 | + | d.remoteCatalog.replaceCache([]ports.RemoteCatalogCacheEntry{{ | |
| 450 | + | Host: "arch", FetchedAt: time.Unix(10, 0), Sessions: []ports.RemoteCatalogSession{{Name: "work", State: "running"}}, | |
| 451 | + | }}) | |
| 452 | + | const sendErr = "remote attach send failed" | |
| 453 | + | tr := &remotePickerSendErrorTransport{err: errors.New(sendErr)} | |
| 454 | + | ac := &attachedClient{tr: tr, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24}} | |
| 455 | + | ac.initOverlays() | |
| 456 | + | tb := newTab(newQuietPTY(), domain.Size{Cols: 80, Rows: 22}) | |
| 457 | + | sess := &session{sessionCore: sessionCore{id: "local", name: "local", attachments: map[*attachedClient]struct{}{ac: {}}}, tabs: []*tab{tb}} | |
| 458 | + | publishTiledPaneOwners(sess, tb) | |
| 459 | + | ac.setSession(sess) | |
| 460 | + | d.sessions[sess.id] = sess | |
| 461 | + | ac.overlays.pickerMu.Lock() | |
| 462 | + | ac.overlays.picker = picker.New(nil, picker.SelectionConfig{}) | |
| 463 | + | ac.overlays.pickerGeneration++ | |
| 464 | + | ac.overlays.pickerMu.Unlock() | |
| 465 | + | ||
| 466 | + | gone := make(chan struct{}) | |
| 467 | + | d.afterClientGoneDetach = func() { close(gone) } | |
| 468 | + | token := sess.attachmentToken(ac, tr) | |
| 469 | + | effect, admitted := ac.beginAttachmentEffect(token) | |
| 470 | + | require.True(t, admitted) | |
| 471 | + | token.effect = effect | |
| 472 | + | defer effect.End() | |
| 473 | + | key := domain.RemoteSessionKey{Host: "arch", Name: "work"} | |
| 474 | + | target := picker.Target{Session: key.ID(), RemoteKey: &key} | |
| 475 | + | err := d.sendRemoteAttachTargetForAttachment(token, target, key, sessionHandoffGuard{closePicker: true}, "picker-select") | |
| 476 | + | require.EqualError(t, err, sendErr) | |
| 477 | + | require.True(t, ac.overlays.pickerActive(), "failed control send must leave the picker open") | |
| 478 | + | select { | |
| 479 | + | case <-gone: | |
| 480 | + | t.Fatal("failed control send reached clientGoneForAttachment") | |
| 481 | + | default: | |
| 482 | + | } | |
| 483 | + | require.Same(t, sess, ac.currentAttachmentSession()) | |
| 484 | + | } | |
| 485 | + | ||
| 486 | + | type remotePickerSendErrorTransport struct { | |
| 487 | + | err error | |
| 488 | + | } | |
| 489 | + | ||
| 490 | + | func (t *remotePickerSendErrorTransport) Send(ports.Frame) error { return t.err } | |
| 491 | + | func (*remotePickerSendErrorTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF } | |
| 492 | + | func (*remotePickerSendErrorTransport) Close() error { return nil } | |
| 493 | + | ||
| 447 | 494 | func addRemoteRefreshPickerOwner(t *testing.T, d *Daemon, id domain.SessionID) (*session, *attachedClient, chan ports.Frame) { | |
| 448 | 495 | t.Helper() | |
| 449 | 496 | tr, sends := newCapturingTransport(t) | |
| @@ -444,6 +444,53 @@ func TestRemotePickerHandoffSendsTargetAndLeavesNoShadowSession(t *testing.T) { | |||||
| 444 | d.mu.Unlock() | 444 | d.mu.Unlock() | ||
| 445 | } | 445 | } | ||
| 446 | 446 | ||||
| 447 | + | func TestRemotePickerHandoffSendFailureKeepsPickerOpen(t *testing.T) { | |||
| 448 | + | d := newRemotePickerDaemon(nil) | |||
| 449 | + | d.remoteCatalog.replaceCache([]ports.RemoteCatalogCacheEntry{{ | |||
| 450 | + | Host: "arch", FetchedAt: time.Unix(10, 0), Sessions: []ports.RemoteCatalogSession{{Name: "work", State: "running"}}, | |||
| 451 | + | }}) | |||
| 452 | + | const sendErr = "remote attach send failed" | |||
| 453 | + | tr := &remotePickerSendErrorTransport{err: errors.New(sendErr)} | |||
| 454 | + | ac := &attachedClient{tr: tr, output: newOutputStateStream(), size: domain.Size{Cols: 80, Rows: 24}} | |||
| 455 | + | ac.initOverlays() | |||
| 456 | + | tb := newTab(newQuietPTY(), domain.Size{Cols: 80, Rows: 22}) | |||
| 457 | + | sess := &session{sessionCore: sessionCore{id: "local", name: "local", attachments: map[*attachedClient]struct{}{ac: {}}}, tabs: []*tab{tb}} | |||
| 458 | + | publishTiledPaneOwners(sess, tb) | |||
| 459 | + | ac.setSession(sess) | |||
| 460 | + | d.sessions[sess.id] = sess | |||
| 461 | + | ac.overlays.pickerMu.Lock() | |||
| 462 | + | ac.overlays.picker = picker.New(nil, picker.SelectionConfig{}) | |||
| 463 | + | ac.overlays.pickerGeneration++ | |||
| 464 | + | ac.overlays.pickerMu.Unlock() | |||
| 465 | + | ||||
| 466 | + | gone := make(chan struct{}) | |||
| 467 | + | d.afterClientGoneDetach = func() { close(gone) } | |||
| 468 | + | token := sess.attachmentToken(ac, tr) | |||
| 469 | + | effect, admitted := ac.beginAttachmentEffect(token) | |||
| 470 | + | require.True(t, admitted) | |||
| 471 | + | token.effect = effect | |||
| 472 | + | defer effect.End() | |||
| 473 | + | key := domain.RemoteSessionKey{Host: "arch", Name: "work"} | |||
| 474 | + | target := picker.Target{Session: key.ID(), RemoteKey: &key} | |||
| 475 | + | err := d.sendRemoteAttachTargetForAttachment(token, target, key, sessionHandoffGuard{closePicker: true}, "picker-select") | |||
| 476 | + | require.EqualError(t, err, sendErr) | |||
| 477 | + | require.True(t, ac.overlays.pickerActive(), "failed control send must leave the picker open") | |||
| 478 | + | select { | |||
| 479 | + | case <-gone: | |||
| 480 | + | t.Fatal("failed control send reached clientGoneForAttachment") | |||
| 481 | + | default: | |||
| 482 | + | } | |||
| 483 | + | require.Same(t, sess, ac.currentAttachmentSession()) | |||
| 484 | + | } | |||
| 485 | + | ||||
| 486 | + | type remotePickerSendErrorTransport struct { | |||
| 487 | + | err error | |||
| 488 | + | } | |||
| 489 | + | ||||
| 490 | + | func (t *remotePickerSendErrorTransport) Send(ports.Frame) error { return t.err } | |||
| 491 | + | func (*remotePickerSendErrorTransport) Recv() (ports.Frame, error) { return ports.Frame{}, io.EOF } | |||
| 492 | + | func (*remotePickerSendErrorTransport) Close() error { return nil } | |||
| 493 | + | ||||
| 447 | func addRemoteRefreshPickerOwner(t *testing.T, d *Daemon, id domain.SessionID) (*session, *attachedClient, chan ports.Frame) { | 494 | func addRemoteRefreshPickerOwner(t *testing.T, d *Daemon, id domain.SessionID) (*session, *attachedClient, chan ports.Frame) { | ||
| 448 | t.Helper() | 495 | t.Helper() | ||
| 449 | tr, sends := newCapturingTransport(t) | 496 | tr, sends := newCapturingTransport(t) | ||