From c1888a2364ea4f866d90298c75364e973aede90a Mon Sep 17 00:00:00 2001 From: Melody Ruch <396921921@qq.com> Date: Mon, 24 Aug 2026 18:34:49 +0800 Subject: [PATCH] fix(storage): use WAL and immediate transactions for shared SQLite data dir --- crates/dbx-core/src/storage.rs | 114 +++++++++++++++++++++++++++------ 1 file changed, 96 insertions(+), 18 deletions(-) diff --git a/crates/dbx-core/src/storage.rs b/crates/dbx-core/src/storage.rs index 202ca06c91..589947bbf0 100644 --- a/crates/dbx-core/src/storage.rs +++ b/crates/dbx-core/src/storage.rs @@ -3,7 +3,10 @@ use std::path::{Path, PathBuf}; use std::time::{SystemTime, UNIX_EPOCH}; use log::warn; -use rusqlite::{params, params_from_iter, types::Value, Connection, DatabaseName, OpenFlags, OptionalExtension, ToSql}; +use rusqlite::{ + params, params_from_iter, types::Value, Connection, DatabaseName, OpenFlags, OptionalExtension, ToSql, + TransactionBehavior, +}; use serde::{Deserialize, Serialize}; use uuid::Uuid; @@ -466,10 +469,52 @@ impl Storage { let db_path = db_path.to_string_lossy().to_string(); let db = connect_path_create_if_missing(&db_path).await?; let storage = Self { db, path }; + // Best-effort: switching journal mode is itself a lock-sensitive + // operation, so a transient failure here (e.g. another process + // racing to open the same brand-new database file) must never stop + // the app from starting. + storage.enable_wal_mode().await; storage.init_schema().await?; Ok(storage) } + /// Multiple `dbx` processes can end up pointed at the same data directory + /// (e.g. a portable install shared by several users on one machine). + /// WAL mode lets readers and writers proceed without blocking each other, + /// which combined with `TransactionBehavior::Immediate` on every write + /// transaction (see `conn.transaction_with_behavior` call sites below) + /// avoids the instant `SQLITE_BUSY` that a deferred transaction's + /// SHARED-to-RESERVED lock upgrade can trigger under concurrent access. + /// Never fails: this is a best-effort upgrade, retried a handful of + /// times, that logs and gives up rather than blocking startup. + async fn enable_wal_mode(&self) { + const ATTEMPTS: u32 = 5; + for attempt in 1..=ATTEMPTS { + let result = self + .with_conn(|conn| { + conn.query_row("PRAGMA journal_mode=WAL", [], |row| row.get::<_, String>(0)) + .map_err(|e| e.to_string()) + }) + .await; + match result { + Ok(mode) if mode.eq_ignore_ascii_case("wal") => return, + Ok(mode) => { + // Non-lock reasons WAL can't apply (e.g. an in-memory + // database in tests) won't be fixed by retrying. + warn!("dbx.db journal_mode did not switch to WAL (got '{mode}'); concurrent multi-process access may hit 'database is locked' more often"); + return; + } + Err(_) if attempt < ATTEMPTS => { + std::thread::sleep(std::time::Duration::from_millis(100 * attempt as u64)); + } + Err(error) => { + warn!("dbx.db could not switch journal_mode to WAL after {ATTEMPTS} attempts: {error}; concurrent multi-process access may hit 'database is locked' more often"); + return; + } + } + } + } + /// Directory containing the SQLite database (`dbx.db`). SSH host keys are /// stored in `/known_hosts` so dbx never touches `~/.ssh`. pub fn data_dir(&self) -> &Path { @@ -1248,7 +1293,7 @@ impl Storage { pub async fn save_ai_configs(&self, configs: &[AiConfigItem]) -> Result<(), String> { let configs = configs.to_vec(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; tx.execute("DELETE FROM ai_configs", []).map_err(|e| e.to_string())?; for config in &configs { let json = serde_json::to_string(&config.config).map_err(|e| e.to_string())?; @@ -1317,7 +1362,7 @@ impl Storage { pub async fn set_default_ai_config(&self, config_id: &str) -> Result<(), String> { let config_id = config_id.to_string(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; tx.execute("UPDATE ai_configs SET is_default = 0 WHERE is_default = 1", []).map_err(|e| e.to_string())?; tx.execute("UPDATE ai_configs SET is_default = 1 WHERE id = ?1", params![config_id]) .map_err(|e| e.to_string())?; @@ -1332,7 +1377,7 @@ impl Storage { self.with_conn(move |conn| { let json = serde_json::to_string(&config.config).map_err(|e| e.to_string())?; let models_json = serde_json::to_string(&config.config.models).map_err(|e| e.to_string())?; - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; // 如果设该配置为默认，先清除其他默认，避免与 idx_ai_configs_default 冲突 if config.is_default { @@ -1411,7 +1456,7 @@ impl Storage { } let profiles = profiles.to_vec(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; tx.execute("DELETE FROM tunnel_profiles", []).map_err(|e| e.to_string())?; for profile in &profiles { let json = serde_json::to_string(profile).map_err(|e| e.to_string())?; @@ -2501,7 +2546,7 @@ impl Storage { ) -> Result<(), String> { let configs = configs.to_vec(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; let replacement_ids = configs.iter().map(|config| config.id.clone()).collect::>(); let mut retained_ids = preserve_unreadable_connections_for_replacement(&tx, &replacement_ids)?; @@ -2541,7 +2586,7 @@ impl Storage { pub async fn save_connections(&self, configs: &[ConnectionConfig]) -> Result<(), String> { let configs = configs.to_vec(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; let replacement_ids = configs.iter().map(|config| config.id.clone()).collect::>(); let mut retained_ids = preserve_unreadable_connections_for_replacement(&tx, &replacement_ids)?; @@ -2560,7 +2605,7 @@ impl Storage { pub async fn add_connection_for_mcp(&self, config: ConnectionConfig) -> Result { let config = config.canonicalized(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; ensure_mcp_connection_change_allowed_in_tx(&tx, None)?; persist_connection_in_tx(&tx, &config)?; tx.commit().map_err(|e| e.to_string())?; @@ -2580,7 +2625,8 @@ impl Storage { let copied_id = copy_id.clone(); let copy_name = copy_name.to_string(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|error| error.to_string())?; + let tx = + conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|error| error.to_string())?; ensure_mcp_connection_change_allowed_in_tx(&tx, Some(&source_id))?; let copy_name_lower = copy_name.to_lowercase(); let mut names = tx.prepare("SELECT config_json FROM connections").map_err(|error| error.to_string())?; @@ -2628,7 +2674,7 @@ impl Storage { pub async fn remove_connection_for_mcp(&self, connection_id: &str) -> Result { let connection_id = connection_id.to_string(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; ensure_mcp_connection_change_allowed_in_tx(&tx, Some(&connection_id))?; let removed = tx.execute("DELETE FROM connections WHERE id = ?1", [&connection_id]).map_err(|e| e.to_string())? > 0; @@ -2969,7 +3015,7 @@ impl Storage { pub async fn replace_saved_sql_library(&self, library: &SavedSqlLibrary) -> Result<(), String> { let library = library.clone(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; tx.execute("DELETE FROM saved_sql_files", []).map_err(|e| e.to_string())?; tx.execute("DELETE FROM saved_sql_folders", []).map_err(|e| e.to_string())?; @@ -3236,7 +3282,7 @@ impl Storage { pub async fn delete_saved_sql_folder(&self, id: &str) -> Result<(), String> { let id = id.to_string(); self.with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; let mut folder_ids = vec![id.clone()]; let mut index = 0; while index < folder_ids.len() { @@ -3573,7 +3619,8 @@ impl Storage { let owner_id = schema_cache_owner(&cache_key).to_string(); let now_ms = unix_timestamp_millis(); self.with_conn(move |conn| { - let transaction = conn.transaction().map_err(|error| error.to_string())?; + let transaction = + conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|error| error.to_string())?; transaction .execute( "INSERT INTO schema_cache ( @@ -3600,7 +3647,9 @@ impl Storage { let now_ms = unix_timestamp_millis(); let json: Option = self .with_conn(move |conn| { - let transaction = conn.transaction().map_err(|error| error.to_string())?; + let transaction = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(|error| error.to_string())?; let json = transaction .query_row( "SELECT payload_json FROM schema_cache @@ -3897,7 +3946,8 @@ impl Storage { let deleted_bytes = entries.iter().filter(|(key, _, _, _)| deleted.contains(key)).map(|(_, bytes, _, _)| *bytes).sum(); - let transaction = conn.transaction().map_err(|e| e.to_string())?; + let transaction = + conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; for key in &deleted { transaction .execute("DELETE FROM tab_runtime_cache WHERE cache_key = ?1", [key]) @@ -4349,7 +4399,7 @@ mod tests { ConnectionConfig, DatabaseConnectionInfo, DatabaseType, SshTunnelConfig, TransportLayerConfig, }; use crate::saved_sql::SavedSqlFile; - use rusqlite::Connection; + use rusqlite::{Connection, TransactionBehavior}; use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; fn temp_db_path(name: &str) -> std::path::PathBuf { @@ -4643,6 +4693,32 @@ mod tests { let _ = std::fs::remove_file(path); } + #[tokio::test] + async fn concurrent_save_connections_from_two_connections_does_not_lock() { + // Regression test for issue #6605: multiple `dbx` processes sharing one + // data directory (e.g. a portable install shared by several users on + // the same machine) each open their own connection to the same + // `dbx.db`. Opening two independent `Storage` instances here exercises + // the same inter-connection SQLite file locking that separate OS + // processes would hit. + let path = temp_db_path("concurrent-save-connections"); + let storage_a = std::sync::Arc::new(Storage::open(&path).await.unwrap()); + let storage_b = std::sync::Arc::new(Storage::open(&path).await.unwrap()); + + let mut tasks = Vec::new(); + for i in 0..20 { + let storage = if i % 2 == 0 { storage_a.clone() } else { storage_b.clone() }; + let config = plain_connection(&format!("concurrent-{i}"), "hunter2"); + tasks.push(tokio::spawn(async move { storage.save_connections(std::slice::from_ref(&config)).await })); + } + + for task in tasks { + task.await.unwrap().expect("concurrent save_connections should not fail with 'database is locked'"); + } + + let _ = std::fs::remove_file(path); + } + #[tokio::test] async fn save_connections_persists_password_when_save_password_true() { let path = temp_db_path("save-password-true"); @@ -5126,7 +5202,7 @@ mod tests { let inserted_json = future_json.clone(); storage .with_conn(move |conn| { - let tx = conn.transaction().map_err(|e| e.to_string())?; + let tx = conn.transaction_with_behavior(TransactionBehavior::Immediate).map_err(|e| e.to_string())?; tx.execute( "INSERT INTO connections (id, config_json) VALUES (?1, ?2)", rusqlite::params!["future", inserted_json], @@ -5727,7 +5803,9 @@ mod tests { let storage = Storage::open(&path).await.unwrap(); storage .with_conn(|conn| { - let transaction = conn.transaction().map_err(|error| error.to_string())?; + let transaction = conn + .transaction_with_behavior(TransactionBehavior::Immediate) + .map_err(|error| error.to_string())?; { let mut statement = transaction .prepare(