ooodc/PicoBot
ooodc/PicoBot
1
0
Fork 0
Code Pull Requests Activity

refactor(scheduler): 任务执行移出 tick 循环，worker_queue_capacity 并发约束生效

Browse Source 
原 process_tick 内联 await execute_job：长耗时 agent 任务（可达数分钟）会串行阻塞整个调度循环，拖后所有其他 job 的触发；worker_queue_capacity 配置项从未被使用。

- 到期 job 的执行与事后状态写入移入 tokio::spawn 后台任务，tick 循环只做派发

- 新增 worker_semaphore（容量=worker_queue_capacity）：槽位耗尽时 job 保持 Scheduled 留待下个 tick，形成有界背压

- job 派发前置为 Running 状态，is_due 要求 Scheduled，天然防止重复派发

- after_execution 状态推进失败时回退为 Scheduled，避免 job 永久卡在 Running

- execute_job 拆分为 execute_job_inner（不依赖 &self，可移入后台任务）
This commit is contained in:
oudecheng 2026-08-15 18:50:40 +08:00
parent 4b93c84447
commit 9b8c64bd21
1 changed files with 129 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

169
src/scheduler/mod.rs
View File

@ -5,7 +5,7 @@ use std::sync::Arc;
use async_trait::async_trait;
use chrono::{DateTime, Duration as ChronoDuration, TimeZone, Utc};
use chrono_tz::Tz;
use tokio::sync::watch;
use tokio::sync::{Semaphore, watch};
use crate::bus::{MessageBus, OutboundMessage};
use crate::config::{
@ -71,6 +71,9 @@ pub struct Scheduler {
jobs: Arc<dyn SchedulerJobRepository>,
agent_task_executor: Arc<dyn AgentTaskExecutor>,
maintenance_executor: Arc<dyn MaintenanceExecutor>,
/// 并发执行槽位：限制同时执行的 job 数量（worker_queue_capacity）。
/// tick 循环只负责派发，job 执行在后台任务中进行，长任务不再阻塞其他 job。
worker_semaphore: Arc<Semaphore>,
}
impl Scheduler {
@ -86,6 +89,7 @@ impl Scheduler {
A: AgentTaskExecutor + 'static,
M: MaintenanceExecutor + 'static,
{
let worker_capacity = config.worker_queue_capacity.max(1);
Self {
bus,
config,
@ -93,6 +97,7 @@ impl Scheduler {
jobs,
agent_task_executor: Arc::new(agent_task_executor),
maintenance_executor: Arc::new(maintenance_executor),
worker_semaphore: Arc::new(Semaphore::new(worker_capacity)),
}
}
@ -224,7 +229,7 @@ impl Scheduler {
let jobs = self.jobs.list_scheduler_jobs(true)?;
for record in jobs {
let Some(mut job) =
let Some(job) =
RuntimeJob::from_record(&record, self.config.misfire_policy, self.timezone)?
else {
continue;
@ -248,6 +253,18 @@ impl Scheduler {
continue;
}
// 尝试获取一个 worker 槽位：并发达到上限（worker_queue_capacity）时
// 不启动该 job——保持 Scheduled 状态，留待下一个 tick 重试。
// 这样长任务不会阻塞 tick 循环，也不会无界堆积并发执行。
let Ok(permit) = self.worker_semaphore.clone().try_acquire_owned() else {
tracing::warn!(
job_id = %job.id,
capacity = self.config.worker_queue_capacity,
"Scheduler worker capacity exhausted, deferring job to next tick"
);
continue;
};
self.jobs.update_scheduler_job_runtime(
&job.id,
SchedulerJobState::Running,
@ -260,62 +277,136 @@ impl Scheduler {
job.completed_at,
)?;
let execution_result = self.execute_job(&job).await;
job.after_execution(
now,
execution_result.as_ref().err().map(|err| err.to_string()),
self.config.misfire_policy,
self.timezone,
)?;
// 执行与事后状态写入移入后台任务：tick 循环只做派发，
// 长耗时任务（agent_task 可能长达数分钟）不再串行阻塞其他 job 的触发。
// job 在 DB 中已是 Running 状态，is_due 要求 Scheduled，因此不会被重复派发。
let bus = self.bus.clone();
let jobs_repo = self.jobs.clone();
let agent_executor = self.agent_task_executor.clone();
let maintenance_executor = self.maintenance_executor.clone();
let misfire_policy = self.config.misfire_policy;
let timezone = self.timezone;
let fire_at = now;
tokio::spawn(async move {
let execution_result = Scheduler::execute_job_inner(
&bus,
&*agent_executor,
&*maintenance_executor,
&job,
)
.await;
let status = if execution_result.is_ok() {
Some(SchedulerJobStatus::Ok)
} else {
Some(SchedulerJobStatus::Error)
};
if let Err(error) = &execution_result {
tracing::error!(job_id = %job.id, error = %error, "Scheduler job failed");
}
if let Err(error) = &execution_result {
tracing::error!(job_id = %job.id, error = %error, "Scheduler job failed");
}
let status = if execution_result.is_ok() {
Some(SchedulerJobStatus::Ok)
} else {
Some(SchedulerJobStatus::Error)
};
self.jobs.update_scheduler_job_runtime(
&job.id,
job.state.clone(),
status,
job.last_error.as_deref(),
job.run_count,
job.last_fired_at,
job.next_fire_at,
job.paused_at,
job.completed_at,
)?;
let mut job = job;
match job.after_execution(
fire_at,
execution_result.as_ref().err().map(|err| err.to_string()),
misfire_policy,
timezone,
) {
Ok(()) => {
if let Err(error) = jobs_repo.update_scheduler_job_runtime(
&job.id,
job.state.clone(),
status,
job.last_error.as_deref(),
job.run_count,
job.last_fired_at,
job.next_fire_at,
job.paused_at,
job.completed_at,
) {
tracing::error!(
job_id = %job.id,
error = %error,
"Failed to persist scheduler job state after execution"
);
}
}
Err(error) => {
// 兜底：状态推进失败时回退为 Scheduled，避免 job 永远卡在 Running
tracing::error!(
job_id = %job.id,
error = %error,
"Failed to compute post-execution scheduler state, resetting to Scheduled"
);
if let Err(update_error) = jobs_repo.update_scheduler_job_runtime(
&job.id,
SchedulerJobState::Scheduled,
Some(SchedulerJobStatus::Error),
Some(&error.to_string()),
job.run_count,
job.last_fired_at,
job.next_fire_at,
job.paused_at,
job.completed_at,
) {
tracing::error!(
job_id = %job.id,
error = %update_error,
"Failed to persist scheduler job state after execution failure"
);
}
}
}
// permit 持有到执行与状态写入全部完成后才释放
drop(permit);
});
}
Ok(())
}
/// 执行单个 job（测试直接调用入口；生产派发走 process_tick 的后台任务）。
#[cfg(test)]
async fn execute_job(&self, job: &RuntimeJob) -> anyhow::Result<()> {
Self::execute_job_inner(
&self.bus,
self.agent_task_executor.as_ref(),
self.maintenance_executor.as_ref(),
job,
)
.await
}
/// job 执行主体：不依赖 &self，便于移入 tokio::spawn 的后台任务。
async fn execute_job_inner(
bus: &Arc<MessageBus>,
agent_task_executor: &dyn AgentTaskExecutor,
maintenance_executor: &dyn MaintenanceExecutor,
job: &RuntimeJob,
) -> anyhow::Result<()> {
match job.kind {
SchedulerJobKind::OutboundMessage => {
let message = build_outbound_message(job)?;
// publish_outbound 失败（bus 满或关闭）不视为 job 失败：
// 通知丢弃是预期的背压行为，标记 job 失败会触发 misfire 重试风暴
if let Err(e) = self.bus.publish_outbound(message).await {
if let Err(e) = bus.publish_outbound(message).await {
tracing::warn!(error = %e, job_id = %job.id, "Dropping outbound for scheduler job");
}
}
SchedulerJobKind::InternalEvent => {
execute_internal_event(self.maintenance_executor.as_ref(), job).await?;
execute_internal_event(maintenance_executor, job).await?;
}
SchedulerJobKind::AgentTask => {
let outbound_messages = execute_agent_task(
self.agent_task_executor.as_ref(),
agent_task_executor,
job,
required_notification_chat_id(job, "agent_task")?,
)
.await?;
for message in outbound_messages {
if let Err(e) = self.bus.publish_outbound(message).await {
if let Err(e) = bus.publish_outbound(message).await {
tracing::warn!(error = %e, job_id = %job.id, "Dropping outbound for scheduler agent task");
}
}
@ -339,7 +430,7 @@ impl Scheduler {
Ok(p) => p,
Err(e) => {
if let Err(notify_error) =
self.notify_silent_agent_task_failure(job, &e).await
Self::notify_silent_agent_task_failure(bus, job, &e).await
{
tracing::error!(
job_id = %job.id,
@ -354,7 +445,7 @@ impl Scheduler {
Ok(o) => o,
Err(e) => {
if let Err(notify_error) =
self.notify_silent_agent_task_failure(job, &e).await
Self::notify_silent_agent_task_failure(bus, job, &e).await
{
tracing::error!(
job_id = %job.id,
@ -366,8 +457,7 @@ impl Scheduler {
}
};
if let Err(error) = self
.agent_task_executor
if let Err(error) = agent_task_executor
.execute_silent(
job.target.channel.as_deref().unwrap_or_default(),
&session_chat_id,
@ -378,7 +468,7 @@ impl Scheduler {
.await
{
if let Err(notify_error) =
self.notify_silent_agent_task_failure(job, &error).await
Self::notify_silent_agent_task_failure(bus, job, &error).await
{
tracing::error!(
job_id = %job.id,
@ -395,7 +485,7 @@ impl Scheduler {
}
async fn notify_silent_agent_task_failure(
&self,
bus: &Arc<MessageBus>,
job: &RuntimeJob,
error: &anyhow::Error,
) -> anyhow::Result<()> {
@ -413,8 +503,7 @@ impl Scheduler {
"silent_agent_task".to_string(),
);
if let Err(e) = self
.bus
if let Err(e) = bus
.publish_outbound(OutboundMessage::error_notification(
channel,
chat_id,