From 6911ecd304ff19aa1fb5b6d32c89476854a04917 Mon Sep 17 00:00:00 2001 From: lueyoung Date: Sun, 4 Oct 2026 08:48:41 -0400 Subject: [PATCH] fix(mtls,mtcp,mws): let concurrent redials converge on one session When no live session exists, each request dials a conn of its own and registers it in Dial, then builds the session on it in Handshake, each step under sessionMutex. A Dial that finds a registration still waiting for its handshake takes it for dead (IsClosed reports true while session is nil) and replaces it, so the earlier request's Handshake no longer finds its own conn and fails with "unrecognized connection". Since c210b74 the replacing Dial also closes that conn. Once requests queue for the lock, another request's Dial can sit between every Dial and its Handshake, and no session is completed for as long as requests keep arriving. Seen on a three-hop relay chain whose last hop is mtls: after the client host woke from sleep, nearly every request on a chain with ~28 concurrent clients failed this way for 80 minutes, until the load happened to drop. Dial now closes the base conn only of a session that was established and died; a session still waiting for its handshake belongs to the request that dialed it, whose Handshake either builds the session on it or closes it. Handshake no longer fails on a mismatch: if the registered session is live it uses that and closes its own conn, otherwise it builds the session on its own conn and registers it, and the other request's Handshake then finds a live session to use. The tests dial twice before either handshake, in both orders, against the package's own listener. On master they fail with "unrecognized connection", and with "use of closed network connection" when only Handshake is changed. Co-Authored-By: Claude Opus 5.5 --- dialer/mtcp/dialer.go | 26 +++++--- dialer/mtcp/dialer_test.go | 93 ++++++++++++++++++++++++++++ dialer/mtls/dialer.go | 26 +++++--- dialer/mtls/dialer_test.go | 123 +++++++++++++++++++++++++++++++++++++ dialer/mws/dialer.go | 28 ++++++--- dialer/mws/dialer_test.go | 93 ++++++++++++++++++++++++++++ 6 files changed, 366 insertions(+), 23 deletions(-) create mode 100644 dialer/mtcp/dialer_test.go create mode 100644 dialer/mtls/dialer_test.go create mode 100644 dialer/mws/dialer_test.go diff --git a/dialer/mtcp/dialer.go b/dialer/mtcp/dialer.go index 496e420..449c13d 100644 --- a/dialer/mtcp/dialer.go +++ b/dialer/mtcp/dialer.go @@ -2,7 +2,6 @@ package mtcp import ( "context" - "errors" "net" "sync" "time" @@ -61,11 +60,17 @@ func (d *mtcpDialer) Dial(ctx context.Context, addr string, opts ...dialer.DialO session, ok := d.sessions[addr] if session != nil && session.IsClosed() { - session.Close() - if session.conn != nil { - session.conn.Close() // base conn would otherwise leak (and hold a p2p tunnel open) + // A session still waiting for its handshake is not dead: its conn + // belongs to the request that dialed it, whose Handshake either + // builds the session on it or closes it. Only an established + // session's base conn is closed here. + if session.session != nil { + session.Close() + if session.conn != nil { + session.conn.Close() // base conn would otherwise leak (and hold a p2p tunnel open) + } } - delete(d.sessions, addr) // session is dead + delete(d.sessions, addr) ok = false } if !ok { @@ -118,8 +123,15 @@ func (d *mtcpDialer) Handshake(ctx context.Context, conn net.Conn, options ...di session, ok := d.sessions[opts.Addr] if session != nil && session.conn != conn { - conn.Close() - return nil, errors.New("mtcp: unrecognized connection") + // Another Dial registered a conn of its own while this one waited + // for the lock. If it has become a live session, use that and drop + // this conn; otherwise build the session on this conn, and the other + // request's Handshake will find it here and use it. + if session.IsClosed() { + ok = false + } else { + conn.Close() + } } if !ok || session.session == nil { diff --git a/dialer/mtcp/dialer_test.go b/dialer/mtcp/dialer_test.go new file mode 100644 index 0000000..30869c6 --- /dev/null +++ b/dialer/mtcp/dialer_test.go @@ -0,0 +1,93 @@ +package mtcp + +import ( + "context" + "io" + "net" + "testing" + + "github.com/go-gost/core/dialer" + "github.com/go-gost/core/listener" + mtcp_listener "github.com/go-gost/x/listener/mtcp" + xlogger "github.com/go-gost/x/logger" + xmd "github.com/go-gost/x/metadata" +) + +type netDialer struct{} + +func (netDialer) Dial(ctx context.Context, network, addr string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, network, addr) +} + +// echoServer serves mtcp on a local port and echoes every stream. +func echoServer(t *testing.T) string { + t.Helper() + ln := mtcp_listener.NewListener( + listener.AddrOption("127.0.0.1:0"), + listener.LoggerOption(xlogger.Nop()), + ) + if err := ln.Init(xmd.NewMetadata(nil)); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go func() { + defer conn.Close() + io.Copy(conn, conn) + }() + } + }() + return ln.Addr().String() +} + +// Concurrent requests that find no live session each dial a conn of their +// own, and a Dial that finds a session still waiting for its handshake +// replaces it. Both handshakes must still end on one usable session, +// whichever runs first; they used to fail with "unrecognized connection", +// or on a conn the later Dial had closed. +func TestHandshakeAfterAnotherDial(t *testing.T) { + for _, first := range []string{"earlier", "later"} { + t.Run(first+" conn first", func(t *testing.T) { + addr := echoServer(t) + d := NewDialer( + dialer.LoggerOption(xlogger.Nop()), + ) + if err := d.Init(xmd.NewMetadata(nil)); err != nil { + t.Fatal(err) + } + ctx := context.Background() + + earlier, err := d.Dial(ctx, addr, dialer.NetDialerDialOption(netDialer{})) + if err != nil { + t.Fatal(err) + } + later, err := d.Dial(ctx, addr, dialer.NetDialerDialOption(netDialer{})) + if err != nil { + t.Fatal(err) + } + conns := []net.Conn{earlier, later} + if first == "later" { + conns = []net.Conn{later, earlier} + } + for i, conn := range conns { + cc, err := d.(dialer.Handshaker).Handshake(ctx, conn, dialer.AddrHandshakeOption(addr)) + if err != nil { + t.Fatalf("handshake %d: %v", i+1, err) + } + if _, err := cc.Write([]byte("ping")); err != nil { + t.Fatalf("handshake %d: write: %v", i+1, err) + } + if _, err := io.ReadFull(cc, make([]byte, 4)); err != nil { + t.Fatalf("handshake %d: read: %v", i+1, err) + } + cc.Close() + } + }) + } +} diff --git a/dialer/mtls/dialer.go b/dialer/mtls/dialer.go index 3d08064..ae1f615 100644 --- a/dialer/mtls/dialer.go +++ b/dialer/mtls/dialer.go @@ -3,7 +3,6 @@ package mtls import ( "context" "crypto/tls" - "errors" "net" "sync" "time" @@ -62,11 +61,17 @@ func (d *mtlsDialer) Dial(ctx context.Context, addr string, opts ...dialer.DialO session, ok := d.sessions[addr] if session != nil && session.IsClosed() { - session.Close() - if session.conn != nil { - session.conn.Close() // base conn would otherwise leak (and hold a p2p tunnel open) + // A session still waiting for its handshake is not dead: its conn + // belongs to the request that dialed it, whose Handshake either + // builds the session on it or closes it. Only an established + // session's base conn is closed here. + if session.session != nil { + session.Close() + if session.conn != nil { + session.conn.Close() // base conn would otherwise leak (and hold a p2p tunnel open) + } } - delete(d.sessions, addr) // session is dead + delete(d.sessions, addr) ok = false } if !ok { @@ -119,8 +124,15 @@ func (d *mtlsDialer) Handshake(ctx context.Context, conn net.Conn, options ...di session, ok := d.sessions[opts.Addr] if session != nil && session.conn != conn { - conn.Close() - return nil, errors.New("mtls: unrecognized connection") + // Another Dial registered a conn of its own while this one waited + // for the lock. If it has become a live session, use that and drop + // this conn; otherwise build the session on this conn, and the other + // request's Handshake will find it here and use it. + if session.IsClosed() { + ok = false + } else { + conn.Close() + } } if !ok || session.session == nil { diff --git a/dialer/mtls/dialer_test.go b/dialer/mtls/dialer_test.go new file mode 100644 index 0000000..34f906f --- /dev/null +++ b/dialer/mtls/dialer_test.go @@ -0,0 +1,123 @@ +package mtls + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "io" + "math/big" + "net" + "testing" + "time" + + "github.com/go-gost/core/dialer" + "github.com/go-gost/core/listener" + mtls_listener "github.com/go-gost/x/listener/mtls" + xlogger "github.com/go-gost/x/logger" + xmd "github.com/go-gost/x/metadata" +) + +type netDialer struct{} + +func (netDialer) Dial(ctx context.Context, network, addr string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, network, addr) +} + +// echoServer serves mtls on a local port and echoes every stream. +func echoServer(t *testing.T) string { + t.Helper() + ln := mtls_listener.NewListener( + listener.AddrOption("127.0.0.1:0"), + listener.TLSConfigOption(&tls.Config{Certificates: []tls.Certificate{selfSignedCert(t)}}), + listener.LoggerOption(xlogger.Nop()), + ) + if err := ln.Init(xmd.NewMetadata(nil)); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go func() { + defer conn.Close() + io.Copy(conn, conn) + }() + } + }() + return ln.Addr().String() +} + +func selfSignedCert(t *testing.T) tls.Certificate { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: "127.0.0.1"}, + IPAddresses: []net.IP{net.IPv4(127, 0, 0, 1)}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + return tls.Certificate{Certificate: [][]byte{der}, PrivateKey: key} +} + +// Concurrent requests that find no live session each dial a conn of their +// own, and a Dial that finds a session still waiting for its handshake +// replaces it. Both handshakes must still end on one usable session, +// whichever runs first; they used to fail with "unrecognized connection", +// or on a conn the later Dial had closed. +func TestHandshakeAfterAnotherDial(t *testing.T) { + for _, first := range []string{"earlier", "later"} { + t.Run(first+" conn first", func(t *testing.T) { + addr := echoServer(t) + d := NewDialer( + dialer.TLSConfigOption(&tls.Config{InsecureSkipVerify: true}), + dialer.LoggerOption(xlogger.Nop()), + ) + if err := d.Init(xmd.NewMetadata(nil)); err != nil { + t.Fatal(err) + } + ctx := context.Background() + + earlier, err := d.Dial(ctx, addr, dialer.NetDialerDialOption(netDialer{})) + if err != nil { + t.Fatal(err) + } + later, err := d.Dial(ctx, addr, dialer.NetDialerDialOption(netDialer{})) + if err != nil { + t.Fatal(err) + } + conns := []net.Conn{earlier, later} + if first == "later" { + conns = []net.Conn{later, earlier} + } + for i, conn := range conns { + cc, err := d.(dialer.Handshaker).Handshake(ctx, conn, dialer.AddrHandshakeOption(addr)) + if err != nil { + t.Fatalf("handshake %d: %v", i+1, err) + } + if _, err := cc.Write([]byte("ping")); err != nil { + t.Fatalf("handshake %d: write: %v", i+1, err) + } + if _, err := io.ReadFull(cc, make([]byte, 4)); err != nil { + t.Fatalf("handshake %d: read: %v", i+1, err) + } + cc.Close() + } + }) + } +} diff --git a/dialer/mws/dialer.go b/dialer/mws/dialer.go index 05f8876..4ceee52 100644 --- a/dialer/mws/dialer.go +++ b/dialer/mws/dialer.go @@ -2,7 +2,6 @@ package mws import ( "context" - "errors" "net" "net/url" "sync" @@ -76,11 +75,17 @@ func (d *mwsDialer) Dial(ctx context.Context, addr string, opts ...dialer.DialOp session, ok := d.sessions[addr] if session != nil && session.IsClosed() { - session.Close() - if session.conn != nil { - session.conn.Close() // base conn would otherwise leak (and hold a p2p tunnel open) + // A session still waiting for its handshake is not dead: its conn + // belongs to the request that dialed it, whose Handshake either + // builds the session on it or closes it. Only an established + // session's base conn is closed here. + if session.session != nil { + session.Close() + if session.conn != nil { + session.conn.Close() // base conn would otherwise leak (and hold a p2p tunnel open) + } } - delete(d.sessions, addr) // session is dead + delete(d.sessions, addr) ok = false } if !ok { @@ -133,10 +138,15 @@ func (d *mwsDialer) Handshake(ctx context.Context, conn net.Conn, options ...dia session, ok := d.sessions[opts.Addr] if session != nil && session.conn != conn { - err := errors.New("mws: unrecognized connection") - log.Error(err) - conn.Close() - return nil, err + // Another Dial registered a conn of its own while this one waited + // for the lock. If it has become a live session, use that and drop + // this conn; otherwise build the session on this conn, and the other + // request's Handshake will find it here and use it. + if session.IsClosed() { + ok = false + } else { + conn.Close() + } } if !ok || session.session == nil { diff --git a/dialer/mws/dialer_test.go b/dialer/mws/dialer_test.go new file mode 100644 index 0000000..be9a626 --- /dev/null +++ b/dialer/mws/dialer_test.go @@ -0,0 +1,93 @@ +package mws + +import ( + "context" + "io" + "net" + "testing" + + "github.com/go-gost/core/dialer" + "github.com/go-gost/core/listener" + mws_listener "github.com/go-gost/x/listener/mws" + xlogger "github.com/go-gost/x/logger" + xmd "github.com/go-gost/x/metadata" +) + +type netDialer struct{} + +func (netDialer) Dial(ctx context.Context, network, addr string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, network, addr) +} + +// echoServer serves mws on a local port and echoes every stream. +func echoServer(t *testing.T) string { + t.Helper() + ln := mws_listener.NewListener( + listener.AddrOption("127.0.0.1:0"), + listener.LoggerOption(xlogger.Nop()), + ) + if err := ln.Init(xmd.NewMetadata(nil)); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { ln.Close() }) + go func() { + for { + conn, err := ln.Accept() + if err != nil { + return + } + go func() { + defer conn.Close() + io.Copy(conn, conn) + }() + } + }() + return ln.Addr().String() +} + +// Concurrent requests that find no live session each dial a conn of their +// own, and a Dial that finds a session still waiting for its handshake +// replaces it. Both handshakes must still end on one usable session, +// whichever runs first; they used to fail with "unrecognized connection", +// or on a conn the later Dial had closed. +func TestHandshakeAfterAnotherDial(t *testing.T) { + for _, first := range []string{"earlier", "later"} { + t.Run(first+" conn first", func(t *testing.T) { + addr := echoServer(t) + d := NewDialer( + dialer.LoggerOption(xlogger.Nop()), + ) + if err := d.Init(xmd.NewMetadata(nil)); err != nil { + t.Fatal(err) + } + ctx := context.Background() + + earlier, err := d.Dial(ctx, addr, dialer.NetDialerDialOption(netDialer{})) + if err != nil { + t.Fatal(err) + } + later, err := d.Dial(ctx, addr, dialer.NetDialerDialOption(netDialer{})) + if err != nil { + t.Fatal(err) + } + conns := []net.Conn{earlier, later} + if first == "later" { + conns = []net.Conn{later, earlier} + } + for i, conn := range conns { + cc, err := d.(dialer.Handshaker).Handshake(ctx, conn, dialer.AddrHandshakeOption(addr)) + if err != nil { + t.Fatalf("handshake %d: %v", i+1, err) + } + if _, err := cc.Write([]byte("ping")); err != nil { + t.Fatalf("handshake %d: write: %v", i+1, err) + } + if _, err := io.ReadFull(cc, make([]byte, 4)); err != nil { + t.Fatalf("handshake %d: read: %v", i+1, err) + } + cc.Close() + } + }) + } +}