From 9be5fcf4e165d369006dbc9908202ab24fc1cfc3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E4=B8=96=E7=95=8C?= Date: Sat, 19 Sep 2026 22:15:13 +0800 Subject: [PATCH] Fix dial contexts cancelled while connections are still in use --- common/dialer/tfo.go | 15 ++++- common/httpclient/http3_transport.go | 83 +++++++++++++++---------- protocol/cloudflare/inbound.go | 93 ++++++++++++++++++++++++++-- protocol/group/urltest.go | 3 + route/network.go | 6 +- transport/v2raygrpc/client.go | 11 +++- transport/v2raygrpclite/client.go | 28 ++++++++- transport/v2raygrpclite/conn.go | 11 +++- transport/v2rayhttp/client.go | 21 ++++++- transport/v2rayhttp/conn.go | 10 ++- 10 files changed, 229 insertions(+), 52 deletions(-) diff --git a/common/dialer/tfo.go b/common/dialer/tfo.go index 90b7b65f..fec2aaf2 100644 --- a/common/dialer/tfo.go +++ b/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() diff --git a/common/httpclient/http3_transport.go b/common/httpclient/http3_transport.go index 7b94560c..e312e519 100644 --- a/common/httpclient/http3_transport.go +++ b/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) diff --git a/protocol/cloudflare/inbound.go b/protocol/cloudflare/inbound.go index 6902b006..e190d37e 100644 --- a/protocol/cloudflare/inbound.go +++ b/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) +) diff --git a/protocol/group/urltest.go b/protocol/group/urltest.go index ba3197fb..1bb6b113 100644 --- a/protocol/group/urltest.go +++ b/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 { diff --git a/route/network.go b/route/network.go index ae950e44..9d74fed3 100644 --- a/route/network.go +++ b/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 { diff --git a/transport/v2raygrpc/client.go b/transport/v2raygrpc/client.go index 7c5e87ca..293373ae 100644 --- a/transport/v2raygrpc/client.go +++ b/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 } diff --git a/transport/v2raygrpclite/client.go b/transport/v2raygrpclite/client.go index b2aab911..7a42cf41 100644 --- a/transport/v2raygrpclite/client.go +++ b/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() diff --git a/transport/v2raygrpclite/conn.go b/transport/v2raygrpclite/conn.go index 5feafbb6..b1f45222 100644 --- a/transport/v2raygrpclite/conn.go +++ b/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 { diff --git a/transport/v2rayhttp/client.go b/transport/v2rayhttp/client.go index 6c327cd6..fdd9210c 100644 --- a/transport/v2rayhttp/client.go +++ b/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() diff --git a/transport/v2rayhttp/conn.go b/transport/v2rayhttp/conn.go index b339a753..5f5cadb9 100644 --- a/transport/v2rayhttp/conn.go +++ b/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 {