Fix dial contexts cancelled while connections are still in use

common/dialer/tfo.go

@@ -20,6 +20,7 @@ import (
 type slowOpenConn struct {
 	dialer      *tfo.Dialer
 	ctx         context.Context
+	cancel      context.CancelFunc
 	network     string
 	destination M.Socksaddr
 	conn        atomic.Pointer[net.TCPConn]
@@ -39,9 +40,11 @@ func DialSlowContext(dialer *tfo.Dialer, ctx context.Context, network string, de
 			return dialer.Dialer.DialContext(ctx, network, destination.AddrString())
 		}
 	}
+	dialCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
 	return &slowOpenConn{
 		dialer:      dialer,
-		ctx:         ctx,
+		ctx:         dialCtx,
+		cancel:      cancel,
 		network:     network,
 		destination: destination,
 		create:      make(chan struct{}),
@@ -86,7 +89,14 @@ func (c *slowOpenConn) Write(b []byte) (n int, err error) {
 	if err != nil {
 		c.err = err
 	} else {
-		c.conn.Store(conn.(*net.TCPConn))
+		select {
+		case <-c.done:
+			conn.Close()
+			err = os.ErrClosed
+			c.err = err
+		default:
+			c.conn.Store(conn.(*net.TCPConn))
+		}
 	}
 	n = len(b)
 	close(c.create)
@@ -96,6 +106,7 @@ func (c *slowOpenConn) Write(b []byte) (n int, err error) {
 func (c *slowOpenConn) Close() error {
 	c.closeOnce.Do(func() {
 		close(c.done)
+		c.cancel()
 		conn := c.conn.Load()
 		if conn != nil {
 			conn.Close()

common/httpclient/http3_transport.go

@@ -6,6 +6,7 @@ import (
 	"context"
 	stdTLS "crypto/tls"
 	"errors"
+	"io"
 	"net/http"
 	"sync"
 	"time"
@@ -172,41 +173,49 @@ func (t *http3FallbackTransport) roundTripHTTP3(request *http.Request) (*http.Re
 }
 
 func (t *http3FallbackTransport) roundTripHTTP3Race(request *http.Request, authority string) (*http.Response, error) {
-	ctx, cancel := context.WithCancel(request.Context())
-	defer cancel()
 	type result struct {
 		response *http.Response
 		err      error
 		h3       bool
 	}
-	results := make(chan result, 2)
-	startRoundTrip := func(request *http.Request, useH3 bool) {
-		request = request.WithContext(ctx)
-		var (
-			response *http.Response
-			err      error
-		)
-		if useH3 {
-			response, err = t.h3Transport.RoundTrip(request)
-		} else {
-			response, err = t.h2FallbackRoundTrip(request)
-		}
-		results <- result{response: response, err: err, h3: useH3}
+	var (
+		results  = make(chan result, 2)
+		cancels  []context.CancelFunc
+		received int
+	)
+	startRoundTrip := func(useH3 bool) {
+		ctx, cancel := context.WithCancel(request.Context())
+		cancels = append(cancels, cancel)
+		raceRequest := cloneRequestForRetry(request).WithContext(ctx)
+		go func() {
+			var (
+				response *http.Response
+				err      error
+			)
+			if useH3 {
+				response, err = t.h3Transport.RoundTrip(raceRequest)
+			} else {
+				response, err = t.h2FallbackRoundTrip(raceRequest)
+			}
+			results <- result{response: response, err: err, h3: useH3}
+		}()
 	}
-	goroutines := 1
-	received := 0
-	drainRemaining := func() {
-		cancel()
-		for range goroutines - received {
+	finish := func(winner int) {
+		for index, cancel := range cancels {
+			if index != winner {
+				cancel()
+			}
+		}
+		for range len(cancels) - received {
 			go func() {
 				loser := <-results
-				if loser.response != nil && loser.response.Body != nil {
+				if loser.response != nil {
 					loser.response.Body.Close()
 				}
 			}()
 		}
 	}
-	go startRoundTrip(cloneRequestForRetry(request), true)
+	startRoundTrip(true)
 	timer := time.NewTimer(t.fallbackDelay)
 	defer timer.Stop()
 	var (
@@ -216,39 +225,40 @@ func (t *http3FallbackTransport) roundTripHTTP3Race(request *http.Request, autho
 	for {
 		select {
 		case <-timer.C:
-			if goroutines == 1 {
-				goroutines++
-				go startRoundTrip(cloneRequestForRetry(request), false)
+			if len(cancels) == 1 {
+				startRoundTrip(false)
 			}
 		case raceResult := <-results:
 			received++
 			if raceResult.err == nil {
+				winner := 1
 				if raceResult.h3 {
+					winner = 0
 					t.clearH3Broken(authority)
 				}
-				drainRemaining()
+				finish(winner)
+				raceResult.response.Body = &cancelOnCloseBody{ReadCloser: raceResult.response.Body, cancel: cancels[winner]}
 				return raceResult.response, nil
 			}
 			if raceResult.h3 {
 				t.markH3Broken(authority)
 				h3Err = raceResult.err
-				if goroutines == 1 {
-					goroutines++
+				if len(cancels) == 1 {
 					if !timer.Stop() {
 						select {
 						case <-timer.C:
 						default:
 						}
 					}
-					go startRoundTrip(cloneRequestForRetry(request), false)
+					startRoundTrip(false)
 				}
 			} else {
 				fallbackErr = raceResult.err
 			}
-			if received < goroutines {
+			if received < len(cancels) {
 				continue
 			}
-			drainRemaining()
+			finish(-1)
 			switch {
 			case h3Err != nil && fallbackErr != nil:
 				return nil, E.Errors(h3Err, fallbackErr)
@@ -261,6 +271,17 @@ func (t *http3FallbackTransport) roundTripHTTP3Race(request *http.Request, autho
 	}
 }
 
+type cancelOnCloseBody struct {
+	io.ReadCloser
+	cancel context.CancelFunc
+}
+
+func (b *cancelOnCloseBody) Close() error {
+	err := b.ReadCloser.Close()
+	b.cancel()
+	return err
+}
+
 func (t *http3FallbackTransport) h2FallbackRoundTrip(request *http.Request) (*http.Response, error) {
 	if fallback, isFallback := t.h2Fallback.(*http2FallbackTransport); isFallback {
 		return fallback.roundTrip(request, true)

protocol/cloudflare/inbound.go

@@ -53,7 +53,7 @@ func NewInbound(ctx context.Context, router adapter.Router, logger log.ContextLo
 
 	service, err := cloudflared.NewService(cloudflared.ServiceOptions{
 		Logger:           logger,
-		ConnectionDialer: &routerDialer{router: router, tag: tag},
+		ConnectionDialer: &routerDialer{ctx: ctx, router: router, tag: tag},
 		ControlDialer:    controlDialer,
 		TunnelDialer:     tunnelDialer,
 		ControlResolver:  controlResolver,
@@ -101,10 +101,82 @@ func (i *Inbound) Close() error {
 }
 
 type routerDialer struct {
+	ctx    context.Context
 	router adapter.Router
 	tag    string
 }
 
+// cloudflared cancels the dial context as soon as DialContext/ListenPacket returns
+// (router_pipe.go dialRouterTCPWithMetadata, origin_dial.go dialWarpPacketConnection), while the
+// routing and the outbound dial behind the pipe are still in progress, so only its deadline is
+// honoured, and only until the outbound handshake is reported.
+func (d *routerDialer) routeContext(ctx context.Context) (context.Context, func(), func()) {
+	routeCtx, cancel := context.WithCancelCause(context.WithoutCancel(ctx))
+	stopShutdown := context.AfterFunc(d.ctx, func() {
+		cancel(context.Cause(d.ctx))
+	})
+	var connectTimeout *time.Timer
+	if deadline, hasDeadline := ctx.Deadline(); hasDeadline {
+		connectTimeout = time.AfterFunc(time.Until(deadline), func() {
+			cancel(context.DeadlineExceeded)
+		})
+	}
+	connected := func() {
+		if connectTimeout != nil {
+			connectTimeout.Stop()
+		}
+	}
+	closed := func() {
+		connected()
+		stopShutdown()
+	}
+	return routeCtx, connected, closed
+}
+
+type routedConn struct {
+	net.Conn
+	connected func()
+}
+
+func (c *routedConn) HandshakeSuccess() error {
+	c.connected()
+	return nil
+}
+
+func (c *routedConn) ReaderReplaceable() bool {
+	return true
+}
+
+func (c *routedConn) WriterReplaceable() bool {
+	return true
+}
+
+func (c *routedConn) Upstream() any {
+	return c.Conn
+}
+
+type routedPacketConn struct {
+	N.PacketConn
+	connected func()
+}
+
+func (c *routedPacketConn) HandshakeSuccess() error {
+	c.connected()
+	return nil
+}
+
+func (c *routedPacketConn) ReaderReplaceable() bool {
+	return true
+}
+
+func (c *routedPacketConn) WriterReplaceable() bool {
+	return true
+}
+
+func (c *routedPacketConn) Upstream() any {
+	return c.PacketConn
+}
+
 func (d *routerDialer) newMetadata(network string, destination M.Socksaddr) adapter.InboundContext {
 	return adapter.InboundContext{
 		Inbound:     d.tag,
@@ -116,7 +188,9 @@ func (d *routerDialer) newMetadata(network string, destination M.Socksaddr) adap
 
 func (d *routerDialer) DialContext(ctx context.Context, network string, destination M.Socksaddr) (net.Conn, error) {
 	input, output := pipe.Pipe()
-	go d.router.RouteConnectionEx(ctx, output, d.newMetadata(N.NetworkTCP, destination), N.OnceClose(func(it error) {
+	routeCtx, connected, closed := d.routeContext(ctx)
+	go d.router.RouteConnectionEx(routeCtx, &routedConn{Conn: output, connected: connected}, d.newMetadata(N.NetworkTCP, destination), N.OnceClose(func(it error) {
+		closed()
 		input.Close()
 	}))
 	return input, nil
@@ -124,8 +198,10 @@ func (d *routerDialer) DialContext(ctx context.Context, network string, destinat
 
 func (d *routerDialer) ListenPacket(ctx context.Context, destination M.Socksaddr) (net.PacketConn, error) {
 	input, output := pipe.Pipe()
-	routerConn := bufio.NewUnbindPacketConn(output)
-	go d.router.RoutePacketConnectionEx(ctx, routerConn, d.newMetadata(N.NetworkUDP, destination), N.OnceClose(func(it error) {
+	routeCtx, connected, closed := d.routeContext(ctx)
+	routerConn := &routedPacketConn{PacketConn: bufio.NewUnbindPacketConn(output), connected: connected}
+	go d.router.RoutePacketConnectionEx(routeCtx, routerConn, d.newMetadata(N.NetworkUDP, destination), N.OnceClose(func(it error) {
+		closed()
 		input.Close()
 	}))
 	return bufio.NewUnbindPacketConn(input), nil
@@ -178,3 +254,12 @@ func (h *icmpRouterHandler) RouteICMPFlow(source netip.Addr, destination netip.A
 		return nil, E.New("no direct route")
 	}
 }
+
+var (
+	_ N.HandshakeSuccess   = (*routedConn)(nil)
+	_ N.ReaderWithUpstream = (*routedConn)(nil)
+	_ N.WriterWithUpstream = (*routedConn)(nil)
+	_ N.HandshakeSuccess   = (*routedPacketConn)(nil)
+	_ N.ReaderWithUpstream = (*routedPacketConn)(nil)
+	_ N.WriterWithUpstream = (*routedPacketConn)(nil)
+)

protocol/group/urltest.go

@@ -464,6 +464,9 @@ func (b *urlTestBatch) test(outbounds []adapter.Outbound, link string, interval
 					testResult.err = testCtx.Err()
 				}
 				if testResult.err != nil {
+					if b.ctx.Err() != nil {
+						return nil, nil
+					}
 					b.logger.Debug("outbound ", tag, " unavailable: ", testResult.err)
 					b.history.DeleteURLTestHistory(tag)
 				} else {

route/network.go

@@ -534,10 +534,7 @@ func (r *NetworkManager) notifyInterfaceUpdate(defaultInterface *control.Interfa
 	if previousCancel != nil {
 		previousCancel()
 	}
-	go func() {
-		defer updateCancel()
-		r.updateInterface(updateContext, defaultInterface)
-	}()
+	go r.updateInterface(updateContext, defaultInterface)
 }
 
 func (r *NetworkManager) updateInterface(ctx context.Context, defaultInterface *control.Interface) {
@@ -606,7 +603,6 @@ func (r *NetworkManager) notifyWindowsPowerEvent(event int) {
 			previousCancel()
 		}
 		go func() {
-			defer updateCancel()
 			r.resetRunAccess.Lock()
 			defer r.resetRunAccess.Unlock()
 			if updateContext.Err() != nil {

transport/v2raygrpc/client.go

@@ -102,12 +102,19 @@ func (c *Client) DialContext(ctx context.Context) (net.Conn, error) {
 		return nil, err
 	}
 	client := NewGunServiceClient(clientConn).(GunServiceCustomNameClient)
-	ctx, cancel := context.WithCancelCause(ctx)
-	stream, err := client.TunCustomName(ctx, c.serviceName)
+	streamCtx, cancel := context.WithCancelCause(context.WithoutCancel(ctx))
+	stopPropagation := context.AfterFunc(ctx, func() {
+		cancel(context.Cause(ctx))
+	})
+	stream, err := client.TunCustomName(streamCtx, c.serviceName)
 	if err != nil {
+		stopPropagation()
 		cancel(err)
 		return nil, err
 	}
+	if !stopPropagation() {
+		return nil, context.Cause(ctx)
+	}
 	return NewGRPCConn(stream, cancel), nil
 }
 

transport/v2raygrpclite/client.go

@@ -5,11 +5,15 @@ import (
 	"io"
 	"net"
 	"net/http"
+	"net/http/httptrace"
 	"net/url"
+	"os"
+	"sync/atomic"
 	"time"
 
 	"github.com/sagernet/sing-box/adapter"
 	"github.com/sagernet/sing-box/common/tls"
+	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing-box/option"
 	"github.com/sagernet/sing-box/transport/v2rayhttp"
 	E "github.com/sagernet/sing/common/exceptions"
@@ -79,6 +83,7 @@ func NewClient(ctx context.Context, dialer N.Dialer, serverAddr M.Socksaddr, opt
 
 func (c *Client) DialContext(ctx context.Context) (net.Conn, error) {
 	pipeInReader, pipeInWriter := io.Pipe()
+	requestCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
 	request := &http.Request{
 		Method: http.MethodPost,
 		Body:   pipeInReader,
@@ -86,11 +91,30 @@ func (c *Client) DialContext(ctx context.Context) (net.Conn, error) {
 		Header: defaultClientHeader,
 		Host:   c.host,
 	}
-	request = request.WithContext(ctx)
-	conn := newLateGunConn(pipeInWriter)
+	conn := newLateGunConn(pipeInWriter, cancel)
+	handshakeTimeout := C.TCPTimeout
+	if deadline, hasDeadline := ctx.Deadline(); hasDeadline {
+		handshakeTimeout = time.Until(deadline)
+	}
+	var handshakeTimedOut atomic.Bool
+	handshakeTimer := time.AfterFunc(handshakeTimeout, func() {
+		handshakeTimedOut.Store(true)
+		cancel()
+	})
+	// gRPC servers send the response headers together with the first message, so RoundTrip
+	// returning does not mark the stream as established; the request headers being written does.
+	request = request.WithContext(httptrace.WithClientTrace(requestCtx, &httptrace.ClientTrace{
+		WroteHeaders: func() {
+			handshakeTimer.Stop()
+		},
+	}))
 	go func() {
 		response, err := c.transport.RoundTrip(request)
+		handshakeTimer.Stop()
 		if err != nil {
+			if handshakeTimedOut.Load() {
+				err = os.ErrDeadlineExceeded
+			}
 			conn.setup(nil, err)
 		} else if response.StatusCode != 200 {
 			response.Body.Close()

transport/v2raygrpclite/conn.go

@@ -2,6 +2,7 @@ package v2raygrpclite
 
 import (
 	std_bufio "bufio"
+	"context"
 	"encoding/binary"
 	"io"
 	"net"
@@ -27,6 +28,7 @@ type GunConn struct {
 	flusher       http.Flusher
 	create        chan struct{}
 	err           error
+	cancel        context.CancelFunc
 	readRemaining int
 }
 
@@ -39,10 +41,11 @@ func newGunConn(reader io.Reader, writer io.Writer, flusher http.Flusher) *GunCo
 	}
 }
 
-func newLateGunConn(writer io.Writer) *GunConn {
+func newLateGunConn(writer io.Writer, cancel context.CancelFunc) *GunConn {
 	return &GunConn{
 		create: make(chan struct{}),
 		writer: writer,
+		cancel: cancel,
 	}
 }
 
@@ -141,7 +144,11 @@ func (c *GunConn) FrontHeadroom() int {
 }
 
 func (c *GunConn) Close() error {
-	return common.Close(c.rawReader, c.writer)
+	err := common.Close(c.rawReader, c.writer)
+	if c.cancel != nil {
+		c.cancel()
+	}
+	return err
 }
 
 func (c *GunConn) LocalAddr() net.Addr {

transport/v2rayhttp/client.go

@@ -7,11 +7,14 @@ import (
 	"net"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/sagernet/sing-box/adapter"
 	"github.com/sagernet/sing-box/common/tls"
+	C "github.com/sagernet/sing-box/constant"
 	"github.com/sagernet/sing-box/option"
 	E "github.com/sagernet/sing/common/exceptions"
 	M "github.com/sagernet/sing/common/metadata"
@@ -120,13 +123,14 @@ func (c *Client) dialHTTP(ctx context.Context) (net.Conn, error) {
 
 func (c *Client) dialHTTP2(ctx context.Context) (net.Conn, error) {
 	pipeInReader, pipeInWriter := io.Pipe()
+	requestCtx, cancel := context.WithCancel(context.WithoutCancel(ctx))
 	request := &http.Request{
 		Method: c.method,
 		Body:   pipeInReader,
 		URL:    &c.requestURL,
 		Header: c.headers.Clone(),
 	}
-	request = request.WithContext(ctx)
+	request = request.WithContext(requestCtx)
 	switch hostLen := len(c.host); hostLen {
 	case 0:
 		// https://github.com/v2fly/v2ray-core/blob/master/transport/internet/http/config.go#L13
@@ -136,10 +140,23 @@ func (c *Client) dialHTTP2(ctx context.Context) (net.Conn, error) {
 	default:
 		request.Host = c.host[rand.Intn(hostLen)]
 	}
-	conn := NewLateHTTPConn(pipeInWriter)
+	conn := NewLateHTTPConn(pipeInWriter, cancel)
+	handshakeTimeout := C.TCPTimeout
+	if deadline, hasDeadline := ctx.Deadline(); hasDeadline {
+		handshakeTimeout = time.Until(deadline)
+	}
+	var handshakeTimedOut atomic.Bool
+	handshakeTimer := time.AfterFunc(handshakeTimeout, func() {
+		handshakeTimedOut.Store(true)
+		cancel()
+	})
 	go func() {
 		response, err := c.transport.RoundTrip(request)
+		handshakeTimer.Stop()
 		if err != nil {
+			if handshakeTimedOut.Load() {
+				err = os.ErrDeadlineExceeded
+			}
 			conn.Setup(nil, err)
 		} else if response.StatusCode != 200 {
 			response.Body.Close()

transport/v2rayhttp/conn.go

@@ -134,6 +134,7 @@ type HTTP2Conn struct {
 	writer io.Writer
 	create chan struct{}
 	err    error
+	cancel context.CancelFunc
 }
 
 func NewHTTPConn(reader io.Reader, writer io.Writer) HTTP2Conn {
@@ -143,10 +144,11 @@ func NewHTTPConn(reader io.Reader, writer io.Writer) HTTP2Conn {
 	}
 }
 
-func NewLateHTTPConn(writer io.Writer) *HTTP2Conn {
+func NewLateHTTPConn(writer io.Writer, cancel context.CancelFunc) *HTTP2Conn {
 	return &HTTP2Conn{
 		create: make(chan struct{}),
 		writer: writer,
+		cancel: cancel,
 	}
 }
 
@@ -173,7 +175,11 @@ func (c *HTTP2Conn) Write(b []byte) (n int, err error) {
 }
 
 func (c *HTTP2Conn) Close() error {
-	return common.Close(c.reader, c.writer)
+	err := common.Close(c.reader, c.writer)
+	if c.cancel != nil {
+		c.cancel()
+	}
+	return err
 }
 
 func (c *HTTP2Conn) LocalAddr() net.Addr {