diff --git a/frontend/src/lib/api/types.ts b/frontend/src/lib/api/types.ts index 3845e86e..8875c590 100644 --- a/frontend/src/lib/api/types.ts +++ b/frontend/src/lib/api/types.ts @@ -543,6 +543,19 @@ export type JobOutcome = * Cadence + last-run bookkeeping. `interval_ms` / `next_run_at` are * `undefined` on on-demand jobs (serde skips `Option::None`). */ +/** + * Enough info about a paused recoverable run for the admin panel to + * render "Resume (scanned/total)" on the job row without opening the + * drawer. Absent when no `Paused` row exists for this job. `total` + * is absent when the tenant didn't seed a countable subject — + * fallback UI is just "Resume". + */ +export interface PausedRunBrief { + id: string; + scanned: number; + total?: number; +} + export interface JobSummary { name: string; interval_ms?: number; @@ -560,6 +573,10 @@ export interface JobSummary { * row-expand until this flag was added). */ recoverable: boolean; + /** Populated iff a `Paused` row exists in `jobs.recoverable_runs` + * for this job. Distinct from `running` — a paused run is + * resumable via the same trigger endpoint. */ + paused_run?: PausedRunBrief; } /** diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte index 5ddb5896..a0b3c69b 100644 --- a/frontend/src/lib/components/AdminJobsPanel.svelte +++ b/frontend/src/lib/components/AdminJobsPanel.svelte @@ -616,14 +616,37 @@ {/if} - - {#if supportsDeep(job.name)} + {#if job.paused_run} + {@const p = job.paused_run} + {@const label = + p.total && p.total > 0 + ? t( + 'admin.jobs.resume_progress', + { scanned: p.scanned, total: p.total }, + 'Resume ({{scanned}}/{{total}})' + ) + : t('admin.jobs.resume', 'Resume')} + + {:else} + + {/if} + {#if supportsDeep(job.name) && !job.paused_run} + {#if isRecoverable(job)} + + + {:else} + + {/if} {/if} diff --git a/src/infrastructure/scheduler/mod.rs b/src/infrastructure/scheduler/mod.rs index 16d9963a..cb2e50c5 100644 --- a/src/infrastructure/scheduler/mod.rs +++ b/src/infrastructure/scheduler/mod.rs @@ -37,5 +37,5 @@ pub use recoverable::{ RecoverableJobHandler, RunOutcome, RunProgress, RunStatus, RunSummary, derive_progress, record_or_log, run_or_resume, }; -pub use registry::{JobEntry, JobRegistry, JobSummary, RegisterError}; +pub use registry::{JobEntry, JobRegistry, JobSummary, PausedRunBrief, RegisterError}; pub use types::{ErrCause, JobOutcome, JobRunArgs}; diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index 47086284..f42c983a 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -320,6 +320,18 @@ pub trait JobStore: Send + Sync { /// stamped. async fn get_string_param(&self, key: &str) -> Result, DomainError>; + /// Current `stats.scanned_count` for this run. Used by handlers + /// on a Resume path to reconstruct progress state that isn't + /// persisted in `params` — e.g. `backend_migration` seeds its + /// user-facing `MigrationProgress` counter with this so the + /// admin banner shows continued progress across a restart + /// instead of resetting to 0. + /// + /// Returns `0` if the key is absent (fresh row) or not a + /// number. Callers on a Fresh run can safely skip this — the + /// answer is trivially 0 and the write path starts fresh. + async fn scanned_count(&self) -> Result; + /// Persist one finding to `jobs.run_findings` and bump /// `stats.finding_count` on the parent run. Consistency handlers /// call this in place of the transitional diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs index 9b5f1349..2c0e1eb2 100644 --- a/src/infrastructure/scheduler/registry.rs +++ b/src/infrastructure/scheduler/registry.rs @@ -226,6 +226,11 @@ impl JobRegistry { last_outcome, running: state.current_run_start.is_some(), recoverable: entry.handler.is_recoverable(), + // Populated in `list_jobs` handler via a single + // DB round-trip — kept out of the registry + // snapshot to avoid pulling a DB dependency into + // the in-memory scheduler state. + paused_run: None, } }) .collect() @@ -293,6 +298,24 @@ pub enum RegisterError { /// `jobs.recoverable_runs`. Consumed by the admin UI to decide /// whether the row is expandable (drawer with run history + /// findings) and to gate the retention/purge action. +/// Enough info about a paused recoverable run for the admin panel +/// to render "Resume (scanned/total)" on the job row without opening +/// the drawer. Populated by `list_jobs` in the admin handler from a +/// single `SELECT job_name, id, stats->>'scanned_count', +/// params->>'total_rows' FROM jobs.recoverable_runs WHERE status = +/// 'Paused'` — indexed by the `one_active_run_per_job` partial UNIQUE. +/// +/// `total` is `None` when the tenant doesn't seed a countable subject +/// (`RecoverableJobHandler::count_total`); the UI then shows just +/// "Resume" without progress. +#[derive(Debug, Clone, Serialize)] +pub struct PausedRunBrief { + pub id: uuid::Uuid, + pub scanned: u64, + #[serde(skip_serializing_if = "Option::is_none")] + pub total: Option, +} + #[derive(Debug, Clone, Serialize)] pub struct JobSummary { pub name: String, @@ -306,6 +329,12 @@ pub struct JobSummary { pub last_outcome: Option, pub running: bool, pub recoverable: bool, + /// Populated iff a `Paused` row exists in `jobs.recoverable_runs` + /// for this job. Distinct from `running` — a paused run is + /// resumable via the same trigger endpoint (`run_or_resume` + /// picks Resume when the latest row is Paused). + #[serde(skip_serializing_if = "Option::is_none")] + pub paused_run: Option, } #[cfg(test)] diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index a12d19a2..67972521 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -30,7 +30,7 @@ use crate::application::ports::plugin_ports::{LogQuery, PluginManagementPort, Pl use crate::common::di::AppState; use crate::domain::repositories::drive_repository::DriveRepository; use crate::domain::services::authorization::{Resource, Subject}; -use crate::infrastructure::scheduler::JobStoreProvider; +use crate::infrastructure::scheduler::{JobStoreProvider, PausedRunBrief}; use crate::interfaces::api::handlers::dedup_handler::{get_stats, recalculate_stats}; use crate::interfaces::api::handlers::search_handler::clear_search_cache; use crate::interfaces::errors::AppError; @@ -2339,7 +2339,54 @@ pub async fn delete_drive_admin( tag = "admin" )] pub async fn list_jobs(State(state): State>) -> impl IntoResponse { - let summary = state.core.job_registry.snapshot().await; + let mut summary = state.core.job_registry.snapshot().await; + + // Enrich with paused-run info for recoverable jobs so the admin + // panel can render "Resume (scanned/total)" on the row instead of + // just "Run". One indexed SELECT hits `jobs.recoverable_runs` + // (`one_active_run_per_job` partial UNIQUE keys the lookup); + // failures fall back to the pre-enrichment shape so the endpoint + // stays useful when the jobs DB is temporarily unreachable. + if let Some(pool) = state.db_pool.as_ref() { + let paused_rows: Vec<(String, uuid::Uuid, Option, Option)> = sqlx::query_as( + r#" + SELECT + job_name, + id, + (stats ->> 'scanned_count')::BIGINT AS scanned, + (params ->> 'total_rows')::BIGINT AS total + FROM jobs.recoverable_runs + WHERE status = 'Paused' + "#, + ) + .fetch_all(pool.as_ref()) + .await + .unwrap_or_default(); + + let by_name: std::collections::HashMap = paused_rows + .into_iter() + .map(|(name, id, scanned, total)| { + ( + name, + PausedRunBrief { + id, + scanned: scanned.unwrap_or(0).max(0) as u64, + total: total.filter(|t| *t > 0).map(|t| t as u64), + }, + ) + }) + .collect(); + + for job in summary.iter_mut() { + if job.recoverable + && !job.running + && let Some(paused) = by_name.get(&job.name) + { + job.paused_run = Some(paused.clone()); + } + } + } + (StatusCode::OK, Json(summary)).into_response() }