From 171c7cd8d7d3f1a4aca044e67f60a2b7d7176463 Mon Sep 17 00:00:00 2001 From: James-Leong Date: Mon, 8 Jun 2026 17:25:14 +0000 Subject: [PATCH] fix(ssh): prevent duplicate SSH tunnels from concurrent pool creation (#891) start_tunnel() and start_chain() had a check-then-act race condition: multiple concurrent callers could all see an empty cache and each spawn a new SSH tunnel. The tunnel that lost the insert race was orphaned, causing queries routed through it to fail. Use double-check locking with stale entry eviction: hold the lock for the cache lookup, release it during the slow SSH handshake, then re-check under the lock before inserting. If another caller already created the tunnel, abort the duplicate and return the existing port. Also check handle liveness on cache hits so that callers don't get a dead port back when the background tunnel task has exited. Add a matching re-check in the MongoDB connection pool creation path. Add idle_timeout_secs connection option (default 60s) to preempt server-side connection idle timeouts that cause "unexpected end of file" errors on pooled MongoDB connections. Reject MongoDB queries in the generic SQL execution path before any pool or session-key creation. Previously, a typo in a MongoDB shell command would fall through to executeMulti / execute_sql_statement, which called get_or_create_pool_for_session and leaked a session-scoped MongoDB Client (and SSH tunnel resources) before eventually returning "Use MongoDB-specific commands". Co-Authored-By: Claude Opus 4.7 --- .../connection/ConnectionDialog.vue | 15 ++++++ apps/desktop/src/i18n/locales/en.ts | 1 + apps/desktop/src/i18n/locales/es.ts | 1 + apps/desktop/src/i18n/locales/it.ts | 1 + apps/desktop/src/i18n/locales/pt-BR.ts | 1 + apps/desktop/src/i18n/locales/zh-CN.ts | 1 + apps/desktop/src/i18n/locales/zh-TW.ts | 1 + apps/desktop/src/stores/connectionStore.ts | 1 + apps/desktop/src/types/database.ts | 1 + crates/dbx-core/src/connection.rs | 10 +++- crates/dbx-core/src/db/mongo_driver.rs | 8 ++- crates/dbx-core/src/db/ssh_tunnel.rs | 53 +++++++++++++++---- crates/dbx-core/src/models/connection.rs | 9 ++++ crates/dbx-core/src/query.rs | 18 +++++++ src-tauri/src/commands/connection.rs | 6 ++- 15 files changed, 111 insertions(+), 16 deletions(-) diff --git a/apps/desktop/src/components/connection/ConnectionDialog.vue b/apps/desktop/src/components/connection/ConnectionDialog.vue index 4a8bddab79..a42e8a8fb9 100644 --- a/apps/desktop/src/components/connection/ConnectionDialog.vue +++ b/apps/desktop/src/components/connection/ConnectionDialog.vue @@ -122,6 +122,7 @@ const defaultForm = (): ConnectionForm => ({ transport_layers: [], connect_timeout_secs: 5, query_timeout_secs: 30, + idle_timeout_secs: 60, ssl: false, ca_cert_path: "", client_cert_path: "", @@ -530,6 +531,7 @@ watch( transport_layers: transportLayersForConfig(legacyConfig), connect_timeout_secs: config.connect_timeout_secs || 5, query_timeout_secs: config.query_timeout_secs ?? 30, + idle_timeout_secs: config.idle_timeout_secs ?? 60, ssl: config.ssl || false, ca_cert_path: config.ca_cert_path || "", client_cert_path: config.client_cert_path || "", @@ -1039,6 +1041,8 @@ function connectionConfigForSubmit(id: string): ConnectionConfig { config.connect_timeout_secs = Number.isFinite(connectTimeout) && connectTimeout > 0 ? connectTimeout : 5; const queryTimeout = Number(config.query_timeout_secs); config.query_timeout_secs = Number.isFinite(queryTimeout) && queryTimeout >= 0 ? queryTimeout : 30; + const idleTimeout = Number(config.idle_timeout_secs); + config.idle_timeout_secs = Number.isFinite(idleTimeout) && idleTimeout >= 0 ? idleTimeout : 60; if (!config.one_time) config.one_time = undefined; if (config.db_type === "mongodb" && !mongoUseUrl.value) { config.connection_string = undefined; @@ -3053,6 +3057,17 @@ function openExternalUrl(url: string) { class="col-span-3" /> +
+ + +
diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index b9c3a22a60..c053ae3a95 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -273,6 +273,7 @@ export default { sshHopInvalidTimeout: "{hop}: SSH timeout must be between 1 and 300 seconds", connectTimeout: "Connection Timeout (seconds)", queryTimeout: "Query Timeout (seconds)", + idleTimeout: "Idle Timeout (seconds)", proxy: "Proxy", proxyEnable: "Connect database through proxy", proxyType: "Proxy Type", diff --git a/apps/desktop/src/i18n/locales/es.ts b/apps/desktop/src/i18n/locales/es.ts index d89fc92d6c..bc1dc3b868 100644 --- a/apps/desktop/src/i18n/locales/es.ts +++ b/apps/desktop/src/i18n/locales/es.ts @@ -254,6 +254,7 @@ export default { sshHopInvalidTimeout: "{hop}: el tiempo de espera SSH debe estar entre 1 y 300 segundos", connectTimeout: "Tiempo de espera de conexión (segundos)", queryTimeout: "Tiempo de espera de consulta (segundos)", + idleTimeout: "Tiempo de espera inactivo (segundos)", proxy: "Proxy", proxyEnable: "Conectar la base de datos mediante proxy", proxyType: "Tipo de proxy", diff --git a/apps/desktop/src/i18n/locales/it.ts b/apps/desktop/src/i18n/locales/it.ts index 0bb0c5d4c2..88165a7516 100644 --- a/apps/desktop/src/i18n/locales/it.ts +++ b/apps/desktop/src/i18n/locales/it.ts @@ -265,6 +265,7 @@ export default { sshHopInvalidTimeout: "{hop}: Il timeout SSH deve essere compreso tra 1 e 300 secondi", connectTimeout: "Timeout Connessione (secondi)", queryTimeout: "Timeout Query (secondi)", + idleTimeout: "Timeout Inattività (secondi)", proxy: "Proxy", proxyEnable: "Connetti database tramite proxy", proxyType: "Tipo Proxy", diff --git a/apps/desktop/src/i18n/locales/pt-BR.ts b/apps/desktop/src/i18n/locales/pt-BR.ts index dd2e398e37..ef957ae8f8 100644 --- a/apps/desktop/src/i18n/locales/pt-BR.ts +++ b/apps/desktop/src/i18n/locales/pt-BR.ts @@ -264,6 +264,7 @@ export default { sshHopInvalidTimeout: "{hop}: o timeout SSH deve estar entre 1 e 300 segundos", connectTimeout: "Timeout de Conexão (segundos)", queryTimeout: "Timeout de Consulta (segundos)", + idleTimeout: "Timeout de Inatividade (segundos)", proxy: "Proxy", proxyEnable: "Conectar ao banco de dados via proxy", proxyType: "Tipo de Proxy", diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index 4305cddf55..20a3fc76fb 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -269,6 +269,7 @@ export default { sshHopInvalidTimeout: "{hop}：SSH 超时时间必须在 1 到 300 秒之间", connectTimeout: "连接超时（秒）", queryTimeout: "查询超时（秒）", + idleTimeout: "空闲超时（秒）", proxy: "代理", proxyEnable: "通过代理连接数据库", proxyType: "代理类型", diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index 87e95a4515..05e49e4ad9 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -258,6 +258,7 @@ export default { sshHopInvalidTimeout: "{hop}：SSH 逾時時間必須在 1 到 300 秒之間", connectTimeout: "連線逾時（秒）", queryTimeout: "查詢逾時（秒）", + idleTimeout: "閒置逾時（秒）", proxy: "代理伺服器", proxyEnable: "藉由代理伺服器連線資料庫", proxyType: "代理伺服器類型", diff --git a/apps/desktop/src/stores/connectionStore.ts b/apps/desktop/src/stores/connectionStore.ts index 753d947994..b9411081f3 100644 --- a/apps/desktop/src/stores/connectionStore.ts +++ b/apps/desktop/src/stores/connectionStore.ts @@ -305,6 +305,7 @@ export const useConnectionStore = defineStore("connection", () => { transport_layers: Array.isArray(config.transport_layers) ? config.transport_layers : [], connect_timeout_secs: config.connect_timeout_secs || 5, query_timeout_secs: config.query_timeout_secs ?? 30, + idle_timeout_secs: config.idle_timeout_secs ?? 60, }; } diff --git a/apps/desktop/src/types/database.ts b/apps/desktop/src/types/database.ts index d8bfd712e1..5656c1a730 100644 --- a/apps/desktop/src/types/database.ts +++ b/apps/desktop/src/types/database.ts @@ -75,6 +75,7 @@ export interface ConnectionConfig { transport_layers?: TransportLayerConfig[]; connect_timeout_secs?: number; query_timeout_secs?: number; + idle_timeout_secs?: number; ssl?: boolean; ca_cert_path?: string; client_cert_path?: string; diff --git a/crates/dbx-core/src/connection.rs b/crates/dbx-core/src/connection.rs index b105c0c192..10695ab7c2 100644 --- a/crates/dbx-core/src/connection.rs +++ b/crates/dbx-core/src/connection.rs @@ -324,6 +324,7 @@ impl AppState { probe_connection_endpoint(&db_config, &host, port).await?; let url = connection_url_for_endpoint(&db_config, &host, port); let connect_timeout = std::time::Duration::from_secs(db_config.effective_connect_timeout_secs()); + let idle_timeout = std::time::Duration::from_secs(db_config.idle_timeout_secs); let mysql_pool_max_connections = if normalize_client_session_id(client_session_id).is_some() { 1 } else { 3 }; let pool = match db_config.db_type { DatabaseType::Mysql => { @@ -401,7 +402,7 @@ impl AppState { PoolKind::DuckDb(con) } DatabaseType::MongoDb => { - let native_err = match db::mongo_driver::connect(&url, connect_timeout).await { + let native_err = match db::mongo_driver::connect(&url, connect_timeout, idle_timeout).await { Ok(client) => match db::mongo_driver::test_connection( &client, connect_timeout, @@ -410,7 +411,12 @@ impl AppState { .await { Ok(()) => { - self.connections.write().await.insert(pool_key.clone(), PoolKind::MongoDb(client)); + let mut conns = self.connections.write().await; + // Re-check: another task may have created the pool while we were connecting. + if conns.contains_key(&pool_key) { + return Ok(pool_key); + } + conns.insert(pool_key.clone(), PoolKind::MongoDb(client)); return Ok(pool_key); } Err(e) => e, diff --git a/crates/dbx-core/src/db/mongo_driver.rs b/crates/dbx-core/src/db/mongo_driver.rs index a0641f060f..5f6ecf633f 100644 --- a/crates/dbx-core/src/db/mongo_driver.rs +++ b/crates/dbx-core/src/db/mongo_driver.rs @@ -16,7 +16,7 @@ pub struct MongoDocumentResult { pub total: u64, } -pub async fn connect(url: &str, timeout: Duration) -> Result { +pub async fn connect(url: &str, timeout: Duration, idle_timeout: Duration) -> Result { let is_multi_host = is_multi_host_mongo_uri(url); let parse_timeout = if is_multi_host { std::cmp::max(timeout * 2, Duration::from_secs(10)) } else { timeout }; @@ -25,6 +25,12 @@ pub async fn connect(url: &str, timeout: Duration) -> Result { options.connect_timeout = Some(timeout); options.server_selection_timeout = if is_multi_host { Some(std::cmp::max(timeout * 2, Duration::from_secs(10))) } else { Some(timeout) }; + // Close idle connections before the server-side timeout drops them, + // preventing "Broken pipe" (os error 32) or "unexpected end of file". + // 0 means no idle timeout (keep connections alive indefinitely). + if idle_timeout.as_secs() > 0 { + options.max_idle_time = Some(idle_timeout); + } Client::with_options(options).map_err(|e| format!("MongoDB connection failed: {e}")) }) .await diff --git a/crates/dbx-core/src/db/ssh_tunnel.rs b/crates/dbx-core/src/db/ssh_tunnel.rs index 30746f0e04..e937faf463 100644 --- a/crates/dbx-core/src/db/ssh_tunnel.rs +++ b/crates/dbx-core/src/db/ssh_tunnel.rs @@ -309,9 +309,15 @@ impl TunnelManager { remote_port: u16, expose_to_lan: bool, ) -> Result { - if let Some(local_port) = self.local_port(connection_id).await { - return Ok(local_port); + // Check cache under lock to avoid race with concurrent callers. + // Also evict stale entries whose background task has exited. + { + let mut tunnels = self.tunnels.lock().await; + if let Some(port) = Self::get_active_port(&mut tunnels, connection_id) { + return Ok(port); + } } + // Slow SSH connection — do this outside the lock. let (handle, local_port) = spawn_tunnel( ssh_host, ssh_port, @@ -326,11 +332,28 @@ impl TunnelManager { ) .await?; - self.tunnels.lock().await.insert(connection_id.to_string(), TunnelEntry { handles: vec![handle], local_port }); - + // Re-check under lock: another caller may have beaten us. + let mut tunnels = self.tunnels.lock().await; + if let Some(port) = Self::get_active_port(&mut tunnels, connection_id) { + // Another task already created a live tunnel; abort ours. + handle.abort(); + return Ok(port); + } + tunnels.insert(connection_id.to_string(), TunnelEntry { handles: vec![handle], local_port }); Ok(local_port) } + /// Returns the local port for a cached tunnel entry, or `None` if the entry + /// is stale (all background handles have exited). + fn get_active_port(tunnels: &mut HashMap, connection_id: &str) -> Option { + let entry = tunnels.get(connection_id)?; + if entry.handles.iter().all(|h| h.is_finished()) { + tunnels.remove(connection_id); + return None; + } + Some(entry.local_port) + } + pub async fn start_chain( &self, connection_id: &str, @@ -341,8 +364,12 @@ impl TunnelManager { if hops.is_empty() { return Err("No SSH tunnel hops configured".to_string()); } - if let Some(local_port) = self.local_port(connection_id).await { - return Ok(local_port); + // Check cache under lock; evict stale entries. + { + let mut tunnels = self.tunnels.lock().await; + if let Some(port) = Self::get_active_port(&mut tunnels, connection_id) { + return Ok(port); + } } let mut handles = Vec::new(); @@ -379,11 +406,15 @@ impl TunnelManager { next_connect_endpoint = Some(("127.0.0.1".to_string(), local_port)); } - self.tunnels - .lock() - .await - .insert(connection_id.to_string(), TunnelEntry { handles, local_port: final_local_port }); - + // Re-check under lock: another caller may have beaten us. + let mut tunnels = self.tunnels.lock().await; + if let Some(port) = Self::get_active_port(&mut tunnels, connection_id) { + for handle in handles { + handle.abort(); + } + return Ok(port); + } + tunnels.insert(connection_id.to_string(), TunnelEntry { handles, local_port: final_local_port }); Ok(final_local_port) } diff --git a/crates/dbx-core/src/models/connection.rs b/crates/dbx-core/src/models/connection.rs index 717c6bb58e..d6e4e1971a 100644 --- a/crates/dbx-core/src/models/connection.rs +++ b/crates/dbx-core/src/models/connection.rs @@ -30,6 +30,8 @@ pub struct ConnectionConfig { pub connect_timeout_secs: u64, #[serde(default = "default_query_timeout_secs")] pub query_timeout_secs: u64, + #[serde(default = "default_idle_timeout_secs")] + pub idle_timeout_secs: u64, #[serde(default)] pub ssl: bool, #[serde(default, skip_serializing_if = "String::is_empty")] @@ -180,6 +182,10 @@ pub fn default_query_timeout_secs() -> u64 { 30 } +pub fn default_idle_timeout_secs() -> u64 { + 60 +} + fn default_proxy_port() -> u16 { 1080 } @@ -295,6 +301,8 @@ struct ConnectionConfigData { pub connect_timeout_secs: u64, #[serde(default = "default_query_timeout_secs")] pub query_timeout_secs: u64, + #[serde(default = "default_idle_timeout_secs")] + pub idle_timeout_secs: u64, #[serde(default)] pub ssl: bool, #[serde(default)] @@ -355,6 +363,7 @@ impl From for ConnectionConfig { transport_layers: data.transport_layers, connect_timeout_secs: data.connect_timeout_secs, query_timeout_secs: data.query_timeout_secs, + idle_timeout_secs: data.idle_timeout_secs, ssl: data.ssl, ca_cert_path: data.ca_cert_path, client_cert_path: data.client_cert_path, diff --git a/crates/dbx-core/src/query.rs b/crates/dbx-core/src/query.rs index 0d217bf936..05897bc0f2 100644 --- a/crates/dbx-core/src/query.rs +++ b/crates/dbx-core/src/query.rs @@ -15,6 +15,11 @@ pub const QUERY_TIMEOUT: Duration = Duration::from_secs(30); pub const MAX_ROWS: usize = 10000; pub const QUERY_CANCELED: &str = "Query canceled"; +async fn connection_is_mongodb(state: &AppState, connection_id: &str) -> bool { + let configs = state.configs.read().await; + configs.get(connection_id).is_some_and(|config| config.db_type == DatabaseType::MongoDb) +} + async fn connection_database_type(state: &AppState, connection_id: &str) -> Option { let configs = state.configs.read().await; configs.get(connection_id).map(|config| config.db_type) @@ -804,6 +809,14 @@ pub async fn execute_sql_statement_with_options( cancel_token: Option, options: QueryExecutionOptions, ) -> Result { + // MongoDB connections use shell-style commands dispatched through the + // frontend parser. Queries that fall through to the generic SQL executor + // (e.g. typos) must be rejected before any pool/key creation so that + // session-scoped pools do not leak MongoDB Clients and SSH tunnels. + if connection_is_mongodb(state, connection_id).await { + return Err("Use MongoDB-specific commands".to_string()); + } + // When a query tab has a client session, keep even database-less execution // on that tab-scoped pool so connection-level state (for example MySQL @vars) // survives across runs. @@ -893,6 +906,11 @@ pub async fn execute_multi_core_with_options( cancel_token: Option, options: QueryExecutionOptions, ) -> Result, String> { + // Reject MongoDB queries that fall through to the generic executor. + if connection_is_mongodb(state, connection_id).await { + return Err("Use MongoDB-specific commands".to_string()); + } + let pool_key = if database.is_empty() { state.get_or_create_pool_for_session(connection_id, None, options.client_session_id.as_deref()).await? } else { diff --git a/src-tauri/src/commands/connection.rs b/src-tauri/src/commands/connection.rs index f2994d5808..b88e24de14 100644 --- a/src-tauri/src/commands/connection.rs +++ b/src-tauri/src/commands/connection.rs @@ -247,6 +247,7 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection let url = connection_url_for_endpoint(&config, &host, port); let target = redacted_connection_url_for_endpoint(&config, &host, port); let connect_timeout = std::time::Duration::from_secs(config.effective_connect_timeout_secs()); + let idle_timeout = std::time::Duration::from_secs(config.idle_timeout_secs); log::info!("[test_connection] db_type={:?} target={}", config.db_type, target); let result = match probe_result { Err(e) => Err(e), @@ -324,7 +325,7 @@ pub async fn test_connection(state: State<'_, Arc>, config: Connection } } DatabaseType::MongoDb => { - let native_err = match db::mongo_driver::connect(&url, connect_timeout).await { + let native_err = match db::mongo_driver::connect(&url, connect_timeout, idle_timeout).await { Ok(client) => { match db::mongo_driver::test_connection(&client, connect_timeout, config.effective_database()) .await @@ -434,6 +435,7 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi probe_connection_endpoint(&db_config, &host, port).await?; let url = connection_url_for_endpoint(&db_config, &host, port); let connect_timeout = std::time::Duration::from_secs(db_config.effective_connect_timeout_secs()); + let idle_timeout = std::time::Duration::from_secs(db_config.idle_timeout_secs); let pool = match db_config.db_type { DatabaseType::Mysql => { @@ -489,7 +491,7 @@ pub async fn connect_db(state: State<'_, Arc>, config: ConnectionConfi PoolKind::DuckDb(con) } DatabaseType::MongoDb => { - let native_err = match db::mongo_driver::connect(&url, connect_timeout).await { + let native_err = match db::mongo_driver::connect(&url, connect_timeout, idle_timeout).await { Ok(client) => { match db::mongo_driver::test_connection(&client, connect_timeout, db_config.effective_database()) .await