feat(recoverable job): add pause/resume capability
This commit is contained in:
@@ -543,6 +543,19 @@ export type JobOutcome =
|
|||||||
* Cadence + last-run bookkeeping. `interval_ms` / `next_run_at` are
|
* Cadence + last-run bookkeeping. `interval_ms` / `next_run_at` are
|
||||||
* `undefined` on on-demand jobs (serde skips `Option::None`).
|
* `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 {
|
export interface JobSummary {
|
||||||
name: string;
|
name: string;
|
||||||
interval_ms?: number;
|
interval_ms?: number;
|
||||||
@@ -560,6 +573,10 @@ export interface JobSummary {
|
|||||||
* row-expand until this flag was added).
|
* row-expand until this flag was added).
|
||||||
*/
|
*/
|
||||||
recoverable: boolean;
|
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;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|||||||
@@ -616,14 +616,37 @@
|
|||||||
{/if}
|
{/if}
|
||||||
</td>
|
</td>
|
||||||
<td class="jobs-panel__actions">
|
<td class="jobs-panel__actions">
|
||||||
<button
|
{#if job.paused_run}
|
||||||
class="jobs-panel__btn jobs-panel__btn--small"
|
{@const p = job.paused_run}
|
||||||
disabled={busyKeys.has(`trigger:${job.name}`)}
|
{@const label =
|
||||||
onclick={() => onTrigger(job.name)}
|
p.total && p.total > 0
|
||||||
>
|
? t(
|
||||||
{t('admin.jobs.run', 'Run')}
|
'admin.jobs.resume_progress',
|
||||||
</button>
|
{ scanned: p.scanned, total: p.total },
|
||||||
{#if supportsDeep(job.name)}
|
'Resume ({{scanned}}/{{total}})'
|
||||||
|
)
|
||||||
|
: t('admin.jobs.resume', 'Resume')}
|
||||||
|
<button
|
||||||
|
class="jobs-panel__btn jobs-panel__btn--small jobs-panel__btn--primary"
|
||||||
|
disabled={busyKeys.has(`trigger:${job.name}`)}
|
||||||
|
onclick={() => onTrigger(job.name)}
|
||||||
|
title={t(
|
||||||
|
'admin.jobs.resume_title',
|
||||||
|
'Continue the paused run from its last checkpoint.'
|
||||||
|
)}
|
||||||
|
>
|
||||||
|
{label}
|
||||||
|
</button>
|
||||||
|
{:else}
|
||||||
|
<button
|
||||||
|
class="jobs-panel__btn jobs-panel__btn--small"
|
||||||
|
disabled={busyKeys.has(`trigger:${job.name}`)}
|
||||||
|
onclick={() => onTrigger(job.name)}
|
||||||
|
>
|
||||||
|
{t('admin.jobs.run', 'Run')}
|
||||||
|
</button>
|
||||||
|
{/if}
|
||||||
|
{#if supportsDeep(job.name) && !job.paused_run}
|
||||||
<button
|
<button
|
||||||
class="jobs-panel__btn jobs-panel__btn--small"
|
class="jobs-panel__btn jobs-panel__btn--small"
|
||||||
disabled={busyKeys.has(`trigger:${job.name}:deep`)}
|
disabled={busyKeys.has(`trigger:${job.name}:deep`)}
|
||||||
@@ -633,13 +656,33 @@
|
|||||||
</button>
|
</button>
|
||||||
{/if}
|
{/if}
|
||||||
{#if isRunning(job) && canExpand}
|
{#if isRunning(job) && canExpand}
|
||||||
<button
|
{#if isRecoverable(job)}
|
||||||
class="jobs-panel__btn jobs-panel__btn--small jobs-panel__btn--danger"
|
<!-- Recoverable jobs: the "cancel" endpoint just
|
||||||
disabled={busyKeys.has(`cancel:${job.name}`)}
|
flips CancelRequested → handler yields at the
|
||||||
onclick={() => onCancel(job.name)}
|
next batch boundary → status=Paused (resumable
|
||||||
>
|
with a fresh Resume click, cursor preserved).
|
||||||
{t('admin.jobs.cancel', 'Cancel')}
|
Label it "Pause" so admins know it's not
|
||||||
</button>
|
destructive. -->
|
||||||
|
<button
|
||||||
|
class="jobs-panel__btn jobs-panel__btn--small"
|
||||||
|
disabled={busyKeys.has(`cancel:${job.name}`)}
|
||||||
|
onclick={() => onCancel(job.name)}
|
||||||
|
title={t(
|
||||||
|
'admin.jobs.pause_title',
|
||||||
|
'Signal a graceful pause at the next batch boundary. Run row stays as `Paused` — Resume picks up from the checkpoint.'
|
||||||
|
)}
|
||||||
|
>
|
||||||
|
{t('admin.jobs.pause', 'Pause')}
|
||||||
|
</button>
|
||||||
|
{:else}
|
||||||
|
<button
|
||||||
|
class="jobs-panel__btn jobs-panel__btn--small jobs-panel__btn--danger"
|
||||||
|
disabled={busyKeys.has(`cancel:${job.name}`)}
|
||||||
|
onclick={() => onCancel(job.name)}
|
||||||
|
>
|
||||||
|
{t('admin.jobs.cancel', 'Cancel')}
|
||||||
|
</button>
|
||||||
|
{/if}
|
||||||
{/if}
|
{/if}
|
||||||
</td>
|
</td>
|
||||||
</tr>
|
</tr>
|
||||||
|
|||||||
@@ -37,5 +37,5 @@ pub use recoverable::{
|
|||||||
RecoverableJobHandler, RunOutcome, RunProgress, RunStatus, RunSummary, derive_progress,
|
RecoverableJobHandler, RunOutcome, RunProgress, RunStatus, RunSummary, derive_progress,
|
||||||
record_or_log, run_or_resume,
|
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};
|
pub use types::{ErrCause, JobOutcome, JobRunArgs};
|
||||||
|
|||||||
@@ -320,6 +320,18 @@ pub trait JobStore: Send + Sync {
|
|||||||
/// stamped.
|
/// stamped.
|
||||||
async fn get_string_param(&self, key: &str) -> Result<Option<String>, DomainError>;
|
async fn get_string_param(&self, key: &str) -> Result<Option<String>, 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<u64, DomainError>;
|
||||||
|
|
||||||
/// Persist one finding to `jobs.run_findings` and bump
|
/// Persist one finding to `jobs.run_findings` and bump
|
||||||
/// `stats.finding_count` on the parent run. Consistency handlers
|
/// `stats.finding_count` on the parent run. Consistency handlers
|
||||||
/// call this in place of the transitional
|
/// call this in place of the transitional
|
||||||
|
|||||||
@@ -226,6 +226,11 @@ impl JobRegistry {
|
|||||||
last_outcome,
|
last_outcome,
|
||||||
running: state.current_run_start.is_some(),
|
running: state.current_run_start.is_some(),
|
||||||
recoverable: entry.handler.is_recoverable(),
|
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()
|
.collect()
|
||||||
@@ -293,6 +298,24 @@ pub enum RegisterError {
|
|||||||
/// `jobs.recoverable_runs`. Consumed by the admin UI to decide
|
/// `jobs.recoverable_runs`. Consumed by the admin UI to decide
|
||||||
/// whether the row is expandable (drawer with run history +
|
/// whether the row is expandable (drawer with run history +
|
||||||
/// findings) and to gate the retention/purge action.
|
/// 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<u64>,
|
||||||
|
}
|
||||||
|
|
||||||
#[derive(Debug, Clone, Serialize)]
|
#[derive(Debug, Clone, Serialize)]
|
||||||
pub struct JobSummary {
|
pub struct JobSummary {
|
||||||
pub name: String,
|
pub name: String,
|
||||||
@@ -306,6 +329,12 @@ pub struct JobSummary {
|
|||||||
pub last_outcome: Option<JobOutcome>,
|
pub last_outcome: Option<JobOutcome>,
|
||||||
pub running: bool,
|
pub running: bool,
|
||||||
pub recoverable: 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<PausedRunBrief>,
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
|
|||||||
@@ -30,7 +30,7 @@ use crate::application::ports::plugin_ports::{LogQuery, PluginManagementPort, Pl
|
|||||||
use crate::common::di::AppState;
|
use crate::common::di::AppState;
|
||||||
use crate::domain::repositories::drive_repository::DriveRepository;
|
use crate::domain::repositories::drive_repository::DriveRepository;
|
||||||
use crate::domain::services::authorization::{Resource, Subject};
|
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::dedup_handler::{get_stats, recalculate_stats};
|
||||||
use crate::interfaces::api::handlers::search_handler::clear_search_cache;
|
use crate::interfaces::api::handlers::search_handler::clear_search_cache;
|
||||||
use crate::interfaces::errors::AppError;
|
use crate::interfaces::errors::AppError;
|
||||||
@@ -2339,7 +2339,54 @@ pub async fn delete_drive_admin(
|
|||||||
tag = "admin"
|
tag = "admin"
|
||||||
)]
|
)]
|
||||||
pub async fn list_jobs(State(state): State<Arc<AppState>>) -> impl IntoResponse {
|
pub async fn list_jobs(State(state): State<Arc<AppState>>) -> 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<i64>, Option<i64>)> = 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<String, PausedRunBrief> = 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()
|
(StatusCode::OK, Json(summary)).into_response()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user