5e9cf03 fix(remote): bound udp bootstrap readiness · vev-vt

M internal/adapters/dgram/ssh_dialer.go +64 -10


@@ -19,8 +19,9 @@ import (
19	19		)
20	20
21	21		const (
22		-	maxBootstrapStderr = 64 * 1024
23		-	defaultProbeTimeout = 3 * time.Second
	22	+	maxBootstrapStderr = 64 * 1024
	23	+	defaultBootstrapTimeout = 10 * time.Second
	24	+	defaultProbeTimeout = 3 * time.Second
24	25		)
25	26
26	27		type bootstrapProcess interface {

@@ -68,10 +69,11 @@ func (b *limitedBuffer) String() string { return string(b.buf) }
68	69		// RemoteDialer connects to the authenticated datagram bootstrap for UDP remote
69	70		// attach. SSH stdio is selected only through the explicit stdio transport mode.
70	71		type RemoteDialer struct {
71		-	Target string
72		-	Session string
73		-	ProbeTimeout time.Duration
74		-	Log *slog.Logger
	72	+	Target string
	73	+	Session string
	74	+	BootstrapTimeout time.Duration
	75	+	ProbeTimeout time.Duration
	76	+	Log *slog.Logger
75	77		}
76	78
77	79		func NewRemoteDialer(target, session string) RemoteDialer {

@@ -86,12 +88,19 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) {
86	88		if err := ctx.Err(); err != nil {
87	89		return nil, err
88	90		}
	91	+	bootstrapTimeout := d.BootstrapTimeout
	92	+	if bootstrapTimeout <= 0 {
	93	+	bootstrapTimeout = defaultBootstrapTimeout
	94	+	}
89	95		probeTimeout := d.ProbeTimeout
90	96		if probeTimeout <= 0 {
91	97		probeTimeout = defaultProbeTimeout
92	98		}
	99	+	bootstrapCtx, bootstrapCancel := context.WithTimeout(ctx, bootstrapTimeout)
	100	+	defer bootstrapCancel()
	101	+
93	102		var stderr limitedBuffer
94		-	proc := startUDPBootstrap(ctx, d.Target, d.Session, &stderr)
	103	+	proc := startUDPBootstrap(bootstrapCtx, d.Target, d.Session, &stderr)
95	104		stdout, err := proc.StdoutPipe()
96	105		if err != nil {
97	106		return nil, udpUnavailable("bootstrap stdout", err, &stderr)

@@ -112,7 +121,7 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) {
112	121		_ = stdout.Close()
113	122		}()
114	123
115		-	ready, err := readUDPReady(stdout)
	124	+	ready, err := readUDPReadyContext(bootstrapCtx, stdout)
116	125		if err != nil {
117	126		_ = proc.Kill()
118	127		_ = proc.Wait()

@@ -120,14 +129,14 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) {
120	129		cleanup = false
121	130		return nil, udpUnavailable("read bootstrap readiness", err, &stderr)
122	131		}
123		-	if err := proc.Wait(); err != nil {
	132	+	if err := waitBootstrapContext(bootstrapCtx, proc); err != nil {
124	133		waited = true
125	134		cleanup = false
126	135		return nil, udpUnavailable("wait bootstrap", err, &stderr)
127	136		}
128	137		waited = true
129	138		cleanup = false
130		-	peer, err := net.ResolveUDPAddr("udp", net.JoinHostPort(sshTargetHost(d.Target), strconv.Itoa(ready.port)))
	139	+	peer, err := resolveUDPPeer(bootstrapCtx, d.Target, ready.port)
131	140		if err != nil {
132	141		return nil, udpUnavailable("resolve UDP peer", err, &stderr)
133	142		}

@@ -150,6 +159,51 @@ func (d RemoteDialer) Dial(ctx context.Context) (ports.Transport, error) {
150	159		return t, nil
151	160		}
152	161
	162	+	func readUDPReadyContext(ctx context.Context, r io.Reader) (udpReady, error) {
	163	+	type result struct {
	164	+	ready udpReady
	165	+	err error
	166	+	}
	167	+	ch := make(chan result, 1)
	168	+	go func() {
	169	+	ready, err := readUDPReady(r)
	170	+	ch <- result{ready: ready, err: err}
	171	+	}()
	172	+	select {
	173	+	case res := <-ch:
	174	+	return res.ready, res.err
	175	+	case <-ctx.Done():
	176	+	return udpReady{}, ctx.Err()
	177	+	}
	178	+	}
	179	+
	180	+	func waitBootstrapContext(ctx context.Context, proc bootstrapProcess) error {
	181	+	ch := make(chan error, 1)
	182	+	go func() { ch <- proc.Wait() }()
	183	+	select {
	184	+	case err := <-ch:
	185	+	return err
	186	+	case <-ctx.Done():
	187	+	_ = proc.Kill()
	188	+	return ctx.Err()
	189	+	}
	190	+	}
	191	+
	192	+	func resolveUDPPeer(ctx context.Context, target string, port int) (*net.UDPAddr, error) {
	193	+	host := sshTargetHost(target)
	194	+	if ip := net.ParseIP(host); ip != nil {
	195	+	return &net.UDPAddr{IP: ip, Port: port}, nil
	196	+	}
	197	+	addrs, err := net.DefaultResolver.LookupIPAddr(ctx, host)
	198	+	if err != nil {
	199	+	return nil, err
	200	+	}
	201	+	if len(addrs) == 0 {
	202	+	return nil, fmt.Errorf("no UDP peer addresses for %q", host)
	203	+	}
	204	+	return &net.UDPAddr{IP: addrs[0].IP, Zone: addrs[0].Zone, Port: port}, nil
	205	+	}
	206	+
153	207		type udpReady struct {
154	208		port int
155	209		key []byte

 )
 const (
-	maxBootstrapStderr  = 64 * 1024
+	maxBootstrapStderr      = 64 * 1024
-	defaultProbeTimeout = 3 * time.Second
+	defaultBootstrapTimeout = 10 * time.Second
+	defaultProbeTimeout     = 3 * time.Second
 )
 type bootstrapProcess interface {
 // RemoteDialer connects to the authenticated datagram bootstrap for UDP remote
 // attach. SSH stdio is selected only through the explicit stdio transport mode.
 type RemoteDialer struct {
-	Target       string
+	Target           string
-	Session      string
+	Session          string
-	ProbeTimeout time.Duration
+	BootstrapTimeout time.Duration
-	Log          *slog.Logger
+	ProbeTimeout     time.Duration
+	Log              *slog.Logger
 }
 func NewRemoteDialer(target, session string) RemoteDialer {
 	if err := ctx.Err(); err != nil {
 		return nil, err
 	}
+	bootstrapTimeout := d.BootstrapTimeout
+	if bootstrapTimeout <= 0 {
+		bootstrapTimeout = defaultBootstrapTimeout
+	}
 	probeTimeout := d.ProbeTimeout
 	if probeTimeout <= 0 {
 		probeTimeout = defaultProbeTimeout
 	}
+	bootstrapCtx, bootstrapCancel := context.WithTimeout(ctx, bootstrapTimeout)
+	defer bootstrapCancel()
 	var stderr limitedBuffer
-	proc := startUDPBootstrap(ctx, d.Target, d.Session, &stderr)
+	proc := startUDPBootstrap(bootstrapCtx, d.Target, d.Session, &stderr)
 	stdout, err := proc.StdoutPipe()
 	if err != nil {
 		return nil, udpUnavailable("bootstrap stdout", err, &stderr)
 		_ = stdout.Close()
 	}()
-	ready, err := readUDPReady(stdout)
+	ready, err := readUDPReadyContext(bootstrapCtx, stdout)
 	if err != nil {
 		_ = proc.Kill()
 		_ = proc.Wait()
 		cleanup = false
 		return nil, udpUnavailable("read bootstrap readiness", err, &stderr)
 	}
-	if err := proc.Wait(); err != nil {
+	if err := waitBootstrapContext(bootstrapCtx, proc); err != nil {
 		waited = true
 		cleanup = false
 		return nil, udpUnavailable("wait bootstrap", err, &stderr)
 	}
 	waited = true
 	cleanup = false
-	peer, err := net.ResolveUDPAddr("udp", net.JoinHostPort(sshTargetHost(d.Target), strconv.Itoa(ready.port)))
+	peer, err := resolveUDPPeer(bootstrapCtx, d.Target, ready.port)
 	if err != nil {
 		return nil, udpUnavailable("resolve UDP peer", err, &stderr)
 	}
 	return t, nil
 }
+func readUDPReadyContext(ctx context.Context, r io.Reader) (udpReady, error) {
+	type result struct {
+		ready udpReady
+		err   error
+	}
+	ch := make(chan result, 1)
+	go func() {
+		ready, err := readUDPReady(r)
+		ch <- result{ready: ready, err: err}
+	}()
+	select {
+	case res := <-ch:
+		return res.ready, res.err
+	case <-ctx.Done():
+		return udpReady{}, ctx.Err()
+	}
+}
+func waitBootstrapContext(ctx context.Context, proc bootstrapProcess) error {
+	ch := make(chan error, 1)
+	go func() { ch <- proc.Wait() }()
+	select {
+	case err := <-ch:
+		return err
+	case <-ctx.Done():
+		_ = proc.Kill()
+		return ctx.Err()
+	}
+}
+func resolveUDPPeer(ctx context.Context, target string, port int) (*net.UDPAddr, error) {
+	host := sshTargetHost(target)
+	if ip := net.ParseIP(host); ip != nil {
+		return &net.UDPAddr{IP: ip, Port: port}, nil
+	}
+	addrs, err := net.DefaultResolver.LookupIPAddr(ctx, host)
+	if err != nil {
+		return nil, err
+	}
+	if len(addrs) == 0 {
+		return nil, fmt.Errorf("no UDP peer addresses for %q", host)
+	}
+	return &net.UDPAddr{IP: addrs[0].IP, Zone: addrs[0].Zone, Port: port}, nil
+}
 type udpReady struct {
 	port int
 	key  []byte

M internal/adapters/dgram/ssh_dialer_test.go +2 -1


@@ -119,7 +119,8 @@ func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) {
119	119		}
120	120		return proc
121	121		})
122		-	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
	122	+	var lc net.ListenConfig
	123	+	pc, err := lc.ListenPacket(context.Background(), "udp", "127.0.0.1:0")
123	124		if err != nil {
124	125		t.Fatal(err)
125	126		}

 		}
 		return proc
 	})
-	pc, err := net.ListenPacket("udp", "127.0.0.1:0")
+	var lc net.ListenConfig
+	pc, err := lc.ListenPacket(context.Background(), "udp", "127.0.0.1:0")
 	if err != nil {
 		t.Fatal(err)
 	}

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


@@ -655,6 +655,8 @@ func runUDPBootstrap(ctx context.Context, session string) error {
655	655		defer func() { _ = devNull.Close() }()
656	656
657	657		cmd := udpProxyCommand(ctx, exe, args...)
	658	+	// _udp-proxy writes diagnostics through configureLogging(logging.Stdio, false);
	659	+	// stdio is detached here so the bootstrap SSH channel can close.
658	660		cmd.Stdin = devNull
659	661		cmd.Stdout = w
660	662		cmd.Stderr = devNull


@@ -655,6 +655,8 @@ func runUDPBootstrap(ctx context.Context, session string) error {
655	defer func() { _ = devNull.Close() }()	655		defer func() { _ = devNull.Close() }()
656		656
657	cmd := udpProxyCommand(ctx, exe, args...)	657		cmd := udpProxyCommand(ctx, exe, args...)
		658	+	// _udp-proxy writes diagnostics through configureLogging(logging.Stdio, false);
		659	+	// stdio is detached here so the bootstrap SSH channel can close.
658	cmd.Stdin = devNull	660		cmd.Stdin = devNull
659	cmd.Stdout = w	661		cmd.Stdout = w
660	cmd.Stderr = devNull	662		cmd.Stderr = devNull

M internal/app/run_test.go +32 -0


@@ -601,3 +601,35 @@ func TestRunUDPBootstrapForwardsReadinessAndExits(t *testing.T) {
601	601		t.Fatalf("proxy inherited SSH stdio handles")
602	602		}
603	603		}
	604	+
	605	+	func TestRunUDPBootstrapReturnsReadinessEOF(t *testing.T) {
	606	+	oldTimeout := udpBootstrapTimeout
	607	+	udpBootstrapTimeout = time.Second
	608	+	t.Cleanup(func() { udpBootstrapTimeout = oldTimeout })
	609	+	oldCommand := udpProxyCommand
	610	+	udpProxyCommand = func(context.Context, string, ...string) *exec.Cmd {
	611	+	return exec.Command("/bin/sh", "-c", "exit 7")
	612	+	}
	613	+	t.Cleanup(func() { udpProxyCommand = oldCommand })
	614	+
	615	+	err := runUDPBootstrap(context.Background(), "work")
	616	+	if !errors.Is(err, io.EOF) {
	617	+	t.Fatalf("runUDPBootstrap() error = %v, want EOF", err)
	618	+	}
	619	+	}
	620	+
	621	+	func TestRunUDPBootstrapTimesOutWaitingForReadiness(t *testing.T) {
	622	+	oldTimeout := udpBootstrapTimeout
	623	+	udpBootstrapTimeout = 10 * time.Millisecond
	624	+	t.Cleanup(func() { udpBootstrapTimeout = oldTimeout })
	625	+	oldCommand := udpProxyCommand
	626	+	udpProxyCommand = func(context.Context, string, ...string) *exec.Cmd {
	627	+	return exec.Command("/bin/sh", "-c", "sleep 5")
	628	+	}
	629	+	t.Cleanup(func() { udpProxyCommand = oldCommand })
	630	+
	631	+	err := runUDPBootstrap(context.Background(), "work")
	632	+	if !errors.Is(err, context.DeadlineExceeded) {
	633	+	t.Fatalf("runUDPBootstrap() error = %v, want deadline exceeded", err)
	634	+	}
	635	+	}


@@ -601,3 +601,35 @@ func TestRunUDPBootstrapForwardsReadinessAndExits(t *testing.T) {
601	t.Fatalf("proxy inherited SSH stdio handles")	601		t.Fatalf("proxy inherited SSH stdio handles")
602	}	602		}
603	}	603		}
		604	+
		605	+	func TestRunUDPBootstrapReturnsReadinessEOF(t *testing.T) {
		606	+	oldTimeout := udpBootstrapTimeout
		607	+	udpBootstrapTimeout = time.Second
		608	+	t.Cleanup(func() { udpBootstrapTimeout = oldTimeout })
		609	+	oldCommand := udpProxyCommand
		610	+	udpProxyCommand = func(context.Context, string, ...string) *exec.Cmd {
		611	+	return exec.Command("/bin/sh", "-c", "exit 7")
		612	+	}
		613	+	t.Cleanup(func() { udpProxyCommand = oldCommand })
		614	+
		615	+	err := runUDPBootstrap(context.Background(), "work")
		616	+	if !errors.Is(err, io.EOF) {
		617	+	t.Fatalf("runUDPBootstrap() error = %v, want EOF", err)
		618	+	}
		619	+	}
		620	+
		621	+	func TestRunUDPBootstrapTimesOutWaitingForReadiness(t *testing.T) {
		622	+	oldTimeout := udpBootstrapTimeout
		623	+	udpBootstrapTimeout = 10 * time.Millisecond
		624	+	t.Cleanup(func() { udpBootstrapTimeout = oldTimeout })
		625	+	oldCommand := udpProxyCommand
		626	+	udpProxyCommand = func(context.Context, string, ...string) *exec.Cmd {
		627	+	return exec.Command("/bin/sh", "-c", "sleep 5")
		628	+	}
		629	+	t.Cleanup(func() { udpProxyCommand = oldCommand })
		630	+
		631	+	err := runUDPBootstrap(context.Background(), "work")
		632	+	if !errors.Is(err, context.DeadlineExceeded) {
		633	+	t.Fatalf("runUDPBootstrap() error = %v, want deadline exceeded", err)
		634	+	}
		635	+	}

M internal/usecase/client/reconnect_toast_test.go +4 -1


@@ -183,7 +183,10 @@ func TestReconnectSleepWithResizeEventsRedrawsUntilTimerFires(t *testing.T) {
183	183
184	184		func TestRemoteReconnectToastLifecycleWithWrappedTransportError(t *testing.T) {
185	185		linkDead := errors.New("remote link dead")
186		-	wrappedLinkDead := errors.Join(fmt.Errorf("remote transport receive failed: %w", linkDead), errors.New("read side closed"))
	186	+	wrappedLinkDead := errors.Join(
	187	+	fmt.Errorf("remote transport receive failed: %w", io.EOF),
	188	+	linkDead,
	189	+	)
187	190		oldSleep := reconnectSleep
188	191		oldSleepWithResize := reconnectSleepWithResize
189	192		reconnectSleep = func(context.Context, ports.Clock, time.Duration) bool { return true }

 func TestRemoteReconnectToastLifecycleWithWrappedTransportError(t *testing.T) {
 	linkDead := errors.New("remote link dead")
-	wrappedLinkDead := errors.Join(fmt.Errorf("remote transport receive failed: %w", linkDead), errors.New("read side closed"))
+	wrappedLinkDead := errors.Join(
+		fmt.Errorf("remote transport receive failed: %w", io.EOF),
+		linkDead,
+	)
 	oldSleep := reconnectSleep
 	oldSleepWithResize := reconnectSleepWithResize
 	reconnectSleep = func(context.Context, ports.Clock, time.Duration) bool { return true }