🐛 fix(ssh): 为共享端口转发增加引用计数与失败回滚

2026-08-07 15:13:51 +08:00 · 2026-07-27 00:07:42 +08:00
parent 546b71945d
commit 76249c00b9
28 changed files with 795 additions and 198 deletions
--- a/internal/db/chroma_impl.go
+++ b/internal/db/chroma_impl.go
@@ -56,16 +56,17 @@ type chromaGetResponse struct {
 	Included   []string                 `json:"included"`
 }

-func (c *ChromaDB) Connect(config connection.ConnectionConfig) error {
-	if c.forwarder != nil {
-		_ = c.forwarder.Close()
-		c.forwarder = nil
-	}
-	c.client = nil
+func (c *ChromaDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = c.Close()
+	defer func() {
+		if err != nil {
+			_ = c.Close()
+		}
+	}()

 	runConfig := normalizeChromaConfig(config)
 	if runConfig.UseSSH {
-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -100,7 +101,7 @@ func (c *ChromaDB) Connect(config connection.ConnectionConfig) error {

 func (c *ChromaDB) Close() error {
 	if c.forwarder != nil {
-		if err := c.forwarder.Close(); err != nil {
+		if err := c.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 Chroma SSH 端口转发失败：%v", err)
 		}
 		c.forwarder = nil
--- a/internal/db/clickhouse_impl.go
+++ b/internal/db/clickhouse_impl.go
@@ -701,7 +701,14 @@ func clickHouseProtocolsForAttempt(config connection.ConnectionConfig) []clickho
 	return []clickhouse.Protocol{primaryProtocol, clickhouse.Native}
 }

-func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error {
+func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = c.Close()
+	defer func() {
+		if err != nil {
+			_ = c.Close()
+		}
+	}()
+
 	if supported, reason := DriverRuntimeSupportStatus("clickhouse"); !supported {
 		if strings.TrimSpace(reason) == "" {
 			reason = localizedDriverRuntimeText("driver_manager.backend.status.optional_disabled", map[string]any{"name": "ClickHouse"})
@@ -709,19 +716,6 @@ func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error {
 		return fmt.Errorf("%s", reason)
 	}

-	if c.forwarder != nil {
-		_ = c.forwarder.Close()
-		c.forwarder = nil
-	}
-	if c.conn != nil {
-		_ = c.conn.Close()
-		c.conn = nil
-	}
-	if c.legacyHTTP != nil {
-		_ = c.legacyHTTP.Close()
-		c.legacyHTTP = nil
-	}
-
 	runConfig := normalizeClickHouseConfig(config)
 	c.pingTimeout = getConnectTimeout(runConfig)
 	c.database = runConfig.Database
@@ -734,7 +728,7 @@ func (c *ClickHouseDB) Connect(config connection.ConnectionConfig) error {
 			runConfig.ClickHouseProtocol = clickHouseProtocolHTTP
 		}
 		logger.Infof("ClickHouse 使用 SSH 连接：地址=%s:%d 用户=%s", runConfig.Host, runConfig.Port, runConfig.User)
-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -869,7 +863,7 @@ func (c *ClickHouseDB) connectClickHouseLegacyHTTP(opts *clickhouse.Options) (*c

 func (c *ClickHouseDB) Close() error {
 	if c.forwarder != nil {
-		if err := c.forwarder.Close(); err != nil {
+		if err := c.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 ClickHouse SSH 端口转发失败：%v", err)
 		}
 		c.forwarder = nil
--- a/internal/db/dameng_impl.go
+++ b/internal/db/dameng_impl.go
@@ -60,7 +60,14 @@ func (d *DamengDB) getDSN(config connection.ConnectionConfig) string {
 	return dsn + "?" + encoded
 }

-func (d *DamengDB) Connect(config connection.ConnectionConfig) error {
+func (d *DamengDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = d.Close()
+	defer func() {
+		if err != nil {
+			_ = d.Close()
+		}
+	}()
+
 	runConfig := config
 	if runConfig.UseSSL {
 		if strings.TrimSpace(runConfig.SSLCertPath) == "" || strings.TrimSpace(runConfig.SSLKeyPath) == "" {
@@ -72,7 +79,7 @@ func (d *DamengDB) Connect(config connection.ConnectionConfig) error {
 		// Create SSH tunnel with local port forwarding
 		logger.Infof("达梦数据库使用 SSH 连接：地址=%s:%d 用户=%s", config.Host, config.Port, config.User)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, config.Host, config.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, config.Host, config.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -132,7 +139,7 @@ func (d *DamengDB) Connect(config connection.ConnectionConfig) error {
 func (d *DamengDB) Close() error {
 	// Close SSH forwarder first if exists
 	if d.forwarder != nil {
-		if err := d.forwarder.Close(); err != nil {
+		if err := d.forwarder.Release(); err != nil {
 			logger.Warnf("关闭达梦数据库 SSH 端口转发失败：%v", err)
 		}
 		d.forwarder = nil
--- a/internal/db/elasticsearch_impl.go
+++ b/internal/db/elasticsearch_impl.go
@@ -50,13 +50,13 @@ func (e *esHTTPStatusError) Error() string {
 }

 // Connect 建立到 Elasticsearch 集群的连接。
-func (e *ElasticsearchDB) Connect(config connection.ConnectionConfig) error {
-	// 清理旧连接
-	if e.forwarder != nil {
-		_ = e.forwarder.Close()
-		e.forwarder = nil
-	}
-	e.client = nil
+func (e *ElasticsearchDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = e.Close()
+	defer func() {
+		if err != nil {
+			_ = e.Close()
+		}
+	}()

 	runConfig := normalizeElasticsearchConfig(config)
 	e.pingTimeout = getConnectTimeout(runConfig)
@@ -68,7 +68,7 @@ func (e *ElasticsearchDB) Connect(config connection.ConnectionConfig) error {
 	// SSH 隧道支持
 	if runConfig.UseSSH {
 		logger.Infof("Elasticsearch 使用 SSH 连接：地址=%s:%d", runConfig.Host, runConfig.Port)
-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -132,7 +132,7 @@ func (e *ElasticsearchDB) Connect(config connection.ConnectionConfig) error {
 // Close 关闭 Elasticsearch 连接并释放底层资源。
 func (e *ElasticsearchDB) Close() error {
 	if e.forwarder != nil {
-		if err := e.forwarder.Close(); err != nil {
+		if err := e.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 Elasticsearch SSH 端口转发失败：%v", err)
 		}
 		e.forwarder = nil
--- a/internal/db/gaussdb_impl.go
+++ b/internal/db/gaussdb_impl.go
@@ -102,7 +102,14 @@ func (g *GaussDB) getDSN(config connection.ConnectionConfig) string {
 	return u.String()
 }

-func (g *GaussDB) Connect(config connection.ConnectionConfig) error {
+func (g *GaussDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = g.Close()
+	defer func() {
+		if err != nil {
+			_ = g.Close()
+		}
+	}()
+
 	if supported, reason := DriverRuntimeSupportStatus("gaussdb"); !supported {
 		if strings.TrimSpace(reason) == "" {
 			reason = localizedDriverRuntimeText("driver_manager.backend.status.optional_disabled", map[string]any{"name": "GaussDB"})
@@ -113,25 +120,10 @@ func (g *GaussDB) Connect(config connection.ConnectionConfig) error {
 	runConfig := applyGaussDBURI(config)
 	g.pingTimeout = getConnectTimeout(runConfig)

-	cleanupOnFailure := true
-	defer func() {
-		if !cleanupOnFailure {
-			return
-		}
-		if g.conn != nil {
-			_ = g.conn.Close()
-			g.conn = nil
-		}
-		if g.forwarder != nil {
-			_ = g.forwarder.Close()
-			g.forwarder = nil
-		}
-	}()
-
 	if runConfig.UseSSH {
 		logger.Infof("GaussDB 使用 SSH 连接：地址=%s:%d 用户=%s", runConfig.Host, runConfig.Port, runConfig.User)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -198,7 +190,6 @@ func (g *GaussDB) Connect(config connection.ConnectionConfig) error {

 			g.ensureSearchPath(dsn)

-			cleanupOnFailure = false
 			return nil
 		}
 	}
--- a/internal/db/highgo_impl.go
+++ b/internal/db/highgo_impl.go
@@ -59,13 +59,20 @@ func (h *HighGoDB) getDSN(config connection.ConnectionConfig) string {
 	return u.String()
 }

-func (h *HighGoDB) Connect(config connection.ConnectionConfig) error {
+func (h *HighGoDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = h.Close()
+	defer func() {
+		if err != nil {
+			_ = h.Close()
+		}
+	}()
+
 	runConfig := config

 	if config.UseSSH {
 		logger.Infof("HighGo 使用 SSH 连接：地址=%s:%d 用户=%s", config.Host, config.Port, config.User)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, config.Host, config.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, config.Host, config.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -122,7 +129,7 @@ func (h *HighGoDB) Connect(config connection.ConnectionConfig) error {

 func (h *HighGoDB) Close() error {
 	if h.forwarder != nil {
-		if err := h.forwarder.Close(); err != nil {
+		if err := h.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 HighGo SSH 端口转发失败：%v", err)
 		}
 		h.forwarder = nil
--- a/internal/db/iotdb_impl.go
+++ b/internal/db/iotdb_impl.go
@@ -115,16 +115,17 @@ type IoTDBDB struct {
 	pingTimeout time.Duration
 }

-func (i *IoTDBDB) Connect(config connection.ConnectionConfig) error {
-	if i.forwarder != nil {
-		_ = i.forwarder.Close()
-		i.forwarder = nil
-	}
-	i.session = nil
+func (i *IoTDBDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = i.Close()
+	defer func() {
+		if err != nil {
+			_ = i.Close()
+		}
+	}()

 	runConfig := normalizeIoTDBConfig(config)
 	if runConfig.UseSSH {
-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -160,7 +161,7 @@ func (i *IoTDBDB) Connect(config connection.ConnectionConfig) error {

 func (i *IoTDBDB) Close() error {
 	if i.forwarder != nil {
-		if err := i.forwarder.Close(); err != nil {
+		if err := i.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 IoTDB SSH 端口转发失败：%v", err)
 		}
 		i.forwarder = nil
--- a/internal/db/iris_impl.go
+++ b/internal/db/iris_impl.go
@@ -92,31 +92,23 @@ func (i *IrisDB) getDSN(config connection.ConnectionConfig) string {
 	return u.String()
 }

-func (i *IrisDB) Connect(config connection.ConnectionConfig) error {
+func (i *IrisDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = i.Close()
+	defer func() {
+		if err != nil {
+			_ = i.Close()
+		}
+	}()
+
 	runConfig := applyIRISURI(config)
 	if runConfig.Port <= 0 {
 		runConfig.Port = defaultIRISPort
 	}
 	i.namespace = normalizeIRISNamespace(runConfig.Database)

-	cleanupOnFailure := true
-	defer func() {
-		if !cleanupOnFailure {
-			return
-		}
-		if i.conn != nil {
-			_ = i.conn.Close()
-			i.conn = nil
-		}
-		if i.forwarder != nil {
-			_ = i.forwarder.Close()
-			i.forwarder = nil
-		}
-	}()
-
 	if runConfig.UseSSH {
 		logger.Infof("InterSystems IRIS 使用 SSH 连接：地址=%s:%d 用户=%s", runConfig.Host, runConfig.Port, runConfig.User)
-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -149,13 +141,12 @@ func (i *IrisDB) Connect(config connection.ConnectionConfig) error {
 		i.conn = nil
 		return wrapDatabaseConnectionVerifyError(err)
 	}
-	cleanupOnFailure = false
 	return nil
 }

 func (i *IrisDB) Close() error {
 	if i.forwarder != nil {
-		if err := i.forwarder.Close(); err != nil {
+		if err := i.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 InterSystems IRIS SSH 端口转发失败：%v", err)
 		}
 		i.forwarder = nil
--- a/internal/db/kafka_impl.go
+++ b/internal/db/kafka_impl.go
@@ -163,7 +163,7 @@ func (k *KafkaDB) Close() error {
 		if forwarder == nil {
 			continue
 		}
-		if err := forwarder.Close(); err != nil && firstErr == nil {
+		if err := forwarder.Release(); err != nil && firstErr == nil {
 			firstErr = err
 		}
 	}
@@ -631,19 +631,29 @@ func kafkaForwardBrokersOverSSH(config connection.ConnectionConfig) (connection.
 	}
 	runConfig := config
 	forwarders := make([]*ssh.LocalForwarder, 0, len(brokers))
+	cleanupForwarders := true
+	defer func() {
+		if !cleanupForwarders {
+			return
+		}
+		for _, forwarder := range forwarders {
+			_ = forwarder.Release()
+		}
+	}()
 	rewritten := make([]string, 0, len(brokers))
 	for _, broker := range brokers {
 		host, port, ok := parseHostPortWithDefault(broker, defaultKafkaPort)
 		if !ok {
 			return connection.ConnectionConfig{}, nil, nil, fmt.Errorf("解析 Kafka broker 地址失败：%s", broker)
 		}
-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, host, port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, host, port)
 		if err != nil {
 			return connection.ConnectionConfig{}, nil, nil, fmt.Errorf("创建 Kafka SSH 隧道失败：%w", err)
 		}
 		forwarders = append(forwarders, forwarder)
 		rewritten = append(rewritten, forwarder.LocalAddr)
 	}
+	cleanupForwarders = false
 	return runConfig, rewritten, forwarders, nil
 }

--- a/internal/db/kingbase_impl.go
+++ b/internal/db/kingbase_impl.go
@@ -110,14 +110,21 @@ func (k *KingbaseDB) getDSN(config connection.ConnectionConfig) string {
 	return strings.Join(parts, " ")
 }

-func (k *KingbaseDB) Connect(config connection.ConnectionConfig) error {
+func (k *KingbaseDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = k.Close()
+	defer func() {
+		if err != nil {
+			_ = k.Close()
+		}
+	}()
+
 	runConfig := config

 	if config.UseSSH {
 		// Create SSH tunnel with local port forwarding
 		logger.Infof("人大金仓使用 SSH 连接：地址=%s:%d 用户=%s", config.Host, config.Port, config.User)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, config.Host, config.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, config.Host, config.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -250,7 +257,7 @@ func (k *KingbaseDB) getSearchPathStr() string {
 func (k *KingbaseDB) Close() error {
 	// Close SSH forwarder first if exists
 	if k.forwarder != nil {
-		if err := k.forwarder.Close(); err != nil {
+		if err := k.forwarder.Release(); err != nil {
 			logger.Warnf("关闭人大金仓 SSH 端口转发失败：%v", err)
 		}
 		k.forwarder = nil
--- a/internal/db/milvus_impl.go
+++ b/internal/db/milvus_impl.go
@@ -50,16 +50,17 @@ type MilvusDB struct {
 	forwarder   *ssh.LocalForwarder
 }

-func (m *MilvusDB) Connect(config connection.ConnectionConfig) error {
-	if m.forwarder != nil {
-		_ = m.forwarder.Close()
-		m.forwarder = nil
-	}
-	m.client = nil
+func (m *MilvusDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = m.Close()
+	defer func() {
+		if err != nil {
+			_ = m.Close()
+		}
+	}()

 	runConfig := normalizeMilvusConfig(config)
 	if runConfig.UseSSH {
-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("create Milvus SSH tunnel: %w", err)
 		}
@@ -93,7 +94,7 @@ func (m *MilvusDB) Connect(config connection.ConnectionConfig) error {

 func (m *MilvusDB) Close() error {
 	if m.forwarder != nil {
-		if err := m.forwarder.Close(); err != nil {
+		if err := m.forwarder.Release(); err != nil {
 			logger.Warnf("close Milvus SSH port forwarding failed: %v", err)
 		}
 		m.forwarder = nil
--- a/internal/db/mongodb_impl.go
+++ b/internal/db/mongodb_impl.go
@@ -324,7 +324,14 @@ func mongoAttemptSSLLabel(config connection.ConnectionConfig, fallbackToPlain bo
 	return "明文"
 }

-func (m *MongoDB) Connect(config connection.ConnectionConfig) error {
+func (m *MongoDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = m.Close()
+	defer func() {
+		if err != nil {
+			_ = m.Close()
+		}
+	}()
+
 	runConfig := applyMongoURI(config)
 	connectConfig := runConfig
 	sshRouteHint := ""
@@ -345,7 +352,7 @@ func (m *MongoDB) Connect(config connection.ConnectionConfig) error {

 		logger.Infof("MongoDB 使用 SSH 连接：地址=%s:%d", targetHost, targetPort)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, targetHost, targetPort)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, targetHost, targetPort)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -472,7 +479,7 @@ func (m *MongoDB) Connect(config connection.ConnectionConfig) error {

 func (m *MongoDB) Close() error {
 	if m.forwarder != nil {
-		if err := m.forwarder.Close(); err != nil {
+		if err := m.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 MongoDB SSH 端口转发失败：%v", err)
 		}
 		m.forwarder = nil
--- a/internal/db/mongodb_impl_v1.go
+++ b/internal/db/mongodb_impl_v1.go
@@ -325,7 +325,14 @@ func mongoAttemptSSLLabel(config connection.ConnectionConfig, fallbackToPlain bo
 	return "明文"
 }

-func (m *MongoDBV1) Connect(config connection.ConnectionConfig) error {
+func (m *MongoDBV1) Connect(config connection.ConnectionConfig) (err error) {
+	_ = m.Close()
+	defer func() {
+		if err != nil {
+			_ = m.Close()
+		}
+	}()
+
 	runConfig := applyMongoURI(config)
 	connectConfig := runConfig
 	sshRouteHint := ""
@@ -346,7 +353,7 @@ func (m *MongoDBV1) Connect(config connection.ConnectionConfig) error {

 		logger.Infof("MongoDB 使用 SSH 连接：地址=%s:%d", targetHost, targetPort)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, targetHost, targetPort)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, targetHost, targetPort)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -475,7 +482,7 @@ func (m *MongoDBV1) Connect(config connection.ConnectionConfig) error {

 func (m *MongoDBV1) Close() error {
 	if m.forwarder != nil {
-		if err := m.forwarder.Close(); err != nil {
+		if err := m.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 MongoDB SSH 端口转发失败：%v", err)
 		}
 		m.forwarder = nil
--- a/internal/db/mqtt_impl.go
+++ b/internal/db/mqtt_impl.go
@@ -174,7 +174,7 @@ func (m *MQTTDB) Close() error {
 		if forwarder == nil {
 			continue
 		}
-		if err := forwarder.Close(); err != nil && firstErr == nil {
+		if err := forwarder.Release(); err != nil && firstErr == nil {
 			firstErr = err
 		}
 	}
@@ -708,19 +708,29 @@ func mqttForwardBrokersOverSSH(config connection.ConnectionConfig) (connection.C
 	}
 	runConfig := config
 	forwarders := make([]*ssh.LocalForwarder, 0, len(brokers))
+	cleanupForwarders := true
+	defer func() {
+		if !cleanupForwarders {
+			return
+		}
+		for _, forwarder := range forwarders {
+			_ = forwarder.Release()
+		}
+	}()
 	rewritten := make([]string, 0, len(brokers))
 	for _, broker := range brokers {
 		host, port, ok := parseHostPortWithDefault(broker, defaultMQTTPort)
 		if !ok {
 			return connection.ConnectionConfig{}, nil, nil, fmt.Errorf("解析 MQTT broker 地址失败：%s", broker)
 		}
-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, host, port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, host, port)
 		if err != nil {
 			return connection.ConnectionConfig{}, nil, nil, fmt.Errorf("创建 MQTT SSH 隧道失败：%w", err)
 		}
 		forwarders = append(forwarders, forwarder)
 		rewritten = append(rewritten, forwarder.LocalAddr)
 	}
+	cleanupForwarders = false
 	return runConfig, rewritten, forwarders, nil
 }

--- a/internal/db/oceanbase_impl.go
+++ b/internal/db/oceanbase_impl.go
@@ -568,6 +568,7 @@ func (o *OceanBaseDB) connectOracleViaTNS(config connection.ConnectionConfig) er
 	}
 	oracleDB := &OracleDB{scanDialect: oceanBaseOracleScanDialect}
 	if err := oracleDB.Connect(runConfig); err != nil {
+		_ = oracleDB.Close()
 		return annotateOceanBaseOracleConnectError(err)
 	}
 	o.oracle = oracleDB
@@ -587,6 +588,13 @@ func (o *OceanBaseDB) connectOracleViaOBClient(config connection.ConnectionConfi
 	var errorDetails []string
 	for index, address := range addresses {
 		candidateConfig := config
+		var forwarder *ssh.LocalForwarder
+		releaseForwarder := func() {
+			if forwarder != nil {
+				_ = forwarder.Release()
+				forwarder = nil
+			}
+		}
 		host, port, ok := parseHostPortWithDefault(address, defaultOceanBasePort)
 		if !ok {
 			continue
@@ -596,7 +604,8 @@ func (o *OceanBaseDB) connectOracleViaOBClient(config connection.ConnectionConfi
 		candidateConfig.User, candidateConfig.Password = resolveMySQLCredential(config, index)

 		if candidateConfig.UseSSH {
-			forwarder, err := ssh.GetOrCreateLocalForwarder(candidateConfig.SSH, host, port)
+			var err error
+			forwarder, err = ssh.AcquireLocalForwarder(candidateConfig.SSH, host, port)
 			if err != nil {
 				errorDetails = append(errorDetails, fmt.Sprintf("%s 创建 SSH 本地转发失败：%v", address, err))
 				continue
@@ -604,6 +613,7 @@ func (o *OceanBaseDB) connectOracleViaOBClient(config connection.ConnectionConfi
 			localHost, localPort, ok := parseHostPortWithDefault(forwarder.LocalAddr, defaultOceanBasePort)
 			if !ok {
 				errorDetails = append(errorDetails, fmt.Sprintf("%s 解析 SSH 本地转发地址失败：%s", address, forwarder.LocalAddr))
+				releaseForwarder()
 				continue
 			}
 			candidateConfig.Host = localHost
@@ -614,11 +624,13 @@ func (o *OceanBaseDB) connectOracleViaOBClient(config connection.ConnectionConfi
 		dsn, err := buildOceanBaseOracleOBClientDSN(candidateConfig)
 		if err != nil {
 			errorDetails = append(errorDetails, fmt.Sprintf("%s 生成连接串失败：%v", address, err))
+			releaseForwarder()
 			continue
 		}
 		db, err := sql.Open(oceanbaseOracleOBClientDriver, dsn)
 		if err != nil {
 			errorDetails = append(errorDetails, fmt.Sprintf("%s 打开失败：%v", address, err))
+			releaseForwarder()
 			continue
 		}
 		configureSQLConnectionPool(db, "oceanbase")
@@ -630,10 +642,16 @@ func (o *OceanBaseDB) connectOracleViaOBClient(config connection.ConnectionConfi
 		if pingErr != nil {
 			_ = db.Close()
 			errorDetails = append(errorDetails, formatOceanBaseOBClientAttemptError(address, pingErr))
+			releaseForwarder()
 			continue
 		}

 		o.bindConnectedDatabase(db, timeout, oceanBaseProtocolOracle)
+		if o.oracle != nil {
+			o.oracle.forwarder = forwarder
+			forwarder = nil
+		}
+		releaseForwarder()
 		return nil
 	}

@@ -670,7 +688,14 @@ func (o *OceanBaseDB) bindConnectedDatabase(db *sql.DB, timeout time.Duration, p
 	o.protocol = oceanBaseProtocolMySQL
 }

-func (o *OceanBaseDB) Connect(config connection.ConnectionConfig) error {
+func (o *OceanBaseDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = o.Close()
+	defer func() {
+		if err != nil {
+			_ = o.Close()
+		}
+	}()
+
 	o.oracle = nil
 	o.conn = nil
 	o.protocol = oceanBaseProtocolMySQL
--- a/internal/db/oracle_impl.go
+++ b/internal/db/oracle_impl.go
@@ -119,7 +119,14 @@ func annotateOracleValidationError(err error) error {
 	return fmt.Errorf("%w（Oracle 连接在验证阶段被服务端关闭或被驱动超时中断；请检查监听端口是否为 Oracle 协议端口、Service Name 是否正确、认证参数如 DBA_PRIVILEGE/AUTH_TYPE 是否匹配）", err)
 }

-func (o *OracleDB) Connect(config connection.ConnectionConfig) error {
+func (o *OracleDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = o.Close()
+	defer func() {
+		if err != nil {
+			_ = o.Close()
+		}
+	}()
+
 	runConfig := config
 	serviceName := strings.TrimSpace(config.Database)
 	if serviceName == "" {
@@ -130,7 +137,7 @@ func (o *OracleDB) Connect(config connection.ConnectionConfig) error {
 		// Create SSH tunnel with local port forwarding
 		logger.Infof("Oracle 使用 SSH 连接：地址=%s:%d 用户=%s", config.Host, config.Port, config.User)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, config.Host, config.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, config.Host, config.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -191,7 +198,7 @@ func (o *OracleDB) Connect(config connection.ConnectionConfig) error {
 func (o *OracleDB) Close() error {
 	// Close SSH forwarder first if exists
 	if o.forwarder != nil {
-		if err := o.forwarder.Close(); err != nil {
+		if err := o.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 Oracle SSH 端口转发失败：%v", err)
 		}
 		o.forwarder = nil
--- a/internal/db/postgres_impl.go
+++ b/internal/db/postgres_impl.go
@@ -79,7 +79,14 @@ func (p *PostgresDB) getDSN(config connection.ConnectionConfig) string {
 	return u.String()
 }

-func (p *PostgresDB) Connect(config connection.ConnectionConfig) error {
+func (p *PostgresDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = p.Close()
+	defer func() {
+		if err != nil {
+			_ = p.Close()
+		}
+	}()
+
 	if supported, reason := DriverRuntimeSupportStatus("postgres"); !supported {
 		if strings.TrimSpace(reason) == "" {
 			reason = localizedDriverRuntimeText("driver_manager.backend.status.optional_disabled", map[string]any{"name": "PostgreSQL"})
@@ -90,26 +97,11 @@ func (p *PostgresDB) Connect(config connection.ConnectionConfig) error {
 	runConfig := config
 	p.pingTimeout = getConnectTimeout(config)

-	cleanupOnFailure := true
-	defer func() {
-		if !cleanupOnFailure {
-			return
-		}
-		if p.conn != nil {
-			_ = p.conn.Close()
-			p.conn = nil
-		}
-		if p.forwarder != nil {
-			_ = p.forwarder.Close()
-			p.forwarder = nil
-		}
-	}()
-
 	if config.UseSSH {
 		// Create SSH tunnel with local port forwarding
 		logger.Infof("PostgreSQL 使用 SSH 连接：地址=%s:%d 用户=%s", config.Host, config.Port, config.User)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, config.Host, config.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, config.Host, config.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -180,7 +172,6 @@ func (p *PostgresDB) Connect(config connection.ConnectionConfig) error {
 			// 设置 search_path，使所有用户 schema 下的表可以不带 schema 前缀访问
 			p.ensureSearchPath(dsn)

-			cleanupOnFailure = false
 			return nil
 		}
 	}
@@ -194,7 +185,7 @@ func (p *PostgresDB) Connect(config connection.ConnectionConfig) error {
 func (p *PostgresDB) Close() error {
 	// Close SSH forwarder first if exists
 	if p.forwarder != nil {
-		if err := p.forwarder.Close(); err != nil {
+		if err := p.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 PostgreSQL SSH 端口转发失败：%v", err)
 		}
 		p.forwarder = nil
--- a/internal/db/qdrant_impl.go
+++ b/internal/db/qdrant_impl.go
@@ -75,16 +75,17 @@ type qdrantCountResponse struct {
 	} `json:"result"`
 }

-func (q *QdrantDB) Connect(config connection.ConnectionConfig) error {
-	if q.forwarder != nil {
-		_ = q.forwarder.Close()
-		q.forwarder = nil
-	}
-	q.client = nil
+func (q *QdrantDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = q.Close()
+	defer func() {
+		if err != nil {
+			_ = q.Close()
+		}
+	}()

 	runConfig := normalizeQdrantConfig(config)
 	if runConfig.UseSSH {
-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -118,7 +119,7 @@ func (q *QdrantDB) Connect(config connection.ConnectionConfig) error {

 func (q *QdrantDB) Close() error {
 	if q.forwarder != nil {
-		if err := q.forwarder.Close(); err != nil {
+		if err := q.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 Qdrant SSH 端口转发失败：%v", err)
 		}
 		q.forwarder = nil
--- a/internal/db/rabbitmq_impl.go
+++ b/internal/db/rabbitmq_impl.go
@@ -41,16 +41,17 @@ type RabbitMQDB struct {
 	forwarder       *ssh.LocalForwarder
 }

-func (r *RabbitMQDB) Connect(config connection.ConnectionConfig) error {
-	if r.forwarder != nil {
-		_ = r.forwarder.Close()
-		r.forwarder = nil
-	}
-	r.client = nil
+func (r *RabbitMQDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = r.Close()
+	defer func() {
+		if err != nil {
+			_ = r.Close()
+		}
+	}()

 	runConfig := normalizeRabbitMQConfig(config)
 	if runConfig.UseSSH {
-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -88,7 +89,7 @@ func (r *RabbitMQDB) Connect(config connection.ConnectionConfig) error {

 func (r *RabbitMQDB) Close() error {
 	if r.forwarder != nil {
-		if err := r.forwarder.Close(); err != nil {
+		if err := r.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 RabbitMQ SSH 端口转发失败：%v", err)
 		}
 		r.forwarder = nil
--- a/internal/db/sqlserver_impl.go
+++ b/internal/db/sqlserver_impl.go
@@ -144,13 +144,20 @@ func (s *SqlServerDB) getDSN(config connection.ConnectionConfig) string {
 	return u.String()
 }

-func (s *SqlServerDB) Connect(config connection.ConnectionConfig) error {
+func (s *SqlServerDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = s.Close()
+	defer func() {
+		if err != nil {
+			_ = s.Close()
+		}
+	}()
+
 	var dsn string

 	if config.UseSSH {
 		logger.Infof("SQL Server 使用 SSH 连接：地址=%s:%d 用户=%s", config.Host, config.Port, config.User)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, config.Host, config.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, config.Host, config.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -195,7 +202,7 @@ func (s *SqlServerDB) Connect(config connection.ConnectionConfig) error {

 func (s *SqlServerDB) Close() error {
 	if s.forwarder != nil {
-		if err := s.forwarder.Close(); err != nil {
+		if err := s.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 SQL Server SSH 端口转发失败：%v", err)
 		}
 		s.forwarder = nil
--- a/internal/db/tdengine_impl.go
+++ b/internal/db/tdengine_impl.go
@@ -56,13 +56,20 @@ func (t *TDengineDB) getDSN(config connection.ConnectionConfig) string {
 	return dsn + "?" + query
 }

-func (t *TDengineDB) Connect(config connection.ConnectionConfig) error {
+func (t *TDengineDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = t.Close()
+	defer func() {
+		if err != nil {
+			_ = t.Close()
+		}
+	}()
+
 	runConfig := config

 	if config.UseSSH {
 		logger.Infof("TDengine 使用 SSH 连接：地址=%s:%d 用户=%s", config.Host, config.Port, config.User)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, config.Host, config.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, config.Host, config.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -118,7 +125,7 @@ func (t *TDengineDB) Connect(config connection.ConnectionConfig) error {

 func (t *TDengineDB) Close() error {
 	if t.forwarder != nil {
-		if err := t.forwarder.Close(); err != nil {
+		if err := t.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 TDengine SSH 端口转发失败：%v", err)
 		}
 		t.forwarder = nil
--- a/internal/db/trino_impl.go
+++ b/internal/db/trino_impl.go
@@ -295,15 +295,19 @@ func buildTrinoDSN(config connection.ConnectionConfig, customClientName string)
 }

 func (t *TrinoDB) Close() error {
+	var firstErr error
 	if t.conn != nil {
 		if err := t.conn.Close(); err != nil {
-			return err
+			firstErr = err
 		}
 		t.conn = nil
 	}
 	if t.forwarder != nil {
-		if err := t.forwarder.Close(); err != nil {
+		if err := t.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 Trino SSH 端口转发失败：%v", err)
+			if firstErr == nil {
+				firstErr = err
+			}
 		}
 		t.forwarder = nil
 	}
@@ -312,7 +316,7 @@ func (t *TrinoDB) Close() error {
 		t.customClientName = ""
 	}
 	t.namespace = ""
-	return nil
+	return firstErr
 }

 func (t *TrinoDB) Connect(config connection.ConnectionConfig) error {
@@ -322,7 +326,7 @@ func (t *TrinoDB) Connect(config connection.ConnectionConfig) error {
 	t.pingTimeout = getConnectTimeout(runConfig)

 	if runConfig.UseSSH {
-		forwarder, err := ssh.GetOrCreateLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(runConfig.SSH, runConfig.Host, runConfig.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
--- a/internal/db/trino_impl_test.go
+++ b/internal/db/trino_impl_test.go
@@ -0,0 +1,65 @@
+//go:build gonavi_full_drivers || gonavi_trino_driver
+
+package db
+
+import (
+	"database/sql"
+	"database/sql/driver"
+	"errors"
+	"sync"
+	"testing"
+)
+
+var (
+	trinoCloseTestDriverOnce sync.Once
+	errTrinoCloseTest        = errors.New("trino close test error")
+)
+
+type trinoCloseTestDriver struct{}
+
+func (trinoCloseTestDriver) Open(string) (driver.Conn, error) {
+	return trinoCloseTestConn{}, nil
+}
+
+type trinoCloseTestConn struct{}
+
+func (trinoCloseTestConn) Prepare(string) (driver.Stmt, error) {
+	return nil, driver.ErrSkip
+}
+
+func (trinoCloseTestConn) Close() error {
+	return errTrinoCloseTest
+}
+
+func (trinoCloseTestConn) Begin() (driver.Tx, error) {
+	return nil, driver.ErrSkip
+}
+
+func TestTrinoCloseCleansStateWhenDatabaseCloseFails(t *testing.T) {
+	const driverName = "gonavi_trino_close_test"
+	trinoCloseTestDriverOnce.Do(func() {
+		sql.Register(driverName, trinoCloseTestDriver{})
+	})
+
+	conn, err := sql.Open(driverName, "")
+	if err != nil {
+		t.Fatalf("open test database: %v", err)
+	}
+	if err := conn.Ping(); err != nil {
+		t.Fatalf("ping test database: %v", err)
+	}
+
+	trino := &TrinoDB{
+		conn:      conn,
+		namespace: "catalog.schema",
+	}
+	if err := trino.Close(); !errors.Is(err, errTrinoCloseTest) {
+		t.Fatalf("Close() error = %v, want %v", err, errTrinoCloseTest)
+	}
+	if trino.conn != nil {
+		t.Fatal("Close() did not clear the database handle after an error")
+	}
+	if trino.namespace != "" {
+		t.Fatalf("Close() namespace = %q, want empty", trino.namespace)
+	}
+}
--- a/internal/db/vastbase_impl.go
+++ b/internal/db/vastbase_impl.go
@@ -50,13 +50,20 @@ func (v *VastbaseDB) getDSN(config connection.ConnectionConfig) string {
 	return u.String()
 }

-func (v *VastbaseDB) Connect(config connection.ConnectionConfig) error {
+func (v *VastbaseDB) Connect(config connection.ConnectionConfig) (err error) {
+	_ = v.Close()
+	defer func() {
+		if err != nil {
+			_ = v.Close()
+		}
+	}()
+
 	runConfig := config

 	if config.UseSSH {
 		logger.Infof("Vastbase 使用 SSH 连接：地址=%s:%d 用户=%s", config.Host, config.Port, config.User)

-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, config.Host, config.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, config.Host, config.Port)
 		if err != nil {
 			return fmt.Errorf("创建 SSH 隧道失败：%w", err)
 		}
@@ -113,7 +120,7 @@ func (v *VastbaseDB) Connect(config connection.ConnectionConfig) error {

 func (v *VastbaseDB) Close() error {
 	if v.forwarder != nil {
-		if err := v.forwarder.Close(); err != nil {
+		if err := v.forwarder.Release(); err != nil {
 			logger.Warnf("关闭 Vastbase SSH 端口转发失败：%v", err)
 		}
 		v.forwarder = nil
--- a/internal/redis/redis_impl.go
+++ b/internal/redis/redis_impl.go
@@ -280,19 +280,20 @@ func sanitizeRedisPassword(password string) string {
 }

 // Connect establishes a connection to Redis
-func (r *RedisClientImpl) Connect(config connection.ConnectionConfig) error {
+func (r *RedisClientImpl) Connect(config connection.ConnectionConfig) (err error) {
+	_ = r.Close()
+	defer func() {
+		if err != nil {
+			_ = r.Close()
+		}
+	}()
+
 	config.Password = sanitizeRedisPassword(config.Password)
 	config.RedisSentinelPassword = sanitizeRedisPassword(config.RedisSentinelPassword)
 	r.config = config
 	if r.config.RedisDB < 0 {
 		r.config.RedisDB = 0
 	}
-	r.forwarder = nil
-	r.client = nil
-	r.singleClient = nil
-	r.clusterClient = nil
-	r.isCluster = false
-
 	seedAddrs, err := buildRedisSeedAddrs(config)
 	if err != nil {
 		return err
@@ -424,7 +425,7 @@ func (r *RedisClientImpl) Connect(config connection.ConnectionConfig) error {

 	addr := seedAddrs[0]
 	if config.UseSSH {
-		forwarder, err := ssh.GetOrCreateLocalForwarder(config.SSH, config.Host, config.Port)
+		forwarder, err := ssh.AcquireLocalForwarder(config.SSH, config.Host, config.Port)
 		if err != nil {
 			return localizedRedisBackendError("redis.backend.error.ssh_tunnel_create_failed", map[string]any{
 				"detail": err.Error(),
@@ -491,17 +492,22 @@ func (r *RedisClientImpl) Connect(config connection.ConnectionConfig) error {

 // Close closes the Redis connection
 func (r *RedisClientImpl) Close() error {
+	var firstErr error
 	if r.client != nil {
-		err := r.client.Close()
-		r.client = nil
-		r.singleClient = nil
-		r.clusterClient = nil
-		r.isCluster = false
-		r.seedAddrs = nil
-		r.forwarder = nil
-		return err
+		firstErr = r.client.Close()
 	}
-	return nil
+	r.client = nil
+	r.singleClient = nil
+	r.clusterClient = nil
+	r.isCluster = false
+	r.seedAddrs = nil
+	if r.forwarder != nil {
+		if err := r.forwarder.Release(); err != nil && firstErr == nil {
+			firstErr = err
+		}
+		r.forwarder = nil
+	}
+	return firstErr
 }

 // Ping tests the connection
--- a/internal/ssh/ssh.go
+++ b/internal/ssh/ssh.go
@@ -19,6 +19,7 @@ import (

 	"github.com/go-sql-driver/mysql"
 	"golang.org/x/crypto/ssh"
+	"golang.org/x/sync/singleflight"
 )

 // ViaSSHDialer registers a custom network for MySQL that proxies through SSH
@@ -162,6 +163,8 @@ func DialContextThroughSSH(ctx context.Context, config connection.SSHConfig, net
 var (
 	sshClientCache   = make(map[sshClientCacheKey]*ssh.Client)
 	sshClientCacheMu sync.RWMutex
+	sshClientFlights singleflight.Group
+	connectSSHClient = connectSSH
 	localForwarders  = make(map[forwarderCacheKey]*LocalForwarder)
 	forwarderMu      sync.RWMutex
 )
@@ -219,9 +222,16 @@ type LocalForwarder struct {
 	SSHClient  *ssh.Client
 	listener   net.Listener
 	closeChan  chan struct{}
-	closeOnce  sync.Once // 防止重复关闭
-	closed     bool      // 关闭状态标记
+	closeOnce  sync.Once
+	closed     bool
 	closedMu   sync.RWMutex
+
+	// shared/cacheKey identify a lease returned by AcquireLocalForwarder.
+	// The cached forwarder itself keeps shared nil and owns the listener.
+	shared    *LocalForwarder
+	cacheKey  forwarderCacheKey
+	leaseOnce sync.Once
+	refCount  int // guarded by forwarderMu; meaningful only on the cached forwarder
 }

 // NewLocalForwarder creates a new local port forwarder
@@ -314,8 +324,29 @@ func (f *LocalForwarder) handleConnection(localConn net.Conn) {
 	<-errc
 }

-// Close closes the forwarder (thread-safe, can be called multiple times)
+// Close releases a cached lease, or closes a standalone forwarder created by
+// NewLocalForwarder. It is thread-safe and can be called multiple times.
 func (f *LocalForwarder) Close() error {
+	if f == nil {
+		return nil
+	}
+	if f.shared != nil {
+		var err error
+		f.leaseOnce.Do(func() {
+			err = releaseLocalForwarder(f.cacheKey, f.shared)
+		})
+		return err
+	}
+	return f.closeUnderlying()
+}
+
+// Release releases this acquisition. It is an explicit lifecycle alias for
+// callers that obtained the forwarder through AcquireLocalForwarder.
+func (f *LocalForwarder) Release() error {
+	return f.Close()
+}
+
+func (f *LocalForwarder) closeUnderlying() error {
 	var err error
 	f.closeOnce.Do(func() {
 		f.closedMu.Lock()
@@ -333,13 +364,21 @@ func (f *LocalForwarder) Close() error {

 // IsClosed returns whether the forwarder is closed
 func (f *LocalForwarder) IsClosed() bool {
+	if f == nil {
+		return true
+	}
+	if f.shared != nil {
+		return f.shared.IsClosed()
+	}
 	f.closedMu.RLock()
 	defer f.closedMu.RUnlock()
 	return f.closed
 }

-// GetOrCreateLocalForwarder returns a cached forwarder or creates a new one
-func GetOrCreateLocalForwarder(sshConfig connection.SSHConfig, remoteHost string, remotePort int) (*LocalForwarder, error) {
+// AcquireLocalForwarder acquires a lease on a cached forwarder or creates one.
+// Each successful call must be paired with Release. The shared listener is
+// closed and evicted only after the last lease is released.
+func AcquireLocalForwarder(sshConfig connection.SSHConfig, remoteHost string, remotePort int) (*LocalForwarder, error) {
 	key := forwarderCacheKey{
 		ssh:        newSSHClientCacheKey(sshConfig),
 		remoteHost: remoteHost,
@@ -348,22 +387,15 @@ func GetOrCreateLocalForwarder(sshConfig connection.SSHConfig, remoteHost string
 	logKey := fmt.Sprintf("%s:%d:%s->%s:%d",
 		sshConfig.Host, sshConfig.Port, sshConfig.User, remoteHost, remotePort)

-	forwarderMu.RLock()
-	forwarder, exists := localForwarders[key]
-	forwarderMu.RUnlock()
-
-	// Check if exists and is still valid
-	if exists && forwarder != nil && !forwarder.IsClosed() {
-		logger.Infof("复用已有端口转发：%s", logKey)
-		return forwarder, nil
-	}
-
-	// Remove stale forwarder from cache
-	if exists {
-		forwarderMu.Lock()
-		delete(localForwarders, key)
+	forwarderMu.Lock()
+	if forwarder := localForwarders[key]; forwarder != nil && !forwarder.IsClosed() {
+		lease := acquireForwarderLeaseLocked(key, forwarder)
 		forwarderMu.Unlock()
+		logger.Infof("复用已有端口转发：%s", logKey)
+		return lease, nil
 	}
+	delete(localForwarders, key)
+	forwarderMu.Unlock()

 	forwarder, err := NewLocalForwarder(sshConfig, remoteHost, remotePort)
 	if err != nil {
@@ -371,20 +403,65 @@ func GetOrCreateLocalForwarder(sshConfig connection.SSHConfig, remoteHost string
 	}

 	forwarderMu.Lock()
+	if existing := localForwarders[key]; existing != nil && !existing.IsClosed() {
+		lease := acquireForwarderLeaseLocked(key, existing)
+		forwarderMu.Unlock()
+		_ = forwarder.closeUnderlying()
+		logger.Infof("复用已有端口转发：%s", logKey)
+		return lease, nil
+	}
+	delete(localForwarders, key)
 	localForwarders[key] = forwarder
+	lease := acquireForwarderLeaseLocked(key, forwarder)
 	forwarderMu.Unlock()

-	return forwarder, nil
+	return lease, nil
 }

-// CloseAllForwarders closes all local forwarders
+// GetOrCreateLocalForwarder is kept for internal compatibility. New callers
+// should use AcquireLocalForwarder so the lease ownership is explicit.
+func GetOrCreateLocalForwarder(sshConfig connection.SSHConfig, remoteHost string, remotePort int) (*LocalForwarder, error) {
+	return AcquireLocalForwarder(sshConfig, remoteHost, remotePort)
+}
+
+func acquireForwarderLeaseLocked(key forwarderCacheKey, shared *LocalForwarder) *LocalForwarder {
+	shared.refCount++
+	return &LocalForwarder{
+		LocalAddr:  shared.LocalAddr,
+		RemoteAddr: shared.RemoteAddr,
+		SSHClient:  shared.SSHClient,
+		shared:     shared,
+		cacheKey:   key,
+	}
+}
+
+func releaseLocalForwarder(key forwarderCacheKey, shared *LocalForwarder) error {
+	forwarderMu.Lock()
+	if shared.refCount > 0 {
+		shared.refCount--
+	}
+	if shared.refCount > 0 {
+		forwarderMu.Unlock()
+		return nil
+	}
+	if localForwarders[key] == shared {
+		delete(localForwarders, key)
+	}
+	forwarderMu.Unlock()
+
+	return shared.closeUnderlying()
+}
+
+// CloseAllForwarders force-closes all cached local forwarders regardless of
+// active leases.
 func CloseAllForwarders() {
 	forwarderMu.Lock()
 	defer forwarderMu.Unlock()

 	for _, forwarder := range localForwarders {
 		if forwarder != nil {
-			_ = forwarder.Close()
+			forwarder.refCount = 0
+			_ = forwarder.closeUnderlying()
 			logger.Infof("已关闭端口转发：本地 %s -> 远程 %s", forwarder.LocalAddr, forwarder.RemoteAddr)
 		}
 	}
@@ -394,7 +471,20 @@ func CloseAllForwarders() {
 // GetOrCreateSSHClient returns a cached SSH client or creates a new one
 func GetOrCreateSSHClient(config connection.SSHConfig) (*ssh.Client, error) {
 	key := newSSHClientCacheKey(config)
+	value, err, _ := sshClientFlights.Do(sshClientFlightKey(key), func() (interface{}, error) {
+		return getOrCreateSSHClient(config, key)
+	})
+	if err != nil {
+		return nil, err
+	}
+	client, ok := value.(*ssh.Client)
+	if !ok || client == nil {
+		return nil, fmt.Errorf("SSH client creation returned an invalid result")
+	}
+	return client, nil
+}

+func getOrCreateSSHClient(config connection.SSHConfig, key sshClientCacheKey) (*ssh.Client, error) {
 	sshClientCacheMu.RLock()
 	client, exists := sshClientCache[key]
 	sshClientCacheMu.RUnlock()
@@ -417,7 +507,7 @@ func GetOrCreateSSHClient(config connection.SSHConfig) (*ssh.Client, error) {
 	}

 	// Create new SSH client
-	client, err := connectSSH(config)
+	client, err := connectSSHClient(config)
 	if err != nil {
 		return nil, err
 	}
@@ -431,6 +521,10 @@ func GetOrCreateSSHClient(config connection.SSHConfig) (*ssh.Client, error) {
 	return client, nil
 }

+func sshClientFlightKey(key sshClientCacheKey) string {
+	return fmt.Sprintf("%q\x00%d\x00%q\x00%s", key.host, key.port, key.user, key.auth)
+}
+
 // DialThroughSSH creates a connection through SSH tunnel
 // This is a generic dialer that can be used by any database driver
 func DialThroughSSH(config connection.SSHConfig, network, address string) (net.Conn, error) {
--- a/internal/ssh/ssh_client_singleflight_test.go
+++ b/internal/ssh/ssh_client_singleflight_test.go
@@ -0,0 +1,97 @@
+package ssh
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
+
+	cryptossh "golang.org/x/crypto/ssh"
+)
+
+func TestGetOrCreateSSHClientCoalescesConcurrentColdStarts(t *testing.T) {
+	config := connection.SSHConfig{
+		Host:     "jump.example.test",
+		Port:     22,
+		User:     "tester",
+		Password: "test-password",
+	}
+	key := newSSHClientCacheKey(config)
+	fakeClient := &cryptossh.Client{}
+
+	sshClientCacheMu.Lock()
+	previousCache := sshClientCache
+	sshClientCache = make(map[sshClientCacheKey]*cryptossh.Client)
+	sshClientCacheMu.Unlock()
+	previousConnect := connectSSHClient
+	t.Cleanup(func() {
+		connectSSHClient = previousConnect
+		sshClientCacheMu.Lock()
+		delete(sshClientCache, key)
+		sshClientCache = previousCache
+		sshClientCacheMu.Unlock()
+	})
+
+	connectStarted := make(chan struct{})
+	releaseConnect := make(chan struct{})
+	var connectOnce sync.Once
+	var connectCalls atomic.Int32
+	connectSSHClient = func(connection.SSHConfig) (*cryptossh.Client, error) {
+		connectCalls.Add(1)
+		connectOnce.Do(func() {
+			close(connectStarted)
+		})
+		<-releaseConnect
+		return fakeClient, nil
+	}
+
+	const callers = 32
+	start := make(chan struct{})
+	var ready sync.WaitGroup
+	var done sync.WaitGroup
+	ready.Add(callers)
+	done.Add(callers)
+	results := make(chan *cryptossh.Client, callers)
+	errorsFound := make(chan error, callers)
+	for range callers {
+		go func() {
+			defer done.Done()
+			ready.Done()
+			<-start
+			client, err := GetOrCreateSSHClient(config)
+			if err != nil {
+				errorsFound <- err
+				return
+			}
+			results <- client
+		}()
+	}
+	ready.Wait()
+	close(start)
+	select {
+	case <-connectStarted:
+	case <-time.After(time.Second):
+		t.Fatal("cold SSH client creation did not start")
+	}
+	// Keep the leader blocked long enough for every released caller to join
+	// the same singleflight instead of observing the synthetic cached client.
+	time.Sleep(100 * time.Millisecond)
+	close(releaseConnect)
+	done.Wait()
+	close(results)
+	close(errorsFound)
+
+	for err := range errorsFound {
+		t.Fatalf("GetOrCreateSSHClient returned error: %v", err)
+	}
+	for client := range results {
+		if client != fakeClient {
+			t.Fatalf("caller received unexpected SSH client %p, want %p", client, fakeClient)
+		}
+	}
+	if got := connectCalls.Load(); got != 1 {
+		t.Fatalf("connectSSH called %d times for one cold cache key, want 1", got)
+	}
+}
--- a/internal/ssh/ssh_forwarder_lease_test.go
+++ b/internal/ssh/ssh_forwarder_lease_test.go
@@ -0,0 +1,251 @@
+package ssh
+
+import (
+	"net"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"GoNavi-Wails/internal/connection"
+)
+
+func TestSharedLocalForwarderStaysOpenUntilLastLeaseCloses(t *testing.T) {
+	sshConfig := connection.SSHConfig{
+		Host:     "jump.example.test",
+		Port:     22,
+		User:     "tester",
+		Password: "test-password",
+	}
+	const (
+		remoteHost = "database.internal.test"
+		remotePort = 5432
+	)
+
+	shared := installCachedLocalForwarderForTest(t, sshConfig, remoteHost, remotePort)
+
+	first, err := AcquireLocalForwarder(sshConfig, remoteHost, remotePort)
+	if err != nil {
+		t.Fatalf("first AcquireLocalForwarder() error = %v", err)
+	}
+	second, err := AcquireLocalForwarder(sshConfig, remoteHost, remotePort)
+	if err != nil {
+		t.Fatalf("second AcquireLocalForwarder() error = %v", err)
+	}
+	if first == second {
+		t.Fatal("separate acquisitions returned the same lease object")
+	}
+	if first.LocalAddr != second.LocalAddr {
+		t.Fatalf("leases did not share one listener: first=%s second=%s", first.LocalAddr, second.LocalAddr)
+	}
+
+	if err := first.Release(); err != nil {
+		t.Fatalf("first Release() error = %v", err)
+	}
+	if second.IsClosed() {
+		t.Fatal("closing one lease closed the shared forwarder while another lease was active")
+	}
+	assertTCPListenerAcceptsConnections(t, second.LocalAddr)
+
+	if err := second.Release(); err != nil {
+		t.Fatalf("second Release() error = %v", err)
+	}
+	if !shared.IsClosed() {
+		t.Fatal("last lease did not close the shared forwarder")
+	}
+	assertTCPListenerClosed(t, shared.LocalAddr)
+	assertForwarderEvicted(t, sshConfig, remoteHost, remotePort)
+}
+
+func TestCloseAllForwardersForceClosesActiveLeases(t *testing.T) {
+	sshConfig := connection.SSHConfig{
+		Host:     "jump.example.test",
+		Port:     22,
+		User:     "tester",
+		Password: "test-password",
+	}
+	const (
+		remoteHost = "database.internal.test"
+		remotePort = 5432
+	)
+
+	shared := installCachedLocalForwarderForTest(t, sshConfig, remoteHost, remotePort)
+	first, err := AcquireLocalForwarder(sshConfig, remoteHost, remotePort)
+	if err != nil {
+		t.Fatalf("first AcquireLocalForwarder() error = %v", err)
+	}
+	second, err := AcquireLocalForwarder(sshConfig, remoteHost, remotePort)
+	if err != nil {
+		t.Fatalf("second AcquireLocalForwarder() error = %v", err)
+	}
+
+	CloseAllForwarders()
+
+	if !shared.IsClosed() || !first.IsClosed() || !second.IsClosed() {
+		t.Fatal("CloseAllForwarders() did not force-close the shared forwarder")
+	}
+	assertTCPListenerClosed(t, shared.LocalAddr)
+	assertForwarderEvicted(t, sshConfig, remoteHost, remotePort)
+
+	if err := first.Release(); err != nil {
+		t.Fatalf("releasing first lease after CloseAllForwarders() error = %v", err)
+	}
+	if err := second.Release(); err != nil {
+		t.Fatalf("releasing second lease after CloseAllForwarders() error = %v", err)
+	}
+}
+
+func TestLocalForwarderLeaseReferenceCountIsThreadSafe(t *testing.T) {
+	sshConfig := connection.SSHConfig{
+		Host:     "jump.example.test",
+		Port:     22,
+		User:     "tester",
+		Password: "test-password",
+	}
+	const (
+		remoteHost = "database.internal.test"
+		remotePort = 5432
+		leaseCount = 32
+	)
+
+	shared := installCachedLocalForwarderForTest(t, sshConfig, remoteHost, remotePort)
+	keeper, err := AcquireLocalForwarder(sshConfig, remoteHost, remotePort)
+	if err != nil {
+		t.Fatalf("keeper AcquireLocalForwarder() error = %v", err)
+	}
+
+	leases := make(chan *LocalForwarder, leaseCount)
+	errs := make(chan error, leaseCount)
+	var acquireWG sync.WaitGroup
+	for range leaseCount {
+		acquireWG.Add(1)
+		go func() {
+			defer acquireWG.Done()
+			lease, acquireErr := AcquireLocalForwarder(sshConfig, remoteHost, remotePort)
+			if acquireErr != nil {
+				errs <- acquireErr
+				return
+			}
+			leases <- lease
+		}()
+	}
+	acquireWG.Wait()
+	close(leases)
+	close(errs)
+	for acquireErr := range errs {
+		t.Fatalf("AcquireLocalForwarder() error = %v", acquireErr)
+	}
+
+	var releaseWG sync.WaitGroup
+	releaseErrs := make(chan error, leaseCount*2)
+	for lease := range leases {
+		lease := lease
+		for range 2 {
+			releaseWG.Add(1)
+			go func() {
+				defer releaseWG.Done()
+				if releaseErr := lease.Release(); releaseErr != nil {
+					releaseErrs <- releaseErr
+				}
+			}()
+		}
+	}
+	releaseWG.Wait()
+	close(releaseErrs)
+	for releaseErr := range releaseErrs {
+		t.Fatalf("Release() error = %v", releaseErr)
+	}
+
+	if shared.IsClosed() {
+		t.Fatal("concurrent release closed the forwarder while the keeper lease was active")
+	}
+	assertTCPListenerAcceptsConnections(t, keeper.LocalAddr)
+
+	if err := keeper.Release(); err != nil {
+		t.Fatalf("keeper Release() error = %v", err)
+	}
+	if !shared.IsClosed() {
+		t.Fatal("releasing the final keeper lease did not close the shared forwarder")
+	}
+	assertForwarderEvicted(t, sshConfig, remoteHost, remotePort)
+}
+
+func installCachedLocalForwarderForTest(
+	t *testing.T,
+	sshConfig connection.SSHConfig,
+	remoteHost string,
+	remotePort int,
+) *LocalForwarder {
+	t.Helper()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatalf("net.Listen() error = %v", err)
+	}
+	forwarder := &LocalForwarder{
+		LocalAddr:  listener.Addr().String(),
+		RemoteAddr: net.JoinHostPort(remoteHost, strconv.Itoa(remotePort)),
+		listener:   listener,
+		closeChan:  make(chan struct{}),
+	}
+	key := forwarderCacheKey{
+		ssh:        newSSHClientCacheKey(sshConfig),
+		remoteHost: remoteHost,
+		remotePort: remotePort,
+	}
+
+	forwarderMu.Lock()
+	previous := localForwarders
+	localForwarders = map[forwarderCacheKey]*LocalForwarder{key: forwarder}
+	forwarderMu.Unlock()
+
+	t.Cleanup(func() {
+		CloseAllForwarders()
+		forwarderMu.Lock()
+		localForwarders = previous
+		forwarderMu.Unlock()
+	})
+	return forwarder
+}
+
+func assertTCPListenerAcceptsConnections(t *testing.T, address string) {
+	t.Helper()
+
+	conn, err := net.DialTimeout("tcp", address, time.Second)
+	if err != nil {
+		t.Fatalf("expected active lease to keep %s reachable: %v", address, err)
+	}
+	_ = conn.Close()
+}
+
+func assertTCPListenerClosed(t *testing.T, address string) {
+	t.Helper()
+
+	conn, err := net.DialTimeout("tcp", address, 100*time.Millisecond)
+	if err != nil {
+		return
+	}
+	_ = conn.Close()
+	t.Fatalf("expected closed forwarder listener at %s to reject connections", address)
+}
+
+func assertForwarderEvicted(
+	t *testing.T,
+	sshConfig connection.SSHConfig,
+	remoteHost string,
+	remotePort int,
+) {
+	t.Helper()
+
+	key := forwarderCacheKey{
+		ssh:        newSSHClientCacheKey(sshConfig),
+		remoteHost: remoteHost,
+		remotePort: remotePort,
+	}
+	forwarderMu.RLock()
+	_, exists := localForwarders[key]
+	forwarderMu.RUnlock()
+	if exists {
+		t.Fatal("closed shared forwarder remained in cache")
+	}
+}