From 54d8e28229c986745c8a09c07a5aedc69aff610f Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 14 May 2026 14:16:37 +0000 Subject: [PATCH] fix: issues with backfill stage/status separation and progress tracking Signed-off-by: Trezy --- src/admin/backfill.rs | 52 +++++++++++++++++++++++++--------------------------- web/src/app/dashboard/backfill/page.tsx | 66 +++++++++++++++++++++++++++++------------------------------------- 2 file(s) changed, 54 insertion(s)(+), 64 deletion(s)(-) diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -125,7 +125,7 @@ async fn fail_job(state: &AppState, job_id: &str, error: &str) { let now = now_rfc3339(); let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'failed', stage = 'failed', completed_at = ?, error = ? WHERE id = ?", + "UPDATE backfill_jobs SET status = 'failed', completed_at = ?, error = ? WHERE id = ?", state.db_backend, ); let _ = sqlx::query(&sql) @@ -162,7 +162,7 @@ async fn finalise_cancel(state: &AppState, job_id: &str) { let now = now_rfc3339(); let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'cancelled', stage = 'cancelled', completed_at = ?, error = 'cancelled by user' WHERE id = ?", + "UPDATE backfill_jobs SET status = 'cancelled', completed_at = ?, error = 'cancelled by user' WHERE id = ?", state.db_backend, ); let _ = sqlx::query(&sql) @@ -370,7 +370,7 @@ // Phase 3: Fetch records from PDS instances // --------------------------------------------------------------------------- -async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[String]) { +async fn run_fetching_phase(state: &AppState, job_id: &str, collections: &[String]) -> (i32, i32) { set_stage(state, job_id, "fetching_records").await; // Load pending repos grouped by PDS @@ -400,6 +400,9 @@ .await .map(|(c,)| c) .unwrap_or(0); + + // Reset processed_repos for the fetching phase + update_job_counter(state, job_id, "processed_repos", already_completed).await; let processed_repos = Arc::new(AtomicI32::new(already_completed)); let total_records = Arc::new(AtomicI32::new(0)); @@ -497,6 +500,23 @@ } }) .await; + + let final_repos = processed_repos.load(Ordering::Relaxed); + let final_records = total_records.load(Ordering::Relaxed); + + // Persist final counts so they're accurate regardless of batch size + let sql = adapt_sql( + "UPDATE backfill_jobs SET processed_repos = ?, total_records = ? WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(final_repos) + .bind(final_records) + .bind(job_id) + .execute(&state.db) + .await; + + (final_repos, final_records) } /// Fetch all records for a given DID and collection from a PDS via @@ -680,35 +700,13 @@ } } - run_fetching_phase(&state, &job_id, &collections).await; + let (final_processed, final_records) = run_fetching_phase(&state, &job_id, &collections).await; if is_cancelled(&state, &job_id).await { tracing::info!(job_id, "backfill job cancelled"); + finalise_cancel(&state, &job_id).await; return; } - - // Read final counters from backfill_repos before cleanup - let sql = adapt_sql( - "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ? AND status = 'completed'", - state.db_backend, - ); - let final_processed: i32 = sqlx::query_as::<_, (i32,)>(&sql) - .bind(&job_id) - .fetch_one(&state.db) - .await - .map(|(c,)| c) - .unwrap_or(0); - - let sql = adapt_sql( - "SELECT total_records FROM backfill_jobs WHERE id = ?", - state.db_backend, - ); - let final_records: i32 = sqlx::query_as::<_, (i32,)>(&sql) - .bind(&job_id) - .fetch_one(&state.db) - .await - .map(|(c,)| c) - .unwrap_or(0); complete_job(&state, &job_id, final_processed, final_records, None).await; diff --git a/web/src/app/dashboard/backfill/page.tsx b/web/src/app/dashboard/backfill/page.tsx --- a/web/src/app/dashboard/backfill/page.tsx +++ b/web/src/app/dashboard/backfill/page.tsx @@ -50,30 +50,14 @@ TableRow, } from "@/components/ui/table"; -const STAGES = [ - "pending", +const PROGRESS_PHASES = [ "discovering_repos", "resolving_pds", "fetching_records", - "completed", - "failed", - "cancelled", - "cancelling", ] as const; -const STAGE_LABELS: Record = { - pending: "Pending", - discovering_repos: "Discovering repos", - resolving_pds: "Resolving PDS", - fetching_records: "Fetching records", - completed: "Completed", - failed: "Failed", - cancelled: "Cancelled", - cancelling: "Cancelling", -}; - -function stageBadge(stage: string) { - switch (stage) { +function statusBadge(job: BackfillJob) { + switch (job.status) { case "completed": return ( @@ -94,20 +78,22 @@ cancelling ); - case "pending": - return pending; - default: + case "running": return ( - {STAGE_LABELS[stage] ?? stage} + {job.stage === "pending" ? "starting" : job.stage.replace(/_/g, " ")} ); + default: + return {job.status}; } } -function stageIndex(stage: string): number { - const idx = STAGES.indexOf(stage as (typeof STAGES)[number]); - return idx === -1 ? 0 : idx; +function phaseIndex(stage: string): number { + const idx = PROGRESS_PHASES.indexOf( + stage as (typeof PROGRESS_PHASES)[number], + ); + return idx; } export default function BackfillPage() { @@ -183,7 +169,7 @@ {job.did ?? "All"} - {stageBadge(job.stage)} + {statusBadge(job)} {job.started_at ? new Date(job.started_at).toLocaleString() @@ -229,8 +215,14 @@ onCancel: () => Promise; }) { const [cancelling, setCancelling] = useState(false); - const current = stageIndex(job.stage); + const current = phaseIndex(job.stage); + const allDone = job.status === "completed"; const isActive = job.status === "running" || job.status === "cancelling"; + + function hasReached(phase: (typeof PROGRESS_PHASES)[number]): boolean { + if (allDone) return true; + return current >= phaseIndex(phase); + } async function handleCancel() { setCancelling(true); @@ -300,17 +292,17 @@
= stageIndex("discovering_repos")} + active={isActive && job.stage === "discovering_repos"} + reached={hasReached("discovering_repos")} value={job.total_repos?.toLocaleString()} suffix="repos found" /> = stageIndex("resolving_pds")} + active={isActive && job.stage === "resolving_pds"} + reached={hasReached("resolving_pds")} value={ - current >= stageIndex("resolving_pds") + hasReached("resolving_pds") ? `${job.processed_repos?.toLocaleString() ?? "0"} / ${job.total_repos?.toLocaleString() ?? "0"}` : undefined } @@ -318,15 +310,15 @@ /> = stageIndex("fetching_records")} + active={isActive && job.stage === "fetching_records"} + reached={hasReached("fetching_records")} value={ - current >= stageIndex("fetching_records") + hasReached("fetching_records") ? `${job.processed_repos?.toLocaleString() ?? "0"} / ${job.total_repos?.toLocaleString() ?? "0"} repos` : undefined } suffix={ - current >= stageIndex("fetching_records") + hasReached("fetching_records") ? `${job.total_records?.toLocaleString() ?? "0"} records` : undefined } -- tangled.sh