diff --git a/src/scheduler/mod.rs b/src/scheduler/mod.rs index f8255aa..cb7c6e5 100644 --- a/src/scheduler/mod.rs +++ b/src/scheduler/mod.rs @@ -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, agent_task_executor: Arc, maintenance_executor: Arc, + /// 并发执行槽位：限制同时执行的 job 数量（worker_queue_capacity）。 + /// tick 循环只负责派发，job 执行在后台任务中进行，长任务不再阻塞其他 job。 + worker_semaphore: Arc, } 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, + 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, 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,