test(scheduler): 调度器非阻塞派发与并发容量回归测试

- 慢执行器(400ms sleep)双任务:process_tick 必须立即返回且两任务并发完成(串行需 >=800ms,断言 <700ms)

- worker_queue_capacity=1:槽位耗尽时多余到期任务被推迟(保持 Scheduled),槽位释放后下个 tick 正常派发

- 执行完成后状态机验证:回到 Scheduled、status=ok、next_fire_at 推进
This commit is contained in:
oudecheng 2026-08-15 19:04:54 +08:00
parent 9b8c64bd21
commit 82b6a882a2

View File

@ -2032,4 +2032,194 @@ mod tests {
assert_eq!(convert_cron_weekday("*"), "*"); assert_eq!(convert_cron_weekday("*"), "*");
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<Vec<OutboundMessage>> {
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<Vec<OutboundMessage>> {
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<SessionStore>, 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;
}
} }