perf(execution): prewarm tab session pools and report perceived wait
This commit is contained in:
17 files changed
+558
-29
No files matched your search
@@ -58,6 +58,12 @@ const DEFAULT_AGENT_CONNECT_TIMEOUT_SECS: u64 = 30;
|
||||
const ACCESS_AGENT_CONNECT_TIMEOUT_SECS: u64 = 30;
|
||||
const POOL_CLOSE_TIMEOUT_SECS: u64 = 3;
|
||||
const HEALTH_CHECK_POOL_ACQUIRE_TIMEOUT: Duration = Duration::from_millis(500);
|
||||
/// Upper bound for the "is this checked-out connection still alive" query that
|
||||
/// follows a successful health checkout. Windows keeps retransmitting on a
|
||||
/// half-open TCP connection for ~21s before the read fails, so a probe without
|
||||
/// its own budget would make `check_connection_health` (and therefore
|
||||
/// `ensureConnected`) hang for the whole OS retry window.
|
||||
const HEALTH_CHECK_PROBE_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
pub(crate) const METADATA_POOL_ACQUIRE_TIMEOUT: Duration = Duration::from_secs(5);
|
||||
pub(crate) const METADATA_POOL_DEFAULT_LIMIT: usize = 6;
|
||||
pub(crate) const METADATA_POOL_SQLSERVER_LIMIT: usize = 1;
|
||||
@@ -4006,7 +4012,6 @@ impl AppState {
|
||||
}
|
||||
PoolKind::Postgres(pool) => {
|
||||
let pool = pool.clone();
|
||||
let timeout = crate::db::connection_timeout();
|
||||
match db::postgres::checkout_postgres_client_classified(
|
||||
&pool,
|
||||
None,
|
||||
@@ -4014,20 +4019,33 @@ impl AppState {
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(client) => match tokio::time::timeout(timeout, client.simple_query("SELECT 1")).await {
|
||||
Ok(Ok(_)) => false,
|
||||
Ok(Err(err)) => {
|
||||
log::warn!("PostgreSQL connection pool '{pool_key}' is stale: {err}");
|
||||
true
|
||||
Ok(client) => {
|
||||
match tokio::time::timeout(HEALTH_CHECK_PROBE_TIMEOUT, client.simple_query("SELECT 1"))
|
||||
.await
|
||||
{
|
||||
Ok(Ok(_)) => false,
|
||||
Ok(Err(err)) => {
|
||||
log::warn!("PostgreSQL connection pool '{pool_key}' is stale: {err}");
|
||||
true
|
||||
}
|
||||
Err(_) => {
|
||||
log::warn!(
|
||||
"PostgreSQL connection pool '{pool_key}' is stale: health check timed out"
|
||||
);
|
||||
true
|
||||
}
|
||||
}
|
||||
Err(_) => {
|
||||
log::warn!("PostgreSQL connection pool '{pool_key}' is stale: health check timed out");
|
||||
true
|
||||
}
|
||||
},
|
||||
Err(err) if err.is_pool_saturation() => {
|
||||
}
|
||||
// The 500 ms probe budget is intentionally shorter than a foreground checkout. A timeout
|
||||
// while waiting, creating, or recycling is inconclusive: slow remote handshakes, pool
|
||||
// re-creation after a keepalive eviction, and active metadata exports can legitimately
|
||||
// exceed it. Removing the pool here would start competing reconnects while useful work is
|
||||
// still running, which is exactly how a sub-second probe turns into a multi-second wait on
|
||||
// the user's next statement. Keep the pool and let the executor's ReconnectAndRetry path
|
||||
// decide, matching the MySQL branch above.
|
||||
Err(err @ db::PoolCheckoutError::Timeout { .. }) => {
|
||||
log::debug!(
|
||||
"PostgreSQL connection pool '{pool_key}' is busy; skipping health probe: {err}"
|
||||
"PostgreSQL connection pool '{pool_key}' did not finish a health checkout; keeping pool: {err}"
|
||||
);
|
||||
false
|
||||
}
|
||||
@@ -5162,6 +5180,33 @@ impl AppState {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Warm the driver/pool a tab is about to use, off the user's critical path.
|
||||
///
|
||||
/// The first statement of a session pays costs that the user perceives as
|
||||
/// "the query is still loading" but that never appear in the reported
|
||||
/// statement duration: creating the pool, spawning a JDBC/agent driver
|
||||
/// session (JVM startup for external drivers such as Oracle), opening
|
||||
/// tunnels, and completing TLS/startup handshakes. `get_or_create_pool_*`
|
||||
/// performs exactly that work and verifies connectivity before returning, so
|
||||
/// calling it while the editor is being opened moves those seconds from the
|
||||
/// first Run to a moment where nobody is waiting on the result.
|
||||
///
|
||||
/// This is deliberately *not* a health probe: it never tears an existing
|
||||
/// pool down. Use `check_connection_health` when the caller needs a verdict.
|
||||
pub async fn prewarm_connection_pool(
|
||||
&self,
|
||||
connection_id: &str,
|
||||
database: Option<&str>,
|
||||
catalog: Option<&str>,
|
||||
client_session_id: Option<&str>,
|
||||
) -> Result<(), String> {
|
||||
let pool_key = self
|
||||
.get_or_create_pool_for_session_with_catalog(connection_id, database, catalog, client_session_id)
|
||||
.await?;
|
||||
self.touch_pool_activity(&pool_key).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn refresh_connections(&self) {
|
||||
// Clone pool handles under a short-lived read lock, then release it
|
||||
// before performing I/O-heavy health checks to avoid blocking writers.
|
||||
@@ -5188,19 +5233,30 @@ impl AppState {
|
||||
},
|
||||
PoolKind::Postgres(p) => {
|
||||
match db::postgres::checkout_postgres_client_classified(p, None, timeout).await {
|
||||
Ok(client) => match tokio::time::timeout(timeout, client.simple_query("SELECT 1")).await {
|
||||
Ok(Ok(_)) => true,
|
||||
Ok(Err(e)) => {
|
||||
log::warn!("PostgreSQL connection pool '{key}' is unhealthy: {e}");
|
||||
false
|
||||
Ok(client) => {
|
||||
match tokio::time::timeout(HEALTH_CHECK_PROBE_TIMEOUT, client.simple_query("SELECT 1"))
|
||||
.await
|
||||
{
|
||||
Ok(Ok(_)) => true,
|
||||
Ok(Err(e)) => {
|
||||
log::warn!("PostgreSQL connection pool '{key}' is unhealthy: {e}");
|
||||
false
|
||||
}
|
||||
Err(_) => {
|
||||
log::warn!(
|
||||
"PostgreSQL connection pool '{key}' is unhealthy: health check timed out"
|
||||
);
|
||||
false
|
||||
}
|
||||
}
|
||||
Err(_) => {
|
||||
log::warn!("PostgreSQL connection pool '{key}' is unhealthy: health check timed out");
|
||||
false
|
||||
}
|
||||
},
|
||||
Err(error) if error.is_pool_saturation() => {
|
||||
log::debug!("PostgreSQL connection pool '{key}' is busy; skipping health probe: {error}");
|
||||
}
|
||||
// A checkout timeout is inconclusive rather than proof of a dead pool: the budget can be
|
||||
// consumed by a concurrent create/recycle, and tearing the pool down here would make the
|
||||
// next foreground statement pay a full reconnect. Mirror `remove_stale_connection_pool`.
|
||||
Err(error @ db::PoolCheckoutError::Timeout { .. }) => {
|
||||
log::debug!(
|
||||
"PostgreSQL connection pool '{key}' did not finish a health checkout; keeping pool: {error}"
|
||||
);
|
||||
true
|
||||
}
|
||||
Err(error) => {
|
||||
@@ -7996,6 +8052,55 @@ mod tests {
|
||||
let _ = std::fs::remove_dir_all(dir);
|
||||
}
|
||||
|
||||
// A health probe that cannot finish its checkout within the 500 ms budget is
|
||||
// inconclusive, not proof of a dead pool. Tearing the pool down used to turn
|
||||
// one slow probe into a full reconnect on the user's next statement, which
|
||||
// showed up as a loading indicator of several seconds next to a summary that
|
||||
// only counted the statement itself.
|
||||
#[tokio::test]
|
||||
async fn postgres_health_check_keeps_pool_when_checkout_budget_is_exhausted() {
|
||||
// Accept connections but never answer the PostgreSQL startup packet, so
|
||||
// every attempt to create a connection runs into its own timeout.
|
||||
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
|
||||
let address = listener.local_addr().unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let mut held = Vec::new();
|
||||
while let Ok((socket, _)) = listener.accept().await {
|
||||
held.push(socket);
|
||||
}
|
||||
});
|
||||
let mut pg_config = tokio_postgres::Config::new();
|
||||
pg_config.host(address.ip().to_string()).port(address.port()).user("health-probe").dbname("health-probe");
|
||||
let manager = deadpool_postgres::Manager::new(pg_config, tokio_postgres::NoTls);
|
||||
let pool = deadpool_postgres::Pool::builder(manager)
|
||||
.runtime(deadpool_postgres::Runtime::Tokio1)
|
||||
.max_size(2)
|
||||
.wait_timeout(Some(Duration::from_millis(200)))
|
||||
.create_timeout(Some(Duration::from_millis(200)))
|
||||
.recycle_timeout(Some(Duration::from_millis(200)))
|
||||
.build()
|
||||
.expect("build PostgreSQL health probe pool");
|
||||
let (state, dir) = test_app_state().await;
|
||||
state.connections.write().await.insert("conn".to_string(), PoolKind::Postgres(pool.clone()));
|
||||
|
||||
let started = Instant::now();
|
||||
assert!(!state.remove_stale_connection_pool("conn").await);
|
||||
|
||||
assert!(
|
||||
started.elapsed() >= Duration::from_millis(150),
|
||||
"probe must actually run into the checkout budget, took {:?}",
|
||||
started.elapsed()
|
||||
);
|
||||
assert!(
|
||||
matches!(state.connections.read().await.get("conn"), Some(PoolKind::Postgres(_))),
|
||||
"an inconclusive PostgreSQL probe must not remove the pool"
|
||||
);
|
||||
state.connections.write().await.remove("conn");
|
||||
server.abort();
|
||||
pool.close();
|
||||
let _ = std::fs::remove_dir_all(dir);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mysql_health_check_keeps_pool_when_connection_creation_exceeds_probe_budget() {
|
||||
let listener = tokio::net::TcpListener::bind(("127.0.0.1", 0)).await.unwrap();
|
||||
|
||||
@@ -1720,6 +1720,13 @@ async fn do_execute_typed(
|
||||
let pool_db_type = connection_database_type_for_pool_key(state, pool_key).await;
|
||||
let mysql_catalog_dialect = connection_mysql_catalog_dialect_for_pool_key(state, pool_key).await;
|
||||
let pool = state.pool_handle(pool_key).await.ok_or("Connection not found")?;
|
||||
// Everything the driver dispatch does for this statement — pool checkout,
|
||||
// schema/search_path setup, the statement, and the post-statement cleanup —
|
||||
// is what the user actually waited for. Report that span instead of the
|
||||
// driver-internal timer, which starts only after a client is in hand and
|
||||
// therefore hides connection-pool stalls from the summary (#6097 fixed the
|
||||
// same mismatch for SQL Server's shared-connection lock).
|
||||
let dispatch_start = std::time::Instant::now();
|
||||
|
||||
let mut typed_agent_error = None;
|
||||
#[cfg(feature = "duckdb-sidecar")]
|
||||
@@ -2232,6 +2239,10 @@ async fn do_execute_typed(
|
||||
PoolKind::Consul(_) => Err("SQL execution is not supported for Consul connections".to_string()),
|
||||
};
|
||||
result
|
||||
.map(|mut result| {
|
||||
result.execution_time_ms = result.execution_time_ms.max(dispatch_start.elapsed().as_millis());
|
||||
result
|
||||
})
|
||||
.map_err(|error| {
|
||||
#[cfg(feature = "duckdb-sidecar")]
|
||||
if let Some(duckdb_error) = typed_duckdb_error {
|
||||
@@ -2721,6 +2732,10 @@ pub(crate) async fn execute_sql_statement_with_options_typed_with_outcome(
|
||||
let db_type = connection_database_type(state, connection_id).await;
|
||||
let invalidate = crate::object_cache::sql_may_change_object_metadata(sql, db_type);
|
||||
let mut transaction = None;
|
||||
// Same contract as the multi-statement entry point: a single statement is
|
||||
// reported with the duration of the whole request, including creating or
|
||||
// reconnecting the pool, so the summary matches the loading indicator.
|
||||
let request_start = std::time::Instant::now();
|
||||
let result = execute_sql_statement_with_options_typed_inner(
|
||||
state,
|
||||
connection_id,
|
||||
@@ -2735,7 +2750,10 @@ pub(crate) async fn execute_sql_statement_with_options_typed_with_outcome(
|
||||
if invalidate {
|
||||
crate::object_cache::invalidate_connection_object_cache(&state.storage, connection_id).await;
|
||||
}
|
||||
result.map(|result| (result, transaction))
|
||||
result.map(|mut result| {
|
||||
result.execution_time_ms = result.execution_time_ms.max(request_start.elapsed().as_millis());
|
||||
(result, transaction)
|
||||
})
|
||||
}
|
||||
|
||||
async fn recover_postgres_create_table_after_connection_error(
|
||||
@@ -3183,6 +3201,11 @@ pub async fn execute_multi_core_with_options_for_client_and_progress_typed(
|
||||
) -> Result<Vec<ExecuteMultiResult>, QueryExecutionError> {
|
||||
let db_type = connection_database_type(state, connection_id).await;
|
||||
let invalidate = crate::object_cache::sql_may_change_object_metadata(sql, db_type);
|
||||
// A run that produced a single statement result is reported back as one
|
||||
// duration, so make it the duration of the whole request: pool creation or
|
||||
// reconnection for a cold pool happens outside the statement dispatcher and
|
||||
// would otherwise be invisible while the loading indicator counts it.
|
||||
let request_start = std::time::Instant::now();
|
||||
let result = execute_multi_core_with_options_for_client_and_progress_typed_inner(
|
||||
state,
|
||||
connection_id,
|
||||
@@ -3197,7 +3220,14 @@ pub async fn execute_multi_core_with_options_for_client_and_progress_typed(
|
||||
if invalidate {
|
||||
crate::object_cache::invalidate_connection_object_cache(&state.storage, connection_id).await;
|
||||
}
|
||||
result
|
||||
result.map(|mut results| {
|
||||
if results.len() == 1 {
|
||||
if let Some(item) = results.first_mut() {
|
||||
item.result.execution_time_ms = item.result.execution_time_ms.max(request_start.elapsed().as_millis());
|
||||
}
|
||||
}
|
||||
results
|
||||
})
|
||||
}
|
||||
|
||||
async fn execute_multi_core_with_options_for_client_and_progress_typed_inner(
|
||||
|
||||
@@ -389,6 +389,7 @@ async fn main() {
|
||||
.route("/connection/final-proxy-port", post(routes::connection::connection_final_proxy_port))
|
||||
.route("/connection/disconnect", post(routes::connection::disconnect_db))
|
||||
.route("/connection/check-health", post(routes::connection::check_connection_health))
|
||||
.route("/connection/prewarm", post(routes::connection::prewarm_connection))
|
||||
.route("/connection/session-credential-status", post(routes::connection::session_credential_status))
|
||||
.route("/connection/forget-session-credential", post(routes::connection::forget_session_credential))
|
||||
.route(
|
||||
|
||||
@@ -106,6 +106,15 @@ pub struct DisconnectRequest {
|
||||
pub client_attempt: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct PrewarmConnectionRequest {
|
||||
pub connection_id: String,
|
||||
pub database: Option<String>,
|
||||
pub catalog: Option<String>,
|
||||
pub client_session_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct CloseDatabaseConnectionRequest {
|
||||
@@ -602,6 +611,21 @@ pub async fn check_connection_health(
|
||||
Ok(Json(()))
|
||||
}
|
||||
|
||||
pub async fn prewarm_connection(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(body): Json<PrewarmConnectionRequest>,
|
||||
) -> Result<Json<()>, AppError> {
|
||||
let database = body.database.as_deref().filter(|value| !value.is_empty());
|
||||
let catalog = body.catalog.as_deref().filter(|value| !value.is_empty());
|
||||
let client_session_id = body.client_session_id.as_deref().filter(|value| !value.is_empty());
|
||||
state
|
||||
.app
|
||||
.prewarm_connection_pool(&body.connection_id, database, catalog, client_session_id)
|
||||
.await
|
||||
.map_err(AppError::from)?;
|
||||
Ok(Json(()))
|
||||
}
|
||||
|
||||
/// 查询连接在本次运行期是否已输入并暂存密码(`save_password=false`)。
|
||||
/// 供前端决定是否需要弹密码框;仅返回布尔状态,不泄露密码本身。
|
||||
/// 按当前登录会话(owner)查询,不会暴露其他会话的凭据状态。
|
||||
|
||||
Reference in new issue
Block a user