detached 1999182

feat(quicnettest): configurable listen address with family-matched upstream

Add Config.ListenAddr and bind the upstream socket in the server address family, reused on Rebind.
commit
1999182e5d650463f11dc04d97ab40a5f6c3b238
author
brice · · 2026-10-07 05:34 UTC
parent
4e35cb5
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
3 files · +83 -32
Unified
UnifiedSplit
A internal/adapters/quicnettest/listen_test.go +47 -0
@@ -0,0 +1,47 @@
1+package quicnettest
2+
3+import (
4+ "net"
5+ "testing"
6+ "time"
7+
8+ "github.com/stretchr/testify/require"
9+)
10+
11+func TestExplicitListenSurvivesRebind(t *testing.T) {
12+ for _, ip := range []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback} {
13+ t.Run(ip.String(), func(t *testing.T) {
14+ server, err := net.ListenUDP("udp", &net.UDPAddr{IP: ip})
15+ if err != nil && ip.To4() == nil {
16+ t.Skipf("IPv6 loopback unavailable: %v", err)
17+ }
18+ require.NoError(t, err)
19+ defer server.Close()
20+ proxy, err := New(Config{ServerAddr: server.LocalAddr().(*net.UDPAddr), ListenAddr: &net.UDPAddr{IP: ip}})
21+ require.NoError(t, err)
22+ defer proxy.Close()
23+ require.True(t, proxy.Addr().IP.Equal(ip), "requested listener address")
24+ client, err := net.ListenUDP("udp", &net.UDPAddr{IP: ip})
25+ require.NoError(t, err)
26+ defer client.Close()
27+ for attempt := 0; attempt < 2; attempt++ {
28+ if attempt == 1 {
29+ require.NoError(t, proxy.Rebind())
30+ }
31+ _, err = client.WriteToUDP([]byte("fixture"), proxy.Addr())
32+ require.NoError(t, err)
33+ require.NoError(t, server.SetReadDeadline(time.Now().Add(time.Second)))
34+ data := make([]byte, 32)
35+ n, from, err := server.ReadFromUDP(data)
36+ require.NoError(t, err)
37+ require.Equal(t, "fixture", string(data[:n]))
38+ _, err = server.WriteToUDP(data[:n], from)
39+ require.NoError(t, err)
40+ require.NoError(t, client.SetReadDeadline(time.Now().Add(time.Second)))
41+ n, _, err = client.ReadFromUDP(data)
42+ require.NoError(t, err)
43+ require.Equal(t, "fixture", string(data[:n]))
44+ }
45+ })
46+ }
47+}
@@ -0,0 +1,47 @@
1+package quicnettest
2+
3+import (
4+ "net"
5+ "testing"
6+ "time"
7+
8+ "github.com/stretchr/testify/require"
9+)
10+
11+func TestExplicitListenSurvivesRebind(t *testing.T) {
12+ for _, ip := range []net.IP{net.IPv4(127, 0, 0, 1), net.IPv6loopback} {
13+ t.Run(ip.String(), func(t *testing.T) {
14+ server, err := net.ListenUDP("udp", &net.UDPAddr{IP: ip})
15+ if err != nil && ip.To4() == nil {
16+ t.Skipf("IPv6 loopback unavailable: %v", err)
17+ }
18+ require.NoError(t, err)
19+ defer server.Close()
20+ proxy, err := New(Config{ServerAddr: server.LocalAddr().(*net.UDPAddr), ListenAddr: &net.UDPAddr{IP: ip}})
21+ require.NoError(t, err)
22+ defer proxy.Close()
23+ require.True(t, proxy.Addr().IP.Equal(ip), "requested listener address")
24+ client, err := net.ListenUDP("udp", &net.UDPAddr{IP: ip})
25+ require.NoError(t, err)
26+ defer client.Close()
27+ for attempt := 0; attempt < 2; attempt++ {
28+ if attempt == 1 {
29+ require.NoError(t, proxy.Rebind())
30+ }
31+ _, err = client.WriteToUDP([]byte("fixture"), proxy.Addr())
32+ require.NoError(t, err)
33+ require.NoError(t, server.SetReadDeadline(time.Now().Add(time.Second)))
34+ data := make([]byte, 32)
35+ n, from, err := server.ReadFromUDP(data)
36+ require.NoError(t, err)
37+ require.Equal(t, "fixture", string(data[:n]))
38+ _, err = server.WriteToUDP(data[:n], from)
39+ require.NoError(t, err)
40+ require.NoError(t, client.SetReadDeadline(time.Now().Add(time.Second)))
41+ n, _, err = client.ReadFromUDP(data)
42+ require.NoError(t, err)
43+ require.Equal(t, "fixture", string(data[:n]))
44+ }
45+ })
46+ }
47+}
M internal/adapters/quicnettest/proxy.go +33 -32
@@ -2,7 +2,6 @@ package quicnettest
22
33 import (
44 "container/heap"
5- "context"
65 "errors"
76 "fmt"
87 "net"
@@ -19,9 +18,10 @@ var ErrClosed = errors.New("quicnettest: proxy is closed")
1918 // Config.ServerAddr from a separate upstream socket that can be rebound to a
2019 // new source address at any time.
2120 type Proxy struct {
22- cfg Config
23- clock Clock
24- server *net.UDPAddr
21+ cfg Config
22+ clock Clock
23+ server *net.UDPAddr
24+ upstreamAddr *net.UDPAddr
2525
2626 clientConn *net.UDPConn
2727 upstream atomic.Pointer[net.UDPConn]
@@ -128,11 +128,25 @@ func New(cfg Config) (*Proxy, error) {
128128 if err := cfg.validate(); err != nil {
129129 return nil, err
130130 }
131- clientConn, err := listenLoopback()
131+ listenAddr := cfg.ListenAddr
132+ if listenAddr == nil {
133+ listenAddr = &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}
134+ }
135+ clientConn, err := net.ListenUDP("udp", listenAddr)
132136 if err != nil {
133137 return nil, fmt.Errorf("quicnettest: listen client socket: %w", err)
134138 }
135- upstream, err := listenLoopback()
139+ upstreamAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}
140+ if !cfg.ServerAddr.IP.IsLoopback() {
141+ upstreamAddr.IP = net.IPv4zero
142+ }
143+ if cfg.ServerAddr.IP.To4() == nil {
144+ upstreamAddr.IP = net.IPv6unspecified
145+ if cfg.ServerAddr.IP.IsLoopback() {
146+ upstreamAddr.IP = net.IPv6loopback
147+ }
148+ }
149+ upstream, err := net.ListenUDP("udp", upstreamAddr)
136150 if err != nil {
137151 _ = clientConn.Close()
138152 return nil, fmt.Errorf("quicnettest: listen upstream socket: %w", err)
@@ -149,17 +163,18 @@ func New(cfg Config) (*Proxy, error) {
149163 return newRNG(cfg.Seed, direction)
150164 }
151165 p := &Proxy{
152- cfg: cfg,
153- clock: cfg.Clock,
154- server: cfg.ServerAddr,
155- clientConn: clientConn,
156- done: make(chan struct{}),
157- slots: make(chan struct{}, cfg.QueueCapacity),
158- queue: make(chan packet, cfg.QueueCapacity),
159- start: start,
160- toServer: impairer{link: cfg.ToServer, rng: rngFor(ToServer), start: start},
161- toClient: impairer{link: cfg.ToClient, rng: rngFor(ToClient), start: start},
162- onPacket: cfg.OnPacket,
166+ cfg: cfg,
167+ clock: cfg.Clock,
168+ server: cfg.ServerAddr,
169+ upstreamAddr: upstreamAddr,
170+ clientConn: clientConn,
171+ done: make(chan struct{}),
172+ slots: make(chan struct{}, cfg.QueueCapacity),
173+ queue: make(chan packet, cfg.QueueCapacity),
174+ start: start,
175+ toServer: impairer{link: cfg.ToServer, rng: rngFor(ToServer), start: start},
176+ toClient: impairer{link: cfg.ToClient, rng: rngFor(ToClient), start: start},
177+ onPacket: cfg.OnPacket,
163178 }
164179 p.upstream.Store(upstream)
165180 p.wg.Add(3)
@@ -206,7 +221,7 @@ func (p *Proxy) Rebind() error {
206221 if p.stopping() {
207222 return ErrClosed
208223 }
209- conn, err := listenLoopback()
224+ conn, err := net.ListenUDP("udp", p.upstreamAddr)
210225 if err != nil {
211226 return fmt.Errorf("quicnettest: rebind: %w", err)
212227 }
@@ -530,20 +545,6 @@ func directionOf(toServer bool) Direction {
530545 return ToClient
531546 }
532547
533-func listenLoopback() (*net.UDPConn, error) {
534- var lc net.ListenConfig
535- conn, err := lc.ListenPacket(context.Background(), "udp", "127.0.0.1:0")
536- if err != nil {
537- return nil, err
538- }
539- udp, ok := conn.(*net.UDPConn)
540- if !ok {
541- _ = conn.Close()
542- return nil, fmt.Errorf("quicnettest: udp listener returned %T", conn)
543- }
544- return udp, nil
545-}
546-
547548 func sameUDPAddr(a, b *net.UDPAddr) bool {
548549 return a != nil && b != nil && a.Port == b.Port && a.IP.Equal(b.IP)
549550 }
@@ -2,7 +2,6 @@ package quicnettest
2 2
3 import (3 import (
4 "container/heap"4 "container/heap"
5- "context"
6 "errors"5 "errors"
7 "fmt"6 "fmt"
8 "net"7 "net"
@@ -19,9 +18,10 @@ var ErrClosed = errors.New("quicnettest: proxy is closed")
19 // Config.ServerAddr from a separate upstream socket that can be rebound to a18 // Config.ServerAddr from a separate upstream socket that can be rebound to a
20 // new source address at any time.19 // new source address at any time.
21 type Proxy struct {20 type Proxy struct {
22- cfg Config21+ cfg Config
23- clock Clock22+ clock Clock
24- server *net.UDPAddr23+ server *net.UDPAddr
24+ upstreamAddr *net.UDPAddr
25 25
26 clientConn *net.UDPConn26 clientConn *net.UDPConn
27 upstream atomic.Pointer[net.UDPConn]27 upstream atomic.Pointer[net.UDPConn]
@@ -128,11 +128,25 @@ func New(cfg Config) (*Proxy, error) {
128 if err := cfg.validate(); err != nil {128 if err := cfg.validate(); err != nil {
129 return nil, err129 return nil, err
130 }130 }
131- clientConn, err := listenLoopback()131+ listenAddr := cfg.ListenAddr
132+ if listenAddr == nil {
133+ listenAddr = &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}
134+ }
135+ clientConn, err := net.ListenUDP("udp", listenAddr)
132 if err != nil {136 if err != nil {
133 return nil, fmt.Errorf("quicnettest: listen client socket: %w", err)137 return nil, fmt.Errorf("quicnettest: listen client socket: %w", err)
134 }138 }
135- upstream, err := listenLoopback()139+ upstreamAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)}
140+ if !cfg.ServerAddr.IP.IsLoopback() {
141+ upstreamAddr.IP = net.IPv4zero
142+ }
143+ if cfg.ServerAddr.IP.To4() == nil {
144+ upstreamAddr.IP = net.IPv6unspecified
145+ if cfg.ServerAddr.IP.IsLoopback() {
146+ upstreamAddr.IP = net.IPv6loopback
147+ }
148+ }
149+ upstream, err := net.ListenUDP("udp", upstreamAddr)
136 if err != nil {150 if err != nil {
137 _ = clientConn.Close()151 _ = clientConn.Close()
138 return nil, fmt.Errorf("quicnettest: listen upstream socket: %w", err)152 return nil, fmt.Errorf("quicnettest: listen upstream socket: %w", err)
@@ -149,17 +163,18 @@ func New(cfg Config) (*Proxy, error) {
149 return newRNG(cfg.Seed, direction)163 return newRNG(cfg.Seed, direction)
150 }164 }
151 p := &Proxy{165 p := &Proxy{
152- cfg: cfg,166+ cfg: cfg,
153- clock: cfg.Clock,167+ clock: cfg.Clock,
154- server: cfg.ServerAddr,168+ server: cfg.ServerAddr,
155- clientConn: clientConn,169+ upstreamAddr: upstreamAddr,
156- done: make(chan struct{}),170+ clientConn: clientConn,
157- slots: make(chan struct{}, cfg.QueueCapacity),171+ done: make(chan struct{}),
158- queue: make(chan packet, cfg.QueueCapacity),172+ slots: make(chan struct{}, cfg.QueueCapacity),
159- start: start,173+ queue: make(chan packet, cfg.QueueCapacity),
160- toServer: impairer{link: cfg.ToServer, rng: rngFor(ToServer), start: start},174+ start: start,
161- toClient: impairer{link: cfg.ToClient, rng: rngFor(ToClient), start: start},175+ toServer: impairer{link: cfg.ToServer, rng: rngFor(ToServer), start: start},
162- onPacket: cfg.OnPacket,176+ toClient: impairer{link: cfg.ToClient, rng: rngFor(ToClient), start: start},
177+ onPacket: cfg.OnPacket,
163 }178 }
164 p.upstream.Store(upstream)179 p.upstream.Store(upstream)
165 p.wg.Add(3)180 p.wg.Add(3)
@@ -206,7 +221,7 @@ func (p *Proxy) Rebind() error {
206 if p.stopping() {221 if p.stopping() {
207 return ErrClosed222 return ErrClosed
208 }223 }
209- conn, err := listenLoopback()224+ conn, err := net.ListenUDP("udp", p.upstreamAddr)
210 if err != nil {225 if err != nil {
211 return fmt.Errorf("quicnettest: rebind: %w", err)226 return fmt.Errorf("quicnettest: rebind: %w", err)
212 }227 }
@@ -530,20 +545,6 @@ func directionOf(toServer bool) Direction {
530 return ToClient545 return ToClient
531 }546 }
532 547
533-func listenLoopback() (*net.UDPConn, error) {
534- var lc net.ListenConfig
535- conn, err := lc.ListenPacket(context.Background(), "udp", "127.0.0.1:0")
536- if err != nil {
537- return nil, err
538- }
539- udp, ok := conn.(*net.UDPConn)
540- if !ok {
541- _ = conn.Close()
542- return nil, fmt.Errorf("quicnettest: udp listener returned %T", conn)
543- }
544- return udp, nil
545-}
546-
547 func sameUDPAddr(a, b *net.UDPAddr) bool {548 func sameUDPAddr(a, b *net.UDPAddr) bool {
548 return a != nil && b != nil && a.Port == b.Port && a.IP.Equal(b.IP)549 return a != nil && b != nil && a.Port == b.Port && a.IP.Equal(b.IP)
549 }550 }
M internal/adapters/quicnettest/simulation.go +3 -0
@@ -143,6 +143,9 @@ func (l LinkConfig) validate(direction Direction) error {
143143
144144 // Config configures a [Proxy].
145145 type Config struct {
146+ // ListenAddr optionally selects the client-facing fixture address.
147+ // Nil retains the ephemeral loopback default.
148+ ListenAddr *net.UDPAddr
146149 // ServerAddr is the UDP server the proxy forwards to. Required.
147150 ServerAddr *net.UDPAddr
148151 // Clock drives all scheduling. Defaults to [SystemClock].
@@ -143,6 +143,9 @@ func (l LinkConfig) validate(direction Direction) error {
143 143
144 // Config configures a [Proxy].144 // Config configures a [Proxy].
145 type Config struct {145 type Config struct {
146+ // ListenAddr optionally selects the client-facing fixture address.
147+ // Nil retains the ephemeral loopback default.
148+ ListenAddr *net.UDPAddr
146 // ServerAddr is the UDP server the proxy forwards to. Required.149 // ServerAddr is the UDP server the proxy forwards to. Required.
147 ServerAddr *net.UDPAddr150 ServerAddr *net.UDPAddr
148 // Clock drives all scheduling. Defaults to [SystemClock].151 // Clock drives all scheduling. Defaults to [SystemClock].

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