diff --git a/frontend/src/lib/api/endpoints/adminJobs.ts b/frontend/src/lib/api/endpoints/adminJobs.ts index 97db6419..067919b9 100644 --- a/frontend/src/lib/api/endpoints/adminJobs.ts +++ b/frontend/src/lib/api/endpoints/adminJobs.ts @@ -17,10 +17,18 @@ const JSON_HEADERS = { 'Content-Type': 'application/json' }; * `ok: true` means "dispatch reached the handler"; the handler's own * pass/fail is in `outcome.outcome`. For `consistency_batch`, per-child * outcomes are inside `outcome.extra.per_check`. + * + * `outcome` is absent for detached jobs (currently only + * `backend_migration`) — the endpoint returns `202 Accepted` with + * `dispatched: true` immediately and the run continues in the + * background. Progress polling shows the state; there's no synchronous + * outcome to surface. */ export interface TriggerResponse { ok: boolean; - outcome: JobOutcome; + outcome?: JobOutcome; + dispatched?: boolean; + detached?: boolean; } /** Envelope from `POST /api/admin/jobs/{name}/cancel`. `run_id` is diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte index a0b3c69b..31060d7c 100644 --- a/frontend/src/lib/components/AdminJobsPanel.svelte +++ b/frontend/src/lib/components/AdminJobsPanel.svelte @@ -237,11 +237,33 @@ const key = `trigger:${name}${opts.deep ? ':deep' : ''}`; markBusy(key, true); try { - const res = await triggerJob(name, opts); + // Fire the trigger + a follow-up loadJobs after a short delay + // in parallel. Long jobs (backend_migration) come back 202 + // immediately; short jobs (consistency checks) come back on + // completion. Either way, the `running` badge / Pause button + // should appear within a render cycle rather than waiting + // for the next 5s poll tick. + const triggerPromise = triggerJob(name, opts); + // Give the backend a moment to register the run's + // `current_run_start` before we ask "is it running?" — this + // races against the trigger acknowledgment for detached + // jobs. 300 ms is well under the 5 s poll cadence and + // invisible to the operator. + setTimeout(() => { + void loadJobs(); + if (expandedJob === name) void loadRuns(expandedJob); + }, 300); + + const res = await triggerPromise; // The trigger envelope carries the child's outcome — surface // its pass/fail immediately so operators don't have to click - // through to see whether the run completed cleanly. - if (res.outcome.outcome === 'ok') { + // through to see whether the run completed cleanly. Detached + // jobs come back with `dispatched: true` and no outcome — + // silence the notify for those (the "started" state is + // already visible via the badge). + if (!res.outcome) { + // dispatched (detached) — no outcome to render + } else if (res.outcome.outcome === 'ok') { ui.notify( t('admin.jobs.triggered_ok', { name }, '{{name}} triggered successfully'), 'success' diff --git a/src/infrastructure/scheduler/pg_job_store.rs b/src/infrastructure/scheduler/pg_job_store.rs index bddaf599..5fff4821 100644 --- a/src/infrastructure/scheduler/pg_job_store.rs +++ b/src/infrastructure/scheduler/pg_job_store.rs @@ -229,6 +229,22 @@ impl JobStore for PgJobStore { Ok(row.and_then(|(v,)| v)) } + async fn scanned_count(&self) -> Result { + // `(stats->>'scanned_count')::BIGINT` — text cast rather than + // `->` numeric extraction because the stored value has been + // written via `((...)::text)::jsonb` in `checkpoint`, which + // may present as either a JSON number or a JSON string + // depending on prior versions. `::BIGINT` handles both. + let row: Option<(Option,)> = sqlx::query_as( + "SELECT (stats ->> 'scanned_count')::BIGINT FROM jobs.recoverable_runs WHERE id = $1", + ) + .bind(self.run_id) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("scanned_count", e))?; + Ok(row.and_then(|(v,)| v).unwrap_or(0).max(0) as u64) + } + async fn merge_stats( &self, extras: &serde_json::Map, diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index f42c983a..64c59cb0 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -1029,6 +1029,9 @@ mod tests { async fn get_string_param(&self, key: &str) -> Result, DomainError> { Ok(self.state.lock().unwrap().string_params.get(key).cloned()) } + async fn scanned_count(&self) -> Result { + Ok(self.state.lock().unwrap().scanned_count) + } async fn merge_stats( &self, extras: &serde_json::Map, diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs index 2c0e1eb2..ba112ce9 100644 --- a/src/infrastructure/scheduler/registry.rs +++ b/src/infrastructure/scheduler/registry.rs @@ -298,24 +298,9 @@ 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, -} - +/// - `paused_run` — populated iff a `Paused` row exists in +/// `jobs.recoverable_runs` for this job. The UI uses it to render +/// "Resume (scanned/total)" instead of "Run". #[derive(Debug, Clone, Serialize)] pub struct JobSummary { pub name: String, @@ -337,6 +322,24 @@ pub struct JobSummary { pub paused_run: Option, } +/// 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, +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/infrastructure/services/backend_migration_service.rs b/src/infrastructure/services/backend_migration_service.rs index b19336d1..61dabc72 100644 --- a/src/infrastructure/services/backend_migration_service.rs +++ b/src/infrastructure/services/backend_migration_service.rs @@ -469,15 +469,31 @@ impl RecoverableJobHandler for BackendMigrationService { .await .map(|n| n.max(0) as u64) .unwrap_or(0); + // On Resume, seed the counter with what's already been done + // in prior sessions — else the banner shows "500 / 1536" + // right after resuming a run that had reached 900/1536, + // which misleads admins into thinking the migration + // regressed. Fresh run reports 0. `stats.scanned_count` + // was written by `checkpoint` after each batch, so it's + // durable across restarts. + let already_scanned = if is_fresh { + 0 + } else { + store.scanned_count().await.unwrap_or(0) + }; { let mut guard = self .migration_progress .write() .unwrap_or_else(std::sync::PoisonError::into_inner); - *guard = Some(crate::common::migration_progress::MigrationProgress::new( + let mut progress = crate::common::migration_progress::MigrationProgress::new( target_name.clone(), total_blobs, - )); + ); + if already_scanned > 0 { + progress.bump(already_scanned); + } + *guard = Some(progress); } let source_kind = self.source.backend_type();