fix(concurrency): 扫描修复——bash 取消泄漏与子代理队列阻塞
- bash: Command 补 kill_on_drop(true)。原先 /stop 的 select! 竞速 drop run_command future 时，Child detach 不杀进程、read_stream 任务因管道永不 EOF 而永久存活，每次取消泄漏孤儿进程+2任务；进程被杀后管道 EOF 同时解决读取任务泄漏 - runtime: sub_done send().await 改 5s 超时保护。队列满且无消费者时永久阻塞会占住全局 Semaphore permit，8 permit 耗尽后整个子代理子系统停摆；超时放弃结果由 DB 状态 + wait 对账兜底 - runtime: cancel_pending_for_topic / reap_orphan_subagents 的 parking_lot registry 锁不再跨 SQLite 调用持有（锁内收集、锁外落库），SQLite 卡顿时不再放大 /stop 延迟
This commit is contained in:
oudecheng
parent
5bc3f64ba2
commit
f010006a66
@ -465,6 +465,11 @@ impl BashTool {
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::piped())
|
||||
// 外部取消安全：run_command future 被 drop 时（/stop 的 select! 竞速、
|
||||
// 子代理超时等），Child drop 会终止 OS 进程。否则子进程变孤儿、
|
||||
// read_stream 任务因管道永不 EOF 而永久存活，每次取消泄漏一组资源。
|
||||
// 进程被杀后管道到达 EOF，read_stream 任务也随之自然退出。
|
||||
.kill_on_drop(true)
|
||||
.current_dir(cwd);
|
||||
|
||||
let mut child = cmd.spawn().map_err(|e| format!("Failed to spawn: {}", e))?;
|
||||
|
||||
@ -51,6 +51,37 @@ impl Drop for CancelRegistryGuard {
|
||||
}
|
||||
}
|
||||
|
||||
/// 超时保护地发送 SubagentResult。
|
||||
///
|
||||
/// sub_done_q 容量有限（32）且仅由 wait 协调器消费。若主代理从不调用 wait，
|
||||
/// 无超时的 `send().await` 会永久阻塞——而本函数运行在持有全局并发许可
|
||||
/// （Semaphore permit）的 spawn 任务末尾，阻塞会永久占住 permit，8 个 permit
|
||||
/// 耗尽后整个子代理子系统停摆。超时后放弃结果：DB 状态照常更新，
|
||||
/// wait 侧另有 query_pending_task_ids 对账兜底，不会丢失完成事实。
|
||||
async fn send_sub_done_timeout(
|
||||
sender: &tokio::sync::mpsc::Sender<SubagentResult>,
|
||||
result: SubagentResult,
|
||||
) {
|
||||
let task_id = result.task_id.clone();
|
||||
match tokio::time::timeout(Duration::from_secs(5), sender.send(result)).await {
|
||||
Ok(Ok(())) => {}
|
||||
Ok(Err(e)) => {
|
||||
tracing::warn!(
|
||||
error = %e,
|
||||
task_id = %task_id,
|
||||
"Failed to send SubagentResult to sub_done_q (receiver dropped?)"
|
||||
);
|
||||
}
|
||||
Err(_) => {
|
||||
tracing::warn!(
|
||||
task_id = %task_id,
|
||||
"Timed out sending SubagentResult (queue full, no consumer); \
|
||||
dropping result — DB status update + wait reconciliation cover this"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
use crate::agent::{
|
||||
AgentLoop, AgentRuntimeConfig, EmittedMessageHandler, PersistingEmittedMessageHandler,
|
||||
SystemPrompt, SystemPromptContext, SystemPromptProvider,
|
||||
@ -1118,7 +1149,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
})
|
||||
.unwrap_or_default(),
|
||||
};
|
||||
let _ = sub_done_sender.send(result).await;
|
||||
let _ = send_sub_done_timeout(&sub_done_sender, result).await;
|
||||
let _ = store.update_pending_subagent_status(&task_id_for_spawn, "failed");
|
||||
// _registry_guard drop 时清理 registry 条目
|
||||
return;
|
||||
@ -1183,13 +1214,7 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
output,
|
||||
pending_task_ids,
|
||||
};
|
||||
if let Err(e) = sub_done_sender.send(result).await {
|
||||
tracing::warn!(
|
||||
error = %e,
|
||||
task_id = %task_id_for_spawn,
|
||||
"Failed to send SubagentResult to sub_done_q (receiver dropped?)"
|
||||
);
|
||||
}
|
||||
send_sub_done_timeout(&sub_done_sender, result).await;
|
||||
|
||||
// UPDATE pending_subagents 状态
|
||||
let status_str = match status {
|
||||
@ -1419,47 +1444,55 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
"Cancelling pending subagents for topic"
|
||||
);
|
||||
|
||||
// 触发每个子代理的 CancellationToken
|
||||
let registry = self.cancel_registry.lock();
|
||||
for record in &running {
|
||||
if let Some(token) = registry.get(&record.task_id) {
|
||||
token.cancel();
|
||||
tracing::info!(
|
||||
task_id = %record.task_id,
|
||||
"Cancelled subagent token"
|
||||
);
|
||||
} else {
|
||||
// token 不在 registry 中（可能已完成但 DB 状态未更新，或进程重启后丢失）
|
||||
// 不变量 1：条件 UPDATE，仅在 status='running' 时转为 cancelled，
|
||||
// 避免 spawn 已完成的终态被覆盖（completed → cancelled 是非法转换）
|
||||
match self.store.try_update_pending_subagent_status(
|
||||
&record.task_id,
|
||||
"running",
|
||||
"cancelled",
|
||||
) {
|
||||
Ok(true) => {
|
||||
tracing::info!(
|
||||
task_id = %record.task_id,
|
||||
"Marked subagent as cancelled in DB (token not in registry)"
|
||||
);
|
||||
}
|
||||
Ok(false) => {
|
||||
tracing::info!(
|
||||
task_id = %record.task_id,
|
||||
"Subagent status already updated by another path, skip cancel"
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
error = %e,
|
||||
task_id = %record.task_id,
|
||||
"Failed to mark subagent as cancelled in DB"
|
||||
);
|
||||
}
|
||||
// 触发每个子代理的 CancellationToken。
|
||||
// 锁内只做内存操作（cancel + 收集），SQLite 调用移到锁外——
|
||||
// registry 是 parking_lot 同步锁，持锁跨 DB IO 会阻塞所有并发的
|
||||
// 注册/Guard::drop，SQLite 卡顿时放大 /stop 延迟。
|
||||
let mut missing_from_registry: Vec<String> = Vec::new();
|
||||
{
|
||||
let registry = self.cancel_registry.lock();
|
||||
for record in &running {
|
||||
if let Some(token) = registry.get(&record.task_id) {
|
||||
token.cancel();
|
||||
tracing::info!(
|
||||
task_id = %record.task_id,
|
||||
"Cancelled subagent token"
|
||||
);
|
||||
} else {
|
||||
missing_from_registry.push(record.task_id.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for task_id in &missing_from_registry {
|
||||
// token 不在 registry 中（可能已完成但 DB 状态未更新，或进程重启后丢失）
|
||||
// 不变量 1：条件 UPDATE，仅在 status='running' 时转为 cancelled，
|
||||
// 避免 spawn 已完成的终态被覆盖（completed → cancelled 是非法转换）
|
||||
match self
|
||||
.store
|
||||
.try_update_pending_subagent_status(task_id, "running", "cancelled")
|
||||
{
|
||||
Ok(true) => {
|
||||
tracing::info!(
|
||||
task_id = %task_id,
|
||||
"Marked subagent as cancelled in DB (token not in registry)"
|
||||
);
|
||||
}
|
||||
Ok(false) => {
|
||||
tracing::info!(
|
||||
task_id = %task_id,
|
||||
"Subagent status already updated by another path, skip cancel"
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
error = %e,
|
||||
task_id = %task_id,
|
||||
"Failed to mark subagent as cancelled in DB"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
drop(registry);
|
||||
|
||||
count
|
||||
}
|
||||
@ -1481,25 +1514,27 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
}
|
||||
|
||||
let mut reaped = 0;
|
||||
let registry = self.cancel_registry.lock();
|
||||
for record in &running {
|
||||
// 执行任务仍在（registry 有 token）→ 真正在跑，保留
|
||||
if registry.contains_key(&record.task_id) {
|
||||
continue;
|
||||
}
|
||||
// 锁内只收集僵尸 task_id，SQLite 条件 UPDATE 移到锁外（理由同
|
||||
// cancel_pending_for_topic：避免持 parking_lot 锁跨 DB IO）。
|
||||
let zombie_ids: Vec<String> = {
|
||||
let registry = self.cancel_registry.lock();
|
||||
running
|
||||
.iter()
|
||||
.filter(|record| !registry.contains_key(&record.task_id))
|
||||
.map(|record| record.task_id.clone())
|
||||
.collect()
|
||||
};
|
||||
for task_id in &zombie_ids {
|
||||
// 僵尸：DB=running 但执行任务已消失。
|
||||
// 条件 UPDATE（仅 running→interrupted），不覆盖已终态。
|
||||
match self.store.try_update_pending_subagent_status(
|
||||
&record.task_id,
|
||||
"running",
|
||||
"interrupted",
|
||||
) {
|
||||
match self
|
||||
.store
|
||||
.try_update_pending_subagent_status(task_id, "running", "interrupted")
|
||||
{
|
||||
Ok(true) => {
|
||||
reaped += 1;
|
||||
tracing::warn!(
|
||||
task_id = %record.task_id,
|
||||
def_name = ?record.def_name,
|
||||
spawned_at = ?record.spawned_at,
|
||||
task_id = %task_id,
|
||||
"Reaped zombie subagent (DB running but executor gone); marked interrupted"
|
||||
);
|
||||
}
|
||||
@ -1507,13 +1542,12 @@ impl SubAgentRuntime for DefaultSubAgentRuntime {
|
||||
Err(e) => {
|
||||
tracing::warn!(
|
||||
error = %e,
|
||||
task_id = %record.task_id,
|
||||
task_id = %task_id,
|
||||
"Failed to reap zombie subagent"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
drop(registry);
|
||||
|
||||
if reaped > 0 {
|
||||
tracing::info!(
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user