feat(core): update pool activity timestamps atomically

This commit is contained in:
vrustx
2026-07-19 08:52:29 +08:00
committed by GitHub
parent 3da3849953
commit 285196894e
+68 -12
View File
@@ -199,10 +199,23 @@ pub struct AppState {
pub mq_registry: crate::mq::MqAdminRegistry,
}
#[derive(Clone, Copy)]
/// 活跃时间以进程内单调时钟的相对毫秒存储(AtomicU64):热路径每条查询都要
/// 更新它,读锁 + 原子写让并发查询不再在全局写锁上串行化。
/// 基准偏移让测试能构造"过去"的时间点(否则进程刚启动时相对毫秒接近 0)。
const POOL_ACTIVITY_BASE_MS: u64 = 86_400_000;
fn pool_activity_epoch() -> Instant {
static EPOCH: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
*EPOCH.get_or_init(Instant::now)
}
fn pool_activity_now_ms() -> u64 {
POOL_ACTIVITY_BASE_MS + pool_activity_epoch().elapsed().as_millis() as u64
}
#[cfg_attr(not(test), allow(dead_code))]
struct PoolActivity {
last_used_at: Instant,
last_used_at_ms: std::sync::atomic::AtomicU64,
}
#[derive(Clone, Copy)]
@@ -213,7 +226,29 @@ struct ConnectionAttemptState {
impl PoolActivity {
fn now() -> Self {
Self { last_used_at: Instant::now() }
Self { last_used_at_ms: std::sync::atomic::AtomicU64::new(pool_activity_now_ms()) }
}
fn touch(&self) {
// fetch_max 防止"先算后写"交错导致时间戳倒退(A 算得 100 被挂起,
// B 写入 200 后 A 再写 100)
self.last_used_at_ms.fetch_max(pool_activity_now_ms(), std::sync::atomic::Ordering::Relaxed);
}
#[cfg_attr(not(test), allow(dead_code))]
fn elapsed(&self) -> Duration {
Duration::from_millis(
pool_activity_now_ms().saturating_sub(self.last_used_at_ms.load(std::sync::atomic::Ordering::Relaxed)),
)
}
#[cfg(test)]
fn idle_for(idle: Duration) -> Self {
Self {
last_used_at_ms: std::sync::atomic::AtomicU64::new(
pool_activity_now_ms().saturating_sub(idle.as_millis() as u64),
),
}
}
}
@@ -233,6 +268,10 @@ impl Drop for PoolActivityTouch {
if !connections.read().await.contains_key(&pool_key) {
return;
}
if let Some(activity) = pool_activity.read().await.get(&pool_key) {
activity.touch();
return;
}
pool_activity.write().await.insert(pool_key, PoolActivity::now());
});
}
@@ -983,6 +1022,11 @@ impl AppState {
}
pub async fn touch_pool_activity(&self, pool_key: &str) {
// 热路径:读锁下原子更新;仅条目缺失(首次/已被清理)才退化为写锁插入
if let Some(activity) = self.pool_activity.read().await.get(pool_key) {
activity.touch();
return;
}
self.pool_activity.write().await.insert(pool_key.to_string(), PoolActivity::now());
}
@@ -4722,16 +4766,17 @@ mod tests {
let pool_key = "conn:session:tab-1";
state.connections.write().await.insert(pool_key.to_string(), PoolKind::Sqlite(pool));
state.pool_activity.write().await.insert(
pool_key.to_string(),
super::PoolActivity { last_used_at: std::time::Instant::now() - std::time::Duration::from_secs(10) },
);
state
.pool_activity
.write()
.await
.insert(pool_key.to_string(), super::PoolActivity::idle_for(std::time::Duration::from_secs(10)));
{
let _touch = state.pool_activity_touch(pool_key);
}
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
let elapsed = state.pool_activity.read().await.get(pool_key).unwrap().last_used_at.elapsed();
let elapsed = state.pool_activity.read().await.get(pool_key).unwrap().elapsed();
assert!(elapsed < std::time::Duration::from_secs(10));
{
@@ -4745,6 +4790,16 @@ mod tests {
let _ = std::fs::remove_dir_all(dir);
}
#[test]
fn pool_activity_touch_never_moves_backwards() {
let activity = super::PoolActivity::idle_for(std::time::Duration::from_secs(0));
let newer = u64::MAX;
activity.last_used_at_ms.store(newer, std::sync::atomic::Ordering::Relaxed);
// touch 的当前时间早于已存值:不得倒退
activity.touch();
assert_eq!(activity.last_used_at_ms.load(std::sync::atomic::Ordering::Relaxed), newer);
}
#[tokio::test]
async fn session_scoped_pool_is_not_closed_by_idle_timeout() {
let (state, dir) = test_app_state().await;
@@ -4755,10 +4810,11 @@ mod tests {
config.keepalive_interval_secs = 0;
state.connections.write().await.insert(pool_key.to_string(), PoolKind::Sqlite(pool));
state.pool_activity.write().await.insert(
pool_key.to_string(),
super::PoolActivity { last_used_at: std::time::Instant::now() - std::time::Duration::from_secs(10) },
);
state
.pool_activity
.write()
.await
.insert(pool_key.to_string(), super::PoolActivity::idle_for(std::time::Duration::from_secs(10)));
let pool = super::clone_pool_kind(state.connections.read().await.get(pool_key).unwrap());
state.start_keepalive_task(pool_key, &pool, &config).await;