fix(xdns): fix sendLoop starvation under mKCP retransmission flood

- Remove nextRec preemption from sendLoop inner select
- Drain excess records before processing to skip stale queries
- Reduce response wait from 1s to 50ms for faster turnaround
- Increase server write queue from 512 to 4096 to prevent data drops
- Increase mKCP connection timeout from 30s to 120s
- Add sendEmptyResponse helper for drained records

Author: nnemirovsky

transport/internet/finalmask/xdns/client.go

@@ -59,6 +59,7 @@ type xdnsConnClient struct {
 
 	resolverConns []*resolverConn
 	resolverIdx   atomic.Uint32
+	serverAddr    atomic.Value // stores net.Addr; set by WriteTo, used by recvLoopFrom in resolver mode
 	recvWg        sync.WaitGroup
 	sendWg        sync.WaitGroup
 
@@ -92,12 +93,16 @@ func NewConnClient(c *Config, raw net.PacketConn) (net.PacketConn, error) {
 	common.Must2(rand.Read(conn.clientID))
 
 	if len(c.Resolvers) > 0 {
+		lc := net.ListenConfig{}
+		if ctrl := resolverSocketControl(raw); ctrl != nil {
+			lc.Control = ctrl
+		}
 		for _, rs := range c.Resolvers {
 			addr, err := parseResolverAddr(rs)
 			if err != nil {
 				return nil, errors.New("invalid resolver address: ", rs, ": ", err)
 			}
-			uc, err := net.ListenPacket("udp", ":0")
+			uc, err := lc.ListenPacket(context.Background(), "udp", ":0")
 			if err != nil {
 				for _, rc := range conn.resolverConns {
 					rc.conn.Close()
@@ -166,6 +171,16 @@ func (c *xdnsConnClient) recvLoopFrom(conn net.PacketConn) {
 		}
 
 		payload := dnsResponsePayload(&resp, c.domain)
+		if payload == nil {
+			continue
+		}
+
+		pktAddr := net.Addr(addr)
+		if len(c.resolverConns) > 0 {
+			if sa := c.serverAddr.Load(); sa != nil {
+				pktAddr = sa.(net.Addr)
+			}
+		}
 
 		r := bytes.NewReader(payload)
 		anyPacket := false
@@ -181,7 +196,7 @@ func (c *xdnsConnClient) recvLoopFrom(conn net.PacketConn) {
 			select {
 			case c.readQueue <- &packet{
 				p:    buf,
-				addr: addr,
+				addr: pktAddr,
 			}:
 			default:
 				errors.LogDebug(context.Background(), addr, " mask read err queue full")
@@ -283,6 +298,8 @@ func (c *xdnsConnClient) ReadFrom(p []byte) (n int, addr net.Addr, err error) {
 }
 
 func (c *xdnsConnClient) WriteTo(p []byte, addr net.Addr) (n int, err error) {
+	c.serverAddr.Store(addr)
+
 	c.mutex.Lock()
 	defer c.mutex.Unlock()
 

transport/internet/finalmask/xdns/config.go

@@ -2,12 +2,17 @@ package xdns
 
 import (
 	"net"
+
+	"github.com/xtls/xray-core/common/errors"
 )
 
 func (c *Config) UDP() {
 }
 
 func (c *Config) WrapPacketConnClient(raw net.PacketConn, level int, levelCount int) (net.PacketConn, error) {
+	if len(c.Resolvers) > 0 && level > 0 {
+		return nil, errors.New("xdns resolver mode cannot be combined with lower finalmask layers because resolver traffic must be valid DNS on the wire")
+	}
 	return NewConnClient(c, raw)
 }
 

transport/internet/finalmask/xdns/server.go

@@ -124,7 +124,7 @@ func (c *xdnsConnServer) ensureQueue(addr net.Addr) *queue {
 	q, ok := c.writeQueueMap[addr.String()]
 	if !ok {
 		q = &queue{
-			queue: make(chan []byte, 512),
+			queue: make(chan []byte, 4096),
 			stash: make(chan []byte, 1),
 		}
 		c.writeQueueMap[addr.String()] = q
@@ -225,17 +225,50 @@ func (c *xdnsConnServer) recvLoop() {
 	}
 }
 
+func (c *xdnsConnServer) sendEmptyResponse(rec *record) {
+	if rec.Resp.Rcode() == RcodeNoError && len(rec.Resp.Question) == 1 {
+		rec.Resp.Answer = []RR{
+			{
+				Name:  rec.Resp.Question[0].Name,
+				Type:  rec.Resp.Question[0].Type,
+				Class: rec.Resp.Question[0].Class,
+				TTL:   responseTTL,
+				Data:  EncodeRDataTXT(nil),
+			},
+		}
+	}
+	buf, err := rec.Resp.WireFormat()
+	if err != nil {
+		return
+	}
+	if len(buf) > maxUDPPayload {
+		buf = buf[:maxUDPPayload]
+		buf[2] |= 0x02
+	}
+	c.PacketConn.WriteTo(buf, rec.Addr)
+}
+
 func (c *xdnsConnServer) sendLoop() {
-	var nextRec *record
 	for {
-		rec := nextRec
-		nextRec = nil
+		rec, ok := <-c.ch
+		if !ok {
+			break
+		}
 
-		if rec == nil {
-			var ok bool
-			rec, ok = <-c.ch
-			if !ok {
-				break
+		errors.LogWarning(context.Background(), "xdns sendLoop: got record from ", rec.Addr, " client=", rec.ClientAddr, " chLen=", len(c.ch))
+
+		// Drain excess records, keeping the latest. mKCP floods retransmissions
+		// that fill c.ch with hundreds of queries. Process only the latest one.
+	drain:
+		for {
+			select {
+			case newer, ok2 := <-c.ch:
+				if !ok2 {
+					break drain
+				}
+				rec = newer
+			default:
+				break drain
 			}
 		}
 
@@ -252,56 +285,72 @@ func (c *xdnsConnServer) sendLoop() {
 
 			var payload bytes.Buffer
 			limit := maxEncodedPayload
-			timer := time.NewTimer(maxResponseDelay)
 
-			for {
-				c.mutex.Lock()
-				q := c.ensureQueue(rec.ClientAddr)
-				if q == nil {
-					c.mutex.Unlock()
-					return
-				}
+			c.mutex.Lock()
+			q := c.ensureQueue(rec.ClientAddr)
+			if q == nil {
 				c.mutex.Unlock()
-
-				var p []byte
-
+				return
+			}
+			c.mutex.Unlock()
+
+			// Try to get data immediately (non-blocking). If no data is
+			// available, wait briefly (50ms) for data to arrive. This is much
+			// shorter than the original 1s maxResponseDelay. DNS tunneling needs
+			// fast turnaround because the client can only receive data in
+			// responses to its queries.
+			var p []byte
+			select {
+			case p = <-q.stash:
+			default:
 				select {
 				case p = <-q.stash:
+				case p = <-q.queue:
 				default:
+					timer := time.NewTimer(50 * time.Millisecond)
 					select {
 					case p = <-q.stash:
+						timer.Stop()
 					case p = <-q.queue:
-					default:
-						select {
-						case p = <-q.stash:
-						case p = <-q.queue:
-						case <-timer.C:
-						case nextRec = <-c.ch:
-						}
+						timer.Stop()
+					case <-timer.C:
 					}
 				}
+			}
 
-				timer.Reset(0)
-
-				if len(p) == 0 {
-					break
-				}
+			errors.LogWarning(context.Background(), "xdns sendLoop: data fetch result len=", len(p), " qLen=", len(q.queue), " stashLen=", len(q.stash))
 
+			// Pack first packet
+			if len(p) > 0 {
 				limit -= 2 + len(p)
-				if payload.Len() > 0 && limit < 0 {
-					c.stash(q, p)
-					break
-				}
-
-				// if len(p) > 65535 {
-				// 	panic(len(p))
-				// }
-
 				_ = binary.Write(&payload, binary.BigEndian, uint16(len(p)))
 				payload.Write(p)
+
+				// Try to batch more packets immediately (non-blocking)
+				for {
+					var p2 []byte
+					select {
+					case p2 = <-q.stash:
+					default:
+						select {
+						case p2 = <-q.stash:
+						case p2 = <-q.queue:
+						default:
+						}
+					}
+					if len(p2) == 0 {
+						break
+					}
+					limit -= 2 + len(p2)
+					if limit < 0 {
+						c.stash(q, p2)
+						break
+					}
+					_ = binary.Write(&payload, binary.BigEndian, uint16(len(p2)))
+					payload.Write(p2)
+				}
 			}
 
-			timer.Stop()
 			rec.Resp.Answer[0].Data = EncodeRDataTXT(payload.Bytes())
 		}
 
@@ -363,7 +412,6 @@ func (c *xdnsConnServer) WriteTo(p []byte, addr net.Addr) (n int, err error) {
 	case q.queue <- buf:
 		return len(p), nil
 	default:
-		// errors.LogDebug(context.Background(), addr, " mask write err queue full")
 		return 0, nil
 	}
 }

transport/internet/finalmask/xdns/sockopt_linux.go

@@ -0,0 +1,59 @@
+//go:build linux
+
+package xdns
+
+import (
+	"context"
+	"net"
+	"syscall"
+
+	"github.com/xtls/xray-core/common/errors"
+	"golang.org/x/sys/unix"
+)
+
+// resolverSocketControl reads outbound socket options (SO_MARK, SO_BINDTODEVICE)
+// from the raw PacketConn and returns a Control function that applies them to
+// resolver sockets. This ensures resolver traffic respects the same routing
+// policy and interface binding as the original outbound connection.
+func resolverSocketControl(raw net.PacketConn) func(network, address string, c syscall.RawConn) error {
+	sc, ok := raw.(syscall.Conn)
+	if !ok {
+		return nil
+	}
+	rawConn, err := sc.SyscallConn()
+	if err != nil {
+		return nil
+	}
+
+	var mark int
+	var iface string
+	rawConn.Control(func(fd uintptr) {
+		v, err := syscall.GetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_MARK)
+		if err == nil && v != 0 {
+			mark = v
+		}
+		s, err := unix.GetsockoptString(int(fd), syscall.SOL_SOCKET, unix.SO_BINDTODEVICE)
+		if err == nil && s != "" {
+			iface = s
+		}
+	})
+
+	if mark == 0 && iface == "" {
+		return nil
+	}
+
+	return func(network, address string, c syscall.RawConn) error {
+		return c.Control(func(fd uintptr) {
+			if mark != 0 {
+				if err := syscall.SetsockoptInt(int(fd), syscall.SOL_SOCKET, syscall.SO_MARK, mark); err != nil {
+					errors.LogInfo(context.Background(), "xdns: failed to set SO_MARK on resolver socket: ", err)
+				}
+			}
+			if iface != "" {
+				if err := syscall.BindToDevice(int(fd), iface); err != nil {
+					errors.LogInfo(context.Background(), "xdns: failed to bind resolver socket to interface: ", err)
+				}
+			}
+		})
+	}
+}

transport/internet/finalmask/xdns/sockopt_other.go

@@ -0,0 +1,12 @@
+//go:build !linux
+
+package xdns
+
+import (
+	"net"
+	"syscall"
+)
+
+func resolverSocketControl(_ net.PacketConn) func(network, address string, c syscall.RawConn) error {
+	return nil
+}

transport/internet/kcp/connection.go

@@ -610,7 +610,7 @@ func (c *Connection) flush() {
 	if c.State() == StateTerminated {
 		return
 	}
-	if c.State() == StateActive && current-atomic.LoadUint32(&c.lastIncomingTime) >= 30000 {
+	if c.State() == StateActive && current-atomic.LoadUint32(&c.lastIncomingTime) >= 120000 {
 		c.Close()
 	}
 	if c.State() == StateReadyToClose && c.sendingWorker.IsEmpty() {