fix(dns): prevent long-run dns stall with bounded async and transport cleanup

Author: MaurUppi

control/control_plane.go

@@ -45,6 +45,12 @@ import (
 	"golang.org/x/sys/unix"
 )
 
+const (
+	// Cap async DNS handling to avoid unbounded goroutine growth under
+	// sustained high-rate DNS traffic.
+	maxAsyncDnsInFlight = 512
+)
+
 type ControlPlane struct {
 	log *logrus.Logger
 
@@ -58,6 +64,7 @@ type ControlPlane struct {
 
 	dnsController    *DnsController
 	onceNetworkReady sync.Once
+	dnsAsyncSem      chan struct{}
 
 	dialMode consts.DialMode
 
@@ -390,6 +397,7 @@ func NewControlPlane(
 		outbounds:         outbounds,
 		dnsController:     nil,
 		onceNetworkReady:  sync.Once{},
+		dnsAsyncSem:       make(chan struct{}, maxAsyncDnsInFlight),
 		dialMode:          dialMode,
 		routingMatcher:    routingMatcher,
 		ctx:               ctx,
@@ -833,20 +841,33 @@ func (c *ControlPlane) Serve(readyChan chan<- bool, listener *Listener) (err err
 				// DNS packets must not block the per-src serial task queue:
 				// a single slow upstream (up to DefaultDialTimeout=8s) would
 				// stall all subsequent packets from the same source IP.
-				// DNS is stateless; parallel handling is safe.
+				// DNS is stateless; parallel handling is safe.  Use a bounded
+				// semaphore to avoid unbounded goroutine growth.
 				if pktDst.Port() == 53 || pktDst.Port() == 5353 {
-					// Transfer buffer ownership to the goroutine; clear defers.
-					gData := data
-					gOob := oob
-					data = nil
-					oob = nil
-					go func() {
-						defer gData.Put()
-						defer gOob.Put()
-						if e := c.handlePkt(udpConn, gData, convergeSrc, common.ConvergeAddrPort(pktDst), common.ConvergeAddrPort(realDst), routingResult, false); e != nil {
+					select {
+					case c.dnsAsyncSem <- struct{}{}:
+						// Transfer buffer ownership to the goroutine; clear defers.
+						gData := data
+						gOob := oob
+						data = nil
+						oob = nil
+						go func() {
+							defer func() {
+								<-c.dnsAsyncSem
+							}()
+							defer gData.Put()
+							defer gOob.Put()
+							if e := c.handlePkt(udpConn, gData, convergeSrc, common.ConvergeAddrPort(pktDst), common.ConvergeAddrPort(realDst), routingResult, false); e != nil {
+								c.log.Warnln("handlePkt(dns):", e)
+							}
+						}()
+					default:
+						// Semaphore saturated: fall back to sync handling here to
+						// apply backpressure instead of spawning unbounded goroutines.
+						if e := c.handlePkt(udpConn, data, convergeSrc, common.ConvergeAddrPort(pktDst), common.ConvergeAddrPort(realDst), routingResult, false); e != nil {
 							c.log.Warnln("handlePkt(dns):", e)
 						}
-					}()
+					}
 					return
 				}
 				if e := c.handlePkt(udpConn, data, convergeSrc, common.ConvergeAddrPort(pktDst), common.ConvergeAddrPort(realDst), routingResult, false); e != nil {

control/dns.go

@@ -76,13 +76,29 @@ type DoH struct {
 	client       *http.Client
 }
 
+func closeDoHClient(client *http.Client) error {
+	if client == nil || client.Transport == nil {
+		return nil
+	}
+	// HTTP/1.1 and HTTP/2 transport.
+	if t, ok := client.Transport.(interface{ CloseIdleConnections() }); ok {
+		t.CloseIdleConnections()
+	}
+	// HTTP/3 transport.
+	if closer, ok := client.Transport.(io.Closer); ok {
+		return closer.Close()
+	}
+	return nil
+}
+
 func (d *DoH) ForwardDNS(ctx context.Context, data []byte) (*dnsmessage.Msg, error) {
 	if d.client == nil {
 		d.client = d.getClient()
 	}
 	msg, err := sendHttpDNS(ctx, d.client, d.dialArgument.bestTarget.String(), &d.Upstream, data)
 	if err != nil {
 		// If failed to send DNS request, we should try to create a new client.
+		_ = closeDoHClient(d.client)
 		d.client = d.getClient()
 		msg, err = sendHttpDNS(ctx, d.client, d.dialArgument.bestTarget.String(), &d.Upstream, data)
 		if err != nil {
@@ -155,7 +171,9 @@ func (d *DoH) getHttp3RoundTripper() *http3.RoundTripper {
 }
 
 func (d *DoH) Close() error {
-	return nil
+	err := closeDoHClient(d.client)
+	d.client = nil
+	return err
 }
 
 type DoQ struct {
@@ -165,6 +183,13 @@ type DoQ struct {
 	connection   quic.EarlyConnection
 }
 
+func closeDoQConnection(conn quic.EarlyConnection) error {
+	if conn == nil {
+		return nil
+	}
+	return conn.CloseWithError(0, "")
+}
+
 func (d *DoQ) ForwardDNS(ctx context.Context, data []byte) (*dnsmessage.Msg, error) {
 	if d.connection == nil {
 		qc, err := d.createConnection(ctx)
@@ -176,7 +201,9 @@ func (d *DoQ) ForwardDNS(ctx context.Context, data []byte) (*dnsmessage.Msg, err
 
 	stream, err := d.connection.OpenStreamSync(ctx)
 	if err != nil {
-		// If failed to open stream, we should try to create a new connection.
+		// If failed to open stream, close old connection and try to create a new one.
+		_ = closeDoQConnection(d.connection)
+		d.connection = nil
 		qc, err := d.createConnection(ctx)
 		if err != nil {
 			return nil, err
@@ -230,7 +257,9 @@ func (d *DoQ) createConnection(ctx context.Context) (quic.EarlyConnection, error
 }
 
 func (d *DoQ) Close() error {
-	return nil
+	err := closeDoQConnection(d.connection)
+	d.connection = nil
+	return err
 }
 
 type DoTLS struct {