From 82b6a882a2488a95c32033024bd545f3b3489327 Mon Sep 17 00:00:00 2001 From: oudecheng <13802883547@139.com> Date: Sat, 15 Aug 2026 19:04:54 +0800 Subject: [PATCH] =?UTF-8?q?test(scheduler):=20=E8=B0=83=E5=BA=A6=E5=99=A8?= =?UTF-8?q?=E9=9D=9E=E9=98=BB=E5=A1=9E=E6=B4=BE=E5=8F=91=E4=B8=8E=E5=B9=B6?= =?UTF-8?q?=E5=8F=91=E5=AE=B9=E9=87=8F=E5=9B=9E=E5=BD=92=E6=B5=8B=E8=AF=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 慢执行器（400ms sleep）双任务：process_tick 必须立即返回且两任务并发完成（串行需 >=800ms，断言 <700ms） - worker_queue_capacity=1：槽位耗尽时多余到期任务被推迟（保持 Scheduled），槽位释放后下个 tick 正常派发 - 执行完成后状态机验证：回到 Scheduled、status=ok、next_fire_at 推进 --- src/scheduler/mod.rs | 190 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 190 insertions(+) diff --git a/src/scheduler/mod.rs b/src/scheduler/mod.rs index cb7c6e5..4bdfac9 100644 --- a/src/scheduler/mod.rs +++ b/src/scheduler/mod.rs @@ -2032,4 +2032,194 @@ mod tests { assert_eq!(convert_cron_weekday("*"), "*"); assert_eq!(convert_cron_weekday("?"), "?"); } + + #[derive(Clone)] + struct SlowAgentTaskExecutor { + delay: std::time::Duration, + } + + #[async_trait::async_trait] + impl AgentTaskExecutor for SlowAgentTaskExecutor { + async fn execute( + &self, + _channel_name: &str, + _chat_id: &str, + _prompt: &str, + _options: ScheduledAgentTaskOptions, + ) -> anyhow::Result> { + tokio::time::sleep(self.delay).await; + Ok(Vec::new()) + } + + async fn execute_silent( + &self, + _channel_name: &str, + _session_chat_id: &str, + _notification_chat_id: Option<&str>, + _prompt: &str, + _options: ScheduledAgentTaskOptions, + ) -> anyhow::Result> { + tokio::time::sleep(self.delay).await; + Ok(Vec::new()) + } + } + + /// 构造一个已到期（next_fire_at 在过去）的 agent_task + fn upsert_due_agent_task(store: &SessionStore, job_id: &str) { + store + .upsert_scheduler_job(&SchedulerJobUpsert { + id: job_id.to_string(), + kind: "agent_task".to_string(), + schedule: serde_json::json!({ + "type": "interval", + "seconds": 3600, + "startup_delay_secs": 0 + }), + interval_secs: 3600, + startup_delay_secs: 0, + target: serde_json::json!({ + "channel": "test-channel", + "chat_id": "oc_demo" + }), + payload: serde_json::json!({ "prompt": "测试任务" }), + enabled: true, + state: SchedulerJobState::Scheduled, + last_status: None, + last_error: None, + run_count: 0, + max_runs: None, + last_fired_at: None, + next_fire_at: Some(1), + paused_at: None, + completed_at: None, + }) + .unwrap(); + } + + fn scheduler_config_with_capacity(capacity: usize) -> SchedulerConfig { + SchedulerConfig { + enabled: true, + tick_resolution_ms: 1000, + worker_queue_capacity: capacity, + misfire_policy: SchedulerMisfirePolicy::Skip, + jobs: Vec::new(), + } + } + + async fn wait_for_run_count(store: &Arc, job_id: &str, expected: i64) { + let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5); + loop { + let record = store.get_scheduler_job(job_id).unwrap().unwrap(); + if record.run_count == expected { + return; + } + assert!( + std::time::Instant::now() < deadline, + "job {} did not reach run_count {} in time (current: {})", + job_id, + expected, + record.run_count + ); + tokio::time::sleep(std::time::Duration::from_millis(20)).await; + } + } + + /// 回归：process_tick 不得内联等待任务执行而阻塞 tick 循环； + /// 多个到期任务应并发执行，且各自正确推进到执行完成状态。 + #[tokio::test] + async fn process_tick_dispatches_jobs_without_blocking() { + let store = Arc::new(SessionStore::in_memory().unwrap()); + upsert_due_agent_task(&store, "job-a"); + upsert_due_agent_task(&store, "job-b"); + + let (_, maintenance_service) = test_scheduler_services(); + let scheduler = Scheduler::new( + MessageBus::new(8), + scheduler_config_with_capacity(64), + chrono_tz::Asia::Shanghai, + store.clone(), + SlowAgentTaskExecutor { + delay: std::time::Duration::from_millis(400), + }, + maintenance_service, + ); + + let started = std::time::Instant::now(); + scheduler.process_tick().await.unwrap(); + let tick_elapsed = started.elapsed(); + // tick 循环必须立即返回（远小于 400ms 的任务执行时长）， + // 否则说明仍在串行等待任务执行 + assert!( + tick_elapsed < std::time::Duration::from_millis(150), + "process_tick blocked on slow jobs: {:?}", + tick_elapsed + ); + + // 两个任务各 sleep 400ms：并发执行约 400ms 完成，串行需 >=800ms。 + // 要求 700ms 内全部完成，证明并发执行。 + wait_for_run_count(&store, "job-a", 1).await; + wait_for_run_count(&store, "job-b", 1).await; + assert!( + started.elapsed() < std::time::Duration::from_millis(700), + "jobs appear to run serially: {:?}", + started.elapsed() + ); + + // 执行完成：状态回到 Scheduled、status=ok、下次触发时间已推进 + let record = store.get_scheduler_job("job-a").unwrap().unwrap(); + assert_eq!(record.state, SchedulerJobState::Scheduled); + assert_eq!(record.last_status, Some(SchedulerJobStatus::Ok)); + assert!(record.next_fire_at.unwrap() > 1); + } + + /// 回归：并发槽位耗尽（worker_queue_capacity）时，多余的到期任务被推迟 + /// （保持 Scheduled 不执行），槽位释放后的 tick 能正常派发。 + #[tokio::test] + async fn process_tick_defers_jobs_when_worker_capacity_exhausted() { + let store = Arc::new(SessionStore::in_memory().unwrap()); + upsert_due_agent_task(&store, "job-a"); + upsert_due_agent_task(&store, "job-b"); + + let (_, maintenance_service) = test_scheduler_services(); + let scheduler = Scheduler::new( + MessageBus::new(8), + scheduler_config_with_capacity(1), + chrono_tz::Asia::Shanghai, + store.clone(), + SlowAgentTaskExecutor { + delay: std::time::Duration::from_millis(500), + }, + maintenance_service, + ); + + // 第一个 tick：只有一个槽位，一个任务开始执行，另一个必须被推迟 + scheduler.process_tick().await.unwrap(); + tokio::time::sleep(std::time::Duration::from_millis(50)).await; + let a = store.get_scheduler_job("job-a").unwrap().unwrap(); + let b = store.get_scheduler_job("job-b").unwrap().unwrap(); + let (started, deferred) = if a.state == SchedulerJobState::Running { + (a, b) + } else { + assert_eq!( + b.state, + SchedulerJobState::Running, + "exactly one job should hold the only worker slot" + ); + (b, a) + }; + assert_eq!(started.run_count, 0, "running job has not finished yet"); + assert_eq!(deferred.run_count, 0, "deferred job must not have executed"); + assert_eq!(deferred.state, SchedulerJobState::Scheduled); + + // 第二个 tick：槽位仍被占用，被推迟任务继续等待（不被派发也不报错） + scheduler.process_tick().await.unwrap(); + let deferred_again = store.get_scheduler_job(&deferred.id).unwrap().unwrap(); + assert_eq!(deferred_again.run_count, 0); + assert_eq!(deferred_again.state, SchedulerJobState::Scheduled); + + // 第一个任务完成释放槽位后，下一个 tick 派发被推迟任务 + wait_for_run_count(&store, &started.id, 1).await; + scheduler.process_tick().await.unwrap(); + wait_for_run_count(&store, &deferred.id, 1).await; + } }