fix(migration): recover the main banner progression on server restart

This commit is contained in:
Edouard Vanbelle
2026-08-02 16:19:09 +02:00
parent 07802e01f8
commit 5eec0fb36e
6 changed files with 92 additions and 24 deletions
@@ -229,6 +229,22 @@ impl JobStore for PgJobStore {
Ok(row.and_then(|(v,)| v))
}
async fn scanned_count(&self) -> Result<u64, DomainError> {
// `(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<i64>,)> = 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<String, serde_json::Value>,
@@ -1029,6 +1029,9 @@ mod tests {
async fn get_string_param(&self, key: &str) -> Result<Option<String>, DomainError> {
Ok(self.state.lock().unwrap().string_params.get(key).cloned())
}
async fn scanned_count(&self) -> Result<u64, DomainError> {
Ok(self.state.lock().unwrap().scanned_count)
}
async fn merge_stats(
&self,
extras: &serde_json::Map<String, serde_json::Value>,
+21 -18
View File
@@ -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<u64>,
}
/// - `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<PausedRunBrief>,
}
/// 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>,
}
#[cfg(test)]
mod tests {
use super::*;
@@ -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();