github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-15 14:52:36 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(query): recover expired manual transaction sessions

Browse Source
This commit is contained in:
t8y2 committed 2026-08-24 12:19:27 +08:00
1 parent 0047d79974
commit e03cc2d269
15 files changed
+252 -24
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/backend_error.rs
+42
View File
@@ -234,6 +234,24 @@ impl BackendError {
)
}
/// Create a retryable envelope for a manual transaction session that DBX
/// already rolled back before the requested SQL was executed.
pub fn from_manual_transaction_session_expired(timeout_secs: u64) -> Self {
let entry = catalog_entry(CatalogCode::TransactionSessionExpired);
let mut params = BTreeMap::new();
params
.insert("timeoutSecs".to_string(), BackendMessageParam::Integer(timeout_secs.min(i64::MAX as u64) as i64));
Self::new(
entry,
BackendErrorSource::LegacyBackend,
BackendErrorOrigin::database(BackendErrorAdapter::Native),
BackendOperationOutcome::NotStarted,
params,
None,
Some(diagnostics_for_local("transaction", AgentErrorStage::Execute)),
)
}
/// Create a SQL failure envelope while retaining bounded native driver detail.
///
/// The caller must pass a typed SQL execution error, not a connection or
@@ -364,6 +382,7 @@ enum CatalogCode {
ConnectionInterrupted,
TimeoutNotStarted,
TimeoutUnknown,
TransactionSessionExpired,
Canceled,
BusyRetryLater,
RuntimeReplaced,
@@ -396,6 +415,7 @@ struct CatalogEntry {
}
const STAGE_PARAM: &[ParamSpec] = &[ParamSpec { name: "stage", kind: ParamKind::String }];
const TRANSACTION_TIMEOUT_PARAM: &[ParamSpec] = &[ParamSpec { name: "timeoutSecs", kind: ParamKind::Integer }];
const NO_PARAMS: &[ParamSpec] = &[];
fn catalog_entry(code: CatalogCode) -> &'static CatalogEntry {
@@ -436,6 +456,15 @@ fn catalog_entry(code: CatalogCode) -> &'static CatalogEntry {
help_url: None,
},
),
(
CatalogCode::TransactionSessionExpired,
CatalogEntry {
code: "DBX-TXN-1001",
message_key: "backendErrors.transaction.sessionExpired",
params: TRANSACTION_TIMEOUT_PARAM,
help_url: None,
},
),
(
CatalogCode::Canceled,
CatalogEntry {
@@ -1031,6 +1060,7 @@ mod tests {
(CatalogCode::ConnectionInterrupted, "DBX-JDBC-1002", "backendErrors.jdbc.connectionInterrupted"),
(CatalogCode::TimeoutNotStarted, "DBX-JDBC-2001", "backendErrors.jdbc.operationTimedOut"),
(CatalogCode::TimeoutUnknown, "DBX-JDBC-2002", "backendErrors.jdbc.operationTimedOut"),
(CatalogCode::TransactionSessionExpired, "DBX-TXN-1001", "backendErrors.transaction.sessionExpired"),
(CatalogCode::Canceled, "DBX-JDBC-2003", "backendErrors.jdbc.operationCanceled"),
(CatalogCode::BusyRetryLater, "DBX-JDBC-3001", "backendErrors.jdbc.busyRetryLater"),
(CatalogCode::RuntimeReplaced, "DBX-JDBC-3002", "backendErrors.jdbc.runtimeReplaced"),
@@ -1049,6 +1079,18 @@ mod tests {
}
}
#[test]
fn manual_transaction_session_expiry_is_retryable_before_execution() {
let error = BackendError::from_manual_transaction_session_expired(300);
let payload = serde_json::to_value(&error).unwrap();
assert_eq!(error.code(), "DBX-TXN-1001");
assert_eq!(error.operation_outcome(), BackendOperationOutcome::NotStarted);
assert_eq!(payload["messageKey"], "backendErrors.transaction.sessionExpired");
assert_eq!(payload["messageParams"]["timeoutSecs"], 300);
assert_eq!(payload["diagnostics"]["category"], "transaction");
}
#[test]
fn catalog_maps_valid_variants_and_emits_allowlisted_params() {
let cases = [
crates/dbx-core/src/query.rs
+22 -21
View File
@@ -33,6 +33,15 @@ pub const MAX_ROWS: usize = 10000;
pub const AGENT_PROTOCOL_MAX_ROWS: usize = i32::MAX as usize;
pub const QUERY_CANCELED: &str = "Query canceled";
pub const METADATA_POOL_BUSY_ERROR: &str = "DBX metadata pool is busy; please retry";
pub const MANUAL_TRANSACTION_IDLE_TIMEOUT_SECS: u64 = 300;
pub const MANUAL_TRANSACTION_SESSION_NOT_FOUND_ERROR: &str =
"Transaction session not found or expired; it may have been auto-rolled back due to inactivity";
pub const MANUAL_TRANSACTION_IDLE_ROLLBACK_ERROR: &str =
"Transaction was auto-rolled back due to 5 minutes of inactivity";
pub fn is_manual_transaction_session_expired_error(error: &str) -> bool {
error.starts_with(MANUAL_TRANSACTION_SESSION_NOT_FOUND_ERROR) || error == MANUAL_TRANSACTION_IDLE_ROLLBACK_ERROR
}
/// Returns true when a metadata request failed because all pool/client slots
/// were temporarily occupied. This is deliberately separate from
@@ -4476,9 +4485,8 @@ pub async fn keep_manual_transaction_alive(
) -> Result<ManualTransactionKeepAlive, String> {
{
let mut sessions = state.transaction_sessions.write().await;
let session = sessions.get_mut(txn_session_id).ok_or_else(|| {
"Transaction session not found or expired; it may have been auto-rolled back due to inactivity".to_string()
})?;
let session =
sessions.get_mut(txn_session_id).ok_or_else(|| MANUAL_TRANSACTION_SESSION_NOT_FOUND_ERROR.to_string())?;
if !session.busy {
session.last_activity = std::time::Instant::now();
}
@@ -4534,16 +4542,14 @@ pub async fn execute_in_manual_transaction_with_options(
max_rows: Option<usize>,
table_data_preview: bool,
) -> Result<Vec<ExecuteMultiResult>, String> {
const TXN_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);
const TXN_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(MANUAL_TRANSACTION_IDLE_TIMEOUT_SECS);
// Resolve statements and validate before taking the per-session connection
// lock. The session stays visible in the map so close/disconnect cleanup can
// remove it and roll back once the current DB operation releases the lock.
let (pool_key, connection_id) = {
let sessions = state.transaction_sessions.read().await;
let session = sessions
.get(txn_session_id)
.ok_or("Transaction session not found or expired; it may have been auto-rolled back due to inactivity")?;
let session = sessions.get(txn_session_id).ok_or(MANUAL_TRANSACTION_SESSION_NOT_FOUND_ERROR)?;
(session.pool_key.clone(), session.connection_id.clone())
};
@@ -4566,10 +4572,7 @@ pub async fn execute_in_manual_transaction_with_options(
let connection = {
let mut sessions = state.transaction_sessions.write().await;
let Some(session) = sessions.get_mut(txn_session_id) else {
return Err(
"Transaction session not found or expired; it may have been auto-rolled back due to inactivity"
.to_string(),
);
return Err(MANUAL_TRANSACTION_SESSION_NOT_FOUND_ERROR.to_string());
};
if session.busy {
return Err("Transaction session is already executing".to_string());
@@ -4587,7 +4590,7 @@ pub async fn execute_in_manual_transaction_with_options(
let mut conn = session.connection.lock().await;
let _ = rollback_manual_txn_connection(&mut conn).await;
release_manual_txn_agent_pool(state, &session.connection_id, &mut conn).await;
return Err("Transaction was auto-rolled back due to 5 minutes of inactivity".to_string());
return Err(MANUAL_TRANSACTION_IDLE_ROLLBACK_ERROR.to_string());
}
let connection = {
@@ -4595,7 +4598,7 @@ pub async fn execute_in_manual_transaction_with_options(
sessions
.get(txn_session_id)
.map(|session| Arc::clone(&session.connection))
.ok_or("Transaction session not found or expired; it may have been auto-rolled back due to inactivity")?
.ok_or(MANUAL_TRANSACTION_SESSION_NOT_FOUND_ERROR)?
};
let row_limit = max_rows.unwrap_or(MAX_ROWS).max(1);
let mut results = Vec::with_capacity(statements.len());
@@ -4674,15 +4677,12 @@ pub async fn stream_rows_in_manual_transaction<F>(
where
F: FnMut(Vec<Vec<serde_json::Value>>) -> Result<(), String> + Send,
{
const TXN_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);
const TXN_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(MANUAL_TRANSACTION_IDLE_TIMEOUT_SECS);
let expired_connection = {
let mut sessions = state.transaction_sessions.write().await;
let Some(session) = sessions.get_mut(txn_session_id) else {
return Err(
"Transaction session not found or expired; it may have been auto-rolled back due to inactivity"
.to_string(),
);
return Err(MANUAL_TRANSACTION_SESSION_NOT_FOUND_ERROR.to_string());
};
if session.busy {
return Err("Transaction session is already executing".to_string());
@@ -4698,7 +4698,7 @@ where
if let Some(connection) = expired_connection {
let mut conn = connection.lock().await;
let _ = rollback_manual_txn_connection(&mut conn).await;
return Err("Transaction was auto-rolled back due to 5 minutes of inactivity".to_string());
return Err(MANUAL_TRANSACTION_IDLE_ROLLBACK_ERROR.to_string());
}
let connection = {
@@ -4706,7 +4706,7 @@ where
sessions
.get(txn_session_id)
.map(|session| Arc::clone(&session.connection))
.ok_or("Transaction session not found or expired; it may have been auto-rolled back due to inactivity")?
.ok_or(MANUAL_TRANSACTION_SESSION_NOT_FOUND_ERROR)?
};
let batch_size = batch_size.max(1);
let mut conn = connection.lock().await;
@@ -4882,7 +4882,8 @@ fn spawn_txn_idle_watcher_for_sessions(
txn_session_id: String,
) {
tokio::spawn(async move {
const TXN_IDLE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(300);
const TXN_IDLE_TIMEOUT: std::time::Duration =
std::time::Duration::from_secs(MANUAL_TRANSACTION_IDLE_TIMEOUT_SECS);
tokio::time::sleep(TXN_IDLE_TIMEOUT).await;
let removed: Option<TransactionSession> = {