detached ed18477

fix(dgram): fall back across remote UDP addresses (#246)

* fix(dgram): fall back across remote UDP addresses

* fix(dgram): harden UDP candidate selection
commit
ed184771bc7d800c7d38db56bc67869911f5b6f3
author
bnema · · 2026-09-02 10:12 UTC
committer
GitHub · · 2026-09-02 10:12 UTC
parent
e591820
signature
GPG UnverifiedB5690EEEBB952194 The key is unknown
4 files · +494 -38
Split
UnifiedSplit
M internal/adapters/dgram/ssh_dialer.go +113 -36
@@ -4,6 +4,7 @@ import (
44 "bufio"
55 "context"
66 "encoding/base64"
7+ "errors"
78 "fmt"
89 "io"
910 "log/slog"
@@ -50,6 +51,9 @@ var (
5051 var lc net.ListenConfig
5152 return lc.ListenPacket(ctx, "udp", ":0")
5253 }
54+ lookupUDPIPAddrs = func(ctx context.Context, host string) ([]net.IPAddr, error) {
55+ return net.DefaultResolver.LookupIPAddr(ctx, host)
56+ }
5357 probeUDPTransport = func(ctx context.Context, t *Transport) error { return t.Probe(ctx) }
5458 )
5559
@@ -166,36 +170,85 @@ func (d RemoteDialer) Dial(ctx context.Context) (wire.Transport, error) {
166170 }
167171 waited = true
168172 cleanup = false
169- peer, err := resolveUDPPeer(bootstrapCtx, d.Target, ready.port)
173+ // Share the existing probe budget across resolution and all candidates so a
174+ // hostname with many addresses cannot extend remote attach indefinitely.
175+ selectionCtx, selectionCancel := context.WithTimeout(ctx, probeTimeout)
176+ defer selectionCancel()
177+ peers, err := resolveUDPPeers(selectionCtx, d.Target, ready.port)
170178 if err != nil {
171- return nil, udpUnavailable("resolve UDP peer", err, &stderr)
179+ return nil, udpUnavailable("resolve UDP peers", err, &stderr)
172180 }
173- pc, err := listenUDPPacket(ctx)
174- if err != nil {
175- return nil, udpUnavailable("listen UDP", err, &stderr)
176- }
177- // RebindPacketConn heals NAT rebinds / stale conntrack paths by hopping to a
178- // fresh local UDP socket once the link is silent long enough to probe. The
179- // bind reuses listenUDPPacket so tests can stub it.
180- t, err := NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{
181- Observe: DiagnosticLogObserver(d.Log),
182- RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
183- return listenUDPPacket(ctx)
184- },
185- }, WithRuntimeObserver(d.RuntimeObserver))
186- pdgram.Erase(ready.key)
187- if err != nil {
188- _ = pc.Close()
189- return nil, udpUnavailable("create UDP transport", err, &stderr)
181+
182+ var (
183+ t *Transport
184+ lastProbeErr error
185+ )
186+ for i, peer := range peers {
187+ if err := selectionCtx.Err(); err != nil {
188+ if lastProbeErr == nil {
189+ lastProbeErr = err
190+ }
191+ break
192+ }
193+ pc, err := listenUDPPacket(selectionCtx)
194+ if err != nil {
195+ if t != nil {
196+ _ = t.Close()
197+ }
198+ return nil, udpUnavailable("listen UDP", err, &stderr)
199+ }
200+ if t == nil {
201+ // RebindPacketConn heals NAT rebinds / stale conntrack paths by hopping to a
202+ // fresh local UDP socket once the link is silent long enough to probe. The
203+ // bind reuses listenUDPPacket so tests can stub it.
204+ t, err = NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{
205+ Observe: DiagnosticLogObserver(d.Log),
206+ RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
207+ return listenUDPPacket(ctx)
208+ },
209+ }, WithRuntimeObserver(d.RuntimeObserver))
210+ pdgram.Erase(ready.key)
211+ if err != nil {
212+ _ = pc.Close()
213+ return nil, udpUnavailable("create UDP transport", err, &stderr)
214+ }
215+ } else if err := t.replaceDialCandidate(pc, peer); err != nil {
216+ _ = pc.Close()
217+ _ = t.Close()
218+ return nil, udpUnavailable("replace UDP peer", err, &stderr)
219+ }
220+
221+ probeCtx, cancel := candidateProbeContext(selectionCtx, len(peers)-i)
222+ err = probeUDPTransport(probeCtx, t)
223+ cancel()
224+ if err == nil {
225+ cleanup = false
226+ return t, nil
227+ }
228+ lastProbeErr = err
190229 }
191- probeCtx, cancel := context.WithTimeout(ctx, probeTimeout)
192- defer cancel()
193- if err := probeUDPTransport(probeCtx, t); err != nil {
230+ if t != nil {
194231 _ = t.Close()
195- return nil, udpProbeUnreachable(d.Target, err, &stderr)
196232 }
197- cleanup = false
198- return t, nil
233+ if lastProbeErr == nil {
234+ lastProbeErr = errors.New("all resolved UDP peer candidates failed")
235+ }
236+ return nil, udpProbeUnreachable(d.Target, sanitizeUDPProbeError(lastProbeErr), &stderr)
237+}
238+
239+func candidateProbeContext(ctx context.Context, remainingCandidates int) (context.Context, context.CancelFunc) {
240+ if remainingCandidates <= 1 {
241+ return context.WithCancel(ctx)
242+ }
243+ deadline, ok := ctx.Deadline()
244+ if !ok {
245+ return context.WithCancel(ctx)
246+ }
247+ remaining := time.Until(deadline)
248+ if remaining <= 0 {
249+ return context.WithCancel(ctx)
250+ }
251+ return context.WithTimeout(ctx, remaining/time.Duration(remainingCandidates))
199252 }
200253
201254 type udpReadyResult struct {
@@ -237,19 +290,29 @@ func waitBootstrapContext(ctx context.Context, proc bootstrapProcess) error {
237290 }
238291 }
239292
240-func resolveUDPPeer(ctx context.Context, target string, port int) (*net.UDPAddr, error) {
293+func resolveUDPPeers(ctx context.Context, target string, port int) ([]*net.UDPAddr, error) {
241294 host := sshTargetHost(target)
242295 if ip := net.ParseIP(host); ip != nil {
243- return &net.UDPAddr{IP: ip, Port: port}, nil
296+ return []*net.UDPAddr{{IP: ip, Port: port}}, nil
244297 }
245- addrs, err := net.DefaultResolver.LookupIPAddr(ctx, host)
298+ addrs, err := lookupUDPIPAddrs(ctx, host)
246299 if err != nil {
247300 return nil, err
248301 }
249- if len(addrs) == 0 {
302+ peers := make([]*net.UDPAddr, 0, len(addrs))
303+ seen := make(map[string]struct{}, len(addrs))
304+ for _, addr := range addrs {
305+ peer := &net.UDPAddr{IP: addr.IP, Zone: addr.Zone, Port: port}
306+ if _, ok := seen[peer.String()]; ok {
307+ continue
308+ }
309+ seen[peer.String()] = struct{}{}
310+ peers = append(peers, peer)
311+ }
312+ if len(peers) == 0 {
250313 return nil, fmt.Errorf("no UDP peer addresses for %q", host)
251314 }
252- return &net.UDPAddr{IP: addrs[0].IP, Zone: addrs[0].Zone, Port: port}, nil
315+ return peers, nil
253316 }
254317
255318 type udpReady struct {
@@ -301,6 +364,22 @@ func readUDPReady(r io.Reader) (udpReady, error) {
301364 return udpReady{port: port, key: out}, nil
302365 }
303366
367+type sanitizedUDPProbeError struct{ cause error }
368+
369+func (e sanitizedUDPProbeError) Error() string { return "UDP probe failed" }
370+func (e sanitizedUDPProbeError) Unwrap() error { return e.cause }
371+
372+func sanitizeUDPProbeError(err error) error {
373+ switch {
374+ case errors.Is(err, context.Canceled):
375+ return context.Canceled
376+ case errors.Is(err, context.DeadlineExceeded):
377+ return context.DeadlineExceeded
378+ default:
379+ return sanitizedUDPProbeError{cause: err}
380+ }
381+}
382+
304383 func udpUnavailable(action string, err error, stderr fmt.Stringer) error {
305384 return udpDialFailure(RemoteDialBootstrapUnavailable, action, err, stderr, "")
306385 }
@@ -312,14 +391,12 @@ func udpDialFailure(kind RemoteDialFailureKind, action string, err error, stderr
312391 return &RemoteDialError{Kind: kind, Action: action, Err: err, Hint: hint}
313392 }
314393
315-// udpProbeUnreachable wraps a probe failure with actionable guidance. A failed
316-// probe almost always means a firewall on the remote is dropping datagrams to
317-// vev's UDP proxy rather than a vev bug, so name the two real fixes instead of
318-// surfacing a bare deadline error.
394+// udpProbeUnreachable wraps a probe failure with actionable guidance without
395+// assuming a particular VPN or firewall implementation.
319396 func udpProbeUnreachable(target string, err error, stderr fmt.Stringer) error {
320397 return udpDialFailure(RemoteDialProbeUnreachable, "probe UDP transport", err, stderr, fmt.Sprintf("\n"+
321- " the remote host is not reachable over UDP; a firewall is likely dropping the datagrams.\n"+
322- " open UDP on the remote (e.g. `sudo ufw allow in on tailscale0`) or attach over SSH with `VEV_REMOTE_TRANSPORT=stdio vev attach %s`", target))
398+ " check UDP routing and the remote firewall for VEV_UDP_PORT_RANGE.\n"+
399+ " retry over SSH with `VEV_REMOTE_TRANSPORT=stdio vev attach %s`", target))
323400 }
324401
325402 func sshTargetHost(target string) string {
@@ -4,6 +4,7 @@ import (
4 "bufio"4 "bufio"
5 "context"5 "context"
6 "encoding/base64"6 "encoding/base64"
7+ "errors"
7 "fmt"8 "fmt"
8 "io"9 "io"
9 "log/slog"10 "log/slog"
@@ -50,6 +51,9 @@ var (
50 var lc net.ListenConfig51 var lc net.ListenConfig
51 return lc.ListenPacket(ctx, "udp", ":0")52 return lc.ListenPacket(ctx, "udp", ":0")
52 }53 }
54+ lookupUDPIPAddrs = func(ctx context.Context, host string) ([]net.IPAddr, error) {
55+ return net.DefaultResolver.LookupIPAddr(ctx, host)
56+ }
53 probeUDPTransport = func(ctx context.Context, t *Transport) error { return t.Probe(ctx) }57 probeUDPTransport = func(ctx context.Context, t *Transport) error { return t.Probe(ctx) }
54 )58 )
55 59
@@ -166,36 +170,85 @@ func (d RemoteDialer) Dial(ctx context.Context) (wire.Transport, error) {
166 }170 }
167 waited = true171 waited = true
168 cleanup = false172 cleanup = false
169- peer, err := resolveUDPPeer(bootstrapCtx, d.Target, ready.port)173+ // Share the existing probe budget across resolution and all candidates so a
174+ // hostname with many addresses cannot extend remote attach indefinitely.
175+ selectionCtx, selectionCancel := context.WithTimeout(ctx, probeTimeout)
176+ defer selectionCancel()
177+ peers, err := resolveUDPPeers(selectionCtx, d.Target, ready.port)
170 if err != nil {178 if err != nil {
171- return nil, udpUnavailable("resolve UDP peer", err, &stderr)179+ return nil, udpUnavailable("resolve UDP peers", err, &stderr)
172 }180 }
173- pc, err := listenUDPPacket(ctx)181+
174- if err != nil {182+ var (
175- return nil, udpUnavailable("listen UDP", err, &stderr)183+ t *Transport
176- }184+ lastProbeErr error
177- // RebindPacketConn heals NAT rebinds / stale conntrack paths by hopping to a185+ )
178- // fresh local UDP socket once the link is silent long enough to probe. The186+ for i, peer := range peers {
179- // bind reuses listenUDPPacket so tests can stub it.187+ if err := selectionCtx.Err(); err != nil {
180- t, err := NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{188+ if lastProbeErr == nil {
181- Observe: DiagnosticLogObserver(d.Log),189+ lastProbeErr = err
182- RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {190+ }
183- return listenUDPPacket(ctx)191+ break
184- },192+ }
185- }, WithRuntimeObserver(d.RuntimeObserver))193+ pc, err := listenUDPPacket(selectionCtx)
186- pdgram.Erase(ready.key)194+ if err != nil {
187- if err != nil {195+ if t != nil {
188- _ = pc.Close()196+ _ = t.Close()
189- return nil, udpUnavailable("create UDP transport", err, &stderr)197+ }
198+ return nil, udpUnavailable("listen UDP", err, &stderr)
199+ }
200+ if t == nil {
201+ // RebindPacketConn heals NAT rebinds / stale conntrack paths by hopping to a
202+ // fresh local UDP socket once the link is silent long enough to probe. The
203+ // bind reuses listenUDPPacket so tests can stub it.
204+ t, err = NewTransportWithOptions(pc, peer, ready.key, 1, 2, Options{
205+ Observe: DiagnosticLogObserver(d.Log),
206+ RebindPacketConn: func(net.PacketConn) (net.PacketConn, error) {
207+ return listenUDPPacket(ctx)
208+ },
209+ }, WithRuntimeObserver(d.RuntimeObserver))
210+ pdgram.Erase(ready.key)
211+ if err != nil {
212+ _ = pc.Close()
213+ return nil, udpUnavailable("create UDP transport", err, &stderr)
214+ }
215+ } else if err := t.replaceDialCandidate(pc, peer); err != nil {
216+ _ = pc.Close()
217+ _ = t.Close()
218+ return nil, udpUnavailable("replace UDP peer", err, &stderr)
219+ }
220+
221+ probeCtx, cancel := candidateProbeContext(selectionCtx, len(peers)-i)
222+ err = probeUDPTransport(probeCtx, t)
223+ cancel()
224+ if err == nil {
225+ cleanup = false
226+ return t, nil
227+ }
228+ lastProbeErr = err
190 }229 }
191- probeCtx, cancel := context.WithTimeout(ctx, probeTimeout)230+ if t != nil {
192- defer cancel()
193- if err := probeUDPTransport(probeCtx, t); err != nil {
194 _ = t.Close()231 _ = t.Close()
195- return nil, udpProbeUnreachable(d.Target, err, &stderr)
196 }232 }
197- cleanup = false233+ if lastProbeErr == nil {
198- return t, nil234+ lastProbeErr = errors.New("all resolved UDP peer candidates failed")
235+ }
236+ return nil, udpProbeUnreachable(d.Target, sanitizeUDPProbeError(lastProbeErr), &stderr)
237+}
238+
239+func candidateProbeContext(ctx context.Context, remainingCandidates int) (context.Context, context.CancelFunc) {
240+ if remainingCandidates <= 1 {
241+ return context.WithCancel(ctx)
242+ }
243+ deadline, ok := ctx.Deadline()
244+ if !ok {
245+ return context.WithCancel(ctx)
246+ }
247+ remaining := time.Until(deadline)
248+ if remaining <= 0 {
249+ return context.WithCancel(ctx)
250+ }
251+ return context.WithTimeout(ctx, remaining/time.Duration(remainingCandidates))
199 }252 }
200 253
201 type udpReadyResult struct {254 type udpReadyResult struct {
@@ -237,19 +290,29 @@ func waitBootstrapContext(ctx context.Context, proc bootstrapProcess) error {
237 }290 }
238 }291 }
239 292
240-func resolveUDPPeer(ctx context.Context, target string, port int) (*net.UDPAddr, error) {293+func resolveUDPPeers(ctx context.Context, target string, port int) ([]*net.UDPAddr, error) {
241 host := sshTargetHost(target)294 host := sshTargetHost(target)
242 if ip := net.ParseIP(host); ip != nil {295 if ip := net.ParseIP(host); ip != nil {
243- return &net.UDPAddr{IP: ip, Port: port}, nil296+ return []*net.UDPAddr{{IP: ip, Port: port}}, nil
244 }297 }
245- addrs, err := net.DefaultResolver.LookupIPAddr(ctx, host)298+ addrs, err := lookupUDPIPAddrs(ctx, host)
246 if err != nil {299 if err != nil {
247 return nil, err300 return nil, err
248 }301 }
249- if len(addrs) == 0 {302+ peers := make([]*net.UDPAddr, 0, len(addrs))
303+ seen := make(map[string]struct{}, len(addrs))
304+ for _, addr := range addrs {
305+ peer := &net.UDPAddr{IP: addr.IP, Zone: addr.Zone, Port: port}
306+ if _, ok := seen[peer.String()]; ok {
307+ continue
308+ }
309+ seen[peer.String()] = struct{}{}
310+ peers = append(peers, peer)
311+ }
312+ if len(peers) == 0 {
250 return nil, fmt.Errorf("no UDP peer addresses for %q", host)313 return nil, fmt.Errorf("no UDP peer addresses for %q", host)
251 }314 }
252- return &net.UDPAddr{IP: addrs[0].IP, Zone: addrs[0].Zone, Port: port}, nil315+ return peers, nil
253 }316 }
254 317
255 type udpReady struct {318 type udpReady struct {
@@ -301,6 +364,22 @@ func readUDPReady(r io.Reader) (udpReady, error) {
301 return udpReady{port: port, key: out}, nil364 return udpReady{port: port, key: out}, nil
302 }365 }
303 366
367+type sanitizedUDPProbeError struct{ cause error }
368+
369+func (e sanitizedUDPProbeError) Error() string { return "UDP probe failed" }
370+func (e sanitizedUDPProbeError) Unwrap() error { return e.cause }
371+
372+func sanitizeUDPProbeError(err error) error {
373+ switch {
374+ case errors.Is(err, context.Canceled):
375+ return context.Canceled
376+ case errors.Is(err, context.DeadlineExceeded):
377+ return context.DeadlineExceeded
378+ default:
379+ return sanitizedUDPProbeError{cause: err}
380+ }
381+}
382+
304 func udpUnavailable(action string, err error, stderr fmt.Stringer) error {383 func udpUnavailable(action string, err error, stderr fmt.Stringer) error {
305 return udpDialFailure(RemoteDialBootstrapUnavailable, action, err, stderr, "")384 return udpDialFailure(RemoteDialBootstrapUnavailable, action, err, stderr, "")
306 }385 }
@@ -312,14 +391,12 @@ func udpDialFailure(kind RemoteDialFailureKind, action string, err error, stderr
312 return &RemoteDialError{Kind: kind, Action: action, Err: err, Hint: hint}391 return &RemoteDialError{Kind: kind, Action: action, Err: err, Hint: hint}
313 }392 }
314 393
315-// udpProbeUnreachable wraps a probe failure with actionable guidance. A failed394+// udpProbeUnreachable wraps a probe failure with actionable guidance without
316-// probe almost always means a firewall on the remote is dropping datagrams to395+// assuming a particular VPN or firewall implementation.
317-// vev's UDP proxy rather than a vev bug, so name the two real fixes instead of
318-// surfacing a bare deadline error.
319 func udpProbeUnreachable(target string, err error, stderr fmt.Stringer) error {396 func udpProbeUnreachable(target string, err error, stderr fmt.Stringer) error {
320 return udpDialFailure(RemoteDialProbeUnreachable, "probe UDP transport", err, stderr, fmt.Sprintf("\n"+397 return udpDialFailure(RemoteDialProbeUnreachable, "probe UDP transport", err, stderr, fmt.Sprintf("\n"+
321- " the remote host is not reachable over UDP; a firewall is likely dropping the datagrams.\n"+398+ " check UDP routing and the remote firewall for VEV_UDP_PORT_RANGE.\n"+
322- " open UDP on the remote (e.g. `sudo ufw allow in on tailscale0`) or attach over SSH with `VEV_REMOTE_TRANSPORT=stdio vev attach %s`", target))399+ " retry over SSH with `VEV_REMOTE_TRANSPORT=stdio vev attach %s`", target))
323 }400 }
324 401
325 func sshTargetHost(target string) string {402 func sshTargetHost(target string) string {
M internal/adapters/dgram/ssh_dialer_test.go +295 -2
@@ -6,6 +6,7 @@ import (
66 "errors"
77 "io"
88 "net"
9+ "slices"
910 "strings"
1011 "testing"
1112 "time"
@@ -47,6 +48,20 @@ func withListenUDP(t *testing.T, fn func(context.Context) (net.PacketConn, error
4748 t.Cleanup(func() { listenUDPPacket = old })
4849 }
4950
51+func withUDPIPLookup(t *testing.T, fn func(context.Context, string) ([]net.IPAddr, error)) {
52+ t.Helper()
53+ old := lookupUDPIPAddrs
54+ lookupUDPIPAddrs = fn
55+ t.Cleanup(func() { lookupUDPIPAddrs = old })
56+}
57+
58+func withUDPProbe(t *testing.T, fn func(context.Context, *Transport) error) {
59+ t.Helper()
60+ old := probeUDPTransport
61+ probeUDPTransport = fn
62+ t.Cleanup(func() { probeUDPTransport = old })
63+}
64+
5065 func TestRemoteDialerUDPBootstrapFailures(t *testing.T) {
5166 key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))
5267 tests := []struct {
@@ -117,6 +132,280 @@ func TestListenUDPPacketUsesDualStackWildcard(t *testing.T) {
117132 }
118133 }
119134
135+func TestResolveUDPPeers(t *testing.T) {
136+ tests := []struct {
137+ name string
138+ target string
139+ addrs []net.IPAddr
140+ wantPeers []string
141+ wantErr string
142+ wantLookups int
143+ }{
144+ {
145+ name: "hostname preserves order and removes duplicates",
146+ target: "remote.example",
147+ addrs: []net.IPAddr{
148+ {IP: net.ParseIP("192.0.2.1")},
149+ {IP: net.ParseIP("2001:db8::1")},
150+ {IP: net.ParseIP("192.0.2.1")},
151+ },
152+ wantPeers: []string{"192.0.2.1:4444", "[2001:db8::1]:4444"},
153+ wantLookups: 1,
154+ },
155+ {
156+ name: "explicit IP bypasses lookup",
157+ target: "[2001:db8::1]",
158+ wantPeers: []string{"[2001:db8::1]:4444"},
159+ wantLookups: 0,
160+ },
161+ {
162+ name: "empty lookup result",
163+ target: "remote.example",
164+ wantErr: `no UDP peer addresses for "remote.example"`,
165+ wantLookups: 1,
166+ },
167+ }
168+ for _, tt := range tests {
169+ t.Run(tt.name, func(t *testing.T) {
170+ lookups := 0
171+ withUDPIPLookup(t, func(_ context.Context, host string) ([]net.IPAddr, error) {
172+ lookups++
173+ if host != "remote.example" {
174+ t.Fatalf("lookup host = %q, want remote.example", host)
175+ }
176+ return tt.addrs, nil
177+ })
178+
179+ peers, err := resolveUDPPeers(context.Background(), tt.target, 4444)
180+ if tt.wantErr != "" {
181+ if err == nil || err.Error() != tt.wantErr {
182+ t.Fatalf("resolveUDPPeers() error = %v, want %q", err, tt.wantErr)
183+ }
184+ } else if err != nil {
185+ t.Fatal(err)
186+ } else {
187+ got := make([]string, 0, len(peers))
188+ for _, peer := range peers {
189+ got = append(got, peer.String())
190+ }
191+ if !slices.Equal(got, tt.wantPeers) {
192+ t.Fatalf("peers = %v, want %v", got, tt.wantPeers)
193+ }
194+ }
195+ if got := lookups; got != tt.wantLookups {
196+ t.Fatalf("lookup calls = %d, want %d", got, tt.wantLookups)
197+ }
198+ })
199+ }
200+}
201+
202+func TestCandidateProbeContext(t *testing.T) {
203+ tests := []struct {
204+ name string
205+ remaining int
206+ newContext func() (context.Context, context.CancelFunc)
207+ wantDeadline bool
208+ wantDone bool
209+ minDeadlineRemain time.Duration
210+ maxDeadlineRemain time.Duration
211+ }{
212+ {
213+ name: "one candidate retains selection deadline",
214+ remaining: 1,
215+ newContext: func() (context.Context, context.CancelFunc) {
216+ return context.WithTimeout(context.Background(), time.Hour)
217+ },
218+ wantDeadline: true,
219+ minDeadlineRemain: 59 * time.Minute,
220+ maxDeadlineRemain: time.Hour,
221+ },
222+ {
223+ name: "no selection deadline",
224+ remaining: 2,
225+ newContext: func() (context.Context, context.CancelFunc) { return context.WithCancel(context.Background()) },
226+ },
227+ {
228+ name: "expired selection deadline",
229+ remaining: 2,
230+ newContext: func() (context.Context, context.CancelFunc) {
231+ return context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
232+ },
233+ wantDeadline: true,
234+ wantDone: true,
235+ },
236+ {
237+ name: "splits remaining budget",
238+ remaining: 4,
239+ newContext: func() (context.Context, context.CancelFunc) {
240+ return context.WithTimeout(context.Background(), time.Hour)
241+ },
242+ wantDeadline: true,
243+ minDeadlineRemain: 14 * time.Minute,
244+ maxDeadlineRemain: 16 * time.Minute,
245+ },
246+ }
247+ for _, tt := range tests {
248+ t.Run(tt.name, func(t *testing.T) {
249+ selectionCtx, selectionCancel := tt.newContext()
250+ defer selectionCancel()
251+ probeCtx, probeCancel := candidateProbeContext(selectionCtx, tt.remaining)
252+ defer probeCancel()
253+
254+ deadline, hasDeadline := probeCtx.Deadline()
255+ if hasDeadline != tt.wantDeadline {
256+ t.Fatalf("probe context has deadline = %v, want %v", hasDeadline, tt.wantDeadline)
257+ }
258+ if hasDeadline && tt.minDeadlineRemain != 0 {
259+ remaining := time.Until(deadline)
260+ if remaining < tt.minDeadlineRemain || remaining > tt.maxDeadlineRemain {
261+ t.Fatalf("probe deadline remaining = %v, want [%v, %v]", remaining, tt.minDeadlineRemain, tt.maxDeadlineRemain)
262+ }
263+ }
264+ if tt.wantDone {
265+ select {
266+ case <-probeCtx.Done():
267+ default:
268+ t.Fatal("probe context is not done")
269+ }
270+ }
271+ })
272+ }
273+}
274+
275+func TestRemoteDialerFallsBackAcrossResolvedUDPPeers(t *testing.T) {
276+ key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))
277+ proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))}
278+ withBootstrapStarter(t, func(context.Context, string, io.Writer) bootstrapProcess { return proc })
279+ withUDPIPLookup(t, func(_ context.Context, host string) ([]net.IPAddr, error) {
280+ if host != "remote.example" {
281+ t.Fatalf("lookup host = %q, want remote.example", host)
282+ }
283+ return []net.IPAddr{
284+ {IP: net.ParseIP("192.0.2.1")},
285+ {IP: net.ParseIP("192.0.2.2")},
286+ }, nil
287+ })
288+
289+ var (
290+ firstClosed bool
291+ secondClosed bool
292+ listenCalls int
293+ )
294+ withListenUDP(t, func(context.Context) (net.PacketConn, error) {
295+ var lc net.ListenConfig
296+ pc, err := lc.ListenPacket(context.Background(), "udp", "127.0.0.1:0")
297+ if err != nil {
298+ return nil, err
299+ }
300+ listenCalls++
301+ switch listenCalls {
302+ case 1:
303+ return closeTrackPacketConn{PacketConn: pc, closed: &firstClosed}, nil
304+ case 2:
305+ return closeTrackPacketConn{PacketConn: pc, closed: &secondClosed}, nil
306+ default:
307+ t.Fatalf("listen UDP calls = %d, want at most 2", listenCalls)
308+ return nil, nil
309+ }
310+ })
311+
312+ var (
313+ peers []string
314+ firstTransport *Transport
315+ )
316+ withUDPProbe(t, func(_ context.Context, tr *Transport) error {
317+ tr.mu.Lock()
318+ peer := tr.peer.String()
319+ tr.mu.Unlock()
320+ peers = append(peers, peer)
321+ if len(peers) == 1 {
322+ firstTransport = tr
323+ return errors.New("first UDP address is unreachable")
324+ }
325+ if tr != firstTransport {
326+ t.Fatal("fallback created a new transport and reset its authenticated state")
327+ }
328+ return nil
329+ })
330+
331+ tr, err := NewRemoteDialer("remote.example", "work").Dial(context.Background())
332+ if err != nil {
333+ t.Fatal(err)
334+ }
335+ if got, want := peers, []string{"192.0.2.1:4444", "192.0.2.2:4444"}; !slices.Equal(got, want) {
336+ t.Fatalf("probed peers = %v, want %v", got, want)
337+ }
338+ if !firstClosed {
339+ t.Fatal("first candidate socket was not closed")
340+ }
341+ if secondClosed {
342+ t.Fatal("winning candidate socket closed before transport close")
343+ }
344+ if err := tr.Close(); err != nil {
345+ t.Fatal(err)
346+ }
347+ if !secondClosed {
348+ t.Fatal("winning candidate socket was not closed with transport")
349+ }
350+}
351+
352+func TestRemoteDialerReportsFailureAfterAllResolvedUDPPeers(t *testing.T) {
353+ key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))
354+ proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))}
355+ withBootstrapStarter(t, func(context.Context, string, io.Writer) bootstrapProcess { return proc })
356+ withUDPIPLookup(t, func(context.Context, string) ([]net.IPAddr, error) {
357+ return []net.IPAddr{
358+ {IP: net.ParseIP("192.0.2.1")},
359+ {IP: net.ParseIP("192.0.2.2")},
360+ }, nil
361+ })
362+
363+ closed := make([]bool, 2)
364+ listenCalls := 0
365+ withListenUDP(t, func(context.Context) (net.PacketConn, error) {
366+ if listenCalls == len(closed) {
367+ t.Fatal("opened more sockets than resolved candidates")
368+ }
369+ var lc net.ListenConfig
370+ pc, err := lc.ListenPacket(context.Background(), "udp", "127.0.0.1:0")
371+ if err != nil {
372+ return nil, err
373+ }
374+ tracked := closeTrackPacketConn{PacketConn: pc, closed: &closed[listenCalls]}
375+ listenCalls++
376+ return tracked, nil
377+ })
378+
379+ probeErr := errors.New("unreachable at 192.0.2.2:4444")
380+ probeCalls := 0
381+ withUDPProbe(t, func(context.Context, *Transport) error {
382+ probeCalls++
383+ return probeErr
384+ })
385+
386+ tr, err := NewRemoteDialer("remote.example", "work").Dial(context.Background())
387+ if err == nil {
388+ t.Fatal("Dial() error = nil, want all candidates unavailable")
389+ }
390+ if tr != nil {
391+ t.Fatalf("Dial() transport = %T, want nil", tr)
392+ }
393+ if got, want := probeCalls, 2; got != want {
394+ t.Fatalf("probe calls = %d, want %d", got, want)
395+ }
396+ for i, wasClosed := range closed {
397+ if !wasClosed {
398+ t.Fatalf("candidate socket %d was not closed", i+1)
399+ }
400+ }
401+ if !errors.Is(err, probeErr) {
402+ t.Fatalf("Dial() error = %v, want final probe cause", err)
403+ }
404+ if strings.Contains(err.Error(), "192.0.2.") {
405+ t.Fatalf("Dial() error leaked candidate address: %q", err)
406+ }
407+}
408+
120409 func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) {
121410 key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))
122411 proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))}
@@ -135,8 +424,9 @@ func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) {
135424 withListenUDP(t, func(context.Context) (net.PacketConn, error) {
136425 return closeTrackPacketConn{PacketConn: pc, closed: &closed}, nil
137426 })
427+ withUDPProbe(t, func(context.Context, *Transport) error { return context.DeadlineExceeded })
138428 d := NewRemoteDialer("127.0.0.1", "work")
139- d.ProbeTimeout = time.Nanosecond
429+ d.ProbeTimeout = time.Second
140430 tr, err := d.Dial(context.Background())
141431 if err == nil || !strings.Contains(err.Error(), "remote UDP transport unavailable: probe UDP transport") {
142432 t.Fatalf("Dial() error = %v, want probe unavailable", err)
@@ -148,7 +438,10 @@ func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) {
148438 if dialErr.Kind != RemoteDialProbeUnreachable {
149439 t.Fatalf("Dial() error kind = %v, want probe unreachable", dialErr.Kind)
150440 }
151- for _, want := range []string{"firewall", "VEV_REMOTE_TRANSPORT=stdio vev attach 127.0.0.1"} {
441+ if !errors.Is(err, context.DeadlineExceeded) {
442+ t.Fatalf("Dial() error = %v, want probe deadline cause", err)
443+ }
444+ for _, want := range []string{"context deadline exceeded", "firewall", "VEV_UDP_PORT_RANGE", "VEV_REMOTE_TRANSPORT=stdio vev attach 127.0.0.1"} {
152445 if !strings.Contains(err.Error(), want) {
153446 t.Fatalf("Dial() error = %q, want actionable hint %q", err, want)
154447 }
@@ -6,6 +6,7 @@ import (
6 "errors"6 "errors"
7 "io"7 "io"
8 "net"8 "net"
9+ "slices"
9 "strings"10 "strings"
10 "testing"11 "testing"
11 "time"12 "time"
@@ -47,6 +48,20 @@ func withListenUDP(t *testing.T, fn func(context.Context) (net.PacketConn, error
47 t.Cleanup(func() { listenUDPPacket = old })48 t.Cleanup(func() { listenUDPPacket = old })
48 }49 }
49 50
51+func withUDPIPLookup(t *testing.T, fn func(context.Context, string) ([]net.IPAddr, error)) {
52+ t.Helper()
53+ old := lookupUDPIPAddrs
54+ lookupUDPIPAddrs = fn
55+ t.Cleanup(func() { lookupUDPIPAddrs = old })
56+}
57+
58+func withUDPProbe(t *testing.T, fn func(context.Context, *Transport) error) {
59+ t.Helper()
60+ old := probeUDPTransport
61+ probeUDPTransport = fn
62+ t.Cleanup(func() { probeUDPTransport = old })
63+}
64+
50 func TestRemoteDialerUDPBootstrapFailures(t *testing.T) {65 func TestRemoteDialerUDPBootstrapFailures(t *testing.T) {
51 key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))66 key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))
52 tests := []struct {67 tests := []struct {
@@ -117,6 +132,280 @@ func TestListenUDPPacketUsesDualStackWildcard(t *testing.T) {
117 }132 }
118 }133 }
119 134
135+func TestResolveUDPPeers(t *testing.T) {
136+ tests := []struct {
137+ name string
138+ target string
139+ addrs []net.IPAddr
140+ wantPeers []string
141+ wantErr string
142+ wantLookups int
143+ }{
144+ {
145+ name: "hostname preserves order and removes duplicates",
146+ target: "remote.example",
147+ addrs: []net.IPAddr{
148+ {IP: net.ParseIP("192.0.2.1")},
149+ {IP: net.ParseIP("2001:db8::1")},
150+ {IP: net.ParseIP("192.0.2.1")},
151+ },
152+ wantPeers: []string{"192.0.2.1:4444", "[2001:db8::1]:4444"},
153+ wantLookups: 1,
154+ },
155+ {
156+ name: "explicit IP bypasses lookup",
157+ target: "[2001:db8::1]",
158+ wantPeers: []string{"[2001:db8::1]:4444"},
159+ wantLookups: 0,
160+ },
161+ {
162+ name: "empty lookup result",
163+ target: "remote.example",
164+ wantErr: `no UDP peer addresses for "remote.example"`,
165+ wantLookups: 1,
166+ },
167+ }
168+ for _, tt := range tests {
169+ t.Run(tt.name, func(t *testing.T) {
170+ lookups := 0
171+ withUDPIPLookup(t, func(_ context.Context, host string) ([]net.IPAddr, error) {
172+ lookups++
173+ if host != "remote.example" {
174+ t.Fatalf("lookup host = %q, want remote.example", host)
175+ }
176+ return tt.addrs, nil
177+ })
178+
179+ peers, err := resolveUDPPeers(context.Background(), tt.target, 4444)
180+ if tt.wantErr != "" {
181+ if err == nil || err.Error() != tt.wantErr {
182+ t.Fatalf("resolveUDPPeers() error = %v, want %q", err, tt.wantErr)
183+ }
184+ } else if err != nil {
185+ t.Fatal(err)
186+ } else {
187+ got := make([]string, 0, len(peers))
188+ for _, peer := range peers {
189+ got = append(got, peer.String())
190+ }
191+ if !slices.Equal(got, tt.wantPeers) {
192+ t.Fatalf("peers = %v, want %v", got, tt.wantPeers)
193+ }
194+ }
195+ if got := lookups; got != tt.wantLookups {
196+ t.Fatalf("lookup calls = %d, want %d", got, tt.wantLookups)
197+ }
198+ })
199+ }
200+}
201+
202+func TestCandidateProbeContext(t *testing.T) {
203+ tests := []struct {
204+ name string
205+ remaining int
206+ newContext func() (context.Context, context.CancelFunc)
207+ wantDeadline bool
208+ wantDone bool
209+ minDeadlineRemain time.Duration
210+ maxDeadlineRemain time.Duration
211+ }{
212+ {
213+ name: "one candidate retains selection deadline",
214+ remaining: 1,
215+ newContext: func() (context.Context, context.CancelFunc) {
216+ return context.WithTimeout(context.Background(), time.Hour)
217+ },
218+ wantDeadline: true,
219+ minDeadlineRemain: 59 * time.Minute,
220+ maxDeadlineRemain: time.Hour,
221+ },
222+ {
223+ name: "no selection deadline",
224+ remaining: 2,
225+ newContext: func() (context.Context, context.CancelFunc) { return context.WithCancel(context.Background()) },
226+ },
227+ {
228+ name: "expired selection deadline",
229+ remaining: 2,
230+ newContext: func() (context.Context, context.CancelFunc) {
231+ return context.WithDeadline(context.Background(), time.Now().Add(-time.Second))
232+ },
233+ wantDeadline: true,
234+ wantDone: true,
235+ },
236+ {
237+ name: "splits remaining budget",
238+ remaining: 4,
239+ newContext: func() (context.Context, context.CancelFunc) {
240+ return context.WithTimeout(context.Background(), time.Hour)
241+ },
242+ wantDeadline: true,
243+ minDeadlineRemain: 14 * time.Minute,
244+ maxDeadlineRemain: 16 * time.Minute,
245+ },
246+ }
247+ for _, tt := range tests {
248+ t.Run(tt.name, func(t *testing.T) {
249+ selectionCtx, selectionCancel := tt.newContext()
250+ defer selectionCancel()
251+ probeCtx, probeCancel := candidateProbeContext(selectionCtx, tt.remaining)
252+ defer probeCancel()
253+
254+ deadline, hasDeadline := probeCtx.Deadline()
255+ if hasDeadline != tt.wantDeadline {
256+ t.Fatalf("probe context has deadline = %v, want %v", hasDeadline, tt.wantDeadline)
257+ }
258+ if hasDeadline && tt.minDeadlineRemain != 0 {
259+ remaining := time.Until(deadline)
260+ if remaining < tt.minDeadlineRemain || remaining > tt.maxDeadlineRemain {
261+ t.Fatalf("probe deadline remaining = %v, want [%v, %v]", remaining, tt.minDeadlineRemain, tt.maxDeadlineRemain)
262+ }
263+ }
264+ if tt.wantDone {
265+ select {
266+ case <-probeCtx.Done():
267+ default:
268+ t.Fatal("probe context is not done")
269+ }
270+ }
271+ })
272+ }
273+}
274+
275+func TestRemoteDialerFallsBackAcrossResolvedUDPPeers(t *testing.T) {
276+ key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))
277+ proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))}
278+ withBootstrapStarter(t, func(context.Context, string, io.Writer) bootstrapProcess { return proc })
279+ withUDPIPLookup(t, func(_ context.Context, host string) ([]net.IPAddr, error) {
280+ if host != "remote.example" {
281+ t.Fatalf("lookup host = %q, want remote.example", host)
282+ }
283+ return []net.IPAddr{
284+ {IP: net.ParseIP("192.0.2.1")},
285+ {IP: net.ParseIP("192.0.2.2")},
286+ }, nil
287+ })
288+
289+ var (
290+ firstClosed bool
291+ secondClosed bool
292+ listenCalls int
293+ )
294+ withListenUDP(t, func(context.Context) (net.PacketConn, error) {
295+ var lc net.ListenConfig
296+ pc, err := lc.ListenPacket(context.Background(), "udp", "127.0.0.1:0")
297+ if err != nil {
298+ return nil, err
299+ }
300+ listenCalls++
301+ switch listenCalls {
302+ case 1:
303+ return closeTrackPacketConn{PacketConn: pc, closed: &firstClosed}, nil
304+ case 2:
305+ return closeTrackPacketConn{PacketConn: pc, closed: &secondClosed}, nil
306+ default:
307+ t.Fatalf("listen UDP calls = %d, want at most 2", listenCalls)
308+ return nil, nil
309+ }
310+ })
311+
312+ var (
313+ peers []string
314+ firstTransport *Transport
315+ )
316+ withUDPProbe(t, func(_ context.Context, tr *Transport) error {
317+ tr.mu.Lock()
318+ peer := tr.peer.String()
319+ tr.mu.Unlock()
320+ peers = append(peers, peer)
321+ if len(peers) == 1 {
322+ firstTransport = tr
323+ return errors.New("first UDP address is unreachable")
324+ }
325+ if tr != firstTransport {
326+ t.Fatal("fallback created a new transport and reset its authenticated state")
327+ }
328+ return nil
329+ })
330+
331+ tr, err := NewRemoteDialer("remote.example", "work").Dial(context.Background())
332+ if err != nil {
333+ t.Fatal(err)
334+ }
335+ if got, want := peers, []string{"192.0.2.1:4444", "192.0.2.2:4444"}; !slices.Equal(got, want) {
336+ t.Fatalf("probed peers = %v, want %v", got, want)
337+ }
338+ if !firstClosed {
339+ t.Fatal("first candidate socket was not closed")
340+ }
341+ if secondClosed {
342+ t.Fatal("winning candidate socket closed before transport close")
343+ }
344+ if err := tr.Close(); err != nil {
345+ t.Fatal(err)
346+ }
347+ if !secondClosed {
348+ t.Fatal("winning candidate socket was not closed with transport")
349+ }
350+}
351+
352+func TestRemoteDialerReportsFailureAfterAllResolvedUDPPeers(t *testing.T) {
353+ key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))
354+ proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))}
355+ withBootstrapStarter(t, func(context.Context, string, io.Writer) bootstrapProcess { return proc })
356+ withUDPIPLookup(t, func(context.Context, string) ([]net.IPAddr, error) {
357+ return []net.IPAddr{
358+ {IP: net.ParseIP("192.0.2.1")},
359+ {IP: net.ParseIP("192.0.2.2")},
360+ }, nil
361+ })
362+
363+ closed := make([]bool, 2)
364+ listenCalls := 0
365+ withListenUDP(t, func(context.Context) (net.PacketConn, error) {
366+ if listenCalls == len(closed) {
367+ t.Fatal("opened more sockets than resolved candidates")
368+ }
369+ var lc net.ListenConfig
370+ pc, err := lc.ListenPacket(context.Background(), "udp", "127.0.0.1:0")
371+ if err != nil {
372+ return nil, err
373+ }
374+ tracked := closeTrackPacketConn{PacketConn: pc, closed: &closed[listenCalls]}
375+ listenCalls++
376+ return tracked, nil
377+ })
378+
379+ probeErr := errors.New("unreachable at 192.0.2.2:4444")
380+ probeCalls := 0
381+ withUDPProbe(t, func(context.Context, *Transport) error {
382+ probeCalls++
383+ return probeErr
384+ })
385+
386+ tr, err := NewRemoteDialer("remote.example", "work").Dial(context.Background())
387+ if err == nil {
388+ t.Fatal("Dial() error = nil, want all candidates unavailable")
389+ }
390+ if tr != nil {
391+ t.Fatalf("Dial() transport = %T, want nil", tr)
392+ }
393+ if got, want := probeCalls, 2; got != want {
394+ t.Fatalf("probe calls = %d, want %d", got, want)
395+ }
396+ for i, wasClosed := range closed {
397+ if !wasClosed {
398+ t.Fatalf("candidate socket %d was not closed", i+1)
399+ }
400+ }
401+ if !errors.Is(err, probeErr) {
402+ t.Fatalf("Dial() error = %v, want final probe cause", err)
403+ }
404+ if strings.Contains(err.Error(), "192.0.2.") {
405+ t.Fatalf("Dial() error leaked candidate address: %q", err)
406+ }
407+}
408+
120 func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) {409 func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) {
121 key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))410 key := base64.StdEncoding.EncodeToString(make([]byte, pdgram.KeySize))
122 proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))}411 proc := &fakeBootstrapProcess{stdout: io.NopCloser(strings.NewReader("VEV-UDP 4444 " + key + "\n"))}
@@ -135,8 +424,9 @@ func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) {
135 withListenUDP(t, func(context.Context) (net.PacketConn, error) {424 withListenUDP(t, func(context.Context) (net.PacketConn, error) {
136 return closeTrackPacketConn{PacketConn: pc, closed: &closed}, nil425 return closeTrackPacketConn{PacketConn: pc, closed: &closed}, nil
137 })426 })
427+ withUDPProbe(t, func(context.Context, *Transport) error { return context.DeadlineExceeded })
138 d := NewRemoteDialer("127.0.0.1", "work")428 d := NewRemoteDialer("127.0.0.1", "work")
139- d.ProbeTimeout = time.Nanosecond429+ d.ProbeTimeout = time.Second
140 tr, err := d.Dial(context.Background())430 tr, err := d.Dial(context.Background())
141 if err == nil || !strings.Contains(err.Error(), "remote UDP transport unavailable: probe UDP transport") {431 if err == nil || !strings.Contains(err.Error(), "remote UDP transport unavailable: probe UDP transport") {
142 t.Fatalf("Dial() error = %v, want probe unavailable", err)432 t.Fatalf("Dial() error = %v, want probe unavailable", err)
@@ -148,7 +438,10 @@ func TestRemoteDialerProbeFailureCleansUpWithoutStdioFallback(t *testing.T) {
148 if dialErr.Kind != RemoteDialProbeUnreachable {438 if dialErr.Kind != RemoteDialProbeUnreachable {
149 t.Fatalf("Dial() error kind = %v, want probe unreachable", dialErr.Kind)439 t.Fatalf("Dial() error kind = %v, want probe unreachable", dialErr.Kind)
150 }440 }
151- for _, want := range []string{"firewall", "VEV_REMOTE_TRANSPORT=stdio vev attach 127.0.0.1"} {441+ if !errors.Is(err, context.DeadlineExceeded) {
442+ t.Fatalf("Dial() error = %v, want probe deadline cause", err)
443+ }
444+ for _, want := range []string{"context deadline exceeded", "firewall", "VEV_UDP_PORT_RANGE", "VEV_REMOTE_TRANSPORT=stdio vev attach 127.0.0.1"} {
152 if !strings.Contains(err.Error(), want) {445 if !strings.Contains(err.Error(), want) {
153 t.Fatalf("Dial() error = %q, want actionable hint %q", err, want)446 t.Fatalf("Dial() error = %q, want actionable hint %q", err, want)
154 }447 }
M internal/adapters/dgram/transport_health.go +38 -0
@@ -1,6 +1,8 @@
11 package dgram
22
33 import (
4+ "errors"
5+ "net"
46 "time"
57
68 "github.com/bnema/vev/internal/ports"
@@ -47,6 +49,42 @@ func (t *Transport) notifyTimerHook(get func() func()) {
4749 }
4850 }
4951
52+// replaceDialCandidate installs a fresh socket and remote peer while retaining
53+// the transport's authenticated codec, counters, and replay state. It is used
54+// only while selecting an initial remote UDP peer.
55+func (t *Transport) replaceDialCandidate(pc net.PacketConn, peer net.Addr) error {
56+ // Keep the same lock order as hopPacketConnOnce. In particular, wait for
57+ // in-flight control writes before retiring their packet connection.
58+ t.writeMu.Lock()
59+ defer t.writeMu.Unlock()
60+ t.controlConnMu.Lock()
61+ defer t.controlConnMu.Unlock()
62+ t.mu.Lock()
63+ if t.closed {
64+ err := t.closeErr
65+ t.mu.Unlock()
66+ if err != nil {
67+ return err
68+ }
69+ return errors.New("dgram: closed")
70+ }
71+ old := t.pc
72+ now := t.clock.Now()
73+ t.pc = pc
74+ t.peer = peer
75+ t.writeDeadlines = newWriteDeadlineState(pc)
76+ t.health = newHealthTracker(now)
77+ t.lastAuthenticatedPacket = now
78+ t.lastCompleteRecord = now
79+ t.lastACKProgress = now
80+ t.setLinkStateLocked(ports.LinkStateConnected)
81+ t.mu.Unlock()
82+
83+ go t.readLoop(pc)
84+ _ = old.Close()
85+ return nil
86+}
87+
5088 func (t *Transport) hopPacketConnOnce(generation uint64) {
5189 t.mu.Lock()
5290 if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
@@ -1,6 +1,8 @@
1 package dgram1 package dgram
2 2
3 import (3 import (
4+ "errors"
5+ "net"
4 "time"6 "time"
5 7
6 "github.com/bnema/vev/internal/ports"8 "github.com/bnema/vev/internal/ports"
@@ -47,6 +49,42 @@ func (t *Transport) notifyTimerHook(get func() func()) {
47 }49 }
48 }50 }
49 51
52+// replaceDialCandidate installs a fresh socket and remote peer while retaining
53+// the transport's authenticated codec, counters, and replay state. It is used
54+// only while selecting an initial remote UDP peer.
55+func (t *Transport) replaceDialCandidate(pc net.PacketConn, peer net.Addr) error {
56+ // Keep the same lock order as hopPacketConnOnce. In particular, wait for
57+ // in-flight control writes before retiring their packet connection.
58+ t.writeMu.Lock()
59+ defer t.writeMu.Unlock()
60+ t.controlConnMu.Lock()
61+ defer t.controlConnMu.Unlock()
62+ t.mu.Lock()
63+ if t.closed {
64+ err := t.closeErr
65+ t.mu.Unlock()
66+ if err != nil {
67+ return err
68+ }
69+ return errors.New("dgram: closed")
70+ }
71+ old := t.pc
72+ now := t.clock.Now()
73+ t.pc = pc
74+ t.peer = peer
75+ t.writeDeadlines = newWriteDeadlineState(pc)
76+ t.health = newHealthTracker(now)
77+ t.lastAuthenticatedPacket = now
78+ t.lastCompleteRecord = now
79+ t.lastACKProgress = now
80+ t.setLinkStateLocked(ports.LinkStateConnected)
81+ t.mu.Unlock()
82+
83+ go t.readLoop(pc)
84+ _ = old.Close()
85+ return nil
86+}
87+
50 func (t *Transport) hopPacketConnOnce(generation uint64) {88 func (t *Transport) hopPacketConnOnce(generation uint64) {
51 t.mu.Lock()89 t.mu.Lock()
52 if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {90 if t.rebind == nil || t.hoppedOffline || t.closed || t.linkState != ports.LinkStateProbing || t.health.generation != generation {
M internal/adapters/dgram/transport_health_test.go +48 -0
@@ -55,6 +55,54 @@ func TestPortHopPreservesPendingReliableMessages(t *testing.T) {
5555 }
5656 }
5757
58+func TestReplaceDialCandidateResetsHealthState(t *testing.T) {
59+ oldPC, oldPeer := newPair()
60+ newPC, newPeer := newPair()
61+ defer func() { _ = oldPeer.Close() }()
62+ defer func() { _ = newPeer.Close() }()
63+ tr, err := NewTransport(oldPC, testAddr("b"), key(), 1, 2)
64+ if err != nil {
65+ t.Fatal(err)
66+ }
67+ defer func() { _ = tr.Close() }()
68+
69+ before := tr.clock.Now()
70+ tr.mu.Lock()
71+ tr.health = healthTracker{
72+ lastPacket: before.Add(-time.Hour),
73+ lastRecord: before.Add(-time.Hour),
74+ lastProgress: before.Add(-time.Hour),
75+ pendingSince: before.Add(-time.Hour),
76+ generation: 42,
77+ }
78+ tr.linkState = ports.LinkStateProbing
79+ tr.hoppedOffline = true
80+ tr.hopGeneration = 42
81+ tr.lastAuthenticatedPacket = before.Add(-time.Hour)
82+ tr.lastCompleteRecord = before.Add(-time.Hour)
83+ tr.lastACKProgress = before.Add(-time.Hour)
84+ tr.mu.Unlock()
85+
86+ if err := tr.replaceDialCandidate(newPC, testAddr("b2")); err != nil {
87+ t.Fatal(err)
88+ }
89+
90+ tr.mu.Lock()
91+ defer tr.mu.Unlock()
92+ if got, want := tr.linkState, ports.LinkStateConnected; got != want {
93+ t.Fatalf("link state = %v, want %v", got, want)
94+ }
95+ if tr.hoppedOffline || tr.hopGeneration != 0 {
96+ t.Fatalf("hop state = (%v, %d), want (false, 0)", tr.hoppedOffline, tr.hopGeneration)
97+ }
98+ if tr.health.lastPacket.Before(before) || tr.health.lastRecord.Before(before) || tr.health.lastProgress.Before(before) {
99+ t.Fatalf("health baseline was not reset: %+v", tr.health)
100+ }
101+ if tr.health.pendingSince != (time.Time{}) {
102+ t.Fatalf("pending since = %v, want zero", tr.health.pendingSince)
103+ }
104+}
105+
58106 func TestPortHopRetriesAfterRebindFailure(t *testing.T) {
59107 aPC, bPC := newPair()
60108 var attempts atomic.Int32
@@ -55,6 +55,54 @@ func TestPortHopPreservesPendingReliableMessages(t *testing.T) {
55 }55 }
56 }56 }
57 57
58+func TestReplaceDialCandidateResetsHealthState(t *testing.T) {
59+ oldPC, oldPeer := newPair()
60+ newPC, newPeer := newPair()
61+ defer func() { _ = oldPeer.Close() }()
62+ defer func() { _ = newPeer.Close() }()
63+ tr, err := NewTransport(oldPC, testAddr("b"), key(), 1, 2)
64+ if err != nil {
65+ t.Fatal(err)
66+ }
67+ defer func() { _ = tr.Close() }()
68+
69+ before := tr.clock.Now()
70+ tr.mu.Lock()
71+ tr.health = healthTracker{
72+ lastPacket: before.Add(-time.Hour),
73+ lastRecord: before.Add(-time.Hour),
74+ lastProgress: before.Add(-time.Hour),
75+ pendingSince: before.Add(-time.Hour),
76+ generation: 42,
77+ }
78+ tr.linkState = ports.LinkStateProbing
79+ tr.hoppedOffline = true
80+ tr.hopGeneration = 42
81+ tr.lastAuthenticatedPacket = before.Add(-time.Hour)
82+ tr.lastCompleteRecord = before.Add(-time.Hour)
83+ tr.lastACKProgress = before.Add(-time.Hour)
84+ tr.mu.Unlock()
85+
86+ if err := tr.replaceDialCandidate(newPC, testAddr("b2")); err != nil {
87+ t.Fatal(err)
88+ }
89+
90+ tr.mu.Lock()
91+ defer tr.mu.Unlock()
92+ if got, want := tr.linkState, ports.LinkStateConnected; got != want {
93+ t.Fatalf("link state = %v, want %v", got, want)
94+ }
95+ if tr.hoppedOffline || tr.hopGeneration != 0 {
96+ t.Fatalf("hop state = (%v, %d), want (false, 0)", tr.hoppedOffline, tr.hopGeneration)
97+ }
98+ if tr.health.lastPacket.Before(before) || tr.health.lastRecord.Before(before) || tr.health.lastProgress.Before(before) {
99+ t.Fatalf("health baseline was not reset: %+v", tr.health)
100+ }
101+ if tr.health.pendingSince != (time.Time{}) {
102+ t.Fatalf("pending since = %v, want zero", tr.health.pendingSince)
103+ }
104+}
105+
58 func TestPortHopRetriesAfterRebindFailure(t *testing.T) {106 func TestPortHopRetriesAfterRebindFailure(t *testing.T) {
59 aPC, bPC := newPair()107 aPC, bPC := newPair()
60 var attempts atomic.Int32108 var attempts atomic.Int32

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close