fix(approval): clean_expired 按 code_hash 定位删除,修复并发 index 错位
P0-5: 原实现快照 (uid, index, code_hash) 后释放锁更新 DB,再重新加锁
按 index remove。若两阶段间 handle_reply 增删了条目,index 指向
错误条目,可能把未过期的审批误标 Expired 并通知等待方。
改为两阶段都按 code_hash 精确定位:收集过期 hash → 更新 DB →
重新加锁按 hash 查找删除,并发回复只会让查找安全跳过。
This commit is contained in:
+23
-11
@@ -192,22 +192,32 @@ impl ApprovalManager {
|
||||
}
|
||||
|
||||
/// 清理过期审批(按时间判定,通知等待方 + 更新 DB)
|
||||
///
|
||||
/// 两阶段加锁:收集过期 code_hash → 更新 DB → 重新加锁按 code_hash 定位删除。
|
||||
/// 不依赖快照 index,避免并发回复导致 index 错位误删。
|
||||
pub async fn clean_expired(&self) {
|
||||
let now = Instant::now();
|
||||
let mut expired = Vec::new();
|
||||
|
||||
// 第一阶段:持锁收集过期条目的 code_hash
|
||||
let mut expired_hashes: Vec<String> = Vec::new();
|
||||
{
|
||||
let map = self.pending.lock().await;
|
||||
for (uid, entries) in map.iter() {
|
||||
for (i, e) in entries.iter().enumerate() {
|
||||
for entries in map.values() {
|
||||
for e in entries {
|
||||
if now > e.expires_at {
|
||||
expired.push((uid.clone(), i, e.code_hash.clone()));
|
||||
expired_hashes.push(e.code_hash.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if expired_hashes.is_empty() {
|
||||
return;
|
||||
}
|
||||
|
||||
// DB 更新(按 code_hash 精确匹配,无错位问题)
|
||||
if let Some(ref pool) = self.pool {
|
||||
for (_uid, _i, code_hash) in &expired {
|
||||
for code_hash in &expired_hashes {
|
||||
let _ = sqlx::query(
|
||||
"UPDATE pending_approvals SET status = 'expired' WHERE code_hash = $1 AND status = 'pending'",
|
||||
)
|
||||
@@ -217,13 +227,15 @@ impl ApprovalManager {
|
||||
}
|
||||
}
|
||||
|
||||
// 第二阶段:重新持锁,按 code_hash 定位并移除
|
||||
// (即使两阶段间 handle_reply 增删了条目,按 hash 查找也能安全跳过)
|
||||
let mut map = self.pending.lock().await;
|
||||
for (uid, i, _) in expired.iter().rev() {
|
||||
if let Some(entries) = map.get_mut(uid)
|
||||
&& *i < entries.len()
|
||||
{
|
||||
let entry = entries.remove(*i);
|
||||
let _ = entry.sender.send(ApprovalDecision::Expired);
|
||||
for code_hash in &expired_hashes {
|
||||
for entries in map.values_mut() {
|
||||
if let Some(idx) = entries.iter().position(|e| &e.code_hash == code_hash) {
|
||||
let entry = entries.remove(idx);
|
||||
let _ = entry.sender.send(ApprovalDecision::Expired);
|
||||
}
|
||||
}
|
||||
}
|
||||
map.retain(|_, v| !v.is_empty());
|
||||
|
||||
Reference in New Issue
Block a user