feat: 加固运行期可靠性并补齐调度与观测接口
- 运行句柄按 run_id 归属，修复取消后重跑时闸门被旧运行删除的竞态 - 调度循环改用取消令牌退出，启动时对账遗留运行，校验 rows_affected - 三类任务统一参与调度，新增 /api/v1/stats 与 /api/v1/schedule - SQLite 启用 WAL 与 busy_timeout，日志清理跳过活跃文件，单行上限 256 KiB - 前端缓存管理令牌并在 403 时重试一次，仪表盘改用统计接口
This commit is contained in:
+53
-3
@@ -1,6 +1,6 @@
|
||||
use axum::{
|
||||
Json, Router,
|
||||
extract::State,
|
||||
extract::{DefaultBodyLimit, State},
|
||||
http::{HeaderValue, Method, header},
|
||||
middleware,
|
||||
response::IntoResponse,
|
||||
@@ -31,8 +31,9 @@ use crate::{
|
||||
state::AppState,
|
||||
tasks::{
|
||||
model::{
|
||||
ConcurrencyPolicy, CreateTaskRequest, HttpMethod, RetryPolicy, TaskAction,
|
||||
TaskActionType, TaskListResponse, TaskResponse, UpdateTaskRequest,
|
||||
ConcurrencyPolicy, CreateTaskRequest, HttpMethod, RetryPolicy, ScheduledTaskResponse,
|
||||
TaskAction, TaskActionType, TaskListResponse, TaskResponse, TaskStatsResponse,
|
||||
UpdateTaskRequest,
|
||||
},
|
||||
routes::{
|
||||
__path_create_task, __path_delete_task, __path_disable_task, __path_enable_task,
|
||||
@@ -162,6 +163,47 @@ pub async fn list_directories(
|
||||
})
|
||||
}
|
||||
|
||||
#[utoipa::path(
|
||||
get,
|
||||
path = "/api/v1/stats",
|
||||
tag = "tasks",
|
||||
responses((status = 200, description = "Dashboard counters", body = TaskStatsResponse))
|
||||
)]
|
||||
pub async fn task_stats(State(state): State<AppState>) -> ApiResult<Json<TaskStatsResponse>> {
|
||||
let counts = crate::tasks::repository::counts(&state.pool).await?;
|
||||
Ok(Json(TaskStatsResponse {
|
||||
total: counts.total,
|
||||
enabled: counts.enabled,
|
||||
running: u64::try_from(state.executor.running_count()).unwrap_or_default(),
|
||||
script: counts.script,
|
||||
http: counts.http,
|
||||
process: counts.process,
|
||||
}))
|
||||
}
|
||||
|
||||
#[utoipa::path(
|
||||
get,
|
||||
path = "/api/v1/schedule",
|
||||
tag = "tasks",
|
||||
responses(
|
||||
(status = 200, description = "Next trigger time per enabled task", body = Vec<ScheduledTaskResponse>)
|
||||
)
|
||||
)]
|
||||
pub async fn list_schedule(
|
||||
State(state): State<AppState>,
|
||||
) -> ApiResult<Json<Vec<ScheduledTaskResponse>>> {
|
||||
let tasks = crate::tasks::repository::list_enabled_tasks(&state.pool).await?;
|
||||
Ok(Json(
|
||||
tasks
|
||||
.into_iter()
|
||||
.map(|task| ScheduledTaskResponse {
|
||||
next_run_at: state.scheduler.next_run_at(&task.id),
|
||||
task_id: task.id,
|
||||
})
|
||||
.collect(),
|
||||
))
|
||||
}
|
||||
|
||||
#[derive(OpenApi)]
|
||||
#[openapi(
|
||||
info(
|
||||
@@ -175,6 +217,8 @@ pub async fn list_directories(
|
||||
preview_cron,
|
||||
probe_python_venv,
|
||||
list_directories,
|
||||
task_stats,
|
||||
list_schedule,
|
||||
list_tasks,
|
||||
create_task,
|
||||
get_task,
|
||||
@@ -202,6 +246,8 @@ pub async fn list_directories(
|
||||
InterpreterKind,
|
||||
DirectoryEntry,
|
||||
DirectoryListingResponse,
|
||||
TaskStatsResponse,
|
||||
ScheduledTaskResponse,
|
||||
ErrorResponse,
|
||||
TaskActionType,
|
||||
HttpMethod,
|
||||
@@ -252,10 +298,14 @@ pub fn router(state: AppState) -> Router {
|
||||
axum::routing::post(probe_python_venv),
|
||||
)
|
||||
.route("/api/v1/system/directories", get(list_directories))
|
||||
.route("/api/v1/stats", get(task_stats))
|
||||
.route("/api/v1/schedule", get(list_schedule))
|
||||
.route("/api/v1/openapi.json", get(openapi_json))
|
||||
.merge(crate::tasks::routes::router())
|
||||
.merge(crate::runs::routes::router())
|
||||
.route("/api/{*path}", any(api_not_found))
|
||||
// 本地工具也不需要接受超大请求体；任务载荷与日志上传都用不到 2 MiB 以上。
|
||||
.layer(DefaultBodyLimit::max(2 * 1024 * 1024))
|
||||
.layer(cors)
|
||||
.route_layer(middleware::from_fn_with_state(
|
||||
state.clone(),
|
||||
|
||||
+16
-9
@@ -1,20 +1,27 @@
|
||||
use std::{error::Error, str::FromStr};
|
||||
use std::{error::Error, str::FromStr, time::Duration};
|
||||
|
||||
use sqlx::{
|
||||
SqlitePool,
|
||||
migrate::MigrateError,
|
||||
sqlite::{SqliteConnectOptions, SqlitePoolOptions},
|
||||
sqlite::{SqliteConnectOptions, SqliteJournalMode, SqlitePoolOptions, SqliteSynchronous},
|
||||
};
|
||||
|
||||
pub async fn connect(database_url: &str) -> Result<SqlitePool, Box<dyn Error>> {
|
||||
let options = SqliteConnectOptions::from_str(database_url)?
|
||||
let in_memory = database_url.contains(":memory:");
|
||||
let mut options = SqliteConnectOptions::from_str(database_url)?
|
||||
.create_if_missing(true)
|
||||
.foreign_keys(true);
|
||||
let max_connections = if database_url == "sqlite::memory:" {
|
||||
1
|
||||
} else {
|
||||
5
|
||||
};
|
||||
.foreign_keys(true)
|
||||
.busy_timeout(Duration::from_secs(5));
|
||||
|
||||
// 运行日志和状态写入与前端查询会并发访问同一个库；WAL 允许读写并行，
|
||||
// busy_timeout 则避免瞬时冲突直接返回 "database is locked"。
|
||||
if !in_memory {
|
||||
options = options
|
||||
.journal_mode(SqliteJournalMode::Wal)
|
||||
.synchronous(SqliteSynchronous::Normal);
|
||||
}
|
||||
|
||||
let max_connections = if in_memory { 1 } else { 5 };
|
||||
let pool = SqlitePoolOptions::new()
|
||||
.max_connections(max_connections)
|
||||
.connect_with(options)
|
||||
|
||||
@@ -79,7 +79,7 @@ pub fn prepare_script_environment(
|
||||
.to_string_lossy()
|
||||
.into_owned(),
|
||||
);
|
||||
environment.insert("PYTHONUNBUFFERED".to_owned(), "1".to_owned());
|
||||
apply_stdio_defaults(&mut environment);
|
||||
return Ok(PreparedPython {
|
||||
interpreter: python,
|
||||
virtual_env: Some(venv.clone()),
|
||||
@@ -93,7 +93,7 @@ pub fn prepare_script_environment(
|
||||
});
|
||||
}
|
||||
|
||||
environment.insert("PYTHONUNBUFFERED".to_owned(), "1".to_owned());
|
||||
apply_stdio_defaults(&mut environment);
|
||||
Ok(PreparedPython {
|
||||
interpreter: interpreter_path,
|
||||
virtual_env: None,
|
||||
@@ -102,6 +102,17 @@ pub fn prepare_script_environment(
|
||||
})
|
||||
}
|
||||
|
||||
/// Python 的标准流默认跟随系统代码页（简体中文为 GBK），一旦打印 emoji 或生僻字符
|
||||
/// 就会 `UnicodeEncodeError` 直接失败。这里默认改用 UTF-8，日志侧本来就优先按 UTF-8 解码。
|
||||
/// 用户在任务里显式设置的 `PYTHONIOENCODING` 仍然优先。
|
||||
fn apply_stdio_defaults(environment: &mut BTreeMap<String, String>) {
|
||||
// 实时日志依赖标准流不缓冲，这一项始终强制。
|
||||
environment.insert("PYTHONUNBUFFERED".to_owned(), "1".to_owned());
|
||||
environment
|
||||
.entry("PYTHONIOENCODING".to_owned())
|
||||
.or_insert_with(|| "utf-8".to_owned());
|
||||
}
|
||||
|
||||
/// 虚拟环境的来源，用于在界面上说明"为什么"关联到了这个环境。
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
@@ -630,6 +641,48 @@ mod tests {
|
||||
assert!(prepared.remove_environment.contains(&"VIRTUAL_ENV_PROMPT"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn injects_utf8_stdio_for_python_tasks() {
|
||||
let temp = TempDir::new().unwrap();
|
||||
|
||||
let prepared = prepare_script_environment(
|
||||
temp.path().to_str().unwrap(),
|
||||
"python.exe",
|
||||
None,
|
||||
&BTreeMap::new(),
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
prepared.environment.get("PYTHONIOENCODING"),
|
||||
Some(&"utf-8".to_owned())
|
||||
);
|
||||
assert_eq!(
|
||||
prepared.environment.get("PYTHONUNBUFFERED"),
|
||||
Some(&"1".to_owned())
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn keeps_user_pythonioencoding_override() {
|
||||
let temp = TempDir::new().unwrap();
|
||||
let mut user_env = BTreeMap::new();
|
||||
user_env.insert("PYTHONIOENCODING".to_owned(), "gbk".to_owned());
|
||||
|
||||
let prepared = prepare_script_environment(
|
||||
temp.path().to_str().unwrap(),
|
||||
"python.exe",
|
||||
None,
|
||||
&user_env,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(
|
||||
prepared.environment.get("PYTHONIOENCODING"),
|
||||
Some(&"gbk".to_owned())
|
||||
);
|
||||
}
|
||||
|
||||
fn probe(working_dir: &str, interpreter: &str, venv: Option<&str>) -> VenvProbeResponse {
|
||||
probe_venv(&VenvProbeRequest {
|
||||
working_dir: working_dir.to_owned(),
|
||||
|
||||
+177
-16
@@ -16,7 +16,7 @@ use reqwest::Method;
|
||||
use serde::Serialize;
|
||||
use sqlx::SqlitePool;
|
||||
use tokio::{
|
||||
io::{AsyncBufReadExt, AsyncRead, AsyncWriteExt, BufReader},
|
||||
io::{AsyncBufRead, AsyncBufReadExt, AsyncRead, AsyncWriteExt, BufReader},
|
||||
process::{Child, Command},
|
||||
sync::{Mutex as AsyncMutex, broadcast},
|
||||
time::{Duration, sleep, timeout},
|
||||
@@ -35,6 +35,7 @@ use crate::{
|
||||
};
|
||||
|
||||
const MAX_LOG_BYTES: u64 = 5 * 1024 * 1024;
|
||||
const MAX_LOG_LINE_BYTES: usize = 256 * 1024;
|
||||
const MAX_HTTP_LOG_BODY_BYTES: usize = 64 * 1024;
|
||||
const TERMINATION_WAIT_SECS: u64 = 5;
|
||||
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
|
||||
@@ -255,7 +256,7 @@ impl TaskExecutor {
|
||||
let run = match repository::create_queued(&self.inner.pool, &task.id, trigger).await {
|
||||
Ok(run) => run,
|
||||
Err(error) => {
|
||||
lock_running(&self.inner.running).remove(&task.id);
|
||||
remove_running_if(&self.inner.running, &task.id, None);
|
||||
return Err(error);
|
||||
}
|
||||
};
|
||||
@@ -301,6 +302,11 @@ impl TaskExecutor {
|
||||
self.inner.events.subscribe()
|
||||
}
|
||||
|
||||
/// 当前正在执行的任务数量（用于统计接口）。
|
||||
pub fn running_count(&self) -> usize {
|
||||
lock_running(&self.inner.running).len()
|
||||
}
|
||||
|
||||
pub fn log_path(&self, task_id: &str, run_id: i64) -> PathBuf {
|
||||
self.inner
|
||||
.logs_dir
|
||||
@@ -320,18 +326,44 @@ impl TaskExecutor {
|
||||
let _guard = self.inner.cleanup_lock.lock().await;
|
||||
let logs_dir = self.inner.logs_dir.clone();
|
||||
let config = self.inner.log_config.clone();
|
||||
tokio::task::spawn_blocking(move || cleanup_logs_sync(&logs_dir, &config))
|
||||
let active = self.active_log_paths();
|
||||
tokio::task::spawn_blocking(move || cleanup_logs_sync(&logs_dir, &config, &active))
|
||||
.await
|
||||
.map_err(std::io::Error::other)?
|
||||
}
|
||||
|
||||
fn active_log_paths(&self) -> std::collections::HashSet<PathBuf> {
|
||||
lock_running(&self.inner.running)
|
||||
.iter()
|
||||
.filter_map(|(task_id, handle)| {
|
||||
handle.run_id.map(|run_id| {
|
||||
self.inner
|
||||
.logs_dir
|
||||
.join(task_id)
|
||||
.join(format!("{run_id}.log"))
|
||||
})
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
async fn execute_run(&self, task: TaskResponse, run_id: i64, cancellation: CancellationToken) {
|
||||
let started_at = Utc::now();
|
||||
let started = Instant::now();
|
||||
if let Err(error) = repository::mark_running(&self.inner.pool, run_id, started_at).await {
|
||||
tracing::error!(run_id, error = %error, "failed to mark task run as running");
|
||||
lock_running(&self.inner.running).remove(&task.id);
|
||||
return;
|
||||
match repository::mark_running(&self.inner.pool, run_id, started_at).await {
|
||||
Ok(true) => {}
|
||||
Ok(false) => {
|
||||
tracing::warn!(
|
||||
run_id,
|
||||
"run row disappeared before it could be marked running"
|
||||
);
|
||||
remove_running_if(&self.inner.running, &task.id, Some(run_id));
|
||||
return;
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::error!(run_id, error = %error, "failed to mark task run as running");
|
||||
remove_running_if(&self.inner.running, &task.id, Some(run_id));
|
||||
return;
|
||||
}
|
||||
}
|
||||
self.emit_run_by_id(run_id).await;
|
||||
|
||||
@@ -339,7 +371,7 @@ impl TaskExecutor {
|
||||
let finished_at = Utc::now();
|
||||
let duration_ms = u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX);
|
||||
|
||||
if let Err(error) = repository::complete(
|
||||
match repository::complete(
|
||||
&self.inner.pool,
|
||||
run_id,
|
||||
outcome.status,
|
||||
@@ -350,11 +382,19 @@ impl TaskExecutor {
|
||||
)
|
||||
.await
|
||||
{
|
||||
tracing::error!(run_id, error = %error, "failed to complete task run");
|
||||
Ok(true) => {}
|
||||
Ok(false) => {
|
||||
tracing::warn!(
|
||||
run_id,
|
||||
"run row disappeared before completion could be recorded"
|
||||
);
|
||||
}
|
||||
Err(error) => tracing::error!(run_id, error = %error, "failed to complete task run"),
|
||||
}
|
||||
|
||||
self.emit_run_by_id(run_id).await;
|
||||
lock_running(&self.inner.running).remove(&task.id);
|
||||
// 只清理属于自己的句柄：取消后立刻重跑时，旧 run 收尾不能删掉新 run 的闸门。
|
||||
remove_running_if(&self.inner.running, &task.id, Some(run_id));
|
||||
if let Err(error) = self.cleanup_logs().await {
|
||||
tracing::warn!(%error, "failed to clean task logs");
|
||||
}
|
||||
@@ -852,6 +892,21 @@ fn lock_running(
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
}
|
||||
|
||||
/// 只在句柄仍属于指定 run 时移除；`None` 表示"刚插入、还没拿到 run id"的占位句柄。
|
||||
fn remove_running_if(
|
||||
running: &Mutex<HashMap<String, RunningHandle>>,
|
||||
task_id: &str,
|
||||
run_id: Option<i64>,
|
||||
) {
|
||||
let mut running = lock_running(running);
|
||||
if running
|
||||
.get(task_id)
|
||||
.is_some_and(|handle| handle.run_id == run_id)
|
||||
{
|
||||
running.remove(task_id);
|
||||
}
|
||||
}
|
||||
|
||||
fn spawn_log_reader<R>(
|
||||
reader: R,
|
||||
stream: &'static str,
|
||||
@@ -864,12 +919,17 @@ where
|
||||
let mut reader = BufReader::new(reader);
|
||||
let mut line = Vec::new();
|
||||
loop {
|
||||
line.clear();
|
||||
match reader.read_until(b'\n', &mut line).await {
|
||||
Ok(0) => break,
|
||||
Ok(_) => {
|
||||
match read_line_capped(&mut reader, &mut line, MAX_LOG_LINE_BYTES).await {
|
||||
Ok((0, _)) => break,
|
||||
Ok((_, truncated)) => {
|
||||
let text = sink.decode_output(&line);
|
||||
sink.append(&format!("[{stream}] {text}")).await;
|
||||
let mut payload = format!("[{stream}] {text}");
|
||||
if truncated {
|
||||
payload.push_str(&format!(
|
||||
"\n[system] 单行输出超过 {MAX_LOG_LINE_BYTES} 字节，已截断\n"
|
||||
));
|
||||
}
|
||||
sink.append(&payload).await;
|
||||
}
|
||||
Err(error) => {
|
||||
tracing::warn!(stream, %error, "failed to read task output");
|
||||
@@ -880,6 +940,38 @@ where
|
||||
})
|
||||
}
|
||||
|
||||
/// 读取一行，最多保留 `max` 字节。超出上限时丢弃剩余部分但继续读到换行，
|
||||
/// 既保住内存上限，也不会让后续输出错位。
|
||||
async fn read_line_capped<R>(
|
||||
reader: &mut R,
|
||||
buffer: &mut Vec<u8>,
|
||||
max: usize,
|
||||
) -> std::io::Result<(usize, bool)>
|
||||
where
|
||||
R: AsyncBufRead + Unpin,
|
||||
{
|
||||
buffer.clear();
|
||||
let mut truncated = false;
|
||||
loop {
|
||||
let chunk = reader.fill_buf().await?;
|
||||
if chunk.is_empty() {
|
||||
return Ok((buffer.len(), truncated));
|
||||
}
|
||||
let newline = chunk.iter().position(|byte| *byte == b'\n');
|
||||
let take = newline.map_or(chunk.len(), |index| index + 1);
|
||||
let remaining = max.saturating_sub(buffer.len());
|
||||
let copied = take.min(remaining);
|
||||
buffer.extend_from_slice(&chunk[..copied]);
|
||||
if copied < take {
|
||||
truncated = true;
|
||||
}
|
||||
reader.consume(take);
|
||||
if newline.is_some() {
|
||||
return Ok((buffer.len(), truncated));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_after_termination(child: &mut Child) {
|
||||
if timeout(Duration::from_secs(TERMINATION_WAIT_SECS), child.wait())
|
||||
.await
|
||||
@@ -935,13 +1027,19 @@ fn failed_outcome(message: impl Into<String>) -> ScriptOutcome {
|
||||
}
|
||||
}
|
||||
|
||||
fn cleanup_logs_sync(root: &Path, config: &LogRetentionConfig) -> std::io::Result<()> {
|
||||
fn cleanup_logs_sync(
|
||||
root: &Path,
|
||||
config: &LogRetentionConfig,
|
||||
active: &std::collections::HashSet<PathBuf>,
|
||||
) -> std::io::Result<()> {
|
||||
if !root.exists() {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
let mut files = Vec::new();
|
||||
collect_log_files(root, &mut files)?;
|
||||
// 正在写入的日志不参与保留期/容量清理。
|
||||
files.retain(|(path, _)| !active.contains(path));
|
||||
let retention = Duration::from_secs(config.retention_days.saturating_mul(86_400));
|
||||
let cutoff = SystemTime::now().checked_sub(retention);
|
||||
|
||||
@@ -1104,6 +1202,7 @@ mod tests {
|
||||
retention_days: 30,
|
||||
max_total_bytes: 10,
|
||||
},
|
||||
&std::collections::HashSet::new(),
|
||||
)
|
||||
.unwrap();
|
||||
let total = std::fs::read_dir(temp.path())
|
||||
@@ -1112,4 +1211,66 @@ mod tests {
|
||||
.sum::<u64>();
|
||||
assert!(total <= 10);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn cleanup_skips_logs_of_running_tasks() {
|
||||
let temp = tempfile::TempDir::new().unwrap();
|
||||
let active = temp.path().join("active.log");
|
||||
std::fs::write(&active, vec![b'x'; 32]).unwrap();
|
||||
std::thread::sleep(std::time::Duration::from_millis(5));
|
||||
let stale = temp.path().join("stale.log");
|
||||
std::fs::write(&stale, vec![b'x'; 32]).unwrap();
|
||||
|
||||
let mut protected = std::collections::HashSet::new();
|
||||
protected.insert(active.clone());
|
||||
cleanup_logs_sync(
|
||||
temp.path(),
|
||||
&LogRetentionConfig {
|
||||
retention_days: 0,
|
||||
max_total_bytes: 1,
|
||||
},
|
||||
&protected,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
assert!(active.exists(), "active log must survive cleanup");
|
||||
assert!(!stale.exists(), "stale log should be removed");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn capped_line_reader_truncates_but_keeps_reading() {
|
||||
let payload = [b'a'; 64];
|
||||
let mut reader = BufReader::new(&payload[..]);
|
||||
let mut buffer = Vec::new();
|
||||
|
||||
let (bytes, truncated) = read_line_capped(&mut reader, &mut buffer, 16)
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(bytes, 16);
|
||||
assert!(truncated);
|
||||
|
||||
let (next, next_truncated) = read_line_capped(&mut reader, &mut buffer, 16)
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(next, 0);
|
||||
assert!(!next_truncated);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn removes_only_the_matching_run_handle() {
|
||||
let running = Mutex::new(HashMap::from([(
|
||||
"task".to_owned(),
|
||||
RunningHandle {
|
||||
run_id: Some(2),
|
||||
cancellation: CancellationToken::new(),
|
||||
},
|
||||
)]));
|
||||
|
||||
remove_running_if(&running, "task", Some(1));
|
||||
assert!(lock_running(&running).contains_key("task"));
|
||||
|
||||
remove_running_if(&running, "task", Some(2));
|
||||
assert!(!lock_running(&running).contains_key("task"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -119,20 +119,22 @@ pub async fn create_skipped(
|
||||
})
|
||||
}
|
||||
|
||||
/// 返回是否真的更新到了运行记录；记录被删除（例如任务被删）时返回 false。
|
||||
pub async fn mark_running(
|
||||
pool: &SqlitePool,
|
||||
run_id: i64,
|
||||
started_at: DateTime<Utc>,
|
||||
) -> ApiResult<()> {
|
||||
sqlx::query("UPDATE task_runs SET status = ?, started_at = ? WHERE id = ?")
|
||||
) -> ApiResult<bool> {
|
||||
let result = sqlx::query("UPDATE task_runs SET status = ?, started_at = ? WHERE id = ?")
|
||||
.bind(RunStatus::Running.as_str())
|
||||
.bind(started_at)
|
||||
.bind(run_id)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
Ok(result.rows_affected() > 0)
|
||||
}
|
||||
|
||||
/// 返回是否真的更新到了运行记录。
|
||||
pub async fn complete(
|
||||
pool: &SqlitePool,
|
||||
run_id: i64,
|
||||
@@ -141,8 +143,8 @@ pub async fn complete(
|
||||
duration_ms: u64,
|
||||
exit_code: Option<i32>,
|
||||
error_message: Option<&str>,
|
||||
) -> ApiResult<()> {
|
||||
sqlx::query(
|
||||
) -> ApiResult<bool> {
|
||||
let result = sqlx::query(
|
||||
"UPDATE task_runs SET
|
||||
status = ?, finished_at = ?, duration_ms = ?,
|
||||
exit_code = ?, error_message = ?
|
||||
@@ -156,7 +158,22 @@ pub async fn complete(
|
||||
.bind(run_id)
|
||||
.execute(pool)
|
||||
.await?;
|
||||
Ok(())
|
||||
Ok(result.rows_affected() > 0)
|
||||
}
|
||||
|
||||
/// 启动时收敛上一次进程遗留下来的非终态运行，避免界面永远显示"运行中"。
|
||||
pub async fn fail_interrupted_runs(pool: &SqlitePool) -> ApiResult<u64> {
|
||||
let result = sqlx::query(
|
||||
"UPDATE task_runs SET status = ?, finished_at = ?, error_message = ?
|
||||
WHERE status IN ('queued', 'running')",
|
||||
)
|
||||
.bind(RunStatus::Cancelled.as_str())
|
||||
.bind(Utc::now())
|
||||
.bind("TaskPilot 进程重启，运行被中断")
|
||||
.execute(pool)
|
||||
.await?;
|
||||
|
||||
Ok(result.rows_affected())
|
||||
}
|
||||
|
||||
pub async fn get(pool: &SqlitePool, run_id: i64) -> ApiResult<Option<TaskRunResponse>> {
|
||||
|
||||
+50
-12
@@ -14,18 +14,23 @@ use tokio_util::sync::CancellationToken;
|
||||
use crate::{
|
||||
error::ApiResult,
|
||||
runs::{executor::TaskExecutor, model::RunTrigger},
|
||||
tasks::{model::TaskAction, model::TaskResponse, repository},
|
||||
tasks::{model::TaskResponse, repository},
|
||||
};
|
||||
|
||||
struct ScheduledTask {
|
||||
cancellation: CancellationToken,
|
||||
handle: JoinHandle<()>,
|
||||
/// 只用来持有任务；取消时通过令牌让循环自行退出，不直接 abort。
|
||||
/// 字段以下划线开头是因为它只承担"保持任务存活"的作用。
|
||||
_handle: JoinHandle<()>,
|
||||
}
|
||||
|
||||
type NextRuns = Arc<Mutex<HashMap<String, DateTime<Utc>>>>;
|
||||
|
||||
struct SchedulerInner {
|
||||
executor: TaskExecutor,
|
||||
shutdown: CancellationToken,
|
||||
jobs: Mutex<HashMap<String, ScheduledTask>>,
|
||||
next_runs: NextRuns,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
@@ -40,19 +45,22 @@ impl Scheduler {
|
||||
executor,
|
||||
shutdown: CancellationToken::new(),
|
||||
jobs: Mutex::new(HashMap::new()),
|
||||
next_runs: Arc::new(Mutex::new(HashMap::new())),
|
||||
}),
|
||||
};
|
||||
|
||||
for task in repository::list_enabled_script_tasks(&pool).await? {
|
||||
for task in repository::list_enabled_tasks(&pool).await? {
|
||||
scheduler.refresh(task);
|
||||
}
|
||||
|
||||
Ok(scheduler)
|
||||
}
|
||||
|
||||
/// 重新注册任务：停用会取消调度，启用则按 cron 新建循环。
|
||||
/// 三类 Action 都参与调度，执行器本身负责区分脚本、HTTP 与进程。
|
||||
pub fn refresh(&self, task: TaskResponse) {
|
||||
self.cancel(&task.id);
|
||||
if !task.enabled || !matches!(task.action, TaskAction::Script { .. }) {
|
||||
if !task.enabled {
|
||||
return;
|
||||
}
|
||||
|
||||
@@ -61,38 +69,52 @@ impl Scheduler {
|
||||
let executor = self.inner.executor.clone();
|
||||
let shutdown = self.inner.shutdown.clone();
|
||||
let cancellation_for_loop = cancellation.clone();
|
||||
let next_runs = Arc::clone(&self.inner.next_runs);
|
||||
let handle = tokio::spawn(async move {
|
||||
schedule_loop(task_for_loop, executor, cancellation_for_loop, shutdown).await;
|
||||
schedule_loop(
|
||||
task_for_loop,
|
||||
executor,
|
||||
cancellation_for_loop,
|
||||
shutdown,
|
||||
next_runs,
|
||||
)
|
||||
.await;
|
||||
});
|
||||
|
||||
lock_jobs(&self.inner.jobs).insert(
|
||||
task.id,
|
||||
ScheduledTask {
|
||||
cancellation,
|
||||
handle,
|
||||
_handle: handle,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
/// 取消注册。循环会在下一个 select 点退出，不会留下半途的 running 记录。
|
||||
pub fn cancel(&self, task_id: &str) {
|
||||
if let Some(job) = lock_jobs(&self.inner.jobs).remove(task_id) {
|
||||
job.cancellation.cancel();
|
||||
job.handle.abort();
|
||||
}
|
||||
lock_next_runs(&self.inner.next_runs).remove(task_id);
|
||||
}
|
||||
|
||||
pub fn is_scheduled(&self, task_id: &str) -> bool {
|
||||
lock_jobs(&self.inner.jobs).contains_key(task_id)
|
||||
}
|
||||
|
||||
/// 下次触发时间；未注册或刚取消时返回 None。
|
||||
pub fn next_run_at(&self, task_id: &str) -> Option<DateTime<Utc>> {
|
||||
lock_next_runs(&self.inner.next_runs).get(task_id).copied()
|
||||
}
|
||||
|
||||
pub fn shutdown(&self) {
|
||||
self.inner.shutdown.cancel();
|
||||
let mut jobs = lock_jobs(&self.inner.jobs);
|
||||
for job in jobs.values() {
|
||||
job.cancellation.cancel();
|
||||
job.handle.abort();
|
||||
}
|
||||
jobs.clear();
|
||||
lock_next_runs(&self.inner.next_runs).clear();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -105,7 +127,6 @@ impl Drop for SchedulerInner {
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner());
|
||||
for job in jobs.values() {
|
||||
job.cancellation.cancel();
|
||||
job.handle.abort();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -115,6 +136,7 @@ async fn schedule_loop(
|
||||
executor: TaskExecutor,
|
||||
cancellation: CancellationToken,
|
||||
shutdown: CancellationToken,
|
||||
next_runs: NextRuns,
|
||||
) {
|
||||
loop {
|
||||
let now = Utc::now();
|
||||
@@ -122,14 +144,22 @@ async fn schedule_loop(
|
||||
Ok(next) => next,
|
||||
Err(error) => {
|
||||
tracing::error!(task_id = %task.id, %error, "cron schedule calculation failed");
|
||||
lock_next_runs(&next_runs).remove(&task.id);
|
||||
return;
|
||||
}
|
||||
};
|
||||
lock_next_runs(&next_runs).insert(task.id.clone(), next);
|
||||
let delay = (next - now).to_std().unwrap_or_default();
|
||||
|
||||
tokio::select! {
|
||||
_ = cancellation.cancelled() => return,
|
||||
_ = shutdown.cancelled() => return,
|
||||
_ = cancellation.cancelled() => {
|
||||
lock_next_runs(&next_runs).remove(&task.id);
|
||||
return;
|
||||
}
|
||||
_ = shutdown.cancelled() => {
|
||||
lock_next_runs(&next_runs).remove(&task.id);
|
||||
return;
|
||||
}
|
||||
_ = tokio::time::sleep(delay) => {
|
||||
if let Err(error) = executor.trigger(&task, RunTrigger::Schedule).await {
|
||||
tracing::error!(task_id = %task.id, %error, "scheduled task trigger failed");
|
||||
@@ -156,11 +186,19 @@ fn lock_jobs(
|
||||
jobs.lock().unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
}
|
||||
|
||||
fn lock_next_runs(
|
||||
next_runs: &Mutex<HashMap<String, DateTime<Utc>>>,
|
||||
) -> MutexGuard<'_, HashMap<String, DateTime<Utc>>> {
|
||||
next_runs
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use chrono::{TimeZone, Timelike};
|
||||
|
||||
use crate::tasks::model::{ConcurrencyPolicy, RetryPolicy};
|
||||
use crate::tasks::model::{ConcurrencyPolicy, RetryPolicy, TaskAction};
|
||||
|
||||
use super::*;
|
||||
|
||||
|
||||
@@ -33,6 +33,13 @@ impl AppState {
|
||||
log_config.max_total_bytes = megabytes.saturating_mul(1024 * 1024);
|
||||
}
|
||||
let executor = TaskExecutor::new(pool.clone(), logs_dir, log_config);
|
||||
let reconciled = crate::runs::repository::fail_interrupted_runs(&pool).await?;
|
||||
if reconciled > 0 {
|
||||
tracing::warn!(
|
||||
count = reconciled,
|
||||
"marked runs interrupted by the previous shutdown as cancelled"
|
||||
);
|
||||
}
|
||||
executor.cleanup_logs().await.map_err(ApiError::internal)?;
|
||||
let scheduler = Scheduler::start(pool.clone(), executor.clone()).await?;
|
||||
|
||||
|
||||
@@ -176,3 +176,28 @@ pub struct ListTasksQuery {
|
||||
pub page: Option<u32>,
|
||||
pub page_size: Option<u32>,
|
||||
}
|
||||
|
||||
/// 仪表盘指标：一次请求拿到全部计数，避免为每个指标各发一次列表查询。
|
||||
#[derive(Debug, Serialize, ToSchema)]
|
||||
pub struct TaskStatsResponse {
|
||||
/// 任务总数
|
||||
pub total: u64,
|
||||
/// 已启用任务数
|
||||
pub enabled: u64,
|
||||
/// 当前正在执行的任务数
|
||||
pub running: u64,
|
||||
/// 脚本类任务数
|
||||
pub script: u64,
|
||||
/// HTTP 类任务数
|
||||
pub http: u64,
|
||||
/// 进程类任务数
|
||||
pub process: u64,
|
||||
}
|
||||
|
||||
/// 已注册调度的任务及其下次触发时间。
|
||||
#[derive(Debug, Serialize, ToSchema)]
|
||||
pub struct ScheduledTaskResponse {
|
||||
pub task_id: String,
|
||||
/// 下次触发时间；调度正在重建时可能为空
|
||||
pub next_run_at: Option<DateTime<Utc>>,
|
||||
}
|
||||
|
||||
@@ -15,6 +15,38 @@ const TASK_COLUMNS: &str = "
|
||||
created_at, updated_at
|
||||
";
|
||||
|
||||
#[derive(Debug, Clone, Copy, Default)]
|
||||
pub struct TaskCounts {
|
||||
pub total: u64,
|
||||
pub enabled: u64,
|
||||
pub script: u64,
|
||||
pub http: u64,
|
||||
pub process: u64,
|
||||
}
|
||||
|
||||
/// 单条 SQL 统计全部指标，供仪表盘使用。
|
||||
pub async fn counts(pool: &SqlitePool) -> ApiResult<TaskCounts> {
|
||||
let (total, enabled, script, http, process): (i64, i64, i64, i64, i64) = sqlx::query_as(
|
||||
"SELECT
|
||||
COUNT(*),
|
||||
COALESCE(SUM(enabled), 0),
|
||||
COALESCE(SUM(action_type = 'script'), 0),
|
||||
COALESCE(SUM(action_type = 'http'), 0),
|
||||
COALESCE(SUM(action_type = 'process'), 0)
|
||||
FROM tasks",
|
||||
)
|
||||
.fetch_one(pool)
|
||||
.await?;
|
||||
|
||||
Ok(TaskCounts {
|
||||
total: u64::try_from(total).unwrap_or_default(),
|
||||
enabled: u64::try_from(enabled).unwrap_or_default(),
|
||||
script: u64::try_from(script).unwrap_or_default(),
|
||||
http: u64::try_from(http).unwrap_or_default(),
|
||||
process: u64::try_from(process).unwrap_or_default(),
|
||||
})
|
||||
}
|
||||
|
||||
pub struct ListFilters {
|
||||
pub q: Option<String>,
|
||||
pub enabled: Option<bool>,
|
||||
@@ -233,14 +265,15 @@ pub async fn list(pool: &SqlitePool, filters: &ListFilters) -> ApiResult<TaskLis
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn list_enabled_script_tasks(pool: &SqlitePool) -> ApiResult<Vec<TaskResponse>> {
|
||||
/// 启动时恢复调度用的列表：所有启用任务都参与调度，不再只限脚本。
|
||||
pub async fn list_enabled_tasks(pool: &SqlitePool) -> ApiResult<Vec<TaskResponse>> {
|
||||
let rows = sqlx::query_as::<_, TaskRow>(
|
||||
"SELECT
|
||||
id, name, cron, timezone, enabled, timeout_secs, concurrency_policy,
|
||||
retry_max_attempts, retry_delay_secs, action_type, action_json,
|
||||
created_at, updated_at
|
||||
FROM tasks
|
||||
WHERE enabled = 1 AND action_type = 'script'
|
||||
WHERE enabled = 1
|
||||
ORDER BY updated_at DESC, id ASC",
|
||||
)
|
||||
.fetch_all(pool)
|
||||
|
||||
+148
-2
@@ -573,7 +573,12 @@ async fn overlapping_trigger_is_skipped_and_timeout_kills_process_tree() {
|
||||
&working_dir,
|
||||
"powershell.exe",
|
||||
"-NoProfile",
|
||||
vec!["-File", &script_path.to_string_lossy()],
|
||||
vec![
|
||||
"-ExecutionPolicy",
|
||||
"Bypass",
|
||||
"-File",
|
||||
&script_path.to_string_lossy(),
|
||||
],
|
||||
"0 0 2 * * *",
|
||||
1,
|
||||
);
|
||||
@@ -603,7 +608,10 @@ async fn overlapping_trigger_is_skipped_and_timeout_kills_process_tree() {
|
||||
assert_eq!(second_run["status"], "skipped");
|
||||
|
||||
let completed = wait_for_run(&app.router, first_run_id).await;
|
||||
assert_eq!(completed["status"], "timed_out");
|
||||
assert_eq!(
|
||||
completed["status"], "timed_out",
|
||||
"unexpected run: {completed}"
|
||||
);
|
||||
assert!(
|
||||
completed["error_message"]
|
||||
.as_str()
|
||||
@@ -821,3 +829,141 @@ async fn long_utf8_log_is_reassembled_across_chunk_boundaries() {
|
||||
);
|
||||
app.state.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn stats_endpoint_aggregates_task_counts() {
|
||||
let app = test_app().await;
|
||||
let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
|
||||
let script = script_task(
|
||||
"stats_script",
|
||||
&working_dir,
|
||||
"cmd.exe",
|
||||
"/C",
|
||||
vec!["exit", "0"],
|
||||
"0 0 2 * * *",
|
||||
5,
|
||||
);
|
||||
let http = http_task("http://127.0.0.1:9/never");
|
||||
for payload in [script.clone(), http.clone()] {
|
||||
let response = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(payload)).await;
|
||||
assert_eq!(response.status(), StatusCode::CREATED);
|
||||
}
|
||||
|
||||
let response = send_json(&app.router, Method::GET, "/api/v1/stats", None).await;
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let stats = response_json(response).await;
|
||||
|
||||
assert_eq!(stats["total"], 2);
|
||||
assert_eq!(stats["enabled"], 2);
|
||||
assert_eq!(stats["script"], 1);
|
||||
assert_eq!(stats["http"], 1);
|
||||
assert_eq!(stats["process"], 0);
|
||||
assert_eq!(stats["running"], 0);
|
||||
app.state.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn schedule_endpoint_reports_next_run_for_enabled_tasks() {
|
||||
let app = test_app().await;
|
||||
let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
|
||||
let task = script_task(
|
||||
"scheduled_stats",
|
||||
&working_dir,
|
||||
"cmd.exe",
|
||||
"/C",
|
||||
vec!["exit", "0"],
|
||||
"*/30 * * * * *",
|
||||
5,
|
||||
);
|
||||
let created = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
|
||||
assert_eq!(created.status(), StatusCode::CREATED);
|
||||
|
||||
let response = send_json(&app.router, Method::GET, "/api/v1/schedule", None).await;
|
||||
assert_eq!(response.status(), StatusCode::OK);
|
||||
let schedule = response_json(response).await;
|
||||
let entries = schedule.as_array().expect("schedule should be an array");
|
||||
let entry = entries
|
||||
.iter()
|
||||
.find(|item| item["task_id"] == "scheduled_stats")
|
||||
.expect("enabled task should appear in the schedule");
|
||||
|
||||
assert!(
|
||||
entry["next_run_at"].is_string(),
|
||||
"next_run_at should be computed for a registered task"
|
||||
);
|
||||
app.state.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn http_tasks_are_registered_for_scheduling() {
|
||||
let app = test_app().await;
|
||||
let task = http_task("http://127.0.0.1:9/never");
|
||||
let created = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
|
||||
assert_eq!(created.status(), StatusCode::CREATED);
|
||||
|
||||
assert!(
|
||||
app.state.scheduler.is_scheduled("http_task"),
|
||||
"HTTP tasks must take part in scheduled runs"
|
||||
);
|
||||
|
||||
let disabled = send_json(
|
||||
&app.router,
|
||||
Method::POST,
|
||||
"/api/v1/tasks/http_task/disable",
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
assert_eq!(disabled.status(), StatusCode::OK);
|
||||
assert!(!app.state.scheduler.is_scheduled("http_task"));
|
||||
app.state.shutdown();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn startup_marks_runs_interrupted_by_restart_as_cancelled() {
|
||||
let app = test_app().await;
|
||||
let working_dir = app._temp_dir.path().to_string_lossy().into_owned();
|
||||
let task = script_task(
|
||||
"interrupted_task",
|
||||
&working_dir,
|
||||
"cmd.exe",
|
||||
"/C",
|
||||
vec!["exit", "0"],
|
||||
"0 0 2 * * *",
|
||||
5,
|
||||
);
|
||||
let created = send_json(&app.router, Method::POST, "/api/v1/tasks", Some(task)).await;
|
||||
assert_eq!(created.status(), StatusCode::CREATED);
|
||||
|
||||
// 模拟上一次进程被强杀后遗留的运行记录。
|
||||
sqlx::query(
|
||||
"INSERT INTO task_runs (task_id, trigger_type, status, created_at)
|
||||
VALUES ('interrupted_task', 'manual', 'running', ?)",
|
||||
)
|
||||
.bind(chrono::Utc::now())
|
||||
.execute(&app.state.pool)
|
||||
.await
|
||||
.expect("stale run should be inserted");
|
||||
|
||||
let reconciled = task_pilot_backend::runs::repository::fail_interrupted_runs(&app.state.pool)
|
||||
.await
|
||||
.expect("reconciliation should run");
|
||||
assert_eq!(reconciled, 1);
|
||||
|
||||
let runs = send_json(
|
||||
&app.router,
|
||||
Method::GET,
|
||||
"/api/v1/tasks/interrupted_task/runs",
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
let body = response_json(runs).await;
|
||||
let item = &body["items"][0];
|
||||
assert_eq!(item["status"], "cancelled");
|
||||
assert!(
|
||||
item["error_message"]
|
||||
.as_str()
|
||||
.unwrap_or_default()
|
||||
.contains("重启")
|
||||
);
|
||||
app.state.shutdown();
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user