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

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"
                   />
                 </div>
+                <div v-show="form.db_type === 'mongodb'" class="grid grid-cols-4 items-center gap-4">
+                  <Label class="text-right text-xs">{{ t("connection.idleTimeout") }}</Label>
+                  <Input
+                    v-model.number="form.idle_timeout_secs"
+                    type="number"
+                    min="0"
+                    max="600"
+                    step="1"
+                    class="col-span-3"
+                  />
+                </div>
               </div>
             </TabsContent>
 

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",

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",

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",

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",

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: "代理类型",

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: "代理伺服器類型",

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,
     };
   }
 

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;

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,

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<Client, String> {
+pub async fn connect(url: &str, timeout: Duration, idle_timeout: Duration) -> Result<Client, String> {
     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<Client, String> {
         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

crates/dbx-core/src/db/ssh_tunnel.rs

@@ -309,9 +309,15 @@ impl TunnelManager {
         remote_port: u16,
         expose_to_lan: bool,
     ) -> Result<u16, String> {
-        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<String, TunnelEntry>, connection_id: &str) -> Option<u16> {
+        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)
     }
 

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<ConnectionConfigData> 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,

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<DatabaseType> {
     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<CancellationToken>,
     options: QueryExecutionOptions,
 ) -> Result<db::QueryResult, String> {
+    // 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<CancellationToken>,
     options: QueryExecutionOptions,
 ) -> Result<Vec<db::QueryResult>, 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 {

src-tauri/src/commands/connection.rs

@@ -247,6 +247,7 @@ pub async fn test_connection(state: State<'_, Arc<AppState>>, 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<AppState>>, 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<AppState>>, 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<AppState>>, 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