49ca2bd fix(remote): address PR review findings · vev-vt

M internal/adapters/sshstdio/transport.go +8 -8


@@ -35,15 +35,15 @@ type closeFunc func() error
35	35		type eofErrFunc func() error
36	36
37	37		// NewTransport wraps separate reader/writer streams as a framed Transport.
38		-	func NewTransport(r io.Reader, w io.Writer, close closeFunc) ports.Transport {
39		-	return newTransport(r, w, close, nil)
	38	+	func NewTransport(r io.Reader, w io.Writer, closeFn closeFunc) ports.Transport {
	39	+	return newTransport(r, w, closeFn, nil)
40	40		}
41	41
42		-	func newTransport(r io.Reader, w io.Writer, close closeFunc, eofErr eofErrFunc) ports.Transport {
43		-	if close == nil {
44		-	close = func() error { return nil }
	42	+	func newTransport(r io.Reader, w io.Writer, closeFn closeFunc, eofErr eofErrFunc) ports.Transport {
	43	+	if closeFn == nil {
	44	+	closeFn = func() error { return nil }
45	45		}
46		-	return &transport{r: r, w: w, close: close, eofErr: eofErr}
	46	+	return &transport{r: r, w: w, close: closeFn, eofErr: eofErr}
47	47		}
48	48
49	49		type transport struct {

@@ -93,7 +93,7 @@ func (t *transport) Recv() (ports.Frame, error) {
93	93		t.readBuf = t.readBuf[:n]
94	94		}
95	95		if _, err := io.ReadFull(t.r, t.readBuf); err != nil {
96		-	return ports.Frame{}, err
	96	+	return ports.Frame{}, t.mapEOFError(err)
97	97		}
98	98
99	99		payload := append([]byte(nil), t.readBuf[1:]...)

@@ -101,7 +101,7 @@ func (t *transport) Recv() (ports.Frame, error) {
101	101		}
102	102
103	103		func (t *transport) mapEOFError(err error) error {
104		-	if t.eofErr == nil \|\| !errors.Is(err, io.EOF) {
	104	+	if t.eofErr == nil \|\| (!errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF)) {
105	105		return err
106	106		}
107	107		if sshErr := t.eofErr(); sshErr != nil {


@@ -35,15 +35,15 @@ type closeFunc func() error
35		type eofErrFunc func() error	35		type eofErrFunc func() error
36			36
37		// NewTransport wraps separate reader/writer streams as a framed Transport.	37		// NewTransport wraps separate reader/writer streams as a framed Transport.
38	-	func NewTransport(r io.Reader, w io.Writer, close closeFunc) ports.Transport {	38	+	func NewTransport(r io.Reader, w io.Writer, closeFn closeFunc) ports.Transport {
39	-	return newTransport(r, w, close, nil)	39	+	return newTransport(r, w, closeFn, nil)
40		}	40		}
41			41
42	-	func newTransport(r io.Reader, w io.Writer, close closeFunc, eofErr eofErrFunc) ports.Transport {	42	+	func newTransport(r io.Reader, w io.Writer, closeFn closeFunc, eofErr eofErrFunc) ports.Transport {
43	-	if close == nil {	43	+	if closeFn == nil {
44	-	close = func() error { return nil }	44	+	closeFn = func() error { return nil }
45		}	45		}
46	-	return &transport{r: r, w: w, close: close, eofErr: eofErr}	46	+	return &transport{r: r, w: w, close: closeFn, eofErr: eofErr}
47		}	47		}
48			48
49		type transport struct {	49		type transport struct {

@@ -93,7 +93,7 @@ func (t *transport) Recv() (ports.Frame, error) {
93		t.readBuf = t.readBuf[:n]	93		t.readBuf = t.readBuf[:n]
94		}	94		}
95		if _, err := io.ReadFull(t.r, t.readBuf); err != nil {	95		if _, err := io.ReadFull(t.r, t.readBuf); err != nil {
96	-	return ports.Frame{}, err	96	+	return ports.Frame{}, t.mapEOFError(err)
97		}	97		}
98			98
99		payload := append([]byte(nil), t.readBuf[1:]...)	99		payload := append([]byte(nil), t.readBuf[1:]...)

@@ -101,7 +101,7 @@ func (t *transport) Recv() (ports.Frame, error) {
101		}	101		}
102			102
103		func (t *transport) mapEOFError(err error) error {	103		func (t *transport) mapEOFError(err error) error {
104	-	if t.eofErr == nil \|\| !errors.Is(err, io.EOF) {	104	+	if t.eofErr == nil \|\| (!errors.Is(err, io.EOF) && !errors.Is(err, io.ErrUnexpectedEOF)) {
105		return err	105		return err
106		}	106		}
107		if sshErr := t.eofErr(); sshErr != nil {	107		if sshErr := t.eofErr(); sshErr != nil {

M internal/adapters/sshstdio/transport_test.go +11 -1


@@ -128,8 +128,18 @@ func TestDialContextCanceledBeforeStart(t *testing.T) {
128	128		}
129	129		}
130	130
	131	+	func TestRecvMapsUnexpectedEOFDuringHeaderToSSHExit(t *testing.T) {
	132	+	sshErr := errors.New("ssh exited with status 255")
	133	+	tr := newTransport(strings.NewReader("\x00"), io.Discard, nil, func() error { return sshErr })
	134	+
	135	+	_, err := tr.Recv()
	136	+	if !errors.Is(err, sshErr) {
	137	+	t.Fatalf("Recv error = %v, want %v", err, sshErr)
	138	+	}
	139	+	}
	140	+
131	141		func TestRecvReportsSSHExitWhenProcessClosesBeforeFrame(t *testing.T) {
132		-	cmd := exec.Command("sh", "-c", "echo connection refused >&2; exit 255")
	142	+	cmd := exec.CommandContext(context.Background(), "sh", "-c", "echo connection refused >&2; exit 255")
133	143		stdin, err := cmd.StdinPipe()
134	144		if err != nil {
135	145		t.Fatalf("StdinPipe: %v", err)


@@ -128,8 +128,18 @@ func TestDialContextCanceledBeforeStart(t *testing.T) {
128		}	128		}
129		}	129		}
130			130
			131	+	func TestRecvMapsUnexpectedEOFDuringHeaderToSSHExit(t *testing.T) {
			132	+	sshErr := errors.New("ssh exited with status 255")
			133	+	tr := newTransport(strings.NewReader("\x00"), io.Discard, nil, func() error { return sshErr })
			134	+
			135	+	_, err := tr.Recv()
			136	+	if !errors.Is(err, sshErr) {
			137	+	t.Fatalf("Recv error = %v, want %v", err, sshErr)
			138	+	}
			139	+	}
			140	+
131		func TestRecvReportsSSHExitWhenProcessClosesBeforeFrame(t *testing.T) {	141		func TestRecvReportsSSHExitWhenProcessClosesBeforeFrame(t *testing.T) {
132	-	cmd := exec.Command("sh", "-c", "echo connection refused >&2; exit 255")	142	+	cmd := exec.CommandContext(context.Background(), "sh", "-c", "echo connection refused >&2; exit 255")
133		stdin, err := cmd.StdinPipe()	143		stdin, err := cmd.StdinPipe()
134		if err != nil {	144		if err != nil {
135		t.Fatalf("StdinPipe: %v", err)	145		t.Fatalf("StdinPipe: %v", err)

M internal/app/run.go +6 -0


@@ -777,6 +777,9 @@ func parseUDPPortRange(value string) (udpPortRange, error) {
777	777		return udpPortRange{}, fmt.Errorf("invalid %s %q: %w", envUDPPortRange, value, err)
778	778		}
779	779		if start == 0 {
	780	+	if hasRange {
	781	+	return udpPortRange{}, fmt.Errorf("invalid %s %q: port 0 (ephemeral) cannot be combined with a range", envUDPPortRange, value)
	782	+	}
780	783		return udpPortRange{start: 0, end: 0}, nil
781	784		}
782	785		end := start

@@ -810,6 +813,9 @@ func listenUDPInRange(ctx context.Context, r udpPortRange) (net.PacketConn, erro
810	813		if r.start == 0 {
811	814		return lc.ListenPacket(ctx, "udp", ":0")
812	815		}
	816	+	if r.start < 0 \|\| r.end < r.start {
	817	+	return nil, fmt.Errorf("invalid UDP port range %d-%d", r.start, r.end)
	818	+	}
813	819		var lastErr error
814	820		for port := r.start; port <= r.end; port++ {
815	821		conn, err := lc.ListenPacket(ctx, "udp", fmt.Sprintf(":%d", port))


@@ -777,6 +777,9 @@ func parseUDPPortRange(value string) (udpPortRange, error) {
777	return udpPortRange{}, fmt.Errorf("invalid %s %q: %w", envUDPPortRange, value, err)	777		return udpPortRange{}, fmt.Errorf("invalid %s %q: %w", envUDPPortRange, value, err)
778	}	778		}
779	if start == 0 {	779		if start == 0 {
		780	+	if hasRange {
		781	+	return udpPortRange{}, fmt.Errorf("invalid %s %q: port 0 (ephemeral) cannot be combined with a range", envUDPPortRange, value)
		782	+	}
780	return udpPortRange{start: 0, end: 0}, nil	783		return udpPortRange{start: 0, end: 0}, nil
781	}	784		}
782	end := start	785		end := start

@@ -810,6 +813,9 @@ func listenUDPInRange(ctx context.Context, r udpPortRange) (net.PacketConn, erro
810	if r.start == 0 {	813		if r.start == 0 {
811	return lc.ListenPacket(ctx, "udp", ":0")	814		return lc.ListenPacket(ctx, "udp", ":0")
812	}	815		}
		816	+	if r.start < 0 \|\| r.end < r.start {
		817	+	return nil, fmt.Errorf("invalid UDP port range %d-%d", r.start, r.end)
		818	+	}
813	var lastErr error	819		var lastErr error
814	for port := r.start; port <= r.end; port++ {	820		for port := r.start; port <= r.end; port++ {
815	conn, err := lc.ListenPacket(ctx, "udp", fmt.Sprintf(":%d", port))	821		conn, err := lc.ListenPacket(ctx, "udp", fmt.Sprintf(":%d", port))

M internal/app/run_test.go +22 -3


@@ -644,6 +644,7 @@ func TestParseUDPPortRange(t *testing.T) {
644	644		}{
645	645		{name: "empty uses default", value: "", want: udpPortRange{start: defaultUDPPortStart, end: defaultUDPPortEnd}},
646	646		{name: "zero is ephemeral", value: "0", want: udpPortRange{start: 0, end: 0}},
	647	+	{name: "zero range rejected", value: "0-100", wantError: true},
647	648		{name: "single port", value: "61000", want: udpPortRange{start: 61000, end: 61000}},
648	649		{name: "range", value: "61000-61023", want: udpPortRange{start: 61000, end: 61023}},
649	650		{name: "range with spaces", value: " 61000 - 61023 ", want: udpPortRange{start: 61000, end: 61023}},

@@ -679,7 +680,15 @@ func TestListenUDPInRange(t *testing.T) {
679	680		t.Fatal(err)
680	681		}
681	682		defer func() { _ = occupied.Close() }()
682		-	busyPort := occupied.LocalAddr().(*net.UDPAddr).Port
	683	+	udpPort := func(t *testing.T, conn net.PacketConn) int {
	684	+	t.Helper()
	685	+	addr, ok := conn.LocalAddr().(*net.UDPAddr)
	686	+	if !ok {
	687	+	t.Fatalf("LocalAddr() = %T, want *net.UDPAddr", conn.LocalAddr())
	688	+	}
	689	+	return addr.Port
	690	+	}
	691	+	busyPort := udpPort(t, occupied)
683	692
684	693		t.Run("skips busy port", func(t *testing.T) {
685	694		conn, err := listenUDPInRange(context.Background(), udpPortRange{start: busyPort, end: busyPort + 8})

@@ -687,7 +696,7 @@ func TestListenUDPInRange(t *testing.T) {
687	696		t.Fatalf("listenUDPInRange error = %v", err)
688	697		}
689	698		defer func() { _ = conn.Close() }()
690		-	got := conn.LocalAddr().(*net.UDPAddr).Port
	699	+	got := udpPort(t, conn)
691	700		if got == busyPort {
692	701		t.Fatalf("bound busy port %d, want a different port in range", busyPort)
693	702		}

@@ -706,13 +715,23 @@ func TestListenUDPInRange(t *testing.T) {
706	715		}
707	716		})
708	717
	718	+	t.Run("invalid direct range errors", func(t *testing.T) {
	719	+	_, err := listenUDPInRange(context.Background(), udpPortRange{start: busyPort + 8, end: busyPort})
	720	+	if err == nil {
	721	+	t.Fatal("listenUDPInRange error = nil, want invalid-range error")
	722	+	}
	723	+	if !strings.Contains(err.Error(), "invalid UDP port range") {
	724	+	t.Fatalf("error = %v, want 'invalid UDP port range'", err)
	725	+	}
	726	+	})
	727	+
709	728		t.Run("ephemeral binds", func(t *testing.T) {
710	729		conn, err := listenUDPInRange(context.Background(), udpPortRange{start: 0, end: 0})
711	730		if err != nil {
712	731		t.Fatalf("listenUDPInRange ephemeral error = %v", err)
713	732		}
714	733		defer func() { _ = conn.Close() }()
715		-	if conn.LocalAddr().(*net.UDPAddr).Port == 0 {
	734	+	if udpPort(t, conn) == 0 {
716	735		t.Fatal("ephemeral bind returned port 0")
717	736		}
718	737		})

 	}{
 		{name: "empty uses default", value: "", want: udpPortRange{start: defaultUDPPortStart, end: defaultUDPPortEnd}},
 		{name: "zero is ephemeral", value: "0", want: udpPortRange{start: 0, end: 0}},
+		{name: "zero range rejected", value: "0-100", wantError: true},
 		{name: "single port", value: "61000", want: udpPortRange{start: 61000, end: 61000}},
 		{name: "range", value: "61000-61023", want: udpPortRange{start: 61000, end: 61023}},
 		{name: "range with spaces", value: " 61000 - 61023 ", want: udpPortRange{start: 61000, end: 61023}},
 		t.Fatal(err)
 	}
 	defer func() { _ = occupied.Close() }()
-	busyPort := occupied.LocalAddr().(*net.UDPAddr).Port
+	udpPort := func(t *testing.T, conn net.PacketConn) int {
+		t.Helper()
+		addr, ok := conn.LocalAddr().(*net.UDPAddr)
+		if !ok {
+			t.Fatalf("LocalAddr() = %T, want *net.UDPAddr", conn.LocalAddr())
+		}
+		return addr.Port
+	}
+	busyPort := udpPort(t, occupied)
 	t.Run("skips busy port", func(t *testing.T) {
 		conn, err := listenUDPInRange(context.Background(), udpPortRange{start: busyPort, end: busyPort + 8})
 			t.Fatalf("listenUDPInRange error = %v", err)
 		}
 		defer func() { _ = conn.Close() }()
-		got := conn.LocalAddr().(*net.UDPAddr).Port
+		got := udpPort(t, conn)
 		if got == busyPort {
 			t.Fatalf("bound busy port %d, want a different port in range", busyPort)
 		}
 		}
 	})
+	t.Run("invalid direct range errors", func(t *testing.T) {
+		_, err := listenUDPInRange(context.Background(), udpPortRange{start: busyPort + 8, end: busyPort})
+		if err == nil {
+			t.Fatal("listenUDPInRange error = nil, want invalid-range error")
+		}
+		if !strings.Contains(err.Error(), "invalid UDP port range") {
+			t.Fatalf("error = %v, want 'invalid UDP port range'", err)
+		}
+	})
 	t.Run("ephemeral binds", func(t *testing.T) {
 		conn, err := listenUDPInRange(context.Background(), udpPortRange{start: 0, end: 0})
 		if err != nil {
 			t.Fatalf("listenUDPInRange ephemeral error = %v", err)
 		}
 		defer func() { _ = conn.Close() }()
-		if conn.LocalAddr().(*net.UDPAddr).Port == 0 {
+		if udpPort(t, conn) == 0 {
 			t.Fatal("ephemeral bind returned port 0")
 		}
 	})

M scripts/debug-remote-attach.sh +3 -3


@@ -109,7 +109,7 @@ printf 'target: %s\nattach_target: %s\nmode: %s\nduration: %s\n' "$target" "$att
109	109		) >"$out_dir/git.txt" 2>&1 \|\| true
110	110
111	111		run_capture local-version "$vev_bin" --version
112		-	ssh -o BatchMode=yes -o ConnectTimeout=8 "$target" 'command -v vev; vev --version; pgrep -a vev \|\| true; state="${XDG_STATE_HOME:-$HOME/.local/state}/vev"; echo "state=$state"' \
	112	+	ssh -o BatchMode=yes -o ConnectTimeout=8 "$target" 'set -e; command -v vev; vev --version; pgrep -a vev \|\| true; state="${XDG_STATE_HOME:-$HOME/.local/state}/vev"; echo "state=$state"' \
113	113		>"$out_dir/remote-version.out" 2>"$out_dir/remote-version.err" \|\| echo "$?" >"$out_dir/remote-version.exit"
114	114
115	115		copy_local_logs before

@@ -119,7 +119,7 @@ attach_log="$out_dir/attach.typescript"
119	119		attach_err="$out_dir/attach.err"
120	120		cmd=("$vev_bin" attach "$attach_target")
121	121		if [[ "$mode" == "default" ]]; then
122		-	env_cmd=(env -u VEV)
	122	+	env_cmd=(env -u VEV -u VEV_REMOTE_TRANSPORT)
123	123		else
124	124		env_cmd=(env -u VEV "VEV_REMOTE_TRANSPORT=$mode")
125	125		fi

@@ -150,7 +150,7 @@ copy_remote_logs after
150	150		sed 's/^/ /' "$attach_err" \| tail -40
151	151		fi
152	152		echo "recent local after logs:"
153		-	find "$out_dir/local-after" -type f -maxdepth 1 -print 2>/dev/null \| while read -r f; do
	153	+	find "$out_dir/local-after" -maxdepth 1 -type f -print0 2>/dev/null \| while IFS= read -r -d '' f; do
154	154		echo " --- $(basename "$f") ---"
155	155		tail -20 "$f" \| sed 's/^/ /'
156	156		done


@@ -109,7 +109,7 @@ printf 'target: %s\nattach_target: %s\nmode: %s\nduration: %s\n' "$target" "$att
109		) >"$out_dir/git.txt" 2>&1 \|\| true	109		) >"$out_dir/git.txt" 2>&1 \|\| true
110			110
111		run_capture local-version "$vev_bin" --version	111		run_capture local-version "$vev_bin" --version
112	-	ssh -o BatchMode=yes -o ConnectTimeout=8 "$target" 'command -v vev; vev --version; pgrep -a vev \|\| true; state="${XDG_STATE_HOME:-$HOME/.local/state}/vev"; echo "state=$state"' \	112	+	ssh -o BatchMode=yes -o ConnectTimeout=8 "$target" 'set -e; command -v vev; vev --version; pgrep -a vev \|\| true; state="${XDG_STATE_HOME:-$HOME/.local/state}/vev"; echo "state=$state"' \
113		>"$out_dir/remote-version.out" 2>"$out_dir/remote-version.err" \|\| echo "$?" >"$out_dir/remote-version.exit"	113		>"$out_dir/remote-version.out" 2>"$out_dir/remote-version.err" \|\| echo "$?" >"$out_dir/remote-version.exit"
114			114
115		copy_local_logs before	115		copy_local_logs before

@@ -119,7 +119,7 @@ attach_log="$out_dir/attach.typescript"
119		attach_err="$out_dir/attach.err"	119		attach_err="$out_dir/attach.err"
120		cmd=("$vev_bin" attach "$attach_target")	120		cmd=("$vev_bin" attach "$attach_target")
121		if [[ "$mode" == "default" ]]; then	121		if [[ "$mode" == "default" ]]; then
122	-	env_cmd=(env -u VEV)	122	+	env_cmd=(env -u VEV -u VEV_REMOTE_TRANSPORT)
123		else	123		else
124		env_cmd=(env -u VEV "VEV_REMOTE_TRANSPORT=$mode")	124		env_cmd=(env -u VEV "VEV_REMOTE_TRANSPORT=$mode")
125		fi	125		fi

@@ -150,7 +150,7 @@ copy_remote_logs after
150		sed 's/^/ /' "$attach_err" \| tail -40	150		sed 's/^/ /' "$attach_err" \| tail -40
151		fi	151		fi
152		echo "recent local after logs:"	152		echo "recent local after logs:"
153	-	find "$out_dir/local-after" -type f -maxdepth 1 -print 2>/dev/null \| while read -r f; do	153	+	find "$out_dir/local-after" -maxdepth 1 -type f -print0 2>/dev/null \| while IFS= read -r -d '' f; do
154		echo " --- $(basename "$f") ---"	154		echo " --- $(basename "$f") ---"
155		tail -20 "$f" \| sed 's/^/ /'	155		tail -20 "$f" \| sed 's/^/ /'
156		done	156		done