From 62bf4544bf2383d670021f9dea4c9d98e2ef0def Mon Sep 17 00:00:00 2001 From: t8y2 Date: Thu, 24 Sep 2026 00:32:04 +0800 Subject: [PATCH] fix(mysql): skip redundant liveness pings and heartbeat resets --- crates/dbx-core/src/connection/mod.rs | 16 +++- crates/dbx-drivers/src/db/mysql.rs | 108 +++++++++++++++++++++++++- 2 files changed, 121 insertions(+), 3 deletions(-) diff --git a/crates/dbx-core/src/connection/mod.rs b/crates/dbx-core/src/connection/mod.rs index 7ba65b7889..fdf5d1a9ef 100644 --- a/crates/dbx-core/src/connection/mod.rs +++ b/crates/dbx-core/src/connection/mod.rs @@ -4044,8 +4044,14 @@ impl AppState { true } Ok(mut conn) => { + // The probe runs no statement, so a shared pool can take + // the connection back without COM_RESET_CONNECTION and the + // setup replay, and a connection verified moments ago (by + // this probe or the checkout that follows it) is not + // pinged again. + conn.reset_connection(false); let timeout = crate::db::connection_timeout(); - match tokio::time::timeout(timeout, conn.ping()).await { + match tokio::time::timeout(timeout, db::mysql::verify_pooled_conn(&pool, &mut conn)).await { Ok(Ok(())) => false, Ok(Err(err)) => { log::warn!("MySQL connection pool '{pool_key}' is stale: {err}"); @@ -5906,8 +5912,14 @@ fn keepalive_target_from_pool(pool: &PoolKind, config: &ConnectionConfig) -> Opt async fn ping_keepalive_target(target: &mut KeepaliveTarget, timeout: Duration) -> Result<(), KeepaliveError> { match target { KeepaliveTarget::Mysql(pool) => { + // The checkout health check is the keepalive round trip: it pings + // the idle connection (unless it was verified moments ago) and + // replaces it when it died. A ping leaves no session state behind, + // so return the connection without the COM_RESET_CONNECTION and + // setup replay a shared pool would run. let mut conn = db::mysql::get_conn_with_health_check(pool).await?; - conn.ping().await.map_err(|error| KeepaliveError::Legacy(error.to_string())) + conn.reset_connection(false); + Ok(()) } KeepaliveTarget::Postgres(pool) => { let client = pool.get().await.map_err(|e| format!("PostgreSQL pool error: {e}"))?; diff --git a/crates/dbx-drivers/src/db/mysql.rs b/crates/dbx-drivers/src/db/mysql.rs index c9b191056b..b1d5691096 100644 --- a/crates/dbx-drivers/src/db/mysql.rs +++ b/crates/dbx-drivers/src/db/mysql.rs @@ -41,6 +41,7 @@ use crate::mysql_event_sql::MysqlEventInfo; pub struct MySqlPool { inner: mysql_async::Pool, max_connections: usize, + checkout_verifications: std::sync::Arc, } impl MySqlPool { @@ -49,7 +50,11 @@ impl MySqlPool { mysql_async::Opts: TryFrom, >::Error: std::error::Error, { - Self { inner: mysql_async::Pool::new(opts), max_connections: max_connections.max(1) } + Self { + inner: mysql_async::Pool::new(opts), + max_connections: max_connections.max(1), + checkout_verifications: Default::default(), + } } /// Whether this is a tab-scoped client-session pool. These pools hold a @@ -86,6 +91,57 @@ impl Deref for MySqlPool { pub trait MySqlPoolAccess { fn driver_pool(&self) -> &mysql_async::Pool; fn checkout_max_connections(&self) -> Option; + /// Liveness evidence for pooled connections; `None` pings every checkout. + fn checkout_verifications(&self) -> Option<&MySqlCheckoutVerifications> { + None + } +} + +/// Connections that passed a `PING` this recently skip the next liveness +/// `PING`, like the Tomcat JDBC pool's `validationInterval` (3 s by default). +/// A burst of checkouts (opening a table, paging a grid, loading metadata) then +/// pays one liveness round trip instead of one each, while a connection that +/// sat idle longer is still verified and replaced when it died. +const MYSQL_CHECKOUT_VERIFY_INTERVAL: Duration = Duration::from_secs(3); + +/// When each pooled connection, keyed by its server connection id, last passed +/// a liveness `PING` (the checkout health check or the pool reuse probe). +#[doc(hidden)] +#[derive(Debug, Default)] +pub struct MySqlCheckoutVerifications { + verified_at: std::sync::Mutex>, +} + +impl MySqlCheckoutVerifications { + fn is_recent(&self, connection_id: u32, now: Instant) -> bool { + let verified_at = self.verified_at.lock().unwrap_or_else(|e| e.into_inner()); + verified_at + .get(&connection_id) + .is_some_and(|verified| now.saturating_duration_since(*verified) < MYSQL_CHECKOUT_VERIFY_INTERVAL) + } + + fn record(&self, connection_id: u32, now: Instant) { + let mut verified_at = self.verified_at.lock().unwrap_or_else(|e| e.into_inner()); + verified_at.retain(|_, verified| now.saturating_duration_since(*verified) < MYSQL_CHECKOUT_VERIFY_INTERVAL); + verified_at.insert(connection_id, now); + } +} + +/// Pings a connection checked out of `pool` unless it passed a `PING` within +/// the verification interval, and remembers a successful `PING`. +pub async fn verify_pooled_conn

(pool: &P, conn: &mut mysql_async::Conn) -> Result<(), mysql_async::Error> +where + P: MySqlPoolAccess + ?Sized, +{ + let verifications = pool.checkout_verifications(); + if verifications.is_some_and(|verifications| verifications.is_recent(conn.id(), Instant::now())) { + return Ok(()); + } + conn.ping().await?; + if let Some(verifications) = verifications { + verifications.record(conn.id(), Instant::now()); + } + Ok(()) } pub async fn get_event_info( @@ -134,6 +190,10 @@ impl MySqlPoolAccess for MySqlPool { fn checkout_max_connections(&self) -> Option { Some(self.max_connections) } + + fn checkout_verifications(&self) -> Option<&MySqlCheckoutVerifications> { + Some(&self.checkout_verifications) + } } impl MySqlPoolAccess for mysql_async::Pool { @@ -4497,9 +4557,16 @@ where P: MySqlPoolAccess + ?Sized, { let start = Instant::now(); + let verifications = pool.checkout_verifications(); let mut conn = get_conn_with_timeout_and_cancel(pool, timeout, cancel_token).await?; + if verifications.is_some_and(|verifications| verifications.is_recent(conn.id(), Instant::now())) { + return Ok(conn); + } match ping_conn_with_timeout_and_cancel(&mut conn, timeout, cancel_token).await { Ok(()) => { + if let Some(verifications) = verifications { + verifications.record(conn.id(), Instant::now()); + } log::debug!( "[db:health.check:done] elapsed_ms={} timeout_ms={}", start.elapsed().as_millis(), @@ -4526,6 +4593,9 @@ where } return Err(err); } + if let Some(verifications) = verifications { + verifications.record(conn.id(), Instant::now()); + } log::info!( "[db:health.check:recovered] elapsed_ms={} timeout_ms={}", start.elapsed().as_millis(), @@ -6504,6 +6574,42 @@ mod tests { assert_eq!(configured_five, Err("MySQL get connection timed out".to_string())); } + #[test] + fn checkout_verification_skips_only_a_recently_verified_connection() { + let verifications = MySqlCheckoutVerifications::default(); + let verified = Instant::now(); + assert!(!verifications.is_recent(7, verified)); + + verifications.record(7, verified); + + assert!(verifications.is_recent(7, verified + MYSQL_CHECKOUT_VERIFY_INTERVAL - Duration::from_millis(1))); + assert!(!verifications.is_recent(7, verified + MYSQL_CHECKOUT_VERIFY_INTERVAL)); + assert!(!verifications.is_recent(8, verified)); + } + + #[test] + fn checkout_verification_forgets_expired_connections() { + let verifications = MySqlCheckoutVerifications::default(); + let first = Instant::now(); + verifications.record(1, first); + + verifications.record(2, first + MYSQL_CHECKOUT_VERIFY_INTERVAL); + + let tracked: Vec = verifications.verified_at.lock().unwrap().keys().copied().collect(); + assert_eq!(tracked, vec![2]); + } + + #[tokio::test] + async fn raw_driver_pool_verifies_every_checkout() { + let raw = mysql_async::Pool::new(mysql_async::OptsBuilder::default()); + let pool = MySqlPool::new(mysql_async::OptsBuilder::default(), 1); + + assert!(raw.checkout_verifications().is_none()); + assert!(pool.checkout_verifications().is_some()); + let _ = tokio::time::timeout(Duration::from_secs(1), raw.disconnect()).await; + let _ = tokio::time::timeout(Duration::from_secs(1), pool.disconnect()).await; + } + #[tokio::test] async fn mysql_checkout_fault_injection_classifies_full_pool_as_wait() { assert!(tokio::time::timeout(Duration::from_millis(1), std::future::pending::<()>()).await.is_err());