udp/darwin: send to busy established peers on connected sockets

On darwin an unconnected sendto runs the address and MAC policy checks
and the EndpointSecurity send check for every datagram, and while
xprotectd throttles it each send takes about 185us, which caps a
tunnel's upload and, through its ACKs, its download. A connected socket
runs those checks once, at connect, and it is the only UDP send that
reports the interface queue's flow control, as ENOBUFS.

Only the current direct remote of an established tunnel is eligible.
udp.EstablishedPeerConn is an optional Conn interface, implemented on
darwin; establishedRemotes refcounts remotes over the main hostmap and
tells each writer on the first and last reference: handshake completion
adds one, deleting or retiring a hostinfo removes it, SetRemote (roaming,
SetRemoteIfPreferred, control, ssh) moves it. Pending and relayed
hostinfos hold none, and a rehandshake's two hostinfos share one.

An eligible peer that Nebula sends NEBULA_CONNSOCK_RUN (64) datagrams to
within NEBULA_CONNSOCK_WINDOW (1s) gets a connected socket: SO_REUSEPORT,
bound to the source address the listener's route to the peer picks on
the listener's port, carrying the listener's IP_BOUND_IF. The peer sees
the same address and port as from the listener, and the socket's reader
delivers through the listener's reader. The listener itself never sets
SO_REUSEPORT (listen routines still do), so no other uid can bind the
port; with a specific listen.host and no routines connected sockets stay
off, since sharing an exact address needs SO_REUSEPORT on both.

All routines listening on one address share one set of connected
sockets: at most NEBULA_CONNSOCKS (8), one per peer, with opens from a
bucket that refills one per NEBULA_CONNSOCK_OPEN_EVERY (1s). A socket
closes when its peer stops being eligible, after NEBULA_CONNSOCK_IDLE
(30s) without traffic, on Rebind (of every routine) and Close, and on an
error that means the connected path is gone. Closing never runs under
the hostmap's or the eligibility lock; the peer can't reopen until the
close finishes. A send that gets ENOBUFS or EAGAIN waits within one
10ms budget shared by every writer since the last send that went
through, then drops; nothing waits in the netpoller without a deadline.
When nothing is eligible or open, a send costs one atomic load before
the listener path.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

control.go

@@ -203,7 +203,8 @@ func (c *Control) RebindUDPServer() {
 	}
 
 	// A failure here means we are likely still pinned to the interface we came up on, so the rest of this is
-	// unlikely to help. Say so instead of silently carrying on as if we rebound.
+	// unlikely to help. Say so instead of silently carrying on as if we rebound. Only the first writer is rebound;
+	// darwin's Rebind clears every listen routine's listener.
 	if err := c.f.outside.Rebind(); err != nil {
 		c.l.Error("Failed to rebind udp socket", "error", err)
 	}

established_remotes.go

@@ -0,0 +1,90 @@
+package nebula
+
+import (
+	"net/netip"
+	"sync"
+
+	"github.com/slackhq/nebula/udp"
+)
+
+// establishedRemotes tells the udp conns that implement udp.EstablishedPeerConn which addresses are the current
+// direct remote of at least one established tunnel. A hostinfo holds a reference on its remote while it is in the
+// main hostmap, and SetRemote moves the reference; a relayed tunnel has no direct remote and holds none.
+type establishedRemotes struct {
+	mu    sync.Mutex
+	conns []udp.EstablishedPeerConn
+	refs  map[netip.AddrPort]int
+}
+
+// newEstablishedRemotes returns nil when no conn wants to know, which every method accepts.
+func newEstablishedRemotes(conns []udp.Conn) *establishedRemotes {
+	var pc []udp.EstablishedPeerConn
+	for _, c := range conns {
+		if p, ok := c.(udp.EstablishedPeerConn); ok {
+			pc = append(pc, p)
+		}
+	}
+	if len(pc) == 0 {
+		return nil
+	}
+	return &establishedRemotes{conns: pc, refs: map[netip.AddrPort]int{}}
+}
+
+// add makes hostinfo hold a reference on its remote, as it joins the main hostmap. A hostinfo holds references in
+// one establishedRemotes at a time, so an add while it is in another is ignored.
+func (e *establishedRemotes) add(hostinfo *HostInfo) {
+	if e == nil {
+		return
+	}
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if !hostinfo.establishedIn.CompareAndSwap(nil, e) && hostinfo.establishedIn.Load() != e {
+		return
+	}
+	e.syncLocked(hostinfo)
+}
+
+// remove drops hostinfo's reference, as it leaves the main hostmap.
+func (e *establishedRemotes) remove(hostinfo *HostInfo) {
+	if e == nil {
+		return
+	}
+	e.mu.Lock()
+	defer e.mu.Unlock()
+	if hostinfo.establishedIn.CompareAndSwap(e, nil) {
+		e.syncLocked(hostinfo)
+	}
+}
+
+// syncLocked moves hostinfo's reference to the remote it should hold now: its current remote while it is
+// established, none otherwise. Reading both under mu makes the last call win however add, remove and SetRemote
+// interleave. Only the establishedRemotes that hostinfo is in, or has just left, may call it, since
+// hostinfo.establishedRemote is a reference in that one's refs.
+func (e *establishedRemotes) syncLocked(hostinfo *HostInfo) {
+	var want netip.AddrPort
+	if hostinfo.establishedIn.Load() == e {
+		want = hostinfo.GetRemote()
+	}
+	held := hostinfo.establishedRemote
+	if want == held {
+		return
+	}
+	if held.IsValid() {
+		e.refs[held]--
+		if e.refs[held] == 0 {
+			delete(e.refs, held)
+			for _, c := range e.conns {
+				c.SetEstablishedPeer(held, false)
+			}
+		}
+	}
+	if want.IsValid() {
+		e.refs[want]++
+		if e.refs[want] == 1 {
+			for _, c := range e.conns {
+				c.SetEstablishedPeer(want, true)
+			}
+		}
+	}
+	hostinfo.establishedRemote = want
+}

established_remotes_test.go

@@ -0,0 +1,173 @@
+package nebula
+
+import (
+	"net/netip"
+	"sync"
+	"testing"
+
+	"github.com/slackhq/nebula/test"
+	"github.com/slackhq/nebula/udp"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// establishedPeerConn records which destinations the hostmap has told it are established peers, and fails the test
+// on a call that doesn't change anything, which would mean a reference count went wrong.
+type establishedPeerConn struct {
+	udp.NoopConn
+	t     *testing.T
+	mu    sync.Mutex
+	peers map[netip.AddrPort]bool
+}
+
+func (c *establishedPeerConn) SetEstablishedPeer(dst netip.AddrPort, established bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	assert.NotEqual(c.t, established, c.peers[dst], "SetEstablishedPeer(%v, %v) repeated", dst, established)
+	if established {
+		c.peers[dst] = true
+	} else {
+		delete(c.peers, dst)
+	}
+}
+
+func (c *establishedPeerConn) established() []netip.AddrPort {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	var out []netip.AddrPort
+	for dst := range c.peers {
+		out = append(out, dst)
+	}
+	return out
+}
+
+// newEstablishedTestHostMap returns a hostmap whose conns are a recording conn and one that doesn't track peers.
+func newEstablishedTestHostMap(t *testing.T) (*HostMap, *establishedPeerConn) {
+	hm := newHostMap(test.NewLogger())
+	hm.preferredRanges.Store(&[]netip.Prefix{netip.MustParsePrefix("192.168.0.0/16")})
+	c := &establishedPeerConn{t: t, peers: map[netip.AddrPort]bool{}}
+	hm.established = newEstablishedRemotes([]udp.Conn{c, udp.NoopConn{}})
+	require.NotNil(t, hm.established)
+	return hm, c
+}
+
+func newEstablishedTestHostInfo(vpnAddr netip.Addr, index uint32) *HostInfo {
+	return &HostInfo{
+		vpnAddrs:     []netip.Addr{vpnAddr},
+		localIndexId: index,
+		remotes:      NewRemoteList([]netip.Addr{vpnAddr}, nil),
+	}
+}
+
+func TestNewEstablishedRemotesWithoutTrackingConns(t *testing.T) {
+	assert.Nil(t, newEstablishedRemotes([]udp.Conn{udp.NoopConn{}}))
+}
+
+// A conn hears about a remote while at least one hostinfo in the main hostmap has it as its direct remote: from
+// the add, through roaming and SetRemoteIfPreferred, until the last such hostinfo is deleted or retired by
+// MaxHostInfosPerVpnIp. Pending and relayed hostinfos never count.
+func TestHostMap_EstablishedRemotes(t *testing.T) {
+	hm, c := newEstablishedTestHostMap(t)
+	f := &Interface{}
+	a := netip.MustParseAddr("10.0.0.1")
+	r1 := netip.MustParseAddrPort("203.0.113.1:4242")
+	r2 := netip.MustParseAddrPort("203.0.113.2:4242")
+	r3 := netip.MustParseAddrPort("192.168.1.3:4242")
+
+	// A pending hostinfo, as a handshake sets its remote, isn't established.
+	h1 := newEstablishedTestHostInfo(a, 1)
+	h1.SetRemote(r1)
+	assert.Empty(t, c.established())
+
+	// Handshake completion adds it to the main hostmap.
+	hm.unlockedAddHostInfo(h1, f)
+	assert.ElementsMatch(t, []netip.AddrPort{r1}, c.established())
+
+	// A relayed tunnel has no direct remote.
+	relayed := newEstablishedTestHostInfo(netip.MustParseAddr("10.0.0.2"), 2)
+	hm.unlockedAddHostInfo(relayed, f)
+	assert.ElementsMatch(t, []netip.AddrPort{r1}, c.established())
+
+	// A rehandshake's new hostinfo shares the remote, which stays established until both are gone.
+	h3 := newEstablishedTestHostInfo(a, 3)
+	h3.SetRemote(r1)
+	hm.unlockedAddHostInfo(h3, f)
+	hm.DeleteHostInfo(h1)
+	assert.ElementsMatch(t, []netip.AddrPort{r1}, c.established())
+
+	// Roaming moves the reference.
+	h3.SetRemote(r2)
+	assert.ElementsMatch(t, []netip.AddrPort{r2}, c.established())
+	require.True(t, h3.SetRemoteIfPreferred(hm, ViaSender{UdpAddr: r3}))
+	assert.ElementsMatch(t, []netip.AddrPort{r3}, c.established())
+
+	// A relayed tunnel that hears from its peer directly roams onto a direct remote.
+	relayed.SetRemote(r1)
+	assert.ElementsMatch(t, []netip.AddrPort{r1, r3}, c.established())
+	hm.DeleteHostInfo(relayed)
+
+	// A deleted hostinfo lets go, and a SetRemote after that changes nothing.
+	hm.DeleteHostInfo(h3)
+	assert.Empty(t, c.established())
+	h3.SetRemote(r2)
+	assert.Empty(t, c.established())
+	hm.DeleteHostInfo(h3)
+
+	// Adding past MaxHostInfosPerVpnIp retires the oldest hostinfo and its remote.
+	var hs []*HostInfo
+	for i := range MaxHostInfosPerVpnIp + 1 {
+		h := newEstablishedTestHostInfo(a, uint32(10+i))
+		h.SetRemote(netip.AddrPortFrom(netip.MustParseAddr("198.51.100.1"), uint16(1000+i)))
+		hm.unlockedAddHostInfo(h, f)
+		hs = append(hs, h)
+	}
+	assert.Len(t, c.established(), MaxHostInfosPerVpnIp)
+	assert.NotContains(t, c.established(), hs[0].GetRemote())
+}
+
+// Roaming on the receive path races tunnel teardown on the connection manager; whatever the order, a deleted
+// hostinfo holds no reference.
+func TestHostMap_EstablishedRemotesRoamRacesDelete(t *testing.T) {
+	hm, c := newEstablishedTestHostMap(t)
+	f := &Interface{}
+	a := netip.MustParseAddr("10.0.0.1")
+	for i := range 200 {
+		h := newEstablishedTestHostInfo(a, uint32(i+1))
+		h.SetRemote(netip.MustParseAddrPort("203.0.113.1:4242"))
+		hm.Lock()
+		hm.unlockedAddHostInfo(h, f)
+		hm.Unlock()
+		var wg sync.WaitGroup
+		wg.Go(func() { h.SetRemote(netip.AddrPortFrom(netip.MustParseAddr("203.0.113.2"), uint16(i+1))) })
+		wg.Go(func() { hm.DeleteHostInfo(h) })
+		wg.Wait()
+		require.Empty(t, c.established(), "round %d", i)
+	}
+}
+
+// A hostinfo holds its reference in the first tracking hostmap it joins; another's add and delete of it change
+// nothing, so neither hostmap's counts go wrong.
+func TestHostMap_EstablishedRemotesSingleOwner(t *testing.T) {
+	hm1, c1 := newEstablishedTestHostMap(t)
+	hm2, c2 := newEstablishedTestHostMap(t)
+	f := &Interface{}
+	r := netip.MustParseAddrPort("203.0.113.1:4242")
+	h := newEstablishedTestHostInfo(netip.MustParseAddr("10.0.0.1"), 1)
+	h.SetRemote(r)
+
+	hm1.unlockedAddHostInfo(h, f)
+	hm2.unlockedAddHostInfo(h, f)
+	assert.ElementsMatch(t, []netip.AddrPort{r}, c1.established())
+	assert.Empty(t, c2.established())
+
+	hm2.DeleteHostInfo(h)
+	assert.Equal(t, map[netip.AddrPort]int{r: 1}, hm1.established.refs)
+	assert.Empty(t, hm2.established.refs)
+	assert.ElementsMatch(t, []netip.AddrPort{r}, c1.established())
+
+	hm1.DeleteHostInfo(h)
+	assert.Empty(t, hm1.established.refs)
+	assert.Empty(t, hm2.established.refs)
+	assert.Empty(t, c1.established())
+	assert.Empty(t, c2.established())
+}

hostmap.go

@@ -69,7 +69,9 @@ type HostMap struct {
 	Hosts           map[netip.Addr]*HostInfo
 	moreHosts       map[netip.Addr][]*HostInfo
 	preferredRanges atomic.Pointer[[]netip.Prefix]
-	l               *slog.Logger
+	// established hears about each hostinfo that joins or leaves the hostmap; nil when no conn wants to know.
+	established *establishedRemotes
+	l           *slog.Logger
 }
 
 // For synchronization, treat the pointed-to Relay struct as immutable. To edit the Relay
@@ -276,6 +278,11 @@ type HostInfo struct {
 	// This value will be behind against actual tunnel utilization in the hot path.
 	// This should only be used by the ConnectionManagers ticker routine.
 	lastUsed time.Time
+
+	// establishedIn is set while the hostinfo is in a main hostmap whose conns track established remotes, and
+	// establishedRemote is the remote it holds a reference on there, guarded by establishedIn's mu.
+	establishedIn     atomic.Pointer[establishedRemotes]
+	establishedRemote netip.AddrPort
 }
 
 type ViaSender struct {
@@ -516,6 +523,7 @@ func (hm *HostMap) unlockedDeleteHostInfo(hostinfo *HostInfo) bool {
 	if len(hm.Indexes) == 0 {
 		hm.Indexes = map[uint32]*HostInfo{}
 	}
+	hm.established.remove(hostinfo)
 
 	if hm.l.Enabled(context.Background(), slog.LevelDebug) {
 		hm.l.Debug("Hostmap hostInfo deleted",
@@ -662,6 +670,7 @@ func (hm *HostMap) unlockedAddHostInfo(hostinfo *HostInfo, f *Interface) {
 
 	hm.Indexes[hostinfo.localIndexId] = hostinfo
 	hm.RemoteIndexes[hostinfo.remoteIndexId] = hostinfo
+	hm.established.add(hostinfo)
 
 	hostinfo.markOut(f.rebindEpoch.Load())
 	if f.connectionManager != nil { // f.connectionManager is only nil in some unit tests
@@ -842,6 +851,13 @@ func (i *HostInfo) SetRemote(remote netip.AddrPort) {
 	if i.GetRemote() != remote {
 		i.remote.Store(&remote)
 		i.remotes.LearnRemote(i.vpnAddrs[0], remote)
+		if e := i.establishedIn.Load(); e != nil {
+			e.mu.Lock()
+			if i.establishedIn.Load() == e {
+				e.syncLocked(i)
+			}
+			e.mu.Unlock()
+		}
 	}
 }
 

main.go

@@ -213,6 +213,7 @@ func Main(c *config.C, configTest bool, buildVersion string, l *slog.Logger, dev
 	}
 
 	hostMap := NewHostMapFromConfig(l, c)
+	hostMap.established = newEstablishedRemotes(udpConns)
 	punchy := NewPunchyFromConfig(l, c, udpConns[0])
 	connManager := newConnectionManagerFromConfig(l, c, hostMap, punchy)
 	lightHouse, err := NewLightHouseFromConfig(ctx, l, c, pki.getCertState(), udpConns[0], punchy)

udp/conn.go

@@ -48,6 +48,14 @@ type Conn interface {
 	Close() error
 }
 
+// EstablishedPeerConn is a Conn that treats the remotes of established tunnels differently from other
+// destinations, as darwin's gives busy ones connected sockets.
+type EstablishedPeerConn interface {
+	// SetEstablishedPeer reports whether dst is the current remote of at least one established tunnel. Calls for
+	// one dst must not race each other.
+	SetEstablishedPeer(dst netip.AddrPort, established bool)
+}
+
 type NoopConn struct{}
 
 func (NoopConn) Rebind() error {

udp/connsock_bench_darwin_test.go

@@ -0,0 +1,115 @@
+//go:build !e2e_testing
+
+package udp
+
+import (
+	"log/slog"
+	"net"
+	"net/netip"
+	"testing"
+	"time"
+)
+
+// benchConnSocks opens a wildcard listener with ListenOut running, with the connected socket knobs pinned to
+// their defaults and then cfg applied, and a loopback peer that never reads, so its sends cost only the sender.
+func benchConnSocks(b *testing.B, s Settings, cfg func(*connSockConfig)) (*StdConn, netip.AddrPort) {
+	b.Helper()
+	for k, v := range map[string]string{
+		"NEBULA_CONNSOCKS":           "8",
+		"NEBULA_CONNSOCK_RUN":        "64",
+		"NEBULA_CONNSOCK_WINDOW":     "1s",
+		"NEBULA_CONNSOCK_IDLE":       "30s",
+		"NEBULA_CONNSOCK_OPEN_EVERY": "1s",
+	} {
+		b.Setenv(k, v)
+	}
+	c, err := NewListener(slog.New(slog.DiscardHandler), s)
+	if err != nil {
+		b.Fatal(err)
+	}
+	b.Cleanup(func() { _ = c.Close() })
+	u := c.(*StdConn)
+	if cfg != nil {
+		cfg(&u.socks.cfg)
+	}
+	go func() { _ = u.ListenOut(func(netip.AddrPort, []byte) {}, func() {}) }()
+	for u.reader.Load() == nil {
+		time.Sleep(time.Millisecond)
+	}
+	pc, err := net.ListenPacket("udp4", "127.0.0.1:0")
+	if err != nil {
+		b.Fatal(err)
+	}
+	b.Cleanup(func() { _ = pc.Close() })
+	return u, pc.LocalAddr().(*net.UDPAddr).AddrPort()
+}
+
+// BenchmarkConnSockWriteTo measures one WriteTo of a full-size datagram down each path: the listener with
+// connected sockets off, to a destination that is no tunnel's remote while another is, to an established peer
+// that never gets busy, and on a connected socket.
+func BenchmarkConnSockWriteTo(b *testing.B) {
+	pkt := make([]byte, 1400)
+	cases := []struct {
+		name  string
+		cfg   func(*connSockConfig)
+		setup func(b *testing.B, u *StdConn, dst netip.AddrPort)
+	}{
+		{name: "listener/off", cfg: func(c *connSockConfig) { c.max = 0 }},
+		{name: "listener/ineligible", setup: func(b *testing.B, u *StdConn, dst netip.AddrPort) {
+			u.SetEstablishedPeer(netip.AddrPortFrom(dst.Addr(), dst.Port()+1), true)
+		}},
+		{name: "listener/eligible-not-busy", cfg: func(c *connSockConfig) { c.run = 1 << 30 },
+			setup: func(b *testing.B, u *StdConn, dst netip.AddrPort) { u.SetEstablishedPeer(dst, true) }},
+		{name: "connected", setup: func(b *testing.B, u *StdConn, dst netip.AddrPort) {
+			u.SetEstablishedPeer(dst, true)
+			for range u.socks.cfg.run {
+				_ = u.WriteTo(pkt, dst)
+			}
+			if u.connSockFor(dst) == nil {
+				b.Fatal("no connected socket")
+			}
+		}},
+	}
+	for _, c := range cases {
+		b.Run(c.name, func(b *testing.B) {
+			u, dst := benchConnSocks(b, Settings{Listen: netip.MustParseAddrPort("0.0.0.0:0")}, c.cfg)
+			if c.setup != nil {
+				c.setup(b, u, dst)
+			}
+			b.ReportAllocs()
+			b.ResetTimer()
+			for range b.N {
+				_ = u.WriteTo(pkt, dst)
+			}
+		})
+	}
+}
+
+// BenchmarkConnSockDispatch measures what WriteTo adds before the listener's sendto, without the syscall: a v4
+// listener refuses a v6 destination before sending, with connected sockets off or with another peer established.
+func BenchmarkConnSockDispatch(b *testing.B) {
+	pkt := make([]byte, 1400)
+	dst := netip.MustParseAddrPort("[::1]:4242")
+	for _, c := range []struct {
+		name      string
+		cfg       func(*connSockConfig)
+		establish bool
+	}{
+		{name: "off", cfg: func(c *connSockConfig) { c.max = 0 }},
+		{name: "ineligible", establish: true},
+	} {
+		b.Run(c.name, func(b *testing.B) {
+			u, peer := benchConnSocks(b, Settings{Listen: netip.MustParseAddrPort("127.0.0.1:0"), Multi: true}, c.cfg)
+			if c.establish {
+				u.SetEstablishedPeer(peer, true)
+			}
+			b.ReportAllocs()
+			b.ResetTimer()
+			for range b.N {
+				if u.WriteTo(pkt, dst) == nil {
+					b.Fatal("sent a v6 datagram on a v4 listener")
+				}
+			}
+		})
+	}
+}

udp/connsock_darwin.go

@@ -0,0 +1,803 @@
+//go:build !e2e_testing
+
+package udp
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log/slog"
+	"maps"
+	"net"
+	"net/netip"
+	"os"
+	"slices"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"time"
+
+	"golang.org/x/sys/unix"
+)
+
+// A connected socket is a UDP socket connected to one busy peer, on the listener's port and bound to the source
+// address the listener's route to that peer picks, so the peer sees the same address and port from either. A send
+// on a connected socket skips the per-datagram address and MAC policy checks, and the EndpointSecurity send check
+// that xprotectd can hold every unconnected sendto on, and it is the only UDP send that reports the interface
+// queue's flow control, as ENOBUFS. A connected socket wins xnu's lookup for its peer's datagrams; it never sees
+// anyone else's.
+//
+// A connected socket sets SO_REUSEPORT and binds a specific address beside a wildcard listener, which sets
+// SO_REUSEPORT only for listen routines. xnu refuses a wildcard SO_REUSEPORT bind beside a holder without the
+// option, and refuses another uid's bind to an address a socket holds specifically, so no other user can join the
+// port while connected sockets open and close: uid nobody got EADDRINUSE on every bind against a root listener
+// with connected sockets in the 09-25 probes. A connected socket is closed, never disconnected, because xnu widens
+// a disconnected socket's local address to the wildcard.
+//
+// Only the current remote of an established tunnel may get one: Nebula names those with SetEstablishedPeer, so an
+// unauthenticated sender can't earn a connected socket by drawing replies such as recv_error or handshake
+// responses.
+type connSock struct {
+	dst  netip.AddrPort
+	peer *connSockPeer
+	conn *net.UDPConn
+	rc   syscall.RawConn
+	// used is the monotonic time in nanoseconds of the connected socket's last send or receive.
+	used atomic.Int64
+	// stallAt is the monotonic time of the first ENOBUFS or EAGAIN since a send last went through, or 0, and
+	// dropping is set once that stall has dropped a datagram, so each stall logs its drops once.
+	stallAt  atomic.Int64
+	dropping atomic.Bool
+	// parkMu serializes the writes that wait for a full send buffer, since the write deadline bounding each wait
+	// is the descriptor's, not the writer's.
+	parkMu sync.Mutex
+	// sys writes one datagram on the descriptor: writeFd, or a stand-in for the interface in tests.
+	sys func(fd int, b []byte) error
+}
+
+func writeFd(fd int, b []byte) error {
+	_, err := syscall.Write(fd, b)
+	return err
+}
+
+// errStallSpent is a write that found the connected socket's stall budget spent, so its datagram is dropped.
+var errStallSpent = errors.New("connected socket stall budget spent")
+
+// connSockWriter carries one write through RawConn.Write. They are pooled with fn bound once, since a closure
+// passed through the RawConn interface escapes and would allocate on every send.
+type connSockWriter struct {
+	s    *connSock
+	b    []byte
+	park bool
+	err  error
+	fn   func(fd uintptr) bool
+}
+
+var connSockWriters = sync.Pool{New: func() any {
+	w := &connSockWriter{}
+	w.fn = w.write
+	return w
+}}
+
+func (w *connSockWriter) write(fd uintptr) bool {
+	for {
+		w.err = w.s.sys(int(fd), w.b)
+		switch {
+		case w.err == syscall.EINTR:
+		case w.err == syscall.EAGAIN && w.park:
+			return false
+		default:
+			return true
+		}
+	}
+}
+
+// try writes b on s once and returns nil, the write's errno, os.ErrDeadlineExceeded, or net.ErrClosed. With park
+// set, EAGAIN waits in the netpoller until the socket is writable or the write deadline passes; without it, EAGAIN
+// is returned.
+func (s *connSock) try(b []byte, park bool) error {
+	w := connSockWriters.Get().(*connSockWriter)
+	w.s, w.b, w.park = s, b, park
+	rerr := s.rc.Write(w.fn)
+	err := w.err
+	*w = connSockWriter{fn: w.fn}
+	connSockWriters.Put(w)
+	switch {
+	case rerr == nil:
+		return err
+	case errors.Is(rerr, os.ErrDeadlineExceeded):
+		return os.ErrDeadlineExceeded
+	default:
+		return net.ErrClosed
+	}
+}
+
+// connSockConfig holds the knobs. connSockConfigFromEnv reads all but backoff and enobufsMax from the environment,
+// so one binary can be tuned without a rebuild.
+type connSockConfig struct {
+	// max is how many connected sockets the listen routines may have open at once; 0 turns connected sockets off.
+	max int
+	// run is how many datagrams to one destination within one window earn it a connected socket. darwin's tun
+	// hands over one packet per read, so a peer's sends arrive one WriteTo or one-datagram batch at a time and
+	// busyness has to be counted across calls.
+	run    int
+	window time.Duration
+	// idle closes a connected socket after it has neither sent nor received for this long.
+	idle time.Duration
+	// backoff keeps a destination whose connected socket failed on the listener for this long.
+	backoff time.Duration
+	// openEvery is how often the open bucket earns back an open once max are spent, which bounds connected socket
+	// churn however fast senders earn connected sockets and errors close them.
+	openEvery time.Duration
+	// enobufsMax is how long a connected socket's sends may wait out ENOBUFS without one going through before they
+	// drop instead, so a suspended interface costs one write per datagram rather than stalling the tun reader.
+	enobufsMax time.Duration
+}
+
+func connSockConfigFromEnv() connSockConfig {
+	c := connSockConfig{
+		max:        8,
+		run:        64,
+		window:     time.Second,
+		idle:       30 * time.Second,
+		backoff:    30 * time.Second,
+		openEvery:  time.Second,
+		enobufsMax: 10 * time.Millisecond,
+	}
+	if v, err := strconv.Atoi(os.Getenv("NEBULA_CONNSOCKS")); err == nil {
+		c.max = v
+	}
+	if v, err := strconv.Atoi(os.Getenv("NEBULA_CONNSOCK_RUN")); err == nil && v > 0 {
+		c.run = v
+	}
+	if v, err := time.ParseDuration(os.Getenv("NEBULA_CONNSOCK_WINDOW")); err == nil && v > 0 {
+		c.window = v
+	}
+	if v, err := time.ParseDuration(os.Getenv("NEBULA_CONNSOCK_IDLE")); err == nil && v > 0 {
+		c.idle = v
+	}
+	if v, err := time.ParseDuration(os.Getenv("NEBULA_CONNSOCK_OPEN_EVERY")); err == nil && v > 0 {
+		c.openEvery = v
+	}
+	return c
+}
+
+// connSockGroups holds the connSocks that the listen routines bound to one address share, so the limits, the
+// connected sockets and Rebind cover every routine, and a peer has at most one connected socket.
+var connSockGroups = struct {
+	sync.Mutex
+	m map[netip.AddrPort]*connSocks
+}{m: map[netip.AddrPort]*connSocks{}}
+
+// joinConnSocks returns the connSocks u shares with the other listen routines bound to la, and makes it for the
+// first.
+func joinConnSocks(u *StdConn, la netip.AddrPort, s Settings) *connSocks {
+	g := &connSockGroups
+	g.Lock()
+	defer g.Unlock()
+	if cs := g.m[la]; s.Multi && cs != nil {
+		cs.mu.Lock()
+		cs.members = append(cs.members, u)
+		cs.mu.Unlock()
+		return cs
+	}
+	cs := &connSocks{cfg: connSockConfigFromEnv(), key: la, members: []*StdConn{u}, clock: monoNow, sleep: time.Sleep}
+	if cs.cfg.max > 0 && !u.listenHost.IsUnspecified() && !s.Multi {
+		// A connected socket on a specific listen.host binds the listener's own address, which xnu allows only when
+		// both sockets set SO_REUSEPORT, and a listener with it lets another uid bind a dual-stack wildcard on the
+		// port.
+		cs.off.Store(true)
+		u.l.Debug("connected sockets: off for a specific listen.host", "addr", s.Listen)
+	}
+	if s.Multi {
+		g.m[la] = cs
+	}
+	return cs
+}
+
+type connSocks struct {
+	cfg connSockConfig
+	// key is the listeners' bound address, which connSockGroups files this under.
+	key netip.AddrPort
+	// off keeps connected sockets from opening, when the listener's address can't be shared with them.
+	off atomic.Bool
+	// table maps each destination that is the remote of an established tunnel, and that a connected socket could
+	// reach, to its peer; it is nil when there are none, so a send with nothing to consider costs one load. Every
+	// send reads it without a lock, and mu guards copying it. An open that holds mu and still finds its peer here
+	// can't outlive the tunnel.
+	table atomic.Pointer[map[netip.AddrPort]*connSockPeer]
+	// reserved counts the connected sockets that are open, opening or closing, against cfg.max; mu guards changing it.
+	reserved atomic.Int32
+	mu       sync.Mutex
+	closed   bool
+	// members are the listen routines' listeners that share these connected sockets.
+	members []*StdConn
+	// pending holds each destination whose connected socket is opening or closing, and not on its peer. xnu
+	// refuses to connect a 4-tuple that is still bound, so a destination in pending can't open another.
+	pending map[netip.AddrPort]struct{}
+	// rebinds counts Rebinds, so an open that copied the listener's interface before one is discarded.
+	rebinds uint64
+	// closers tracks the goroutines that close detached connected sockets.
+	closers sync.WaitGroup
+	// done stops the reaper; it is made when ListenOut starts the connected sockets.
+	done chan struct{}
+	// opens bounds connected socket opens to cfg.max at once and one per cfg.openEvery after.
+	opens gcra
+	// logs bounds connected socket log lines at info or warn; see connSockLogBurst.
+	logs gcra
+	// clock returns monotonic nanoseconds and sleep waits out ENOBUFS: monoNow and time.Sleep, or stand-ins in
+	// tests.
+	clock func() int64
+	sleep func(time.Duration)
+	// dialHook and closeHook, when set, run just before an open's dials and a close's Close, outside mu; tests
+	// use them to hold an open or close in flight.
+	dialHook, closeHook func(dst netip.AddrPort)
+}
+
+// connSockPeer is what connSocks keeps about an established tunnel's remote: its connected socket, how busy it is,
+// and how long a failed connected socket keeps it on the listener.
+type connSockPeer struct {
+	sock atomic.Pointer[connSock]
+	// windowMu guards start, the monotonic time the current window began, and n, the datagrams sent in it, so a
+	// window's reset and its counts can't interleave with another sender's.
+	windowMu sync.Mutex
+	start    int64
+	n        int
+	// retryAt is the monotonic time before which a peer whose connected socket failed stays on the listener;
+	// connSocks.mu guards it.
+	retryAt int64
+}
+
+// busy counts n more datagrams to p and reports whether it has now sent cfg.run within one window. A busy peer's
+// count starts over, so a peer that can't get a connected socket retries at most once per cfg.run sends.
+func (p *connSockPeer) busy(now int64, n int, cfg *connSockConfig) bool {
+	p.windowMu.Lock()
+	defer p.windowMu.Unlock()
+	if time.Duration(now-p.start) > cfg.window {
+		p.start, p.n = now, 0
+	}
+	p.n += n
+	if p.n < cfg.run {
+		return false
+	}
+	p.n = 0
+	return true
+}
+
+// connSockLogBurst and connSockLogEvery size the bucket that lets connected socket opens, closes and failures log
+// at info or warn; past it they log at debug, so churn can't flood the log.
+const (
+	connSockLogBurst = 32
+	connSockLogEvery = 10 * time.Second
+)
+
+var monoStart = time.Now()
+
+func monoNow() int64 { return int64(time.Since(monoStart)) }
+
+// gcra is a token bucket kept as the monotonic time it is next full (the generic cell rate algorithm), so it needs
+// no refill tick.
+type gcra struct{ tat int64 }
+
+// take spends a token from a bucket of burst that earns one back per every, and reports whether there was one to
+// spend.
+func (g *gcra) take(now int64, burst int, every time.Duration) bool {
+	e := int64(every)
+	tat := max(g.tat, now)
+	if tat-now > int64(burst-1)*e {
+		return false
+	}
+	g.tat = tat + e
+	return true
+}
+
+// logLevelLocked returns level while the log bucket has a token, and debug once it is spent.
+func (cs *connSocks) logLevelLocked(level slog.Level) slog.Level {
+	if cs.logs.take(cs.clock(), connSockLogBurst, connSockLogEvery) {
+		return level
+	}
+	return slog.LevelDebug
+}
+
+// startConnSocks starts the idle reaper.
+func (u *StdConn) startConnSocks() {
+	cs := u.socks
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	if cs.closed || cs.done != nil || cs.cfg.max <= 0 || cs.off.Load() {
+		return
+	}
+	cs.done = make(chan struct{})
+	go u.reapConnSocks(cs.done)
+}
+
+// stopConnSocks closes every connected socket and keeps new ones from opening, once any listen routine's listener
+// closes, as they do together at shutdown. It returns once their descriptors are closed.
+func (u *StdConn) stopConnSocks() {
+	cs := u.socks
+	cs.mu.Lock()
+	cs.members = slices.DeleteFunc(cs.members, func(m *StdConn) bool { return m == u })
+	if !cs.closed {
+		cs.closed = true
+		if cs.done != nil {
+			close(cs.done)
+		}
+		u.detachAllLocked("close")
+	}
+	cs.mu.Unlock()
+	connSockGroups.Lock()
+	if connSockGroups.m[cs.key] == cs {
+		delete(connSockGroups.m, cs.key)
+	}
+	connSockGroups.Unlock()
+	cs.closers.Wait()
+}
+
+// siblings returns the other listen routines' listeners.
+func (cs *connSocks) siblings(u *StdConn) []*StdConn {
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	return slices.DeleteFunc(slices.Clone(cs.members), func(m *StdConn) bool { return m == u })
+}
+
+// connSockFor returns dst's connected socket, or nil.
+func (u *StdConn) connSockFor(dst netip.AddrPort) *connSock {
+	if p := u.socks.peer(dst); p != nil {
+		return p.sock.Load()
+	}
+	return nil
+}
+
+func (cs *connSocks) peer(dst netip.AddrPort) *connSockPeer {
+	if t := cs.table.Load(); t != nil {
+		return (*t)[dst]
+	}
+	return nil
+}
+
+// SetEstablishedPeer makes dst eligible for a connected socket while it is the remote of an established tunnel,
+// and closes the one it has once it no longer is. Nebula calls it for every listen routine under the hostmap's
+// write lock, so repeats are no-ops and it makes no syscalls: the descriptor is closed by another goroutine.
+func (u *StdConn) SetEstablishedPeer(dst netip.AddrPort, established bool) {
+	cs := u.socks
+	if cs.cfg.max <= 0 || cs.off.Load() || !u.connSockable(dst) {
+		return
+	}
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	t := cs.table.Load()
+	p := cs.peer(dst)
+	if cs.closed || (p != nil) == established {
+		return
+	}
+	next := map[netip.AddrPort]*connSockPeer{}
+	if t != nil {
+		next = maps.Clone(*t)
+	}
+	if established {
+		p = &connSockPeer{start: cs.clock()}
+		next[dst] = p
+	} else {
+		delete(next, dst)
+		if s := p.sock.Load(); s != nil {
+			u.detachLocked(s, "tunnel gone", false)
+		}
+	}
+	cs.storeTableLocked(next)
+}
+
+// storeTableLocked publishes t, or nil for an empty one.
+func (cs *connSocks) storeTableLocked(t map[netip.AddrPort]*connSockPeer) {
+	if len(t) == 0 {
+		cs.table.Store(nil)
+		return
+	}
+	cs.table.Store(&t)
+}
+
+// connSockable reports whether a connected socket to dst could send what the listener would. A v4 listener can't
+// reach a v6 destination at all, and a connected socket bound to a specific listen.host must share its family.
+func (u *StdConn) connSockable(dst netip.AddrPort) bool {
+	if !dst.IsValid() || dst.Port() == 0 {
+		return false
+	}
+	if u.isV4 {
+		return dst.Addr().Is4()
+	}
+	return u.listenHost.IsUnspecified() || u.listenHost.Unmap().Is4() == dst.Addr().Unmap().Is4()
+}
+
+// connSockRun sends bufs, all to p's destination, on p's connected socket, opening one if they make p busy. It
+// returns how many went out; when that is short of len(bufs), err is the failure of bufs[sent], or nil to leave
+// bufs[sent:] to the listener.
+func (u *StdConn) connSockRun(p *connSockPeer, dst netip.AddrPort, bufs [][]byte) (sent int, err error) {
+	s := p.sock.Load()
+	if s == nil {
+		if s = u.maybeOpenConnSock(p, dst, len(bufs)); s == nil {
+			return 0, nil
+		}
+	}
+	sent, errno, closed := u.connSockWrite(s, bufs)
+	switch {
+	case sent == len(bufs):
+	case connSockDead(errno):
+		u.closeConnSock(s, errno.Error(), true)
+	case !closed:
+		return sent, &net.OpError{Op: "write", Err: errno}
+	}
+	return sent, nil
+}
+
+// maybeOpenConnSock counts n datagrams to p, and opens a connected socket to it once it is busy and nothing rules
+// one out.
+func (u *StdConn) maybeOpenConnSock(p *connSockPeer, dst netip.AddrPort, n int) *connSock {
+	cs := u.socks
+	if n <= 0 || u.reader.Load() == nil || int(cs.reserved.Load()) >= cs.cfg.max || !p.busy(cs.clock(), n, &cs.cfg) {
+		return nil
+	}
+	cs.mu.Lock()
+	if cs.closed || cs.peer(dst) != p {
+		cs.mu.Unlock()
+		return nil
+	}
+	if s := p.sock.Load(); s != nil {
+		cs.mu.Unlock()
+		return s
+	}
+	now := cs.clock()
+	if _, ok := cs.pending[dst]; ok || int(cs.reserved.Load()) >= cs.cfg.max || now < p.retryAt ||
+		!cs.opens.take(now, cs.cfg.max, cs.cfg.openEvery) {
+		cs.mu.Unlock()
+		return nil
+	}
+	if cs.pending == nil {
+		cs.pending = map[netip.AddrPort]struct{}{}
+	}
+	cs.pending[dst] = struct{}{}
+	cs.reserved.Add(1)
+	rebinds := cs.rebinds
+	cs.mu.Unlock()
+
+	if cs.dialHook != nil {
+		cs.dialHook(dst)
+	}
+	s, err := u.dialConnSock(p, dst)
+
+	cs.mu.Lock()
+	if err != nil {
+		delete(cs.pending, dst)
+		cs.reserved.Add(-1)
+		p.retryAt = cs.clock() + int64(cs.cfg.backoff)
+		level := cs.logLevelLocked(slog.LevelWarn)
+		cs.mu.Unlock()
+		u.l.Log(context.Background(), level, "connected socket: open failed, staying on the listener", "udpAddr", dst, "error", err)
+		return nil
+	}
+	if cs.closed || cs.rebinds != rebinds || cs.peer(dst) != p {
+		// The tunnel went away, or a Rebind or shutdown ran, while the socket was opening; it stays in pending until
+		// it is closed.
+		cs.mu.Unlock()
+		u.finishClose(s)
+		u.l.Debug("connected socket: discarded, its tunnel or route changed while it opened", "udpAddr", dst)
+		return nil
+	}
+	delete(cs.pending, dst)
+	p.sock.Store(s)
+	go u.readConnSock(s)
+	level := cs.logLevelLocked(slog.LevelInfo)
+	cs.mu.Unlock()
+	u.l.Log(context.Background(), level, "connected socket: opened", "udpAddr", dst, "local", s.conn.LocalAddr(), "open", cs.reserved.Load())
+	return s
+}
+
+// dialConnSock opens a socket on the listener's port with SO_REUSEPORT, bound to the configured listen.host or,
+// for a wildcard listener, to the source address the route to dst picks, and connected to dst. It carries the
+// listener's IP_BOUND_IF, so both route the same way.
+func (u *StdConn) dialConnSock(p *connSockPeer, dst netip.AddrPort) (*connSock, error) {
+	raddr := net.UDPAddrFromAddrPort(netip.AddrPortFrom(dst.Addr().Unmap(), dst.Port()))
+	network := "udp6"
+	if raddr.IP.To4() != nil {
+		network = "udp4"
+	}
+	ifindex, err := u.boundIf()
+	if err != nil {
+		return nil, err
+	}
+	local := &net.UDPAddr{IP: u.listenHost.AsSlice(), Zone: u.listenHost.Zone()}
+	if u.listenHost.IsUnspecified() {
+		// A connected socket gets the source address xnu would pick for an unconnected send to dst.
+		d := net.Dialer{Control: connSockControl(false, ifindex)}
+		probe, err := d.Dial(network, raddr.String())
+		if err != nil {
+			return nil, err
+		}
+		local = probe.LocalAddr().(*net.UDPAddr)
+		_ = probe.Close()
+	}
+	local.Port = int(u.port)
+	d := net.Dialer{LocalAddr: local, Control: connSockControl(true, ifindex)}
+	c, err := d.Dial(network, raddr.String())
+	if err != nil {
+		return nil, err
+	}
+	uc := c.(*net.UDPConn)
+	rc, err := uc.SyscallConn()
+	if err != nil {
+		_ = uc.Close()
+		return nil, err
+	}
+	s := &connSock{dst: dst, peer: p, conn: uc, rc: rc, sys: writeFd}
+	s.used.Store(u.socks.clock())
+	return s, nil
+}
+
+// connSockControl sets SO_REUSEPORT when reuse is set, and IP_BOUND_IF or IPV6_BOUND_IF when ifindex isn't 0,
+// before the socket binds.
+func connSockControl(reuse bool, ifindex int) func(network, address string, c syscall.RawConn) error {
+	return func(network, _ string, c syscall.RawConn) error {
+		var serr error
+		err := c.Control(func(fd uintptr) {
+			if reuse {
+				serr = unix.SetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT, 1)
+			}
+			if serr == nil && ifindex != 0 {
+				if network == "udp4" {
+					serr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_BOUND_IF, ifindex)
+				} else {
+					serr = unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_BOUND_IF, ifindex)
+				}
+			}
+		})
+		if err != nil {
+			return err
+		}
+		return serr
+	}
+}
+
+// boundIf returns the interface index the listener is scoped to, or 0. An error keeps the peer on the listener,
+// since a connected socket opened without the listener's scope could route out an interface the listener can't.
+func (u *StdConn) boundIf() (int, error) {
+	level, opt := unix.IPPROTO_IPV6, unix.IPV6_BOUND_IF
+	if u.isV4 {
+		level, opt = unix.IPPROTO_IP, unix.IP_BOUND_IF
+	}
+	v, err := unix.GetsockoptInt(int(u.sysFd), level, opt)
+	if err != nil {
+		return 0, fmt.Errorf("reading the listener's bound interface: %w", err)
+	}
+	return v, nil
+}
+
+// closeConnSock closes s if it is still open. With backoff set, the peer stays on the listener until the backoff
+// runs out.
+func (u *StdConn) closeConnSock(s *connSock, why string, backoff bool) {
+	u.socks.mu.Lock()
+	u.detachLocked(s, why, backoff)
+	u.socks.mu.Unlock()
+}
+
+// detachLocked takes s off its peer, if it is still there, and closes it on another goroutine, since callers such
+// as SetEstablishedPeer hold locks that no socket syscall should run under. s.dst stays in pending until the close
+// completes.
+func (u *StdConn) detachLocked(s *connSock, why string, backoff bool) {
+	cs := u.socks
+	if !s.peer.sock.CompareAndSwap(s, nil) {
+		return
+	}
+	if backoff {
+		s.peer.retryAt = cs.clock() + int64(cs.cfg.backoff)
+	}
+	if cs.pending == nil {
+		cs.pending = map[netip.AddrPort]struct{}{}
+	}
+	cs.pending[s.dst] = struct{}{}
+	level := cs.logLevelLocked(slog.LevelInfo)
+	cs.closers.Add(1)
+	go func() {
+		defer cs.closers.Done()
+		open := u.finishClose(s)
+		u.l.Log(context.Background(), level, "connected socket: closed", "udpAddr", s.dst, "reason", why, "open", open)
+	}()
+}
+
+// finishClose closes s, whose destination is in pending, then lets the destination open again, and returns how
+// many connected sockets are left. Closing waits only for a write already in a syscall: a write parked on a full
+// buffer is woken, and one waiting out ENOBUFS holds no reference on the descriptor.
+func (u *StdConn) finishClose(s *connSock) int32 {
+	cs := u.socks
+	if cs.closeHook != nil {
+		cs.closeHook(s.dst)
+	}
+	_ = s.conn.Close()
+	cs.mu.Lock()
+	defer cs.mu.Unlock()
+	delete(cs.pending, s.dst)
+	return cs.reserved.Add(-1)
+}
+
+// detachAllLocked detaches every connected socket, for a network change or shutdown.
+func (u *StdConn) detachAllLocked(why string) {
+	if t := u.socks.table.Load(); t != nil {
+		for _, p := range *t {
+			if s := p.sock.Load(); s != nil {
+				u.detachLocked(s, why, false)
+			}
+		}
+	}
+}
+
+// closeConnSocks detaches every connected socket and discards the ones still opening, for a network change.
+func (u *StdConn) closeConnSocks(why string) {
+	u.socks.mu.Lock()
+	u.socks.rebinds++
+	u.detachAllLocked(why)
+	u.socks.mu.Unlock()
+}
+
+func (u *StdConn) reapConnSocks(done <-chan struct{}) {
+	cs := u.socks
+	idle := cs.cfg.idle
+	tick := time.NewTicker(max(min(idle/4, 5*time.Second), time.Millisecond))
+	defer tick.Stop()
+	for {
+		select {
+		case <-done:
+			return
+		case <-tick.C:
+		}
+		t := cs.table.Load()
+		if t == nil {
+			continue
+		}
+		now := cs.clock()
+		for _, p := range *t {
+			if s := p.sock.Load(); s != nil && time.Duration(now-s.used.Load()) > idle {
+				u.closeConnSock(s, "idle", false)
+			}
+		}
+	}
+}
+
+// connSockDead reports whether a connected socket send or receive error means the connected path to the peer is
+// gone, so its traffic belongs back on the listener. A connected UDP socket reports a peer's ICMP unreachable as
+// ECONNREFUSED on its next call, which the unconnected listener never sees.
+func connSockDead(errno syscall.Errno) bool {
+	switch errno {
+	case unix.ECONNREFUSED, unix.EHOSTUNREACH, unix.EHOSTDOWN, unix.ENETUNREACH, unix.ENETDOWN,
+		unix.EADDRNOTAVAIL, unix.EPIPE, unix.ENOTCONN, unix.EDESTADDRREQ:
+		return true
+	}
+	return false
+}
+
+// enobufsWait is how long a connected socket send sleeps when the interface queue is flow controlled. The kernel
+// posts no wakeup when the advisory lifts, so this polls; 50us kept a gigabit link full without spinning a core in
+// the 09-25 probes.
+const enobufsWait = 50 * time.Microsecond
+
+// connSockWrite writes bufs on s, one datagram per write. A full socket buffer parks the caller in the netpoller
+// until the socket is writable, and a flow-controlled interface makes it wait out ENOBUFS, so a busy link slows
+// the tun reader instead of dropping. Once the connected socket has waited cfg.enobufsMax without a send going
+// through, either one drops the datagram instead, and a dropped datagram counts as sent, as the listener's
+// unreported interface drops do; it never falls back to the listener, which would reorder it. closed reports that
+// the connected socket was closed before bufs[sent] went out; any other error stops the run and is returned with
+// how many went out.
+func (u *StdConn) connSockWrite(s *connSock, bufs [][]byte) (sent int, errno syscall.Errno, closed bool) {
+	cs := u.socks
+	defer func() {
+		if sent > 0 {
+			s.used.Store(cs.clock())
+		}
+	}()
+	for sent < len(bufs) {
+		err := s.try(bufs[sent], false)
+		if err == syscall.EAGAIN || err == os.ErrDeadlineExceeded {
+			// os.ErrDeadlineExceeded here is another writer's parked deadline, passed but not yet cleared.
+			err = u.parkedWrite(s, bufs[sent])
+		}
+		switch err {
+		case nil:
+			sent++
+			if s.stallAt.Load() != 0 {
+				s.stallAt.Store(0)
+				s.dropping.Store(false)
+			}
+			continue
+		case errStallSpent:
+			u.dropStalled(s)
+			sent++
+			continue
+		}
+		errno, ok := err.(syscall.Errno)
+		if !ok {
+			return sent, 0, true
+		}
+		if errno != syscall.ENOBUFS {
+			return sent, errno, false
+		}
+		// The wait holds no reference on the descriptor, so a close doesn't wait for it.
+		if s.stalledFor(cs.clock()) < cs.cfg.enobufsMax {
+			cs.sleep(enobufsWait)
+		} else {
+			u.dropStalled(s)
+			sent++
+		}
+	}
+	return sent, 0, false
+}
+
+// parkedWrite writes b after EAGAIN, parked in the netpoller for what is left of s's stall budget, and returns
+// errStallSpent once none is. golang/go#73919: a darwin UDP write parked on EAGAIN can wait forever for a
+// writability event that never comes, so the wait always has a deadline.
+func (u *StdConn) parkedWrite(s *connSock, b []byte) error {
+	cs := u.socks
+	s.parkMu.Lock()
+	defer s.parkMu.Unlock()
+	left := cs.cfg.enobufsMax - s.stalledFor(cs.clock())
+	if left <= 0 {
+		return errStallSpent
+	}
+	if err := s.conn.SetWriteDeadline(time.Now().Add(left)); err != nil {
+		return net.ErrClosed
+	}
+	err := s.try(b, true)
+	_ = s.conn.SetWriteDeadline(time.Time{})
+	if err == os.ErrDeadlineExceeded {
+		return errStallSpent
+	}
+	return err
+}
+
+// dropStalled logs the first datagram a stall drops.
+func (u *StdConn) dropStalled(s *connSock) {
+	if !s.dropping.Swap(true) {
+		u.l.Debug("connected socket: sends stalled, dropping datagrams until one goes through", "udpAddr", s.dst)
+	}
+}
+
+// stalledFor records an ENOBUFS or EAGAIN on s at now and returns how long its sends have stalled: since the first
+// ENOBUFS or EAGAIN after a send last went through. Every writer measures from that one time, so concurrent writers
+// spend the budget no faster than one, and a send going through ends the stall for every writer, including one
+// asleep in it.
+func (s *connSock) stalledFor(now int64) time.Duration {
+	if s.stallAt.CompareAndSwap(0, now) {
+		return 0
+	}
+	at := s.stallAt.Load()
+	if at == 0 {
+		return 0
+	}
+	return time.Duration(now - at)
+}
+
+// readConnSock delivers a connected socket's datagrams through the listener's reader, serialized with ListenOut,
+// until the connected socket is closed. Any receive error closes it: a connected socket reports its peer's
+// unreachables here, and retrying an unexpected error could spin.
+func (u *StdConn) readConnSock(s *connSock) {
+	rd := u.reader.Load()
+	buf := make([]byte, MTU)
+	for {
+		n, from, err := s.conn.ReadFromUDPAddrPort(buf)
+		if err != nil {
+			if errors.Is(err, net.ErrClosed) {
+				return
+			}
+			why := err.Error()
+			var errno syscall.Errno
+			if errors.As(err, &errno) && connSockDead(errno) {
+				why = errno.Error()
+			} else {
+				u.l.Warn("connected socket: unexpected receive error", "udpAddr", s.dst, "error", err)
+			}
+			u.closeConnSock(s, why, true)
+			return
+		}
+		s.used.Store(u.socks.clock())
+		u.readMu.Lock()
+		rd.r(netip.AddrPortFrom(from.Addr().Unmap(), from.Port()), buf[:n:n])
+		rd.flush()
+		u.readMu.Unlock()
+	}
+}

udp/connsock_darwin_test.go

@@ -0,0 +1,1020 @@
+//go:build !e2e_testing
+
+package udp
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"net"
+	"net/netip"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"syscall"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sys/unix"
+)
+
+// newConnSockListener opens a listener with the connected socket knobs pinned to their defaults, so NEBULA_*
+// variables in the environment can't change what a test sees, applies cfg, and starts ListenOut with r.
+func newConnSockListener(t *testing.T, s Settings, h slog.Handler, r EncReader, cfg ...func(*connSockConfig)) *StdConn {
+	t.Helper()
+	for k, v := range map[string]string{
+		"NEBULA_CONNSOCKS":           "8",
+		"NEBULA_CONNSOCK_RUN":        "64",
+		"NEBULA_CONNSOCK_WINDOW":     "1s",
+		"NEBULA_CONNSOCK_IDLE":       "30s",
+		"NEBULA_CONNSOCK_OPEN_EVERY": "1s",
+	} {
+		t.Setenv(k, v)
+	}
+	if h == nil {
+		h = slog.DiscardHandler
+	}
+	if r == nil {
+		r = func(netip.AddrPort, []byte) {}
+	}
+	c, err := NewListener(slog.New(h), s)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = c.Close() })
+	u := c.(*StdConn)
+	for _, f := range cfg {
+		f(&u.socks.cfg)
+	}
+	go func() { _ = u.ListenOut(r, func() {}) }()
+	require.Eventually(t, func() bool { return u.reader.Load() != nil }, 5*time.Second, time.Millisecond)
+	return u
+}
+
+func listen(s string) Settings { return Settings{Listen: netip.MustParseAddrPort(s)} }
+
+// run builds n datagrams to dst.
+func run(dst netip.AddrPort, n int, tag string) ([][]byte, []netip.AddrPort) {
+	bufs := make([][]byte, n)
+	addrs := make([]netip.AddrPort, n)
+	for i := range bufs {
+		bufs[i] = fmt.Appendf(nil, "%s-%d", tag, i)
+		addrs[i] = dst
+	}
+	return bufs, addrs
+}
+
+func newPeer(t *testing.T, network, addr string) (*net.UDPConn, netip.AddrPort) {
+	t.Helper()
+	pc, err := net.ListenPacket(network, addr)
+	require.NoError(t, err)
+	t.Cleanup(func() { _ = pc.Close() })
+	uc := pc.(*net.UDPConn)
+	return uc, uc.LocalAddr().(*net.UDPAddr).AddrPort()
+}
+
+// newEstablishedPeer opens a peer as newPeer does and tells u it is an established tunnel's remote.
+func newEstablishedPeer(t *testing.T, u *StdConn, network, addr string) (*net.UDPConn, netip.AddrPort) {
+	t.Helper()
+	c, dst := newPeer(t, network, addr)
+	u.SetEstablishedPeer(dst, true)
+	return c, dst
+}
+
+// readFrom reads n datagrams at peer and returns the source of each.
+func readFrom(t *testing.T, peer *net.UDPConn, n int) []netip.AddrPort {
+	t.Helper()
+	buf := make([]byte, MTU)
+	var from []netip.AddrPort
+	for range n {
+		require.NoError(t, peer.SetReadDeadline(time.Now().Add(5*time.Second)))
+		_, f, err := peer.ReadFromUDPAddrPort(buf)
+		require.NoError(t, err)
+		from = append(from, netip.AddrPortFrom(f.Addr().Unmap(), f.Port()))
+	}
+	return from
+}
+
+func reusePort(t *testing.T, c syscall.Conn) int {
+	t.Helper()
+	rc, err := c.SyscallConn()
+	require.NoError(t, err)
+	var v int
+	var serr error
+	require.NoError(t, rc.Control(func(fd uintptr) {
+		v, serr = unix.GetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_REUSEPORT)
+	}))
+	require.NoError(t, serr)
+	return v
+}
+
+// rogueBinds tries each of binds, "network address plain|reuseport", on port and returns the ones that bound.
+func rogueBinds(t *testing.T, port uint16, binds []string) []string {
+	t.Helper()
+	var bound []string
+	for _, b := range binds {
+		f := strings.Fields(b)
+		network, addr, reuse := f[0], f[1], f[2]
+		lc := net.ListenConfig{Control: connSockControl(reuse == "reuseport", 0)}
+		pc, err := lc.ListenPacket(context.Background(), network, net.JoinHostPort(addr, fmt.Sprint(port)))
+		if err == nil {
+			_ = pc.Close()
+			bound = append(bound, b)
+		}
+	}
+	return bound
+}
+
+// The listener never sets SO_REUSEPORT, and a connected socket binds with it only to a specific address, so every
+// bind a lone listener refuses is refused after NewListener returns and while connected sockets are open. Same-uid
+// SO_REUSEPORT binds to a specific address are left out: xnu allows them beside a lone wildcard listener too,
+// which is how a connected socket binds, and refuses them to any other uid.
+func TestConnSocksNoSharedBind(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	la, err := u.LocalAddr()
+	require.NoError(t, err)
+	binds := []string{
+		"udp4 0.0.0.0 plain", "udp4 0.0.0.0 reuseport", "udp4 127.0.0.1 plain",
+		"udp6 :: plain", "udp6 :: reuseport", "udp6 ::1 plain",
+	}
+	assert.Zero(t, reusePort(t, u.UDPConn), "listener")
+	assert.Empty(t, rogueBinds(t, la.Port(), binds), "bound beside the listener")
+
+	peer4, dst4 := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
+	peer6, dst6 := newEstablishedPeer(t, u, "udp6", "[::1]:0")
+	for _, p := range []struct {
+		c   *net.UDPConn
+		dst netip.AddrPort
+	}{{peer4, dst4}, {peer6, dst6}} {
+		bufs, addrs := run(p.dst, u.socks.cfg.run, "p")
+		_, err := u.WriteBatch(bufs, addrs)
+		require.NoError(t, err)
+		readFrom(t, p.c, len(bufs))
+		s := u.connSockFor(p.dst)
+		require.NotNil(t, s, "no connected socket to %v", p.dst)
+		assert.NotZero(t, reusePort(t, s.conn), "connected socket to %v", p.dst)
+	}
+	assert.Empty(t, rogueBinds(t, la.Port(), binds), "bound beside the connected sockets")
+}
+
+// A connected socket binds the source address the listener's route to its peer picks, for either family on a
+// dual-stack listener, so the peer sees one address and port whichever socket sent.
+func TestConnSocksSourceMatchesListener(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	la, err := u.LocalAddr()
+	require.NoError(t, err)
+	for _, p := range []struct{ network, addr string }{{"udp4", "127.0.0.1:0"}, {"udp6", "[::1]:0"}} {
+		peer, dst := newEstablishedPeer(t, u, p.network, p.addr)
+		require.NoError(t, u.WriteTo([]byte("probe"), dst))
+		require.Nil(t, u.connSockFor(dst))
+		viaListener := readFrom(t, peer, 1)[0]
+		assert.Equal(t, la.Port(), viaListener.Port())
+
+		bufs, addrs := run(dst, u.socks.cfg.run, p.network)
+		_, err := u.WriteBatch(bufs, addrs)
+		require.NoError(t, err)
+		s := u.connSockFor(dst)
+		require.NotNil(t, s, "no connected socket to %v", dst)
+		for _, from := range readFrom(t, peer, len(bufs)) {
+			assert.Equal(t, viaListener, from, "source seen by %v", dst)
+		}
+		local := s.conn.LocalAddr().(*net.UDPAddr).AddrPort()
+		assert.Equal(t, viaListener, netip.AddrPortFrom(local.Addr().Unmap(), local.Port()))
+	}
+}
+
+// A run long enough opens a connected socket per peer, its replies reach the listener's reader, and a peer that
+// goes away sends its traffic back through the listener with its connected socket closed, while the next peer gets
+// a connected socket of its own.
+func TestConnSocks(t *testing.T) {
+	var mu sync.Mutex
+	got := map[netip.AddrPort]int{}
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, func(addr netip.AddrPort, _ []byte) {
+		mu.Lock()
+		got[addr]++
+		mu.Unlock()
+	})
+	la, err := u.LocalAddr()
+	require.NoError(t, err)
+	to := netip.AddrPortFrom(netip.MustParseAddr("127.0.0.1"), la.Port())
+
+	var peers [3]*net.UDPConn
+	var dsts [3]netip.AddrPort
+	for i := range peers {
+		peers[i], dsts[i] = newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
+	}
+	for i, dst := range dsts[:2] {
+		bufs, addrs := run(dst, u.socks.cfg.run, fmt.Sprint("p", i))
+		n, err := u.WriteBatch(bufs, addrs)
+		require.NoError(t, err)
+		assert.Equal(t, len(bufs), n)
+		require.NotNil(t, u.connSockFor(dst), "no connected socket to peer %d", i)
+		for _, from := range readFrom(t, peers[i], len(bufs)) {
+			assert.Equal(t, to, from)
+		}
+		for range 3 {
+			_, err := peers[i].WriteToUDPAddrPort([]byte("reply"), to)
+			require.NoError(t, err)
+		}
+	}
+	require.Eventually(t, func() bool {
+		mu.Lock()
+		defer mu.Unlock()
+		return got[dsts[0]] == 3 && got[dsts[1]] == 3
+	}, 5*time.Second, time.Millisecond)
+
+	// The peer's ICMP port unreachable surfaces on the connected socket's next call. The deadline is generous
+	// because xnu rate-limits and delivers the unreachable on its own schedule.
+	gone := dsts[1]
+	goneSock := u.connSockFor(gone)
+	require.NoError(t, peers[1].Close())
+	bufs, addrs := run(gone, u.socks.cfg.run, "gone")
+	require.Eventually(t, func() bool {
+		_, _ = u.WriteBatch(bufs, addrs)
+		return u.connSockFor(gone) == nil
+	}, 10*time.Second, 10*time.Millisecond)
+	assert.NotNil(t, u.connSockFor(dsts[0]))
+	u.socks.closers.Wait()
+	_, err = goneSock.conn.Write([]byte("x"))
+	assert.ErrorIs(t, err, net.ErrClosed, "closed connected socket left open")
+
+	bufs, addrs = run(dsts[2], u.socks.cfg.run, "p2")
+	_, err = u.WriteBatch(bufs, addrs)
+	require.NoError(t, err)
+	require.NotNil(t, u.connSockFor(dsts[2]))
+	for _, from := range readFrom(t, peers[2], len(bufs)) {
+		assert.Equal(t, to, from)
+	}
+
+	require.NoError(t, u.Rebind())
+	for _, dst := range dsts {
+		assert.Nil(t, u.connSockFor(dst), "connected socket left after Rebind")
+	}
+}
+
+// Listen routines on one port share one set of connected sockets: the cap and the open bucket are theirs together,
+// a peer's connected socket carries every routine's sends to it, and Rebind on the first routine, which is all
+// Control.RebindUDPServer calls, clears every routine's listener and closes the connected sockets.
+func TestConnSocksSharedAcrossRoutines(t *testing.T) {
+	s := listen("0.0.0.0:0")
+	s.Multi = true
+	u0 := newConnSockListener(t, s, nil, nil, func(c *connSockConfig) { c.max = 1 })
+	la, err := u0.LocalAddr()
+	require.NoError(t, err)
+	s.Listen = netip.AddrPortFrom(s.Listen.Addr(), la.Port())
+	u1 := newConnSockListener(t, s, nil, nil)
+	require.Same(t, u0.socks, u1.socks)
+	lo, err := net.InterfaceByName("lo0")
+	require.NoError(t, err)
+
+	peerA, a := newPeer(t, "udp4", "127.0.0.1:0")
+	peerB, b := newPeer(t, "udp4", "127.0.0.1:0")
+	for _, dst := range []netip.AddrPort{a, b} {
+		// Nebula tells every writer.
+		u0.SetEstablishedPeer(dst, true)
+		u1.SetEstablishedPeer(dst, true)
+	}
+
+	bufs, addrs := run(a, u1.socks.cfg.run, "a")
+	_, err = u1.WriteBatch(bufs, addrs)
+	require.NoError(t, err)
+	readFrom(t, peerA, len(bufs))
+	s1 := u1.connSockFor(a)
+	require.NotNil(t, s1)
+	var sends atomic.Int64
+	s1.sys = func(fd int, b []byte) error {
+		sends.Add(1)
+		return writeFd(fd, b)
+	}
+	require.NoError(t, u0.WriteTo([]byte("via u0"), a))
+	readFrom(t, peerA, 1)
+	assert.EqualValues(t, 1, sends.Load(), "routine 0 didn't send on routine 1's connected socket")
+
+	bufs, addrs = run(b, 2*u0.socks.cfg.run, "b")
+	_, err = u0.WriteBatch(bufs, addrs)
+	require.NoError(t, err)
+	readFrom(t, peerB, len(bufs))
+	assert.Nil(t, u0.connSockFor(b), "a second connected socket past a cap of 1")
+
+	level, opt := unix.IPPROTO_IPV6, unix.IPV6_BOUND_IF
+	if u0.isV4 {
+		level, opt = unix.IPPROTO_IP, unix.IP_BOUND_IF
+	}
+	boundIf := func(u *StdConn) int {
+		v, err := unix.GetsockoptInt(int(u.sysFd), level, opt)
+		require.NoError(t, err)
+		return v
+	}
+	for _, u := range []*StdConn{u0, u1} {
+		require.NoError(t, unix.SetsockoptInt(int(u.sysFd), level, opt, lo.Index))
+		require.Equal(t, lo.Index, boundIf(u))
+	}
+	require.NoError(t, u0.Rebind())
+	assert.Zero(t, boundIf(u0))
+	assert.Zero(t, boundIf(u1), "Rebind left routine 1 scoped")
+	assert.Nil(t, u1.connSockFor(a), "Rebind left routine 1's connected socket open")
+	u0.socks.closers.Wait()
+	_, err = s1.conn.Write([]byte("x"))
+	assert.ErrorIs(t, err, net.ErrClosed)
+}
+
+// A connected socket on a specific listen.host would need SO_REUSEPORT on the listener, so connected sockets stay
+// off there unless listen routines already set it.
+func TestConnSocksSpecificHost(t *testing.T) {
+	for _, multi := range []bool{false, true} {
+		t.Run(fmt.Sprint("multi=", multi), func(t *testing.T) {
+			s := listen("127.0.0.1:0")
+			s.Multi = multi
+			u := newConnSockListener(t, s, nil, nil)
+			la, err := u.LocalAddr()
+			require.NoError(t, err)
+			if !multi {
+				assert.Zero(t, reusePort(t, u.UDPConn), "listener")
+			}
+			peer, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
+			bufs, addrs := run(dst, u.socks.cfg.run, "p")
+			_, err = u.WriteBatch(bufs, addrs)
+			require.NoError(t, err)
+			for _, from := range readFrom(t, peer, len(bufs)) {
+				assert.Equal(t, la, from)
+			}
+			if multi {
+				assert.NotNil(t, u.connSockFor(dst))
+			} else {
+				assert.Nil(t, u.connSockFor(dst))
+			}
+		})
+	}
+}
+
+// A v4 listener can't send to a v6 destination; however often it's asked, it says so and opens no connected
+// socket.
+func TestConnSocksV6DestinationOnV4Listener(t *testing.T) {
+	s := listen("127.0.0.1:0")
+	s.Multi = true
+	u := newConnSockListener(t, s, nil, nil)
+	require.True(t, u.isV4)
+	dst := netip.MustParseAddrPort("[::1]:9")
+	u.SetEstablishedPeer(dst, true)
+	bufs, _ := run(dst, 2*u.socks.cfg.run, "v6")
+	for _, b := range bufs {
+		assert.ErrorIs(t, u.WriteTo(b, dst), ErrInvalidIPv6RemoteForSocket)
+	}
+	assert.Nil(t, u.connSockFor(dst))
+}
+
+// darwin's tun hands over one packet per read, so a busy peer's traffic reaches the socket as single WriteTo
+// calls. cfg.run of them inside one window earn a connected socket; the same count spread over more than a window
+// doesn't.
+func TestConnSocksOpenFromWriteTo(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	var now atomic.Int64
+	now.Store(int64(time.Hour))
+	u.socks.clock = now.Load
+	_, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
+	pkt := []byte("pkt")
+
+	for range u.socks.cfg.run - 1 {
+		require.NoError(t, u.WriteTo(pkt, dst))
+	}
+	now.Add(int64(u.socks.cfg.window) + 1)
+	require.NoError(t, u.WriteTo(pkt, dst))
+	assert.Nil(t, u.connSockFor(dst), "a connected socket for sends spread over two windows")
+
+	for range u.socks.cfg.run - 2 {
+		require.NoError(t, u.WriteTo(pkt, dst))
+	}
+	assert.Nil(t, u.connSockFor(dst))
+	require.NoError(t, u.WriteTo(pkt, dst))
+	assert.NotNil(t, u.connSockFor(dst), "no connected socket after %d sends in one window", u.socks.cfg.run)
+}
+
+// Only the remote of an established tunnel gets a connected socket, however busy another destination is, and one
+// that stops being established loses its connected socket and has to be busy again to get a new one.
+func TestConnSocksOnlyForEstablishedPeers(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	peer, dst := newPeer(t, "udp4", "127.0.0.1:0")
+	run1 := u.socks.cfg.run
+	send := func(n int) {
+		t.Helper()
+		bufs, addrs := run(dst, n, "p")
+		_, err := u.WriteBatch(bufs, addrs)
+		require.NoError(t, err)
+		readFrom(t, peer, n)
+	}
+
+	send(2 * run1)
+	assert.Nil(t, u.connSockFor(dst), "a connected socket for a busy destination with no tunnel")
+
+	u.SetEstablishedPeer(dst, true)
+	send(run1)
+	s := u.connSockFor(dst)
+	require.NotNil(t, s, "no connected socket for a busy established peer")
+
+	u.SetEstablishedPeer(dst, false)
+	assert.Nil(t, u.connSockFor(dst))
+	u.socks.closers.Wait()
+	_, err := s.conn.Write([]byte("x"))
+	assert.ErrorIs(t, err, net.ErrClosed, "connected socket left open after its tunnel went away")
+	send(2 * run1)
+	assert.Nil(t, u.connSockFor(dst), "a connected socket after the tunnel went away")
+
+	u.SetEstablishedPeer(dst, true)
+	send(run1 - 1)
+	assert.Nil(t, u.connSockFor(dst), "a re-established peer kept its old count")
+	send(1)
+	assert.NotNil(t, u.connSockFor(dst))
+}
+
+// A tunnel going away while its busy remote is opening a connected socket never leaves one open, whether or not
+// the remote is established again before the open finishes. The dial hook holds the open between reserving its
+// destination and dialing until the tunnel has changed, so every round races.
+func TestConnSocksEstablishedRace(t *testing.T) {
+	h := &countingHandler{}
+	u := newConnSockListener(t, listen("0.0.0.0:0"), h, nil, func(c *connSockConfig) {
+		c.run = 1
+		c.openEvery = time.Nanosecond
+	})
+	cs := u.socks
+	peer, dst := newPeer(t, "udp4", "127.0.0.1:0")
+	for i, reestablish := range []bool{false, true} {
+		u.SetEstablishedPeer(dst, true)
+		dialing, release := make(chan struct{}), make(chan struct{})
+		cs.dialHook = func(netip.AddrPort) {
+			close(dialing)
+			<-release
+		}
+		done := make(chan error)
+		go func() { done <- u.WriteTo([]byte("p"), dst) }()
+		<-dialing
+		u.SetEstablishedPeer(dst, false)
+		if reestablish {
+			u.SetEstablishedPeer(dst, true)
+		}
+		close(release)
+		require.NoError(t, <-done)
+		cs.dialHook = nil
+		readFrom(t, peer, 1)
+
+		assert.Nil(t, u.connSockFor(dst), "reestablish=%v", reestablish)
+		assert.EqualValues(t, i+1, h.discarded.Load(), "reestablish=%v", reestablish)
+		cs.mu.Lock()
+		assert.Empty(t, cs.pending)
+		cs.mu.Unlock()
+		assert.Zero(t, cs.reserved.Load())
+		u.SetEstablishedPeer(dst, false)
+	}
+	assert.Zero(t, h.opened.Load())
+}
+
+// Neither an open in flight nor a close holds anything SetEstablishedPeer waits on, since Nebula calls it under
+// the hostmap's write lock; and a peer whose connected socket is still closing can't open another until the close
+// completes. The hooks hold the open in its dials and the close before its Close.
+func TestConnSocksSetEstablishedPeerMakesNoSyscalls(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil, func(c *connSockConfig) { c.openEvery = time.Nanosecond })
+	cs := u.socks
+	peer, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
+	_, other := newPeer(t, "udp4", "127.0.0.1:0")
+
+	dialing, releaseDial := make(chan struct{}), make(chan struct{})
+	cs.dialHook = func(netip.AddrPort) {
+		close(dialing)
+		<-releaseDial
+	}
+	opened := make(chan *connSock)
+	go func() {
+		bufs, addrs := run(dst, cs.cfg.run, "p")
+		_, _ = u.WriteBatch(bufs, addrs)
+		opened <- u.connSockFor(dst)
+	}()
+	<-dialing
+	u.SetEstablishedPeer(other, true)
+	u.SetEstablishedPeer(other, false)
+	close(releaseDial)
+	s := <-opened
+	require.NotNil(t, s)
+	readFrom(t, peer, cs.cfg.run)
+	cs.dialHook = nil
+
+	closing, releaseClose := make(chan struct{}), make(chan struct{})
+	cs.closeHook = func(netip.AddrPort) {
+		close(closing)
+		<-releaseClose
+	}
+	u.SetEstablishedPeer(dst, false)
+	<-closing
+	assert.Nil(t, u.connSockFor(dst))
+	u.SetEstablishedPeer(dst, true)
+	bufs, addrs := run(dst, 2*cs.cfg.run, "q")
+	_, err := u.WriteBatch(bufs, addrs)
+	require.NoError(t, err)
+	readFrom(t, peer, len(bufs))
+	assert.Nil(t, u.connSockFor(dst), "a connected socket opened while the last one to its peer was closing")
+
+	close(releaseClose)
+	cs.closers.Wait()
+	cs.closeHook = nil
+	_, err = s.conn.Write([]byte("x"))
+	assert.ErrorIs(t, err, net.ErrClosed)
+	bufs, addrs = run(dst, cs.cfg.run, "r")
+	_, err = u.WriteBatch(bufs, addrs)
+	require.NoError(t, err)
+	readFrom(t, peer, len(bufs))
+	assert.NotNil(t, u.connSockFor(dst), "no connected socket once the last one closed")
+}
+
+// A listener whose interface scope can't be read gives a busy peer no connected socket, rather than one without
+// the scope; the peer stays on the listener for the backoff. A pipe stands in for the listener's descriptor, so
+// getsockopt fails with ENOTSOCK.
+func TestConnSocksBoundIfFailsClosed(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	_, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
+	var pipe [2]int
+	require.NoError(t, unix.Pipe(pipe[:]))
+	t.Cleanup(func() {
+		_ = unix.Close(pipe[0])
+		_ = unix.Close(pipe[1])
+	})
+	listener := u.sysFd
+	u.sysFd = uintptr(pipe[0])
+	s := u.maybeOpenConnSock(u.socks.peer(dst), dst, u.socks.cfg.run)
+	u.sysFd = listener
+	assert.Nil(t, s)
+	assert.Nil(t, u.connSockFor(dst))
+	assert.Greater(t, u.socks.peer(dst).retryAt, u.socks.clock(), "no backoff after a failed open")
+}
+
+// readTags reads n datagrams at peer and returns their payloads.
+func readTags(t *testing.T, peer *net.UDPConn, n int) []string {
+	t.Helper()
+	buf := make([]byte, MTU)
+	var tags []string
+	for range n {
+		require.NoError(t, peer.SetReadDeadline(time.Now().Add(5*time.Second)))
+		m, _, err := peer.ReadFromUDPAddrPort(buf)
+		require.NoError(t, err)
+		tags = append(tags, string(buf[:m]))
+	}
+	return tags
+}
+
+// One batch can hold runs to a peer with a connected socket, to an established peer without one, and to a
+// destination that is no tunnel's remote, interleaved; each run goes out its own way, in order, and all count.
+func TestConnSocksWriteBatchMixedDestinations(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	s, peerA := openConnSock(t, u)
+	peerB, b := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
+	peerC, c := newPeer(t, "udp4", "127.0.0.1:0")
+	var onSock atomic.Int64
+	s.sys = func(fd int, b []byte) error {
+		onSock.Add(1)
+		return writeFd(fd, b)
+	}
+
+	var bufs [][]byte
+	var addrs []netip.AddrPort
+	for _, r := range []struct {
+		dst netip.AddrPort
+		tag string
+		n   int
+	}{{s.dst, "a", 3}, {b, "b", 2}, {c, "c", 2}, {s.dst, "A", 2}, {c, "C", 1}} {
+		for i := range r.n {
+			bufs = append(bufs, fmt.Appendf(nil, "%s%d", r.tag, i))
+			addrs = append(addrs, r.dst)
+		}
+	}
+	n, err := u.WriteBatch(bufs, addrs)
+	require.NoError(t, err)
+	assert.Equal(t, len(bufs), n)
+	assert.EqualValues(t, 5, onSock.Load(), "datagrams on a's connected socket")
+	assert.Equal(t, []string{"a0", "a1", "a2", "A0", "A1"}, readTags(t, peerA, 5))
+	assert.Equal(t, []string{"b0", "b1"}, readTags(t, peerB, 2))
+	assert.Equal(t, []string{"c0", "c1", "C0"}, readTags(t, peerC, 3))
+	assert.Nil(t, u.connSockFor(b))
+}
+
+// A connected socket error that doesn't mean the path is gone costs only its datagram: WriteBatch skips it and
+// sends the rest of the run on the connected socket, WriteTo returns it, and the connected socket stays open.
+func TestConnSocksWriteError(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	s, peer := openConnSock(t, u)
+	s.sys = func(fd int, b []byte) error {
+		if string(b) == "bad" {
+			return syscall.EMSGSIZE
+		}
+		return writeFd(fd, b)
+	}
+	bufs := [][]byte{[]byte("ok0"), []byte("bad"), []byte("ok2")}
+	n, err := u.WriteBatch(bufs, []netip.AddrPort{s.dst, s.dst, s.dst})
+	require.NoError(t, err)
+	assert.Equal(t, 2, n)
+	assert.Equal(t, []string{"ok0", "ok2"}, readTags(t, peer, 2))
+	assert.ErrorIs(t, u.WriteTo([]byte("bad"), s.dst), syscall.EMSGSIZE)
+	assert.Same(t, s, u.connSockFor(s.dst))
+}
+
+// A connected socket that turns out closed or dead partway through a run leaves the rest of the run to the
+// listener, in order and all counted; a dead one is closed and its peer backs off to the listener.
+func TestConnSocksWriteBatchFallsBackMidRun(t *testing.T) {
+	for _, fail := range []error{net.ErrClosed, syscall.ECONNREFUSED} {
+		t.Run(fail.Error(), func(t *testing.T) {
+			u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+			s, peer := openConnSock(t, u)
+			var calls atomic.Int64
+			s.sys = func(fd int, b []byte) error {
+				if calls.Add(1) > 2 {
+					return fail
+				}
+				return writeFd(fd, b)
+			}
+			bufs, addrs := run(s.dst, 5, "r")
+			n, err := u.WriteBatch(bufs, addrs)
+			require.NoError(t, err)
+			assert.Equal(t, 5, n)
+			assert.Equal(t, []string{"r-0", "r-1", "r-2", "r-3", "r-4"}, readTags(t, peer, 5))
+			assert.EqualValues(t, 3, calls.Load(), "the connected socket was tried again after it failed")
+			if fail == syscall.ECONNREFUSED {
+				assert.Nil(t, u.connSockFor(s.dst))
+				u.socks.mu.Lock()
+				assert.Greater(t, u.socks.peer(s.dst).retryAt, u.socks.clock(), "no backoff after a dead connected socket")
+				u.socks.mu.Unlock()
+			}
+		})
+	}
+}
+
+// openConnSock earns a connected socket to a fresh loopback peer and returns it and the peer.
+func openConnSock(t *testing.T, u *StdConn) (*connSock, *net.UDPConn) {
+	t.Helper()
+	peer, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
+	bufs, addrs := run(dst, u.socks.cfg.run, "p")
+	_, err := u.WriteBatch(bufs, addrs)
+	require.NoError(t, err)
+	readFrom(t, peer, len(bufs))
+	s := u.connSockFor(dst)
+	require.NotNil(t, s)
+	return s, peer
+}
+
+// A flow-controlled interface makes a connected socket's sends wait, up to cfg.enobufsMax without one going
+// through; after that each datagram is dropped without waiting until a send succeeds again, and each stall logs
+// its drops once. s.sys stands in for an interface that answers every write with ENOBUFS until enobufs is
+// cleared, and each sleep advances the clock.
+func TestConnSockWriteENOBUFS(t *testing.T) {
+	h := &countingHandler{}
+	u := newConnSockListener(t, listen("0.0.0.0:0"), h, nil)
+	s, _ := openConnSock(t, u)
+	var now atomic.Int64
+	now.Store(int64(time.Hour))
+	sleeps := 0
+	u.socks.clock = now.Load
+	u.socks.sleep = func(d time.Duration) {
+		sleeps++
+		now.Add(int64(d))
+	}
+	var enobufs atomic.Bool
+	s.sys = func(fd int, b []byte) error {
+		if enobufs.Load() {
+			return syscall.ENOBUFS
+		}
+		return writeFd(fd, b)
+	}
+	perStall := int(u.socks.cfg.enobufsMax / enobufsWait)
+	bufs, _ := run(s.dst, 3, "q")
+
+	for stall := 1; stall <= 2; stall++ {
+		enobufs.Store(true)
+		sent, errno, closed := u.connSockWrite(s, bufs)
+		assert.Equal(t, 3, sent)
+		assert.Zero(t, errno)
+		assert.False(t, closed)
+		assert.Equal(t, stall*perStall, sleeps)
+
+		sent, _, _ = u.connSockWrite(s, bufs)
+		assert.Equal(t, 3, sent)
+		assert.Equal(t, stall*perStall, sleeps, "a stalled connected socket waited again")
+		assert.EqualValues(t, stall, h.dropping.Load(), "drop log lines after stall %d", stall)
+
+		enobufs.Store(false)
+		sent, _, _ = u.connSockWrite(s, bufs[:1])
+		assert.Equal(t, 1, sent)
+		assert.Zero(t, s.stallAt.Load())
+	}
+}
+
+// Concurrent writers share one ENOBUFS budget measured from the first ENOBUFS since a send went through: two
+// writers stalled together wait as long as one would, and a send going through while a writer sleeps gives that
+// writer a fresh budget. The sleep hook parks each writer until the test has moved the clock.
+func TestConnSockWriteENOBUFSConcurrentWriters(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	s, _ := openConnSock(t, u)
+	var now atomic.Int64
+	u.socks.clock = now.Load
+	asleep, wake := make(chan struct{}), make(chan struct{})
+	u.socks.sleep = func(time.Duration) {
+		asleep <- struct{}{}
+		<-wake
+	}
+	// Datagrams starting with 'e' get ENOBUFS while enobufs is set; the rest go out.
+	var enobufs atomic.Bool
+	var wrote atomic.Int64
+	s.sys = func(fd int, b []byte) error {
+		if b[0] == 'e' && enobufs.Load() {
+			return syscall.ENOBUFS
+		}
+		wrote.Add(1)
+		return writeFd(fd, b)
+	}
+	write := func(tag string, done chan<- int) {
+		sent, _, _ := u.connSockWrite(s, [][]byte{[]byte(tag)})
+		done <- sent
+	}
+	budget := int64(u.socks.cfg.enobufsMax)
+
+	t.Run("a send going through ends the stall", func(t *testing.T) {
+		enobufs.Store(true)
+		now.Store(int64(time.Hour))
+		done := make(chan int)
+		go write("e1", done)
+		<-asleep
+		now.Add(budget / 2)
+		sent, _, _ := u.connSockWrite(s, [][]byte{[]byte("ok")})
+		require.Equal(t, 1, sent)
+		now.Add(budget)
+		wake <- struct{}{}
+		<-asleep // e1 hit ENOBUFS again 1.5x the budget after its first, and waited instead of dropping.
+		enobufs.Store(false)
+		before := wrote.Load()
+		wake <- struct{}{}
+		assert.Equal(t, 1, <-done)
+		assert.Equal(t, before+1, wrote.Load(), "e1 was dropped")
+	})
+
+	t.Run("two stalled writers spend one budget", func(t *testing.T) {
+		enobufs.Store(true)
+		now.Store(int64(2 * time.Hour))
+		done := make(chan int)
+		go write("e2", done)
+		go write("e3", done)
+		<-asleep
+		<-asleep
+		now.Add(budget * 6 / 10)
+		wake <- struct{}{}
+		wake <- struct{}{}
+		<-asleep // Both are 0.6x the budget into one stall, not 1.2x, so both wait again.
+		<-asleep
+		now.Add(budget / 2)
+		before := wrote.Load()
+		wake <- struct{}{}
+		wake <- struct{}{}
+		assert.Equal(t, 1, <-done)
+		assert.Equal(t, 1, <-done)
+		assert.Equal(t, before, wrote.Load(), "a datagram went out while ENOBUFS lasted")
+	})
+}
+
+// A full send buffer that never drains parks a connected socket send only for cfg.enobufsMax, the stall budget
+// ENOBUFS gets: golang/go#73919 has darwin UDP writes parked on EAGAIN waiting forever. Past it the datagrams are
+// dropped and counted as sent, concurrent writers spend that one budget, and once a send goes through the
+// connected socket works as before. s.sys stands in for a send buffer that stays full while eagain is set.
+func TestConnSockWriteEAGAINBounded(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	s, peer := openConnSock(t, u)
+	var eagain atomic.Bool
+	eagain.Store(true)
+	s.sys = func(fd int, b []byte) error {
+		if eagain.Load() {
+			return syscall.EAGAIN
+		}
+		return writeFd(fd, b)
+	}
+	budget := u.socks.cfg.enobufsMax
+
+	start := time.Now()
+	sent, errno, closed := u.connSockWrite(s, [][]byte{[]byte("x")})
+	elapsed := time.Since(start)
+	assert.Equal(t, 1, sent)
+	assert.Zero(t, errno)
+	assert.False(t, closed)
+	assert.GreaterOrEqual(t, elapsed, budget, "the send didn't wait for the buffer to drain")
+	assert.Less(t, elapsed, time.Second, "the send waited past its deadline")
+
+	// The budget is spent, so these drop without waiting, however many writers there are.
+	var wg sync.WaitGroup
+	start = time.Now()
+	for range 4 {
+		wg.Go(func() {
+			bufs, _ := run(s.dst, 8, "y")
+			sent, _, _ := u.connSockWrite(s, bufs)
+			assert.Equal(t, 8, sent)
+		})
+	}
+	wg.Wait()
+	assert.Less(t, time.Since(start), time.Second)
+
+	// A send going through starts a fresh budget, which two writers stalled together spend once; the upper bound
+	// is far looser than that, so a descheduled runner doesn't fail it.
+	eagain.Store(false)
+	sent, _, _ = u.connSockWrite(s, [][]byte{[]byte("ok")})
+	require.Equal(t, 1, sent)
+	readTags(t, peer, 1)
+	eagain.Store(true)
+	start = time.Now()
+	for range 2 {
+		wg.Go(func() {
+			sent, _, _ := u.connSockWrite(s, [][]byte{[]byte("w")})
+			assert.Equal(t, 1, sent)
+		})
+	}
+	wg.Wait()
+	elapsed = time.Since(start)
+	assert.GreaterOrEqual(t, elapsed, budget)
+	assert.Less(t, elapsed, time.Second)
+
+	eagain.Store(false)
+	bufs, addrs := run(s.dst, 3, "z")
+	n, err := u.WriteBatch(bufs, addrs)
+	require.NoError(t, err)
+	assert.Equal(t, 3, n)
+	assert.Equal(t, []string{"z-0", "z-1", "z-2"}, readTags(t, peer, 3))
+	assert.Zero(t, s.stallAt.Load())
+	assert.Same(t, s, u.connSockFor(s.dst))
+}
+
+// A send waiting out ENOBUFS holds nothing a close or Close needs, and returns closed once it wakes to a closed
+// connected socket. The sleep hook parks the writer until both closes have finished; after the close, s.sys
+// writes for real, so the closed descriptor answers.
+func TestConnSockWriteDoesNotBlockClose(t *testing.T) {
+	u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil)
+	s, _ := openConnSock(t, u)
+	asleep, wake := make(chan struct{}), make(chan struct{})
+	u.socks.sleep = func(time.Duration) {
+		asleep <- struct{}{}
+		<-wake
+	}
+	var gone atomic.Bool
+	s.sys = func(fd int, b []byte) error {
+		if gone.Load() {
+			return writeFd(fd, b)
+		}
+		return syscall.ENOBUFS
+	}
+	done := make(chan bool)
+	go func() {
+		bufs, _ := run(s.dst, 1, "q")
+		_, _, closed := u.connSockWrite(s, bufs)
+		done <- closed
+	}()
+	<-asleep
+
+	u.closeConnSock(s, "test", false)
+	u.socks.closers.Wait()
+	require.NoError(t, u.Close())
+	gone.Store(true)
+	close(wake)
+	assert.True(t, <-done, "the writer didn't see its connected socket close")
+}
+
+// countingHandler counts connected socket open, close, discard and ENOBUFS drop log lines.
+type countingHandler struct{ opened, closed, discarded, dropping atomic.Int64 }
+
+func (h *countingHandler) Enabled(context.Context, slog.Level) bool { return true }
+func (h *countingHandler) Handle(_ context.Context, r slog.Record) error {
+	switch r.Message {
+	case "connected socket: opened":
+		h.opened.Add(1)
+	case "connected socket: closed":
+		h.closed.Add(1)
+	case "connected socket: discarded, its tunnel or route changed while it opened":
+		h.discarded.Add(1)
+	case "connected socket: sends stalled, dropping datagrams until one goes through":
+		h.dropping.Add(1)
+	}
+	return nil
+}
+func (h *countingHandler) WithAttrs([]slog.Attr) slog.Handler { return h }
+func (h *countingHandler) WithGroup(string) slog.Handler      { return h }
+
+// Anyone can make Nebula reply to them, as recv_error does to every datagram with an unknown index; the reader
+// below stands in for that reply. However many replies a sender draws, it never earns a connected socket without
+// an established tunnel.
+func TestConnSocksNoneForUnestablishedSenders(t *testing.T) {
+	h := &countingHandler{}
+	reply := []byte("reply")
+	var c *StdConn
+	var ready atomic.Bool
+	c = newConnSockListener(t, listen("0.0.0.0:0"), h, func(from netip.AddrPort, _ []byte) {
+		if ready.Load() {
+			_ = c.WriteTo(reply, from)
+		}
+	})
+	ready.Store(true)
+	la, err := c.LocalAddr()
+	require.NoError(t, err)
+	to := netip.AddrPortFrom(netip.MustParseAddr("127.0.0.1"), la.Port())
+	pkt := []byte("junk")
+	buf := make([]byte, 64)
+	for range 10 {
+		pc, err := net.ListenUDP("udp4", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
+		require.NoError(t, err)
+		for range 2 * c.socks.cfg.run {
+			_, err := pc.WriteToUDPAddrPort(pkt, to)
+			require.NoError(t, err)
+			require.NoError(t, pc.SetReadDeadline(time.Now().Add(5*time.Second)))
+			_, _, err = pc.ReadFromUDPAddrPort(buf)
+			require.NoError(t, err)
+		}
+		_ = pc.Close()
+	}
+	assert.Zero(t, h.opened.Load())
+	assert.Nil(t, c.socks.table.Load())
+}
+
+// Established peers that keep going away churn connected sockets at most as fast as the open bucket allows, and
+// what connSocks keeps about each goes when its tunnel does. Each round a fresh peer is established, gets busy and
+// goes away, and the clock moves 10ms.
+func TestConnSocksChurnBounded(t *testing.T) {
+	h := &countingHandler{}
+	u := newConnSockListener(t, listen("0.0.0.0:0"), h, nil)
+	var now atomic.Int64
+	now.Store(int64(time.Hour))
+	u.socks.clock = now.Load
+	const rounds, step = 300, 10 * time.Millisecond
+	for range rounds {
+		peer, dst := newEstablishedPeer(t, u, "udp4", "127.0.0.1:0")
+		bufs, addrs := run(dst, u.socks.cfg.run, "p")
+		n, err := u.WriteBatch(bufs, addrs)
+		require.NoError(t, err)
+		require.Equal(t, len(bufs), n)
+		readFrom(t, peer, len(bufs))
+		u.SetEstablishedPeer(dst, false)
+		u.socks.closers.Wait()
+		require.NoError(t, peer.Close())
+		now.Add(int64(step))
+	}
+	elapsed := time.Duration(rounds) * step
+	budget := int64(u.socks.cfg.max) + int64(elapsed/u.socks.cfg.openEvery) + 1
+	t.Logf("%d rounds over %v: %d opens, %d closes", rounds, elapsed, h.opened.Load(), h.closed.Load())
+	require.GreaterOrEqual(t, h.opened.Load(), int64(u.socks.cfg.max), "the rounds never churned")
+	require.LessOrEqual(t, h.opened.Load(), budget)
+	assert.Equal(t, h.opened.Load(), h.closed.Load())
+	assert.Nil(t, u.socks.table.Load(), "peers left after their tunnels went away")
+	assert.Zero(t, u.socks.reserved.Load())
+}
+
+// No send allocates, down any path: the listener with connected sockets off, to a destination that is no
+// tunnel's remote, to an established peer that isn't busy, or on a connected socket, one datagram or a run.
+func TestConnSockSendsDontAllocate(t *testing.T) {
+	if raceEnabled {
+		t.Skip("-race makes sync.Pool drop writers at random, so connected sends allocate")
+	}
+	pkt := make([]byte, 1400)
+	for _, c := range []struct {
+		name  string
+		cfg   func(*connSockConfig)
+		setup func(u *StdConn, dst netip.AddrPort)
+	}{
+		{name: "off", cfg: func(c *connSockConfig) { c.max = 0 }},
+		{name: "ineligible", setup: func(u *StdConn, dst netip.AddrPort) {
+			u.SetEstablishedPeer(netip.AddrPortFrom(dst.Addr(), dst.Port()+1), true)
+		}},
+		{name: "eligible", cfg: func(c *connSockConfig) { c.run = 1 << 30 },
+			setup: func(u *StdConn, dst netip.AddrPort) { u.SetEstablishedPeer(dst, true) }},
+		{name: "connected", setup: func(u *StdConn, dst netip.AddrPort) {
+			u.SetEstablishedPeer(dst, true)
+			bufs, addrs := run(dst, u.socks.cfg.run, "p")
+			_, err := u.WriteBatch(bufs, addrs)
+			require.NoError(t, err)
+			require.NotNil(t, u.connSockFor(dst))
+		}},
+	} {
+		t.Run(c.name, func(t *testing.T) {
+			var cfg []func(*connSockConfig)
+			if c.cfg != nil {
+				cfg = append(cfg, c.cfg)
+			}
+			u := newConnSockListener(t, listen("0.0.0.0:0"), nil, nil, cfg...)
+			_, dst := newPeer(t, "udp4", "127.0.0.1:0")
+			if c.setup != nil {
+				c.setup(u, dst)
+			}
+			bufs, addrs := run(dst, 8, "b")
+			assert.Zero(t, testing.AllocsPerRun(100, func() { _ = u.WriteTo(pkt, dst) }), "WriteTo")
+			assert.Zero(t, testing.AllocsPerRun(100, func() { _, _ = u.WriteBatch(bufs, addrs) }), "WriteBatch")
+		})
+	}
+}
+
+// Once burst tokens are spent the bucket earns one back per interval.
+func TestGCRA(t *testing.T) {
+	var g gcra
+	now := int64(time.Hour)
+	for i := range 3 {
+		require.True(t, g.take(now, 3, time.Second), "token %d of the burst", i)
+	}
+	require.False(t, g.take(now, 3, time.Second))
+	require.False(t, g.take(now+int64(999*time.Millisecond), 3, time.Second))
+	require.True(t, g.take(now+int64(time.Second), 3, time.Second))
+	require.False(t, g.take(now+int64(time.Second), 3, time.Second))
+	require.True(t, g.take(now+int64(time.Hour), 3, time.Second))
+}

udp/race_off_darwin_test.go

@@ -0,0 +1,6 @@
+//go:build !race && !e2e_testing
+
+package udp
+
+// raceEnabled reports a -race build, whose sync.Pool drops a random share of what is put back.
+const raceEnabled = false

udp/race_on_darwin_test.go

@@ -0,0 +1,6 @@
+//go:build race && !e2e_testing
+
+package udp
+
+// raceEnabled reports a -race build, whose sync.Pool drops a random share of what is put back.
+const raceEnabled = true

udp/udp_darwin.go

@@ -10,6 +10,8 @@ import (
 	"log/slog"
 	"net"
 	"net/netip"
+	"sync"
+	"sync/atomic"
 	"syscall"
 	"unsafe"
 
@@ -22,9 +24,26 @@ type StdConn struct {
 	isV4  bool
 	sysFd uintptr
 	l     *slog.Logger
+
+	// listenHost and port are the configured address and the bound port, which a connected socket shares.
+	listenHost netip.Addr
+	port       uint16
+	// rc reaches the listener's descriptor for another listen routine's Rebind.
+	rc    syscall.RawConn
+	socks *connSocks
+	// reader is ListenOut's reader, which connected socket readers share; readMu serializes them, since an
+	// EncReader isn't safe for concurrent use.
+	reader atomic.Pointer[readerPair]
+	readMu sync.Mutex
+}
+
+type readerPair struct {
+	r     EncReader
+	flush func()
 }
 
 var _ Conn = &StdConn{}
+var _ EstablishedPeerConn = &StdConn{}
 
 func NewListener(l *slog.Logger, s Settings) (Conn, error) {
 	lc := NewListenConfig(s.Multi)
@@ -34,12 +53,13 @@ func NewListener(l *slog.Logger, s Settings) (Conn, error) {
 	}
 
 	if uc, ok := pc.(*net.UDPConn); ok {
-		c := &StdConn{UDPConn: uc, l: l}
+		c := &StdConn{UDPConn: uc, l: l, listenHost: s.Listen.Addr()}
 
 		rc, err := uc.SyscallConn()
 		if err != nil {
 			return nil, fmt.Errorf("failed to open udp socket: %w", err)
 		}
+		c.rc = rc
 
 		err = rc.Control(func(fd uintptr) {
 			c.sysFd = fd
@@ -53,6 +73,8 @@ func NewListener(l *slog.Logger, s Settings) (Conn, error) {
 			return nil, err
 		}
 		c.isV4 = la.Addr().Is4()
+		c.port = la.Port()
+		c.socks = joinConnSocks(c, la, s)
 
 		return c, nil
 	}
@@ -89,6 +111,19 @@ func NewListenConfig(multi bool) net.ListenConfig {
 func sendto(s int, buf []byte, flags int, to unsafe.Pointer, addrlen int32) (err error)
 
 func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error {
+	if t := u.socks.table.Load(); t != nil {
+		if p := (*t)[ap]; p != nil {
+			bufs := [1][]byte{b}
+			if sent, err := u.connSockRun(p, ap, bufs[:]); sent == 1 || err != nil {
+				return err
+			}
+		}
+	}
+	return u.writeTo(b, ap)
+}
+
+// writeTo sends b through the listener.
+func (u *StdConn) writeTo(b []byte, ap netip.AddrPort) error {
 	var sa unsafe.Pointer
 	var addrLen int32
 
@@ -141,15 +176,47 @@ func (u *StdConn) WriteTo(b []byte, ap netip.AddrPort) error {
 
 func (u *StdConn) WriteBatch(bufs [][]byte, addrs []netip.AddrPort) (int, error) {
 	// An un-sendable destination costs its own packet, never the ones behind it in the batch.
-	// TODO: WriteTo maps EWOULDBLOCK to an error, so a full send buffer
-	// silently drops the rest of a burst (linux blocks instead). Poll for
-	// writability on EAGAIN before giving up on the remainder.
+	// TODO: writeTo maps EWOULDBLOCK to an error, so a full listener send buffer
+	// silently drops those packets (linux blocks instead). Poll for
+	// writability on EAGAIN before giving up on them.
 	written := 0
-	for i, b := range bufs {
-		if err := u.WriteTo(b, addrs[i]); err == nil {
-			written++
-		} else {
-			u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)
+	t := u.socks.table.Load()
+	if t == nil {
+		for i, b := range bufs {
+			if err := u.writeTo(b, addrs[i]); err == nil {
+				written++
+			} else {
+				u.l.Debug("failed to write packet in batch", "udpAddr", addrs[i], "error", err)
+			}
+		}
+		return written, nil
+	}
+	for off := 0; off < len(bufs); {
+		// A run of datagrams to one peer goes out together, on its connected socket if it has or earns one.
+		dst := addrs[off]
+		end := off + 1
+		for end < len(bufs) && addrs[end] == dst {
+			end++
+		}
+		if p := (*t)[dst]; p != nil {
+			for off < end {
+				sent, err := u.connSockRun(p, dst, bufs[off:end])
+				written += sent
+				off += sent
+				if err == nil {
+					break
+				}
+				u.l.Debug("failed to write packet in batch", "udpAddr", dst, "error", err)
+				off++
+			}
+		}
+		// The listener sends what no connected socket took.
+		for ; off < end; off++ {
+			if err := u.writeTo(bufs[off], dst); err == nil {
+				written++
+			} else {
+				u.l.Debug("failed to write packet in batch", "udpAddr", dst, "error", err)
+			}
 		}
 	}
 	return written, nil
@@ -182,6 +249,8 @@ func NewUDPStatsEmitter(udpConns []Conn) func() {
 
 func (u *StdConn) ListenOut(r EncReader, flush func()) error {
 	buffer := make([]byte, MTU)
+	u.reader.Store(&readerPair{r: r, flush: flush})
+	u.startConnSocks()
 
 	for {
 		// Just read one packet at a time
@@ -194,18 +263,29 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error {
 			continue
 		}
 
+		u.readMu.Lock()
 		r(netip.AddrPortFrom(rua.Addr().Unmap(), rua.Port()), buffer[:n:n])
 		flush()
+		u.readMu.Unlock()
 	}
 }
 
+// Close closes the connected sockets of every listen routine, then the listener.
+func (u *StdConn) Close() error {
+	u.stopConnSocks()
+	return u.UDPConn.Close()
+}
+
 func (u *StdConn) SupportsMultipleReaders() bool {
 	return false
 }
 
 // Rebind clears the interface the kernel scoped this socket to, so that sends are routed against the current
 // routing table instead of the interface we happened to be on when the socket was created. Darwin pins sockets
-// this way on its own, which is what strands us after the underlying network changes.
+// this way on its own, which is what strands us after the underlying network changes. Control.RebindUDPServer
+// rebinds only the first writer, so this clears the other listen routines' listeners too. Connected sockets are
+// closed after the listeners are cleared, and later ones copy the cleared scope and pick their source addresses
+// afresh.
 func (u *StdConn) Rebind() error {
 	var err error
 	if u.isV4 {
@@ -213,6 +293,23 @@ func (u *StdConn) Rebind() error {
 	} else {
 		err = syscall.SetsockoptInt(int(u.sysFd), syscall.IPPROTO_IPV6, syscall.IPV6_BOUND_IF, 0)
 	}
-
+	for _, m := range u.socks.siblings(u) {
+		err = errors.Join(err, m.clearBoundIf())
+	}
+	u.closeConnSocks("rebind")
 	return err
 }
+
+// clearBoundIf clears the interface the listener is scoped to through its RawConn, which fails on a listener that
+// closed concurrently instead of reaching whatever reused its descriptor number.
+func (u *StdConn) clearBoundIf() error {
+	level, opt := syscall.IPPROTO_IPV6, syscall.IPV6_BOUND_IF
+	if u.isV4 {
+		level, opt = syscall.IPPROTO_IP, syscall.IP_BOUND_IF
+	}
+	var serr error
+	if err := u.rc.Control(func(fd uintptr) { serr = syscall.SetsockoptInt(int(fd), level, opt, 0) }); err != nil {
+		return err
+	}
+	return serr
+}