diff --git a/crates/dbx-core/src/query/mod.rs b/crates/dbx-core/src/query/mod.rs index a78bebf277..12d7f395c9 100644 --- a/crates/dbx-core/src/query/mod.rs +++ b/crates/dbx-core/src/query/mod.rs @@ -1745,7 +1745,7 @@ async fn do_execute_typed( ) .await?; let execution_cancel_token = if options.await_cancel_completion { None } else { cancel_token }; - wait_for_result_opt( + let statement_result = wait_for_result_opt( execution_cancel_token, query_timeout, db::mysql::execute_query_on_conn_with_limits( @@ -1760,7 +1760,30 @@ async fn do_execute_typed( ), ) .await - .map(|result| result.result) + .map(|result| result.result); + + // Client-session pools hold one connection for the whole tab and + // skip COM_RESET_CONNECTION on return so session state survives + // across executions. A single-statement BEGIN / START TRANSACTION + // would therefore leave its transaction open and pin the + // connection's REPEATABLE READ read view, so every later + // auto-commit query in the tab would keep reading the same stale + // snapshot until the connection was closed. Clear it here, exactly + // like the multi-statement MySQL path does, so each execution + // restores the auto-commit contract (ROLLBACK is a server no-op + // when no transaction is open). + if p.is_client_session_pool() { + if let Err(error) = db::mysql::rollback_open_transaction(&mut conn).await { + log::warn!( + "[query][mysql] trace_id={} open_txn_rollback_failed error={}", + options.execution_id.as_deref().unwrap_or_default(), + error + ); + let _ = tokio::time::timeout(Duration::from_secs(5), conn.disconnect()).await; + state.remove_pool_by_key(pool_key).await; + } + } + statement_result } PoolKind::Postgres(p) => { let p = p.clone(); @@ -3716,7 +3739,7 @@ async fn execute_multi_mysql( // server no-op, and a failure here only discards this connection. { let rollback_started_at = std::time::Instant::now(); - match conn.query_drop("ROLLBACK").await { + match db::mysql::rollback_open_transaction(&mut conn).await { Ok(()) => { if rollback_started_at.elapsed() > std::time::Duration::from_millis(5) { log::info!( diff --git a/crates/dbx-core/tests/live_mysql_autocommit_snapshot.rs b/crates/dbx-core/tests/live_mysql_autocommit_snapshot.rs new file mode 100644 index 0000000000..8b88618c9f --- /dev/null +++ b/crates/dbx-core/tests/live_mysql_autocommit_snapshot.rs @@ -0,0 +1,138 @@ +//! Regression coverage for #9479: a single-statement `BEGIN` typed in an +//! auto-commit tab must not pin that tab's read snapshot. +//! +//! Run against a writable MySQL database: +//! `DBX_LIVE_AUTOCOMMIT_MYSQL_HOST=... DBX_LIVE_AUTOCOMMIT_MYSQL_USER=... \ +//! DBX_LIVE_AUTOCOMMIT_MYSQL_PASSWORD=... DBX_LIVE_AUTOCOMMIT_MYSQL_DATABASE=... \ +//! cargo test -p dbx-core --test live_mysql_autocommit_snapshot -- --ignored` + +use dbx_core::connection::AppState; +use dbx_core::models::connection::{ConnectionConfig, DatabaseType}; +use dbx_core::query::{execute_sql_statement, execute_sql_statement_with_options, QueryExecutionOptions}; +use dbx_core::storage::Storage; +use std::sync::Arc; + +fn live_config(prefix: &str) -> ConnectionConfig { + serde_json::from_value(serde_json::json!({ + "id": format!("autocommit-snapshot-{}", uuid::Uuid::new_v4().simple()), + "name": "MySQL auto-commit snapshot regression", + "db_type": DatabaseType::Mysql, + "host": std::env::var(format!("{prefix}_HOST")).expect("live database host"), + "port": std::env::var(format!("{prefix}_PORT")).ok().and_then(|port| port.parse::().ok()).unwrap_or(3306), + "username": std::env::var(format!("{prefix}_USER")).expect("live database user"), + "password": std::env::var(format!("{prefix}_PASSWORD")).expect("live database password"), + "database": std::env::var(format!("{prefix}_DATABASE")).expect("live database name"), + "connect_timeout_secs": 5, + "query_timeout_secs": 30, + "idle_timeout_secs": 60, + "keepalive_interval_secs": 0 + })) + .expect("live connection configuration") +} + +async fn setup(config: &ConnectionConfig) -> (Arc, std::path::PathBuf, String) { + let storage_path = + std::env::temp_dir().join(format!("dbx-autocommit-snapshot-{}.db", uuid::Uuid::new_v4().simple())); + let state = Arc::new(AppState::new(Storage::open(&storage_path).await.expect("temporary storage"))); + let table_name = format!("dbx_issue_9479_{}", uuid::Uuid::new_v4().simple()); + let database = config.database.clone().expect("database"); + state.configs.write().await.insert(config.id.clone(), config.clone()); + execute_sql_statement( + &state, + &config.id, + &database, + &format!("CREATE TABLE {table_name} (id INTEGER PRIMARY KEY)"), + None, + None, + ) + .await + .expect("create isolated regression table"); + execute_sql_statement( + &state, + &config.id, + &database, + &format!("INSERT INTO {table_name} (id) VALUES (1)"), + None, + None, + ) + .await + .expect("seed regression table"); + (state, storage_path, table_name) +} + +fn assert_row_count(rows: &[Vec], expected: i64) { + assert_eq!(rows.len(), 1); + assert_eq!(rows[0].len(), 1); + let count = rows[0][0].as_i64().or_else(|| rows[0][0].as_str().and_then(|value| value.parse().ok())); + assert_eq!(count, Some(expected)); +} + +/// Runs one statement on the tab-scoped client session that pools a single +/// connection across executions, exactly like a query editor tab. +async fn run_in_tab( + state: &AppState, + config: &ConnectionConfig, + database: &str, + client_session_id: &str, + sql: &str, +) -> dbx_core::db::QueryResult { + execute_sql_statement_with_options( + state, + &config.id, + database, + sql, + None, + None, + QueryExecutionOptions { client_session_id: Some(client_session_id.to_string()), ..Default::default() }, + ) + .await + .expect("execute statement in tab client session") +} + +async fn cleanup(state: &AppState, config: &ConnectionConfig, table_name: &str, storage_path: std::path::PathBuf) { + execute_sql_statement( + state, + &config.id, + config.database.as_deref().expect("database"), + &format!("DROP TABLE {table_name}"), + None, + None, + ) + .await + .expect("drop isolated regression table"); + let _ = std::fs::remove_file(storage_path); +} + +#[tokio::test] +#[ignore = "requires DBX_LIVE_AUTOCOMMIT_MYSQL_* pointing at a writable MySQL database"] +async fn live_mysql_single_statement_begin_does_not_pin_the_tab_snapshot() { + let config = live_config("DBX_LIVE_AUTOCOMMIT_MYSQL"); + let database = config.database.clone().expect("database"); + let (state, storage_path, table_name) = setup(&config).await; + let client_session_id = format!("query-tab-{}", uuid::Uuid::new_v4().simple()); + let count_sql = format!("SELECT COUNT(*) AS row_count FROM {table_name}"); + + // A user typing BEGIN as its own statement in an auto-commit tab used to + // leave the transaction open on the tab connection. + run_in_tab(&state, &config, &database, &client_session_id, "BEGIN").await; + let first = run_in_tab(&state, &config, &database, &client_session_id, &count_sql).await; + assert_row_count(&first.rows, 1); + + // Another connection commits a row while the tab keeps its own connection. + execute_sql_statement( + &state, + &config.id, + &database, + &format!("INSERT INTO {table_name} (id) VALUES (2)"), + None, + None, + ) + .await + .expect("external committed write"); + + // The tab must observe the committed row instead of its pinned snapshot. + let second = run_in_tab(&state, &config, &database, &client_session_id, &count_sql).await; + assert_row_count(&second.rows, 2); + + cleanup(&state, &config, &table_name, storage_path).await; +} diff --git a/crates/dbx-drivers/src/db/mysql.rs b/crates/dbx-drivers/src/db/mysql.rs index 1b9d087422..d5f5e7a816 100644 --- a/crates/dbx-drivers/src/db/mysql.rs +++ b/crates/dbx-drivers/src/db/mysql.rs @@ -52,6 +52,14 @@ impl MySqlPool { Self { inner: mysql_async::Pool::new(opts), max_connections: max_connections.max(1) } } + /// Whether this is a tab-scoped client-session pool. These pools hold a + /// single connection and disable `COM_RESET_CONNECTION` on return so that + /// session state (temporary tables, user variables, selected catalog) + /// survives across executions in the same tab. + pub fn is_client_session_pool(&self) -> bool { + self.max_connections == 1 + } + pub async fn disconnect(self) -> Result<(), mysql_async::Error> { self.inner.disconnect().await } @@ -137,6 +145,22 @@ impl MySqlPoolAccess for mysql_async::Pool { None } } + +/// Clears a transaction that is still open on a pooled MySQL connection. +/// +/// Client-session pools keep their connection across executions, so a +/// transaction left open by a statement (a user-typed `BEGIN` / +/// `START TRANSACTION` without `COMMIT`, or a canceled statement) would pin the +/// connection's REPEATABLE READ read view and make every later auto-commit +/// query on that tab read a stale snapshot until the connection is closed. +/// `ROLLBACK` is a server no-op when no transaction is open. +/// +/// An error means the transaction state is unknown, so the caller must discard +/// the connection instead of returning it to the pool. +pub async fn rollback_open_transaction(conn: &mut mysql_async::Conn) -> Result<(), String> { + conn.query_drop("ROLLBACK").await.map_err(|error| error.to_string()) +} + const MYSQL_TCP_KEEPALIVE_MS: u32 = 30_000; const MYSQL_SQL_PACKET_MARGIN_MAX_BYTES: usize = 64 * 1024;