diff --git a/src/common/di.rs b/src/common/di.rs index da5615ef..629029c8 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -2145,7 +2145,12 @@ impl AppServiceFactory { // periodic-triggered recoverable job's first tick sees a clean // slate. See `docs/plan/job-registry.md` Part 2. use crate::infrastructure::scheduler::JobStoreProvider as _; - match app_state.core.job_store_provider.boot_recovery_sweep().await { + match app_state + .core + .job_store_provider + .boot_recovery_sweep() + .await + { Ok(0) => tracing::debug!( target: "oxicloud::scheduler", event = "recoverable.boot_recovery", @@ -2212,8 +2217,7 @@ pub struct CoreServices { /// `svc.register_recoverable_job(®istry, &job_store_provider).await`. /// Boot-time crash-recovery sweep is run in `build_app_state` right /// after this provider is created. - pub job_store_provider: - Arc, + pub job_store_provider: Arc, } /// Container for repository services diff --git a/src/infrastructure/scheduler/mod.rs b/src/infrastructure/scheduler/mod.rs index 2b5cd20d..9afc6426 100644 --- a/src/infrastructure/scheduler/mod.rs +++ b/src/infrastructure/scheduler/mod.rs @@ -34,7 +34,7 @@ pub use handler::JobHandler; pub use pg_job_store::{PgJobStore, PgJobStoreProvider}; pub use recoverable::{ JobStore, JobStoreProvider, OpenedRun, RecoverableAdapter, RecoverableJobHandler, RunOutcome, - RunStatus, run_or_resume, + RunStatus, RunSummary, run_or_resume, }; pub use registry::{JobEntry, JobRegistry, JobSummary, RegisterError}; pub use types::{ErrCause, JobOutcome, JobRunArgs}; diff --git a/src/infrastructure/scheduler/pg_job_store.rs b/src/infrastructure/scheduler/pg_job_store.rs index 5b798e9a..7bebeff1 100644 --- a/src/infrastructure/scheduler/pg_job_store.rs +++ b/src/infrastructure/scheduler/pg_job_store.rs @@ -16,7 +16,7 @@ use uuid::Uuid; use crate::common::errors::DomainError; -use super::recoverable::{JobStore, JobStoreProvider, OpenedRun, RunStatus}; +use super::recoverable::{JobStore, JobStoreProvider, OpenedRun, RunStatus, RunSummary}; // ─── PgJobStore — bound to one run ────────────────────────────────────────── @@ -239,8 +239,132 @@ impl JobStoreProvider for PgJobStoreProvider { .map_err(|e| map_sqlx_err("boot_recovery_sweep", e))?; Ok(result.rows_affected()) } + + async fn list_runs(&self, job_name: &str, limit: u32) -> Result, DomainError> { + // Cap the limit at 100 defensively — the API layer should + // also clamp, but a broken caller shouldn't tank the DB. + let capped = limit.min(100) as i64; + let rows: Vec = sqlx::query_as(RUN_SUMMARY_SELECT_LIST) + .bind(job_name) + .bind(capped) + .fetch_all(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("list_runs", e))?; + rows.into_iter().map(row_to_summary).collect() + } + + async fn get_run_by_id(&self, run_id: Uuid) -> Result, DomainError> { + let row: Option = sqlx::query_as(RUN_SUMMARY_SELECT_BY_ID) + .bind(run_id) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("get_run_by_id", e))?; + row.map(row_to_summary).transpose() + } + + async fn request_cancel(&self, job_name: &str) -> Result, DomainError> { + // Only Running → CancelRequested flips. `Paused` can be + // cancelled by not resuming — no need for a state change. + // `CancelRequested` already is what it is. + // Multiple Running rows shouldn't exist (partial unique index), + // but LIMIT 1 is defensive. + let flipped: Option<(Uuid,)> = sqlx::query_as( + r#" + UPDATE jobs.recoverable_runs + SET status = 'CancelRequested', + last_progress_at = NOW() + WHERE id = ( + SELECT id FROM jobs.recoverable_runs + WHERE job_name = $1 + AND status = 'Running' + ORDER BY started_at DESC + LIMIT 1 + ) + RETURNING id + "#, + ) + .bind(job_name) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("request_cancel", e))?; + Ok(flipped.map(|(id,)| id)) + } } +// ─── Shared row → RunSummary decoder ──────────────────────────────────────── + +/// Row shape returned by the run-summary SELECTs. Kept as a distinct +/// type so both `list_runs` and `get_run_by_id` share the projection +/// (SQL column list + decoder). Order matches the SELECT below. +type RunSummaryRow = ( + Uuid, // id + String, // job_name + String, // status + DateTime, // started_at + DateTime, // last_progress_at + Option>, // completed_at + Option>, // cursor + serde_json::Value, // stats + serde_json::Value, // params + Option, // error_message +); + +const RUN_SUMMARY_COLUMNS: &str = "id, job_name, status, started_at, last_progress_at, completed_at, cursor, stats, params, error_message"; + +// `format!` isn't const, but `concat!` gives us a &'static str at compile +// time — worth it so the SELECT strings show up in tracing / SQL logs +// as one contiguous line instead of a runtime string build. +const RUN_SUMMARY_SELECT_LIST: &str = concat!( + "SELECT id, job_name, status, started_at, last_progress_at, completed_at, cursor, stats, params, error_message ", + "FROM jobs.recoverable_runs ", + "WHERE job_name = $1 ", + "ORDER BY started_at DESC ", + "LIMIT $2" +); + +const RUN_SUMMARY_SELECT_BY_ID: &str = concat!( + "SELECT id, job_name, status, started_at, last_progress_at, completed_at, cursor, stats, params, error_message ", + "FROM jobs.recoverable_runs ", + "WHERE id = $1" +); + +fn row_to_summary(row: RunSummaryRow) -> Result { + let ( + id, + job_name, + status_str, + started_at, + last_progress_at, + completed_at, + cursor, + stats, + params, + error_message, + ) = row; + let status = RunStatus::parse(&status_str).ok_or_else(|| { + DomainError::internal_error("JobStore", format!("unknown status: {status_str}")) + })?; + Ok(RunSummary { + id, + job_name, + status, + started_at, + last_progress_at, + completed_at, + stats, + params, + cursor_hex: cursor.map(hex::encode), + error_message, + }) +} + +// Suppress the dead-code lint on the column list — kept as a +// human-readable constant even though the actual SELECTs currently +// inline it. Future rewrites of the SELECTs (e.g. adding stats +// projection) will use it. +#[allow(dead_code)] +const _RUN_SUMMARY_COLUMNS_UNUSED: &str = RUN_SUMMARY_COLUMNS; + /// Row shape returned by `open_or_start`'s SELECT — factored out /// so clippy's `type_complexity` lint doesn't yell at the query. type ExistingRun = (Uuid, String, DateTime, Option>); @@ -283,10 +407,7 @@ impl PgJobStoreProvider { })?; match status { RunStatus::Running | RunStatus::CancelRequested => { - Ok(OpenedRun::AlreadyActive { - run_id: id, - status, - }) + Ok(OpenedRun::AlreadyActive { run_id: id, status }) } RunStatus::Paused => { // Flip to Running and hand back the cursor. @@ -313,20 +434,15 @@ impl PgJobStoreProvider { .bind(id) .fetch_optional(self.pool.as_ref()) .await - .map_err(|e| { - OpenErr::Fatal(map_sqlx_err("open_or_start.resume", e)) - })?; + .map_err(|e| OpenErr::Fatal(map_sqlx_err("open_or_start.resume", e)))?; let (started_at, cursor_bytes) = row.ok_or_else(|| { OpenErr::Fatal(DomainError::internal_error( "JobStore", format!("run vanished during resume: {id}"), )) })?; - let store: Arc = Arc::new(PgJobStore::new( - self.pool.clone(), - id, - started_at, - )); + let store: Arc = + Arc::new(PgJobStore::new(self.pool.clone(), id, started_at)); Ok(OpenedRun::Resumed { store, cursor: cursor_bytes.unwrap_or_default(), diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index 204b2cf3..e042f97f 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -256,6 +256,54 @@ pub trait JobStoreProvider: Send + Sync { /// via `POST /api/admin/jobs/{name}/trigger`, which calls /// `open_or_start` and picks up the Paused cursor. async fn boot_recovery_sweep(&self) -> Result; + + /// Latest N runs for `job_name`, newest first, terminal + non-terminal + /// both included. Powers `GET /api/admin/jobs/{name}/runs`. `limit` + /// caps the return size; the API layer clamps it too. + async fn list_runs(&self, job_name: &str, limit: u32) -> Result, DomainError>; + + /// Fetch one run by id. Powers `GET /api/admin/jobs/{name}/runs/{id}`. + /// Returns `None` when the id doesn't exist (unknown or pruned). + async fn get_run_by_id(&self, run_id: Uuid) -> Result, DomainError>; + + /// Request cancellation of the CURRENT active run for `job_name` + /// by flipping its status from `Running` → `CancelRequested`. + /// Returns the run's id when a Running row was flipped, `None` + /// when there was no Running row to cancel (nothing in flight, + /// or the latest non-terminal row is already `Paused` / + /// `CancelRequested`). + /// + /// Cooperative — the handler still needs to poll `store.status()` + /// and return `RunOutcome::Paused` at the next safe boundary. If + /// the handler doesn't poll, cancel is a no-op until the run + /// completes naturally. + async fn request_cancel(&self, job_name: &str) -> Result, DomainError>; +} + +/// Serialisable snapshot of one `jobs.recoverable_runs` row, returned +/// by the admin listing + get-run endpoints. +#[derive(Debug, Clone, Serialize)] +pub struct RunSummary { + pub id: Uuid, + pub job_name: String, + pub status: RunStatus, + pub started_at: DateTime, + pub last_progress_at: DateTime, + #[serde(skip_serializing_if = "Option::is_none")] + pub completed_at: Option>, + /// `stats` JSONB dump — job-specific counters (scanned_count, + /// migrated_blobs, findings_this_run, …). + pub stats: serde_json::Value, + /// `params` JSONB dump — per-run params captured at start + /// (grace_window_secs, source_backend, …). + pub params: serde_json::Value, + /// Cursor as hex — omitted when null. Operators occasionally want + /// to inspect this for "where did the scan get to" diagnostics; + /// the raw bytes are opaque per-job so we render as hex. + #[serde(skip_serializing_if = "Option::is_none")] + pub cursor_hex: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub error_message: Option, } /// Result of [`JobStoreProvider::open_or_start`]. @@ -324,11 +372,7 @@ pub async fn run_or_resume( } RunOutcome::Paused { cursor } => { let cursor_hex = hex::encode(&cursor); - log_terminal_write_err( - "mark_paused", - run_id, - store.mark_paused(Some(cursor)).await, - ); + log_terminal_write_err("mark_paused", run_id, store.mark_paused(Some(cursor)).await); JobOutcome::ok_with( 0, serde_json::json!({ @@ -339,11 +383,7 @@ pub async fn run_or_resume( ) } RunOutcome::Failed { message } => { - log_terminal_write_err( - "mark_failed", - run_id, - store.mark_failed(&message).await, - ); + log_terminal_write_err("mark_failed", run_id, store.mark_failed(&message).await); JobOutcome::err(format!("{message} (run_id={run_id})")) } } @@ -461,11 +501,7 @@ mod tests { async fn status(&self) -> Result { Ok(self.state.lock().unwrap().status) } - async fn checkpoint( - &self, - cursor: Vec, - delta_count: u64, - ) -> Result<(), DomainError> { + async fn checkpoint(&self, cursor: Vec, delta_count: u64) -> Result<(), DomainError> { let mut s = self.state.lock().unwrap(); s.cursor = Some(cursor); s.scanned_count += delta_count; @@ -530,15 +566,15 @@ mod tests { /// assertions. fn last_status(&self) -> Option { let stores = self.stores.lock().unwrap(); - stores - .last() - .map(|s| s.state.lock().unwrap().status) + stores.last().map(|s| s.state.lock().unwrap().status) } /// Test-only read — last-created run's cursor. fn last_cursor(&self) -> Option> { let stores = self.stores.lock().unwrap(); - stores.last().and_then(|s| s.state.lock().unwrap().cursor.clone()) + stores + .last() + .and_then(|s| s.state.lock().unwrap().cursor.clone()) } } @@ -596,6 +632,68 @@ mod tests { } Ok(n) } + + async fn list_runs( + &self, + job_name: &str, + limit: u32, + ) -> Result, DomainError> { + let stores = self.stores.lock().unwrap(); + let now = Utc::now(); + let out: Vec = stores + .iter() + .rev() // newest first — MemProvider stores in insertion order + .take(limit as usize) + .map(|s| { + let state = s.state.lock().unwrap(); + RunSummary { + id: s.run_id, + job_name: job_name.to_string(), + status: state.status, + started_at: s.started_at, + last_progress_at: now, + completed_at: None, + stats: serde_json::json!({ "scanned_count": state.scanned_count }), + params: serde_json::json!({}), + cursor_hex: state.cursor.as_ref().map(hex::encode), + error_message: state.error_message.clone(), + } + }) + .collect(); + Ok(out) + } + + async fn get_run_by_id(&self, run_id: Uuid) -> Result, DomainError> { + let stores = self.stores.lock().unwrap(); + let now = Utc::now(); + Ok(stores.iter().find(|s| s.run_id == run_id).map(|s| { + let state = s.state.lock().unwrap(); + RunSummary { + id: s.run_id, + job_name: "mem".to_string(), + status: state.status, + started_at: s.started_at, + last_progress_at: now, + completed_at: None, + stats: serde_json::json!({ "scanned_count": state.scanned_count }), + params: serde_json::json!({}), + cursor_hex: state.cursor.as_ref().map(hex::encode), + error_message: state.error_message.clone(), + } + })) + } + + async fn request_cancel(&self, _job_name: &str) -> Result, DomainError> { + let stores = self.stores.lock().unwrap(); + if let Some(s) = stores.last() { + let mut state = s.state.lock().unwrap(); + if state.status == RunStatus::Running { + state.status = RunStatus::CancelRequested; + return Ok(Some(s.run_id)); + } + } + Ok(None) + } } // ─── Handlers ────────────────────────────────────────────────────────── diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs index 03d9668c..fe74d020 100644 --- a/src/infrastructure/scheduler/registry.rs +++ b/src/infrastructure/scheduler/registry.rs @@ -120,9 +120,7 @@ impl JobRegistry { cadence, ); } - Err(e) => panic!( - "JobRegistry::register({name}) failed — DI wiring bug: {e}" - ), + Err(e) => panic!("JobRegistry::register({name}) failed — DI wiring bug: {e}"), } } diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index 475af99a..0c69c39a 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -147,8 +147,19 @@ pub fn admin_routes() -> Router> { // audit-logged. See `docs/plan/job-registry.md` §Cross-cutting. // Retired the `/internal/trigger-sweep|gc|grant-cleanup` shims // that used to sit here (Stage 2 of the job-registry rollout). + // + // `/jobs` + `/jobs/{name}/trigger` cover every registered + // JobHandler (periodic + recoverable — recoverable ones slot + // in through `RecoverableAdapter`). The `/cancel` + `/runs` + // + `/runs/{id}` triplet is recoverable-only — hitting them + // on a stateless job silently gets an empty list / no-op + // cancel, since no rows in `jobs.recoverable_runs` match. + // See `docs/plan/job-registry.md` Part 2. .route("/jobs", get(list_jobs)) .route("/jobs/{name}/trigger", post(trigger_job)) + .route("/jobs/{name}/cancel", post(cancel_job)) + .route("/jobs/{name}/runs", get(list_job_runs)) + .route("/jobs/{name}/runs/{id}", get(get_job_run)) // Drives — admin-wide view (distinct from `/api/drives` which // is filtered to the caller's role grants). .route("/drives", get(list_all_drives)) @@ -2146,3 +2157,148 @@ pub async fn trigger_job( .into_response(), } } + +/// `POST /api/admin/jobs/{name}/cancel` — cooperative cancel of the +/// currently-running recoverable run for `{name}`. +/// +/// Flips the row's `status` from `Running` → `CancelRequested`. The +/// handler is responsible for polling `store.status()` between batches +/// and returning `RunOutcome::Paused` at the next safe boundary; if it +/// doesn't, the cancel is a no-op until the run completes naturally. +/// +/// Returns 200 with the run_id when a Running row was flipped, 200 with +/// `cancelled: false` when nothing was running (either no runs exist, +/// or the latest is Paused / Completed / Failed / already CancelRequested). +/// Never 404 on "no active run" — the job name is registered and the +/// endpoint just reports the truth. +#[utoipa::path( + post, + path = "/api/admin/jobs/{name}/cancel", + params(("name" = String, Path, description = "Registered job name")), + responses( + (status = 200, description = "Cancel signalled (or no-op if nothing was running)"), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Admin required"), + (status = 500, description = "DB error"), + ), + security(("bearerAuth" = [])), + tag = "admin" +)] +pub async fn cancel_job( + State(state): State>, + axum::extract::Path(name): axum::extract::Path, +) -> impl IntoResponse { + use crate::infrastructure::scheduler::JobStoreProvider as _; + tracing::info!( + target: "audit", + event = "job.cancel_requested", + job = %name, + "👮🏻‍♂️ Admin requested cancel for job {}", + name, + ); + match state.core.job_store_provider.request_cancel(&name).await { + Ok(Some(run_id)) => ( + StatusCode::OK, + Json(serde_json::json!({ + "cancelled": true, + "run_id": run_id.to_string(), + "status": "CancelRequested", + })), + ) + .into_response(), + Ok(None) => ( + StatusCode::OK, + Json(serde_json::json!({ + "cancelled": false, + "reason": "no running run for this job", + })), + ) + .into_response(), + Err(e) => AppError::internal_error(format!("cancel failed: {e}")).into_response(), + } +} + +/// Query parameters for `GET /api/admin/jobs/{name}/runs`. +#[derive(serde::Deserialize)] +pub struct ListRunsQuery { + /// Cap on returned rows. Server-side clamps to 100 defensively. + #[serde(default = "default_runs_limit")] + pub limit: u32, +} + +fn default_runs_limit() -> u32 { + 20 +} + +/// `GET /api/admin/jobs/{name}/runs?limit=N` — history of recoverable +/// runs for a registered job, newest first. Includes terminal + +/// non-terminal rows. +/// +/// Read-only, no audit line — standard admin-middleware auth is enough. +#[utoipa::path( + get, + path = "/api/admin/jobs/{name}/runs", + params( + ("name" = String, Path, description = "Registered job name"), + ("limit" = Option, Query, description = "Max rows to return (default 20, capped at 100)"), + ), + responses( + (status = 200, description = "Runs listed (may be empty)"), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Admin required"), + (status = 500, description = "DB error"), + ), + security(("bearerAuth" = [])), + tag = "admin" +)] +pub async fn list_job_runs( + State(state): State>, + axum::extract::Path(name): axum::extract::Path, + axum::extract::Query(query): axum::extract::Query, +) -> impl IntoResponse { + use crate::infrastructure::scheduler::JobStoreProvider as _; + let limit = query.limit.clamp(1, 100); + match state.core.job_store_provider.list_runs(&name, limit).await { + Ok(runs) => (StatusCode::OK, Json(runs)).into_response(), + Err(e) => AppError::internal_error(format!("list_runs failed: {e}")).into_response(), + } +} + +/// `GET /api/admin/jobs/{name}/runs/{id}` — single-run detail. +/// +/// Returns 404 when the id doesn't exist. `{name}` is not validated +/// against the run's `job_name` — the id is globally unique — but +/// keeping the name in the URL path lets operators build stable +/// per-job history links without knowing individual run ids upfront. +#[utoipa::path( + get, + path = "/api/admin/jobs/{name}/runs/{id}", + params( + ("name" = String, Path, description = "Registered job name"), + ("id" = String, Path, description = "Run UUID"), + ), + responses( + (status = 200, description = "Run detail"), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Admin required"), + (status = 404, description = "Run not found"), + (status = 500, description = "DB error"), + ), + security(("bearerAuth" = [])), + tag = "admin" +)] +pub async fn get_job_run( + State(state): State>, + axum::extract::Path((_name, id)): axum::extract::Path<(String, uuid::Uuid)>, +) -> impl IntoResponse { + use crate::infrastructure::scheduler::JobStoreProvider as _; + match state.core.job_store_provider.get_run_by_id(id).await { + Ok(Some(run)) => (StatusCode::OK, Json(run)).into_response(), + Ok(None) => ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ "error": "run not found", "id": id.to_string() })), + ) + .into_response(), + Err(e) => AppError::internal_error(format!("get_run failed: {e}")).into_response(), + } +} diff --git a/src/interfaces/api/mod.rs b/src/interfaces/api/mod.rs index b98e01ce..64d7308d 100644 --- a/src/interfaces/api/mod.rs +++ b/src/interfaces/api/mod.rs @@ -234,6 +234,9 @@ use crate::interfaces::api::handlers::file_handler::MoveFilePayload; // (docs/plan/job-registry.md §Cross-cutting). handlers::admin_handler::list_jobs, handlers::admin_handler::trigger_job, + handlers::admin_handler::cancel_job, + handlers::admin_handler::list_job_runs, + handlers::admin_handler::get_job_run, // Grant / ReBAC handlers (free functions) handlers::grant_handler::create_grant, handlers::grant_handler::revoke_grant,