fix(tuic): preserve client identity for Xray user routing (#6875)

TUIC traffic reached Xray through a loopback SOCKS relay that took no login,
so Xray never learned which client a connection belonged to and user routing
rules never matched TUIC (#6874). The relay now carries one account per client
email, so per-user rules match on TCP and both UDP modes; admission stays with
the native TUIC server.

Adding, removing or renaming a TUIC client swaps only that inbound's relay
instead of restarting Xray: new sessions wait behind a bounded gate while live
streams and UDP associations continue, accounts are verified after the swap and
rolled back on failure, one failing relay no longer strands the others or blocks
unrelated live changes, and emails over SOCKS5's 255-byte limit are excluded
with a warning.

internal/tuic/manager.go

@@ -104,8 +104,10 @@ func (m *Manager) ensureLocked(inst Instance) error {
 
 func (m *Manager) startLocked(inst Instance) (*Server, error) {
 	relay := &SocksRelay{
-		Addr:     fmt.Sprintf("127.0.0.1:%d", SOCKSPortForInbound(inst.Id)),
-		Password: SocksPassword(),
+		Gate:      ManagedRelayGate(SOCKSPortForInbound(inst.Id)),
+		InboundID: inst.Id,
+		Addr:      fmt.Sprintf("127.0.0.1:%d", SOCKSPortForInbound(inst.Id)),
+		Password:  SocksPassword(),
 	}
 	server, err := NewServer(inst, relay)
 	if err != nil {

internal/tuic/manager_live_traffic_test.go

@@ -220,6 +220,11 @@ func audit3StartSocksForManager(t *testing.T, expectedUser, expectedPass string,
 		t.Fatalf("failed to listen: %v", err)
 	}
 
+	update, err := ManagedRelayGate(SOCKSPortForInbound(inboundID)).BeginUpdate(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	update.Finish(inboundID, []string{expectedUser}, true)
 	stop := make(chan struct{})
 
 	go func() {

internal/tuic/manager_shutdown_test.go

@@ -29,6 +29,11 @@ func TestAudit3ManagerStopMustInterruptIdleTCPRelay(t *testing.T) {
 		t.Fatal(err)
 	}
 	defer listener.Close()
+	update, err := ManagedRelayGate(SOCKSPortForInbound(inboundID)).BeginUpdate(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	update.Finish(inboundID, []string{"close-idle@audit3"}, true)
 	release := make(chan struct{})
 	defer close(release)
 	peerFIN := make(chan struct{})

internal/tuic/relay_gate.go

@@ -0,0 +1,159 @@
+package tuic
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+const (
+	relayOpenTimeout     = 10 * time.Second
+	maxPendingRelayOpens = 256
+)
+
+var ErrRelayQueueFull = errors.New("tuic relay: pending connection limit reached")
+
+var relayGates [relayPortSlots]atomic.Pointer[RelayGate]
+
+type RelayGate struct {
+	mu       sync.Mutex
+	writer   chan struct{}
+	changed  chan struct{}
+	paused   bool
+	ready    bool
+	owner    int
+	accounts map[string]struct{}
+	active   int
+	waiting  int
+}
+
+// ManagedRelayGate uses the same bounded slots as the loopback relay ports.
+// The owner check prevents a reused slot from forwarding through a deleted inbound.
+func ManagedRelayGate(port int) *RelayGate {
+	index := port - SOCKSBasePort - 1
+	if index < 0 || index >= len(relayGates) {
+		return nil
+	}
+	if gate := relayGates[index].Load(); gate != nil {
+		return gate
+	}
+	gate := &RelayGate{writer: make(chan struct{}, 1), changed: make(chan struct{})}
+	if relayGates[index].CompareAndSwap(nil, gate) {
+		return gate
+	}
+	return relayGates[index].Load()
+}
+
+func (g *RelayGate) signalLocked() {
+	close(g.changed)
+	g.changed = make(chan struct{})
+}
+
+func (g *RelayGate) acquire(ctx context.Context, owner int, email string) (func(), error) {
+	g.mu.Lock()
+	if g.waiting >= maxPendingRelayOpens {
+		g.mu.Unlock()
+		return nil, ErrRelayQueueFull
+	}
+	g.waiting++
+	defer func() {
+		g.waiting--
+		g.mu.Unlock()
+	}()
+	for {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		_, admitted := g.accounts[email]
+		if !g.paused && g.ready && g.owner == owner && admitted {
+			g.active++
+			return func() {
+				g.mu.Lock()
+				g.active--
+				g.signalLocked()
+				g.mu.Unlock()
+			}, nil
+		}
+		changed := g.changed
+		g.mu.Unlock()
+		select {
+		case <-changed:
+		case <-ctx.Done():
+		}
+		g.mu.Lock()
+	}
+}
+
+type RelayUpdate struct{ gate *RelayGate }
+
+func (g *RelayGate) ReadyFor(owner int) bool {
+	g.mu.Lock()
+	defer g.mu.Unlock()
+	return g.ready && g.owner == owner
+}
+
+func (g *RelayGate) Invalidate(owner int) {
+	g.mu.Lock()
+	g.owner, g.ready, g.accounts = owner, false, nil
+	g.signalLocked()
+	g.mu.Unlock()
+}
+
+// BeginUpdate blocks new handshakes and drains only local SOCKS authentication.
+// Existing TCP streams and UDP associations do not hold this gate.
+func (g *RelayGate) BeginUpdate(ctx context.Context) (*RelayUpdate, error) {
+	select {
+	case g.writer <- struct{}{}:
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	}
+	g.mu.Lock()
+	g.paused = true
+	g.signalLocked()
+	for {
+		if err := ctx.Err(); err != nil {
+			g.paused = false
+			g.signalLocked()
+			g.mu.Unlock()
+			<-g.writer
+			return nil, err
+		}
+		if g.active == 0 {
+			break
+		}
+		changed := g.changed
+		g.mu.Unlock()
+		select {
+		case <-changed:
+		case <-ctx.Done():
+		}
+		g.mu.Lock()
+	}
+	g.mu.Unlock()
+	return &RelayUpdate{gate: g}, nil
+}
+
+func (u *RelayUpdate) Finish(owner int, emails []string, available bool) {
+	g := u.gate
+	g.mu.Lock()
+	g.owner, g.ready = owner, available
+	g.accounts = make(map[string]struct{}, len(emails))
+	for _, email := range emails {
+		g.accounts[email] = struct{}{}
+	}
+	g.paused = false
+	g.signalLocked()
+	g.mu.Unlock()
+	<-g.writer
+}
+
+func (u *RelayUpdate) Cancel() {
+	g := u.gate
+	g.mu.Lock()
+	g.paused = false
+	g.signalLocked()
+	g.mu.Unlock()
+	<-g.writer
+}

internal/tuic/relay_gate_test.go

@@ -0,0 +1,131 @@
+package tuic
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+	"time"
+)
+
+func testRelayGate() *RelayGate {
+	return &RelayGate{writer: make(chan struct{}, 1), changed: make(chan struct{})}
+}
+
+func waitRelayWaiters(t *testing.T, gate *RelayGate, want int) {
+	t.Helper()
+	deadline := time.Now().Add(2 * time.Second)
+	for {
+		gate.mu.Lock()
+		got := gate.waiting
+		gate.mu.Unlock()
+		if got == want {
+			return
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("pending relay opens = %d, want %d", got, want)
+		}
+		time.Sleep(time.Millisecond)
+	}
+}
+
+func TestRelayWaitQueueRejectsOverflowAndCancelsEveryWaiter(t *testing.T) {
+	gate := testRelayGate()
+	relay := &SocksRelay{Gate: gate, InboundID: 17, Addr: "127.0.0.1:1"}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	done := make(chan error, maxPendingRelayOpens)
+	target := &Address{Type: AddrTypeIPv4, IP: net.IPv4(127, 0, 0, 1), Port: 9}
+	for range maxPendingRelayOpens {
+		go func() {
+			conn, err := relay.DialTCP(ctx, "waiting@test", target)
+			if conn != nil {
+				_ = conn.Close()
+			}
+			done <- err
+		}()
+	}
+	waitRelayWaiters(t, gate, maxPendingRelayOpens)
+	if _, err := relay.DialTCP(ctx, "overflow@test", target); !errors.Is(err, ErrRelayQueueFull) {
+		t.Fatalf("overflow error = %v, want %v", err, ErrRelayQueueFull)
+	}
+	cancel()
+	for range maxPendingRelayOpens {
+		select {
+		case err := <-done:
+			if !errors.Is(err, context.Canceled) {
+				t.Fatalf("canceled pending open returned %v", err)
+			}
+		case <-time.After(time.Second):
+			t.Fatal("canceled relay open remained blocked")
+		}
+	}
+	waitRelayWaiters(t, gate, 0)
+}
+
+func TestRelayUpdateDrainsAuthenticationAndCancellationRestoresAdmission(t *testing.T) {
+	gate := testRelayGate()
+	initial, err := gate.BeginUpdate(context.Background())
+	if err != nil {
+		t.Fatal(err)
+	}
+	initial.Finish(17, []string{"active@test"}, true)
+	release, err := gate.acquire(context.Background(), 17, "active@test")
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	updates := make(chan error, 1)
+	go func() {
+		update, err := gate.BeginUpdate(ctx)
+		if update != nil {
+			update.Cancel()
+		}
+		updates <- err
+	}()
+	select {
+	case err := <-updates:
+		t.Fatalf("relay replacement overtook active authentication: %v", err)
+	case <-time.After(50 * time.Millisecond):
+	}
+	cancel()
+	if err := <-updates; !errors.Is(err, context.Canceled) {
+		t.Fatalf("cancel update = %v, want context.Canceled", err)
+	}
+	release()
+	ctx, cancel = context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	release, err = gate.acquire(ctx, 17, "active@test")
+	if err != nil {
+		t.Fatalf("canceled update left admission paused: %v", err)
+	}
+	release()
+	if _, err := gate.acquire(ctx, 18, "active@test"); !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("reused relay slot admitted another inbound: %v", err)
+	}
+}
+
+func TestServerCloseCancelsQueuedUDPAssociation(t *testing.T) {
+	gate := testRelayGate()
+	server, client, id, _ := startLifecycleTestServerWithRelay(t, &SocksRelay{
+		Gate: gate, InboundID: 99101, Addr: "127.0.0.1:1", Password: "lifecycle-socks-password",
+	}, "closing@test")
+	_, _ = authenticatedServerConnection(t, server, id)
+	audit3SendPacket(t, client, packetTransportDatagram, 99, 1, "queued")
+	waitRelayWaiters(t, gate, 1)
+	closed := make(chan error, 1)
+	go func() { closed <- server.Close() }()
+	select {
+	case err := <-closed:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("server shutdown waited for relay update timeout")
+	}
+	waitRelayWaiters(t, gate, 0)
+	if bytes := server.pendingUdpBudget.bytes.Load(); bytes != 0 {
+		t.Fatalf("shutdown retained %d bytes of queued UDP packets", bytes)
+	}
+}

internal/tuic/server.go

@@ -32,6 +32,7 @@ type Server struct {
 	logLevel          atomic.Uint32
 
 	maxUdpRelayPacketSize int
+	pendingUdpBudget      pendingUdpBudget
 
 	users *UserRegistry
 	relay *SocksRelay
@@ -355,6 +356,7 @@ func (s *Server) handleConn(conn *quic.Conn) {
 		authOnce        sync.Once
 		udpAssociations = newUdpAssociationRegistry(s.maxUdpRelayPacketSize)
 	)
+	udpAssociations.pendingBudget = &s.pendingUdpBudget
 
 	authTimer := time.AfterFunc(s.authTimeout, func() {
 		if authState.CompareAndSwap(0, 2) {
@@ -835,15 +837,35 @@ func (s *Server) forwardUDPPacket(
 	udpAssociations *udpAssociationRegistry,
 	relayWg *sync.WaitGroup,
 ) {
-	if len(payload) > s.maxUdpRelayPacketSize || !isPacketTarget(target) {
-		return
-	}
 	if association == nil || len(payload) > s.maxUdpRelayPacketSize || !isPacketTarget(target) {
 		return
 	}
-	association, created, err := udpAssociations.ensureRelay(ctx, assocID, association, user, s.relay)
+	sess, opening, err := udpAssociations.queuePacket(ctx, assocID, association, target, payload)
 	if err != nil {
-		s.logLimited(tuicLogWarn, "udp-dial", 30*time.Second, "UDP relay could not be opened: %v", err)
+		s.logLimited(tuicLogWarn, "udp-queue", 30*time.Second, "UDP relay packet could not be queued: %v", err)
+		return
+	}
+	if sess != nil {
+		s.sendUDPRequest(user, sess, target, payload)
+		return
+	}
+	if opening == nil {
+		return
+	}
+	relayWg.Add(1)
+	go func() {
+		defer relayWg.Done()
+		s.openUDPRelay(ctx, opening, conn, user, assocID, association, udpAssociations, relayWg)
+	}()
+}
+
+func (s *Server) openUDPRelay(ctx, opening context.Context, conn *quic.Conn, user *User, assocID uint16, association *udpAssociation, udpAssociations *udpAssociationRegistry, relayWg *sync.WaitGroup) {
+	_, created, err := udpAssociations.ensureRelay(opening, assocID, association, user, s.relay)
+	if err != nil {
+		if opening.Err() == nil {
+			s.logLimited(tuicLogWarn, "udp-dial", 30*time.Second, "UDP relay could not be opened: %v", err)
+		}
+		udpAssociations.failPending(association)
 		return
 	}
 	sess := association.relay
@@ -855,7 +877,19 @@ func (s *Server) forwardUDPPacket(
 			s.relayUDPResponses(ctx, conn, user, assocID, association, udpAssociations, sess)
 		}()
 	}
+	for {
+		packet, ok := udpAssociations.nextPending(assocID, association)
+		if !ok {
+			return
+		}
+		if ctx.Err() == nil {
+			s.sendUDPRequest(user, sess, packet.target, packet.payload)
+		}
+		udpAssociations.releasePendingPacket(packet)
+	}
+}
 
+func (s *Server) sendUDPRequest(user *User, sess *udpRelaySession, target *Address, payload []byte) {
 	if _, err := sess.relay.Send(target, payload); err != nil {
 		s.logLimited(tuicLogWarn, "udp-send", 30*time.Second, "UDP relay request failed: %v", err)
 		return

internal/tuic/socks_bridge.go

@@ -20,10 +20,19 @@ var ErrUdpPayloadTooLarge = errors.New("tuic socks: UDP packet exceeds the maxim
 
 const maxSocksUdpDatagramSize = 65507
 
+func ValidateRelayEmail(email string) error {
+	if len(email) == 0 || len(email) > 255 {
+		return errors.New("TUIC client email must contain 1 to 255 UTF-8 bytes")
+	}
+	return nil
+}
+
 // SocksRelay describes the loopback SOCKS5 endpoint where decrypted TUIC traffic is forwarded.
 type SocksRelay struct {
-	Addr     string // e.g. "127.0.0.1:63201"
-	Password string // internal shared password for the SOCKS inbound
+	Gate      *RelayGate
+	InboundID int
+	Addr      string // e.g. "127.0.0.1:63201"
+	Password  string // internal shared password for the SOCKS inbound
 }
 
 // CountingConn wraps a net.Conn and tracks bytes read and written atomically.
@@ -51,18 +60,12 @@ func (c *CountingConn) Write(p []byte) (int, error) {
 
 // DialTCP establishes a SOCKS5 CONNECT tunnel to the target address on behalf of user.
 func (r *SocksRelay) DialTCP(ctx context.Context, user string, target *Address) (net.Conn, error) {
-	dialer := net.Dialer{Timeout: 10 * time.Second}
-	conn, err := dialer.DialContext(ctx, "tcp", r.Addr)
+	conn, err := r.dialAuthenticated(ctx, user)
 	if err != nil {
-		return nil, fmt.Errorf("tuic socks: dial relay %s: %w", r.Addr, err)
-	}
-
-	_ = conn.SetDeadline(time.Now().Add(10 * time.Second))
-
-	if err := socks5Handshake(conn, user, r.Password); err != nil {
-		conn.Close()
 		return nil, err
 	}
+	stop := context.AfterFunc(ctx, func() { _ = conn.Close() })
+	defer stop()
 
 	// Send SOCKS5 CONNECT request
 	req := buildSocks5ConnectRequest(target)
@@ -81,10 +84,43 @@ func (r *SocksRelay) DialTCP(ctx context.Context, user string, target *Address)
 	}
 
 	_ = conn.SetDeadline(time.Time{})
+	if err := ctx.Err(); err != nil {
+		_ = conn.Close()
+		return nil, err
+	}
 
 	return conn, nil
 }
 
+func (r *SocksRelay) dialAuthenticated(ctx context.Context, user string) (net.Conn, error) {
+	if r.Gate != nil {
+		waiting, cancel := context.WithTimeout(ctx, relayOpenTimeout)
+		release, err := r.Gate.acquire(waiting, r.InboundID, user)
+		cancel()
+		if err != nil {
+			return nil, err
+		}
+		defer release()
+	}
+	dialer := net.Dialer{Timeout: 10 * time.Second}
+	conn, err := dialer.DialContext(ctx, "tcp", r.Addr)
+	if err != nil {
+		return nil, fmt.Errorf("tuic socks: dial relay %s: %w", r.Addr, err)
+	}
+	stop := context.AfterFunc(ctx, func() { _ = conn.Close() })
+	defer stop()
+	_ = conn.SetDeadline(time.Now().Add(10 * time.Second))
+	if err := socks5Handshake(conn, user, r.Password); err != nil {
+		_ = conn.Close()
+		return nil, err
+	}
+	if err := ctx.Err(); err != nil {
+		_ = conn.Close()
+		return nil, err
+	}
+	return conn, nil
+}
+
 func buildSocks5ConnectRequest(target *Address) []byte {
 	if target == nil {
 		return nil
@@ -148,18 +184,12 @@ type SocksUDPSession struct {
 
 // DialUDP establishes a SOCKS5 UDP ASSOCIATE tunnel to Xray.
 func (r *SocksRelay) DialUDP(ctx context.Context, user string) (*SocksUDPSession, error) {
-	dialer := net.Dialer{Timeout: 10 * time.Second}
-	ctrl, err := dialer.DialContext(ctx, "tcp", r.Addr)
+	ctrl, err := r.dialAuthenticated(ctx, user)
 	if err != nil {
-		return nil, fmt.Errorf("tuic socks: dial UDP control connection: %w", err)
-	}
-
-	_ = ctrl.SetDeadline(time.Now().Add(10 * time.Second))
-
-	if err := socks5Handshake(ctrl, user, r.Password); err != nil {
-		ctrl.Close()
 		return nil, err
 	}
+	stop := context.AfterFunc(ctx, func() { _ = ctrl.Close() })
+	defer stop()
 
 	// SOCKS5 UDP ASSOCIATE (0x03), dst 0.0.0.0:0
 	if _, err := ctrl.Write([]byte{0x05, 0x03, 0x00, 0x01, 0, 0, 0, 0, 0, 0}); err != nil {
@@ -180,6 +210,11 @@ func (r *SocksRelay) DialUDP(ctx context.Context, user string) (*SocksUDPSession
 		ctrl.Close()
 		return nil, fmt.Errorf("tuic socks: dial UDP relay endpoint %s: %w", bind, err)
 	}
+	if err := ctx.Err(); err != nil {
+		_ = ctrl.Close()
+		_ = udpConn.Close()
+		return nil, err
+	}
 
 	return &SocksUDPSession{
 		ctrl:     ctrl,
@@ -558,19 +593,19 @@ func SocksPassword() string {
 	return socksPassword
 }
 
-// SocksInboundSettings builds the JSON `settings` block for a stock Xray
-// SOCKS5 inbound with one username/password account per email, all sharing
-// password. UDP is enabled for UDP ASSOCIATE proxying.
-func SocksInboundSettings(emails []string, password string) ([]byte, error) {
+// SocksInboundSettings preserves each client email as the Xray routing identity.
+// The caller selects a policy level so relay traffic is not billed twice.
+func SocksInboundSettings(emails []string, password string, userLevel uint32) ([]byte, error) {
 	type account struct {
 		User string `json:"user"`
 		Pass string `json:"pass"`
 	}
 	settings := struct {
-		Auth     string    `json:"auth"`
-		UDP      bool      `json:"udp"`
-		Accounts []account `json:"accounts"`
-	}{Auth: "password", UDP: true}
+		Auth      string    `json:"auth"`
+		UDP       bool      `json:"udp"`
+		UserLevel uint32    `json:"userLevel"`
+		Accounts  []account `json:"accounts"`
+	}{Auth: "password", UDP: true, UserLevel: userLevel}
 	for _, email := range emails {
 		settings.Accounts = append(settings.Accounts, account{User: email, Pass: password})
 	}

internal/tuic/stream_udp_lifecycle_test.go

@@ -15,6 +15,11 @@ import (
 )
 
 func startLifecycleTestServer(t *testing.T, relayAddr, email string) (*Server, *clientquic.Conn, uuid.UUID, string) {
+	t.Helper()
+	return startLifecycleTestServerWithRelay(t, &SocksRelay{Addr: relayAddr, Password: "lifecycle-socks-password"}, email)
+}
+
+func startLifecycleTestServerWithRelay(t *testing.T, relay *SocksRelay, email string) (*Server, *clientquic.Conn, uuid.UUID, string) {
 	t.Helper()
 	cert, key := generateTestCert(t)
 	clientID := uuid.New()
@@ -24,7 +29,7 @@ func startLifecycleTestServer(t *testing.T, relayAddr, email string) (*Server, *
 		Certificate: string(cert), PrivateKey: string(key), ALPN: []string{"h3"},
 		AuthenticationTimeout: 2, MaxIdleTime: 30,
 		Clients: []TuicClientSettings{{UUID: clientID.String(), Password: password, Email: email}},
-	}, &SocksRelay{Addr: relayAddr, Password: "lifecycle-socks-password"})
+	}, relay)
 	if err != nil {
 		t.Fatalf("create TUIC server: %v", err)
 	}

internal/tuic/types.go

@@ -95,6 +95,16 @@ func (inst Instance) FullFingerprint() string {
 }
 
 func InstanceFromInbound(ib *model.Inbound) (Instance, bool) {
+	return instanceFromInbound(ib, false)
+}
+
+// ConfiguredInstanceFromInbound includes disabled clients for stable relay identity.
+// Only the native server decides whether a client may authenticate.
+func ConfiguredInstanceFromInbound(ib *model.Inbound) (Instance, bool) {
+	return instanceFromInbound(ib, true)
+}
+
+func instanceFromInbound(ib *model.Inbound, includeDisabled bool) (Instance, bool) {
 	if ib == nil || ib.Protocol != model.TUIC {
 		return Instance{}, false
 	}
@@ -221,7 +231,7 @@ func InstanceFromInbound(ib *model.Inbound) (Instance, bool) {
 
 	clients := make([]TuicClientSettings, 0, len(parsed.Clients))
 	for _, c := range parsed.Clients {
-		if c.Enable != nil && !*c.Enable {
+		if !includeDisabled && c.Enable != nil && !*c.Enable {
 			continue
 		}
 		uuidVal := strings.ToLower(strings.TrimSpace(c.UUID))

internal/tuic/udp_associations.go

@@ -18,6 +18,9 @@ type udpAssociation struct {
 	responseTransport uint8
 	relay             *udpRelaySession
 	lastActive        time.Time
+	opening           bool
+	openCancel        context.CancelFunc
+	pending           []pendingUdpPacket
 }
 
 type udpAssociationRegistry struct {
@@ -26,6 +29,8 @@ type udpAssociationRegistry struct {
 	datagramReassembler *packetReassembler
 	streamReassembler   *packetReassembler
 	maxPacketSize       int
+	pendingBytes        int
+	pendingBudget       *pendingUdpBudget
 }
 
 func newUdpAssociationRegistry(maxPacketSize int) *udpAssociationRegistry {
@@ -115,6 +120,9 @@ func (r *udpAssociationRegistry) ensureRelay(ctx context.Context, assocID uint16
 func (r *udpAssociationRegistry) dissociate(assocID uint16) bool {
 	r.mu.Lock()
 	association := r.associations[assocID]
+	if association != nil {
+		r.clearPendingLocked(association)
+	}
 	delete(r.associations, assocID)
 	r.datagramReassembler.clearAssociation(assocID)
 	r.streamReassembler.clearAssociation(assocID)
@@ -151,6 +159,7 @@ func (r *udpAssociationRegistry) expireLocked(now time.Time) {
 			continue
 		}
 		delete(r.associations, assocID)
+		r.clearPendingLocked(association)
 		r.datagramReassembler.clearAssociation(assocID)
 		r.streamReassembler.clearAssociation(assocID)
 		if association.relay != nil {
@@ -162,6 +171,7 @@ func (r *udpAssociationRegistry) expireLocked(now time.Time) {
 func (r *udpAssociationRegistry) closeAll() {
 	r.mu.Lock()
 	for _, association := range r.associations {
+		r.clearPendingLocked(association)
 		if association.relay != nil {
 			_ = association.relay.relay.Close()
 		}
@@ -175,6 +185,7 @@ func (r *udpAssociationRegistry) closeAll() {
 func (r *udpAssociationRegistry) release(id uint16, expected *udpAssociation) {
 	r.mu.Lock()
 	if r.associations[id] == expected {
+		r.clearPendingLocked(expected)
 		delete(r.associations, id)
 		r.datagramReassembler.clearAssociation(id)
 		r.streamReassembler.clearAssociation(id)

internal/tuic/udp_pending.go

@@ -0,0 +1,113 @@
+package tuic
+
+import (
+	"context"
+	"errors"
+	"sync/atomic"
+)
+
+const (
+	maxPendingUdpBytesPerInbound = 1 << 20
+	maxPendingUdpBytesPerConn    = 128 << 10
+	maxPendingUdpPackets         = 16
+	udpPacketOverhead            = 256
+)
+
+var errUdpQueueFull = errors.New("tuic: pending UDP packet limit reached")
+
+type pendingUdpBudget struct{ bytes atomic.Int64 }
+
+func (b *pendingUdpBudget) reserve(size int) bool {
+	if b == nil {
+		return true
+	}
+	for {
+		used := b.bytes.Load()
+		if used+int64(size) > maxPendingUdpBytesPerInbound {
+			return false
+		}
+		if b.bytes.CompareAndSwap(used, used+int64(size)) {
+			return true
+		}
+	}
+}
+
+type pendingUdpPacket struct {
+	target  *Address
+	payload []byte
+}
+
+func (r *udpAssociationRegistry) queuePacket(ctx context.Context, id uint16, expected *udpAssociation, target *Address, payload []byte) (*udpRelaySession, context.Context, error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.associations[id] != expected {
+		return nil, nil, errUdpAssociationClosed
+	}
+	if expected.relay != nil && !expected.opening {
+		return expected.relay, nil, nil
+	}
+	charge := len(payload) + udpPacketOverhead
+	if len(expected.pending) >= maxPendingUdpPackets || r.pendingBytes+charge > maxPendingUdpBytesPerConn || !r.pendingBudget.reserve(charge) {
+		return nil, nil, errUdpQueueFull
+	}
+	r.pendingBytes += charge
+	addr := *target
+	addr.IP = append([]byte(nil), target.IP...)
+	expected.pending = append(expected.pending, pendingUdpPacket{target: &addr, payload: append([]byte(nil), payload...)})
+	if expected.opening {
+		return nil, nil, nil
+	}
+	expected.opening = true
+	opening, cancel := context.WithCancel(ctx)
+	expected.openCancel = cancel
+	return nil, opening, nil
+}
+
+func (r *udpAssociationRegistry) nextPending(id uint16, expected *udpAssociation) (pendingUdpPacket, bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.associations[id] != expected || len(expected.pending) == 0 {
+		expected.opening = false
+		if expected.openCancel != nil {
+			expected.openCancel()
+			expected.openCancel = nil
+		}
+		return pendingUdpPacket{}, false
+	}
+	packet := expected.pending[0]
+	expected.pending[0] = pendingUdpPacket{}
+	expected.pending = expected.pending[1:]
+	return packet, true
+}
+
+func (r *udpAssociationRegistry) releasePendingPacketLocked(packet pendingUdpPacket) {
+	charge := len(packet.payload) + udpPacketOverhead
+	r.pendingBytes -= charge
+	if r.pendingBudget != nil {
+		r.pendingBudget.bytes.Add(-int64(charge))
+	}
+}
+
+func (r *udpAssociationRegistry) releasePendingPacket(packet pendingUdpPacket) {
+	r.mu.Lock()
+	r.releasePendingPacketLocked(packet)
+	r.mu.Unlock()
+}
+
+func (r *udpAssociationRegistry) clearPendingLocked(association *udpAssociation) {
+	for _, packet := range association.pending {
+		r.releasePendingPacketLocked(packet)
+	}
+	association.pending = nil
+	association.opening = false
+	if association.openCancel != nil {
+		association.openCancel()
+		association.openCancel = nil
+	}
+}
+
+func (r *udpAssociationRegistry) failPending(expected *udpAssociation) {
+	r.mu.Lock()
+	r.clearPendingLocked(expected)
+	r.mu.Unlock()
+}

internal/tuic/udp_pending_test.go

@@ -0,0 +1,139 @@
+package tuic
+
+import (
+	"context"
+	"errors"
+	"net"
+	"testing"
+	"time"
+)
+
+func pendingTestAssociation(t *testing.T, registry *udpAssociationRegistry, id uint16, size int) (*udpAssociation, *Address, []byte) {
+	t.Helper()
+	target := &Address{Type: AddrTypeIPv4, IP: net.IPv4(127, 0, 0, 1), Port: 53}
+	payload := make([]byte, size)
+	association, _, _, complete := registry.feed(packetTransportDatagram, &PacketHeader{
+		AssocID: id, PktID: 1, FragTotal: 1, Size: uint16(size), Addr: target,
+	}, payload)
+	if association == nil || !complete {
+		t.Fatal("valid UDP packet did not create association")
+	}
+	return association, target, payload
+}
+
+func TestPendingUdpPacketLimitsReleaseCapacityAfterDissociate(t *testing.T) {
+	registry := newUdpAssociationRegistry(1500)
+	registry.pendingBudget = &pendingUdpBudget{}
+	association, target, payload := pendingTestAssociation(t, registry, 1, 1000)
+	for range 16 {
+		if _, _, err := registry.queuePacket(context.Background(), 1, association, target, payload); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, _, err := registry.queuePacket(context.Background(), 1, association, target, payload); !errors.Is(err, errUdpQueueFull) {
+		t.Fatalf("association overflow error = %v, want %v", err, errUdpQueueFull)
+	}
+	registry.dissociate(1)
+	if used := registry.pendingBudget.bytes.Load(); used != 0 {
+		t.Fatalf("dissociation retained %d queued bytes", used)
+	}
+	replacement, _, _ := pendingTestAssociation(t, registry, 1, 1000)
+	if _, _, err := registry.queuePacket(context.Background(), 1, association, target, payload); !errors.Is(err, errUdpAssociationClosed) {
+		t.Fatalf("stale worker queued into reused association: %v", err)
+	}
+	if _, opening, err := registry.queuePacket(context.Background(), 1, replacement, target, payload); err != nil || opening == nil {
+		t.Fatalf("released capacity unavailable to replacement: opening=%v err=%v", opening, err)
+	}
+	registry.closeAll()
+}
+
+func TestPendingUdpByteBudgetsBoundMultipleConnections(t *testing.T) {
+	budget := &pendingUdpBudget{}
+	var registries []*udpAssociationRegistry
+	defer func() {
+		for _, registry := range registries {
+			registry.closeAll()
+		}
+	}()
+	blockedInbound := false
+	for connID := 0; connID < 12; connID++ {
+		registry := newUdpAssociationRegistry(1500)
+		registry.pendingBudget = budget
+		registries = append(registries, registry)
+		blockedConn := false
+		for id := uint16(1); id < 256; id++ {
+			association, target, payload := pendingTestAssociation(t, registry, id, 1500)
+			_, _, err := registry.queuePacket(context.Background(), id, association, target, payload)
+			if err == nil {
+				continue
+			}
+			if !errors.Is(err, errUdpQueueFull) {
+				t.Fatal(err)
+			}
+			if connID == 0 && registry.pendingBytes+1756 <= 128<<10 {
+				t.Fatal("first connection hit a limit before filling its 128 KiB allowance")
+			}
+			blockedConn = registry.pendingBytes+1756 > 128<<10
+			blockedInbound = budget.bytes.Load()+1756 > 1<<20
+			break
+		}
+		if registry.pendingBytes > 128<<10 || budget.bytes.Load() > 1<<20 {
+			t.Fatal("pending UDP payloads exceeded their memory budgets")
+		}
+		if blockedInbound {
+			break
+		}
+		if !blockedConn {
+			t.Fatal("connection queue admitted all associations without enforcing byte budget")
+		}
+	}
+	if !blockedInbound {
+		t.Fatal("separate QUIC connections bypassed the shared inbound byte budget")
+	}
+	for _, registry := range registries {
+		registry.closeAll()
+	}
+	if used := budget.bytes.Load(); used != 0 {
+		t.Fatalf("closing connections retained %d queued bytes", used)
+	}
+}
+
+func TestPendingUdpRemovalCancelsDialAndReleasesQueuedAndInFlightPackets(t *testing.T) {
+	for _, remove := range []string{"dissociate", "expire", "shutdown", "reader-failure", "dial-failure"} {
+		t.Run(remove, func(t *testing.T) {
+			registry := newUdpAssociationRegistry(1500)
+			registry.pendingBudget = &pendingUdpBudget{}
+			association, target, payload := pendingTestAssociation(t, registry, 1, 100)
+			_, opening, err := registry.queuePacket(context.Background(), 1, association, target, payload)
+			if err != nil || opening == nil {
+				t.Fatalf("queue first UDP packet: opening=%v err=%v", opening, err)
+			}
+			if _, _, err := registry.queuePacket(context.Background(), 1, association, target, payload); err != nil {
+				t.Fatal(err)
+			}
+			inFlight, ok := registry.nextPending(1, association)
+			if !ok {
+				t.Fatal("first pending packet disappeared before removal")
+			}
+			switch remove {
+			case "dissociate":
+				registry.dissociate(1)
+			case "expire":
+				registry.reapIdle(time.Now().Add(udpAssociationIdleTimeout + time.Second))
+			case "shutdown":
+				registry.closeAll()
+			case "reader-failure":
+				registry.release(1, association)
+			case "dial-failure":
+				registry.failPending(association)
+			}
+			if !errors.Is(opening.Err(), context.Canceled) {
+				t.Fatal("association removal left its SOCKS dial alive")
+			}
+			registry.releasePendingPacket(inFlight)
+			if used := registry.pendingBudget.bytes.Load(); used != 0 || registry.pendingBytes != 0 {
+				t.Fatalf("removal leaked or double released packet budget: inbound=%d connection=%d", used, registry.pendingBytes)
+			}
+		})
+	}
+}