diff --git a/.gitignore b/.gitignore index 5928732..2fefd1d 100644 --- a/.gitignore +++ b/.gitignore @@ -17,7 +17,9 @@ /data/ /logs/ /.e2e-data/ +/.tmp/ /release/ +/samples/python-task/.venv/ # Editors and operating system files /.idea/ diff --git a/backend/src/api.rs b/backend/src/api.rs index 5ffe169..6ff8a2c 100644 --- a/backend/src/api.rs +++ b/backend/src/api.rs @@ -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) -> ApiResult> { + 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) + ) +)] +pub async fn list_schedule( + State(state): State, +) -> ApiResult>> { + 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(), diff --git a/backend/src/database.rs b/backend/src/database.rs index a54549a..29e6166 100644 --- a/backend/src/database.rs +++ b/backend/src/database.rs @@ -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> { - 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) diff --git a/backend/src/python_env.rs b/backend/src/python_env.rs index e4371b7..c9a2617 100644 --- a/backend/src/python_env.rs +++ b/backend/src/python_env.rs @@ -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) { + // 实时日志依赖标准流不缓冲,这一项始终强制。 + 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(), diff --git a/backend/src/runs/executor.rs b/backend/src/runs/executor.rs index a0dc92f..4c6cf83 100644 --- a/backend/src/runs/executor.rs +++ b/backend/src/runs/executor.rs @@ -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 { + 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>, + task_id: &str, + run_id: Option, +) { + 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( 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( + reader: &mut R, + buffer: &mut Vec, + 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) -> ScriptOutcome { } } -fn cleanup_logs_sync(root: &Path, config: &LogRetentionConfig) -> std::io::Result<()> { +fn cleanup_logs_sync( + root: &Path, + config: &LogRetentionConfig, + active: &std::collections::HashSet, +) -> 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::(); 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")); + } } diff --git a/backend/src/runs/repository.rs b/backend/src/runs/repository.rs index 37e7b87..b305a8f 100644 --- a/backend/src/runs/repository.rs +++ b/backend/src/runs/repository.rs @@ -119,20 +119,22 @@ pub async fn create_skipped( }) } +/// 返回是否真的更新到了运行记录;记录被删除(例如任务被删)时返回 false。 pub async fn mark_running( pool: &SqlitePool, run_id: i64, started_at: DateTime, -) -> ApiResult<()> { - sqlx::query("UPDATE task_runs SET status = ?, started_at = ? WHERE id = ?") +) -> ApiResult { + 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, error_message: Option<&str>, -) -> ApiResult<()> { - sqlx::query( +) -> ApiResult { + 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 { + 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> { diff --git a/backend/src/scheduler.rs b/backend/src/scheduler.rs index aac870a..05489a8 100644 --- a/backend/src/scheduler.rs +++ b/backend/src/scheduler.rs @@ -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>>>; + struct SchedulerInner { executor: TaskExecutor, shutdown: CancellationToken, jobs: Mutex>, + 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> { + 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>>, +) -> MutexGuard<'_, HashMap>> { + 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::*; diff --git a/backend/src/state.rs b/backend/src/state.rs index a053042..2c3a38c 100644 --- a/backend/src/state.rs +++ b/backend/src/state.rs @@ -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?; diff --git a/backend/src/tasks/model.rs b/backend/src/tasks/model.rs index 5726b63..5d4bbbf 100644 --- a/backend/src/tasks/model.rs +++ b/backend/src/tasks/model.rs @@ -176,3 +176,28 @@ pub struct ListTasksQuery { pub page: Option, pub page_size: Option, } + +/// 仪表盘指标:一次请求拿到全部计数,避免为每个指标各发一次列表查询。 +#[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>, +} diff --git a/backend/src/tasks/repository.rs b/backend/src/tasks/repository.rs index b6c1224..6df4a6a 100644 --- a/backend/src/tasks/repository.rs +++ b/backend/src/tasks/repository.rs @@ -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 { + 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, pub enabled: Option, @@ -233,14 +265,15 @@ pub async fn list(pool: &SqlitePool, filters: &ListFilters) -> ApiResult ApiResult> { +/// 启动时恢复调度用的列表:所有启用任务都参与调度,不再只限脚本。 +pub async fn list_enabled_tasks(pool: &SqlitePool) -> ApiResult> { 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) diff --git a/backend/tests/runs.rs b/backend/tests/runs.rs index ddb6a4a..237a3aa 100644 --- a/backend/tests/runs.rs +++ b/backend/tests/runs.rs @@ -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(); +} diff --git a/contracts/openapi.json b/contracts/openapi.json index 81b2a53..d433fb3 100644 --- a/contracts/openapi.json +++ b/contracts/openapi.json @@ -317,6 +317,29 @@ } } }, + "/api/v1/schedule": { + "get": { + "tags": [ + "tasks" + ], + "operationId": "list_schedule", + "responses": { + "200": { + "description": "Next trigger time per enabled task", + "content": { + "application/json": { + "schema": { + "type": "array", + "items": { + "$ref": "#/components/schemas/ScheduledTaskResponse" + } + } + } + } + } + } + } + }, "/api/v1/session": { "get": { "tags": [ @@ -337,6 +360,26 @@ } } }, + "/api/v1/stats": { + "get": { + "tags": [ + "tasks" + ], + "operationId": "task_stats", + "responses": { + "200": { + "description": "Dashboard counters", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/TaskStatsResponse" + } + } + } + } + } + } + }, "/api/v1/system/directories": { "get": { "tags": [ @@ -1088,6 +1131,26 @@ "retry" ] }, + "ScheduledTaskResponse": { + "type": "object", + "description": "已注册调度的任务及其下次触发时间。", + "required": [ + "task_id" + ], + "properties": { + "next_run_at": { + "type": [ + "string", + "null" + ], + "format": "date-time", + "description": "下次触发时间;调度正在重建时可能为空" + }, + "task_id": { + "type": "string" + } + } + }, "SessionResponse": { "type": "object", "required": [ @@ -1426,6 +1489,56 @@ } } }, + "TaskStatsResponse": { + "type": "object", + "description": "仪表盘指标:一次请求拿到全部计数,避免为每个指标各发一次列表查询。", + "required": [ + "total", + "enabled", + "running", + "script", + "http", + "process" + ], + "properties": { + "enabled": { + "type": "integer", + "format": "int64", + "description": "已启用任务数", + "minimum": 0 + }, + "http": { + "type": "integer", + "format": "int64", + "description": "HTTP 类任务数", + "minimum": 0 + }, + "process": { + "type": "integer", + "format": "int64", + "description": "进程类任务数", + "minimum": 0 + }, + "running": { + "type": "integer", + "format": "int64", + "description": "当前正在执行的任务数", + "minimum": 0 + }, + "script": { + "type": "integer", + "format": "int64", + "description": "脚本类任务数", + "minimum": 0 + }, + "total": { + "type": "integer", + "format": "int64", + "description": "任务总数", + "minimum": 0 + } + } + }, "UpdateTaskRequest": { "type": "object", "required": [ diff --git a/docs/decisions/0007-runtime-reliability.md b/docs/decisions/0007-runtime-reliability.md new file mode 100644 index 0000000..7900a66 --- /dev/null +++ b/docs/decisions/0007-runtime-reliability.md @@ -0,0 +1,41 @@ +# ADR-0007:运行期可靠性优化 + +- 状态:已接受 +- 日期:2026-10-25 + +## 背景 + +代码评审与端到端自检暴露出五类问题,都会让"看起来正常"的功能在边界条件下失真: + +1. 取消任务时执行器直接删除 `running` 句柄,而旧运行的收尾逻辑又无条件删除同任务句柄。取消后立刻重跑时,旧运行会把新运行的闸门删掉,跳过策略失效,同一任务可能并发执行。 +2. 调度器用 `JoinHandle::abort()` 取消循环。若恰好卡在 `executor.trigger()` 的数据库 await 中,`running` 表项和 `queued` 运行会永久残留,该任务此后永远被判定为"运行中"并跳过。 +3. 进程被强杀后,数据库里的 `queued`/`running` 运行没有对账,前端会一直显示"运行中"并持续轮询。 +4. `mark_running`/`complete` 不检查 `rows_affected`,运行记录被级联删除后收尾静默失效。 +5. 调度器只注册脚本任务,HTTP 与进程任务"启用"后返回 200 却永远不会被定时触发。 + +## 决策 + +1. 句柄带运行 ID,收尾使用 `remove_running_if(task_id, run_id)`:只有仍属于本次运行的句柄才会被移除;取消时依旧立即释放闸门,允许马上重跑。 +2. 调度循环只用 `CancellationToken` 退出,不再 `abort`;`ScheduledTask` 的 `JoinHandle` 仅用于保活。 +3. 启动时执行 `fail_interrupted_runs`:把遗留的 `queued`/`running` 统一改为 `cancelled`,并写入"进程重启,运行被中断"。 +4. `mark_running`/`complete` 返回是否更新成功,执行器对 0 行更新记警告并同步清理内存句柄。 +5. 三类 Action 统一参与调度:`list_enabled_tasks` 不再按 `action_type` 过滤,`Scheduler::refresh` 只判断 `enabled`。 +6. SQLite 启用 WAL + `busy_timeout(5s)` + `synchronous=NORMAL`,避免执行器写日志时前端查询撞上 `database is locked`。 +7. 日志清理跳过正在写入的文件;单行输出上限 256 KiB,超出后截断并继续读行尾。 +8. 请求体上限 2 MiB;新增 `/api/v1/stats`(一次查询拿到全部计数)与 `/api/v1/schedule`(下次触发时间)。 +9. 前端缓存会话令牌,写请求收到 403 时丢弃缓存并重试一次(覆盖后端重启轮换令牌)。 +10. Python 任务默认注入 `PYTHONIOENCODING=utf-8`(用户显式设置优先),`PYTHONUNBUFFERED=1` 保持强制。 + +## 原因 + +- 句柄竞态与 `abort` 泄漏都会造成"任务不再执行"且无任何报错,是排障成本最高的一类缺陷。 +- 对账让崩溃恢复变成确定行为,前端轮询也能正常收敛。 +- 统一调度消除了"启用但永不触发"的假能力。 +- 统计与调度接口把仪表盘从 4 次请求降到 2 次,并让"下次执行时间"成为可见信息。 + +## 边界 + +- 取消后立即可重跑,意味着旧进程的终止与新进程的启动可能短暂重叠(取决于 Job Object 终止速度)。 +- 没有全局并发上限:多个任务在同一时刻触发会各自启动一个进程。 +- 对账只针对 `queued`/`running`,不会尝试恢复重试或补偿错过的调度。 +- 单行截断后该行内容不完整,日志里会多一条 `[system]` 说明。 diff --git a/frontend/src/App.test.tsx b/frontend/src/App.test.tsx index 6a64e6b..68531d0 100644 --- a/frontend/src/App.test.tsx +++ b/frontend/src/App.test.tsx @@ -68,6 +68,21 @@ function createFetchMock(options: FetchMockOptions = {}) { if (url.endsWith('/api/v1/session')) { return jsonResponse({ token: 'test-token' }) } + if (url.endsWith('/api/v1/stats')) { + return jsonResponse({ + total: 1, + enabled: 1, + running: 0, + script: 1, + http: 0, + process: 0, + }) + } + if (url.endsWith('/api/v1/schedule')) { + return jsonResponse([ + { task_id: 'echo_task', next_run_at: '2026-10-19T02:00:00Z' }, + ]) + } if (url.endsWith('/api/v1/cron/preview')) { return jsonResponse({ timezone: 'Asia/Shanghai', @@ -152,6 +167,13 @@ describe('TaskDashboard', () => { expect(await screen.findByText('在线')).toBeInTheDocument() }) + it('shows the next run time reported by the schedule endpoint', async () => { + vi.stubGlobal('fetch', createFetchMock()) + renderApp() + + expect(await screen.findByText(/下次/)).toBeInTheDocument() + }) + it('creates a script task with the visible form values', async () => { const fetchMock = createFetchMock() vi.stubGlobal('fetch', fetchMock) diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 6dad335..be327e6 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -24,6 +24,8 @@ export type InterpreterKind = components['schemas']['InterpreterKind'] export type DirectoryEntry = components['schemas']['DirectoryEntry'] export type DirectoryListingResponse = components['schemas']['DirectoryListingResponse'] +export type TaskStats = components['schemas']['TaskStatsResponse'] +export type ScheduledTask = components['schemas']['ScheduledTaskResponse'] export interface TaskFilters { q?: string @@ -77,20 +79,27 @@ async function requestJson( path: string, init: RequestInit = {}, ): Promise { - const headers = new Headers(init.headers) - headers.set('Accept', 'application/json') const method = (init.method ?? 'GET').toUpperCase() - if (init.body) { - headers.set('Content-Type', 'application/json') - } - if (!['GET', 'HEAD', 'OPTIONS'].includes(method)) { - headers.set('X-TaskPilot-Token', await fetchSessionToken()) + const mutating = !['GET', 'HEAD', 'OPTIONS'].includes(method) + + const send = async (): Promise => { + const headers = new Headers(init.headers) + headers.set('Accept', 'application/json') + if (init.body) { + headers.set('Content-Type', 'application/json') + } + if (mutating) { + headers.set('X-TaskPilot-Token', await fetchSessionToken()) + } + return fetch(path, { ...init, headers }) } - const response = await fetch(path, { - ...init, - headers, - }) + let response = await send() + if (mutating && response.status === 403) { + // 后端重启会轮换管理令牌:丢弃缓存后重试一次。 + invalidateSessionToken() + response = await send() + } if (!response.ok) { let body: ErrorResponse | undefined @@ -108,15 +117,34 @@ async function requestJson( return (await response.json()) as T } +let cachedSessionToken: string | null = null +let pendingSessionToken: Promise | null = null + +export function invalidateSessionToken(): void { + cachedSessionToken = null +} + async function fetchSessionToken(): Promise { - const response = await fetch('/api/v1/session', { - headers: { Accept: 'application/json' }, - }) - if (!response.ok) { - throw new ApiRequestError(response.status) + if (cachedSessionToken) { + return cachedSessionToken } - const session = (await response.json()) as SessionResponse - return session.token + if (!pendingSessionToken) { + pendingSessionToken = fetch('/api/v1/session', { + headers: { Accept: 'application/json' }, + }) + .then(async (response) => { + if (!response.ok) { + throw new ApiRequestError(response.status) + } + const session = (await response.json()) as SessionResponse + cachedSessionToken = session.token + return session.token + }) + .finally(() => { + pendingSessionToken = null + }) + } + return pendingSessionToken } function withQuery(path: string, values: object): string { @@ -265,3 +293,13 @@ export function listDirectories( { signal }, ) } + +/** 仪表盘计数:一次请求拿到全部指标。 */ +export function fetchStats(signal?: AbortSignal): Promise { + return requestJson('/api/v1/stats', { signal }) +} + +/** 已启用任务的下次触发时间。 */ +export function fetchSchedule(signal?: AbortSignal): Promise { + return requestJson('/api/v1/schedule', { signal }) +} diff --git a/frontend/src/api/generated.ts b/frontend/src/api/generated.ts index 4db9801..546d45d 100644 --- a/frontend/src/api/generated.ts +++ b/frontend/src/api/generated.ts @@ -132,6 +132,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/v1/schedule": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["list_schedule"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/v1/session": { parameters: { query?: never; @@ -148,6 +164,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/v1/stats": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["task_stats"]; + put?: never; + post?: never; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/v1/system/directories": { parameters: { query?: never; @@ -338,6 +370,15 @@ export interface components { RunStatus: "queued" | "running" | "succeeded" | "failed" | "timed_out" | "cancelled" | "skipped"; /** @enum {string} */ RunTrigger: "schedule" | "manual" | "retry"; + /** @description 已注册调度的任务及其下次触发时间。 */ + ScheduledTaskResponse: { + /** + * Format: date-time + * @description 下次触发时间;调度正在重建时可能为空 + */ + next_run_at?: string | null; + task_id: string; + }; SessionResponse: { /** @description Random per-process token required for mutating browser requests. */ token: string; @@ -427,6 +468,39 @@ export interface components { task_id: string; trigger: components["schemas"]["RunTrigger"]; }; + /** @description 仪表盘指标:一次请求拿到全部计数,避免为每个指标各发一次列表查询。 */ + TaskStatsResponse: { + /** + * Format: int64 + * @description 已启用任务数 + */ + enabled: number; + /** + * Format: int64 + * @description HTTP 类任务数 + */ + http: number; + /** + * Format: int64 + * @description 进程类任务数 + */ + process: number; + /** + * Format: int64 + * @description 当前正在执行的任务数 + */ + running: number; + /** + * Format: int64 + * @description 脚本类任务数 + */ + script: number; + /** + * Format: int64 + * @description 任务总数 + */ + total: number; + }; UpdateTaskRequest: { action: components["schemas"]["TaskAction"]; concurrency_policy?: components["schemas"]["ConcurrencyPolicy"]; @@ -718,6 +792,26 @@ export interface operations { }; }; }; + list_schedule: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Next trigger time per enabled task */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ScheduledTaskResponse"][]; + }; + }; + }; + }; session: { parameters: { query?: never; @@ -738,6 +832,26 @@ export interface operations { }; }; }; + task_stats: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Dashboard counters */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["TaskStatsResponse"]; + }; + }; + }; + }; list_directories: { parameters: { query?: { diff --git a/frontend/src/lib/format.ts b/frontend/src/lib/format.ts index 12c9005..6c6cbce 100644 --- a/frontend/src/lib/format.ts +++ b/frontend/src/lib/format.ts @@ -17,6 +17,21 @@ export function formatDateTime(value: string | null | undefined): string { return Number.isNaN(date.getTime()) ? '--' : dateTimeFormatter.format(date) } +const shortDateTimeFormatter = new Intl.DateTimeFormat('zh-CN', { + month: '2-digit', + day: '2-digit', + hour: '2-digit', + minute: '2-digit', +}) + +export function formatShortDateTime(value: string | null | undefined): string { + if (!value) { + return '--' + } + const date = new Date(value) + return Number.isNaN(date.getTime()) ? '--' : shortDateTimeFormatter.format(date) +} + export function formatDuration(value: number | null | undefined): string { if (value === null || value === undefined) { return '--' diff --git a/frontend/src/pages/TaskDashboard.tsx b/frontend/src/pages/TaskDashboard.tsx index 466a58d..73b9e67 100644 --- a/frontend/src/pages/TaskDashboard.tsx +++ b/frontend/src/pages/TaskDashboard.tsx @@ -10,6 +10,8 @@ import { createTask, deleteTask, fetchHealth, + fetchSchedule, + fetchStats, listTasks, runTask, setTaskEnabled, @@ -24,7 +26,7 @@ import { EnabledBadge, } from '../components/StatusBadge' import { TaskEditorModal } from '../components/TaskEditorModal' -import { actionTypeLabel } from '../lib/format' +import { actionTypeLabel, formatShortDateTime } from '../lib/format' import { type TaskFormValues, valuesToCreateRequest, @@ -75,23 +77,19 @@ export function TaskDashboard() { queryFn: ({ signal }) => listTasks(filters, signal), placeholderData: keepPreviousData, }) - const totalTasksQuery = useQuery({ - queryKey: ['tasks', 'summary', 'total'], - queryFn: ({ signal }) => listTasks({ page: 1, page_size: 1 }, signal), + const statsQuery = useQuery({ + queryKey: ['stats'], + queryFn: ({ signal }) => fetchStats(signal), + refetchInterval: 15_000, }) - const enabledTasksQuery = useQuery({ - queryKey: ['tasks', 'summary', 'enabled'], - queryFn: ({ signal }) => - listTasks({ enabled: true, page: 1, page_size: 1 }, signal), - }) - const scriptTasksQuery = useQuery({ - queryKey: ['tasks', 'summary', 'script'], - queryFn: ({ signal }) => - listTasks( - { action_type: 'script', page: 1, page_size: 1 }, - signal, - ), + const scheduleQuery = useQuery({ + queryKey: ['schedule'], + queryFn: ({ signal }) => fetchSchedule(signal), + refetchInterval: 15_000, }) + const nextRuns = new Map( + (scheduleQuery.data ?? []).map((entry) => [entry.task_id, entry.next_run_at]), + ) const saveMutation = useMutation({ mutationFn: ({ @@ -232,18 +230,18 @@ export function TaskDashboard() {
Rust 引擎 @@ -407,6 +405,13 @@ export function TaskDashboard() { {task.cron}
{task.timezone}
+
+ {task.enabled + ? nextRuns.get(task.id) + ? `下次 ${formatShortDateTime(nextRuns.get(task.id))}` + : '调度中…' + : '未调度'} +
@@ -573,5 +578,7 @@ async function invalidateTaskData(queryClient: ReturnType await Promise.all([ queryClient.invalidateQueries({ queryKey: ['tasks'] }), queryClient.invalidateQueries({ queryKey: ['task-runs'] }), + queryClient.invalidateQueries({ queryKey: ['stats'] }), + queryClient.invalidateQueries({ queryKey: ['schedule'] }), ]) } diff --git a/scripts/check.ps1 b/scripts/check.ps1 index 6e8b4c8..3f0a5c6 100644 --- a/scripts/check.ps1 +++ b/scripts/check.ps1 @@ -1,6 +1,7 @@ $ErrorActionPreference = 'Stop' $projectRoot = Split-Path -Parent $PSScriptRoot +$scratch = Join-Path $projectRoot '.tmp' function Assert-ExitCode { param([string]$Step) @@ -10,6 +11,12 @@ function Assert-ExitCode { } } +# 测试统一使用仓库内的临时目录:不依赖系统 %TEMP% 的权限策略, +# 也让清理、归档和排障都留在项目目录里。 +[void](New-Item -ItemType Directory -Path $scratch -Force) +$env:TMP = $scratch +$env:TEMP = $scratch + Push-Location $projectRoot try { pnpm --dir frontend lint @@ -31,4 +38,5 @@ try { } finally { Pop-Location + Remove-Item -LiteralPath $scratch -Recurse -Force -ErrorAction SilentlyContinue }