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 <noreply@anthropic.com>

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() {
+		// 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 {
+		// 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()
-		return nil, errors.New("mtcp: unrecognized connection")
+		}
 	}
 
 	if !ok || session.session == nil {

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()
+			}
+		})
+	}
+}

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() {
+		// 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 {
+		// 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()
-		return nil, errors.New("mtls: unrecognized connection")
+		}
 	}
 
 	if !ok || session.session == nil {

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()
+			}
+		})
+	}
+}

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() {
+		// 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")
+		// Another Dial registered a conn of its own while this one waited
-		log.Error(err)
+		// 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()
-		return nil, err
+		}
 	}
 
 	if !ok || session.session == nil {

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()
+			}
+		})
+	}
+}