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