diff --git a/migrations/postgres/20260513000000_add_backfill_stage.sql b/migrations/postgres/20260513000000_add_backfill_stage.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260513000000_add_backfill_stage.sql @@ -0,0 +1,4 @@ +ALTER TABLE backfill_jobs ADD COLUMN stage TEXT NOT NULL DEFAULT 'pending'; + +UPDATE backfill_jobs SET stage = status WHERE status IN ('completed', 'failed'); +UPDATE backfill_jobs SET stage = 'failed', status = 'failed', error = 'interrupted by restart' WHERE status = 'running'; diff --git a/migrations/postgres/20260513000001_create_backfill_repos.sql b/migrations/postgres/20260513000001_create_backfill_repos.sql new file mode 100644 --- /dev/null +++ b/migrations/postgres/20260513000001_create_backfill_repos.sql @@ -0,0 +1,7 @@ +CREATE TABLE IF NOT EXISTS backfill_repos ( + job_id TEXT NOT NULL REFERENCES backfill_jobs(id) ON DELETE CASCADE, + did TEXT NOT NULL, + pds_endpoint TEXT, + status TEXT NOT NULL DEFAULT 'pending', + PRIMARY KEY (job_id, did) +); diff --git a/migrations/sqlite/20260513000000_add_backfill_stage.sql b/migrations/sqlite/20260513000000_add_backfill_stage.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260513000000_add_backfill_stage.sql @@ -0,0 +1,4 @@ +ALTER TABLE backfill_jobs ADD COLUMN stage TEXT NOT NULL DEFAULT 'pending'; + +UPDATE backfill_jobs SET stage = status WHERE status IN ('completed', 'failed'); +UPDATE backfill_jobs SET stage = 'failed', status = 'failed', error = 'interrupted by restart' WHERE status = 'running'; diff --git a/migrations/sqlite/20260513000001_create_backfill_repos.sql b/migrations/sqlite/20260513000001_create_backfill_repos.sql new file mode 100644 --- /dev/null +++ b/migrations/sqlite/20260513000001_create_backfill_repos.sql @@ -0,0 +1,7 @@ +CREATE TABLE IF NOT EXISTS backfill_repos ( + job_id TEXT NOT NULL REFERENCES backfill_jobs(id) ON DELETE CASCADE, + did TEXT NOT NULL, + pds_endpoint TEXT, + status TEXT NOT NULL DEFAULT 'pending', + PRIMARY KEY (job_id, did) +); diff --git a/packages/docs/content/docs/api-reference/admin/backfill.md b/packages/docs/content/docs/api-reference/admin/backfill.md --- a/packages/docs/content/docs/api-reference/admin/backfill.md +++ b/packages/docs/content/docs/api-reference/admin/backfill.md @@ -96,9 +96,60 @@ ```json { "id": "550e8400-e29b-41d4-a716-446655440000", - "status": "pending" + "status": "running" +} +``` + +## Cancel a backfill job + +``` +POST /admin/backfill/{id}/cancel +``` + +Requests cancellation of a running backfill job. The job status transitions to `cancelling` immediately; the background worker will stop at its next checkpoint and set the final status to `cancelled`. See the [Backfill guide](../../guides/backfill.md#cancelling-a-job) for details on the two-phase process. + +```ts tab="TypeScript" tab-group="language" +const response = await fetch( + `http://127.0.0.1:3000/admin/backfill/${jobId}/cancel`, + { method: "POST", headers }, +); +const data = await response.json(); +``` +```js tab="JavaScript" tab-group="language" +const response = await fetch( + `http://127.0.0.1:3000/admin/backfill/${jobId}/cancel`, + { method: "POST", headers }, +); +const data = await response.json(); +``` +```rust tab="Rust" tab-group="language" +let response = client + .post(format!("http://127.0.0.1:3000/admin/backfill/{job_id}/cancel")) + .bearer_auth(token) + .send() + .await?; +let data: serde_json::Value = response.json().await?; +``` +```go tab="Go" tab-group="language" +url := fmt.Sprintf("http://127.0.0.1:3000/admin/backfill/%s/cancel", jobID) +req, _ := http.NewRequest("POST", url, nil) +req.Header.Set("Authorization", "Bearer "+token) +resp, err := http.DefaultClient.Do(req) +``` +```sh tab="cURL" tab-group="language" +curl -X POST "http://127.0.0.1:3000/admin/backfill/$JOB_ID/cancel" -H "$AUTH" +``` + +**Response**: `200 OK` + +```json +{ + "id": "550e8400-e29b-41d4-a716-446655440000", + "status": "cancelling" } ``` + +Returns `400` if the job is not currently running, or `404` if the job ID is not found. ## List backfill jobs @@ -112,9 +163,10 @@ id: string; collection: string | null; did: string | null; status: string; - total_repos: number; - processed_repos: number; - total_records: number; + stage: string; + total_repos: number | null; + processed_repos: number | null; + total_records: number | null; error: string | null; started_at: string | null; completed_at: string | null; @@ -158,6 +210,7 @@ "id": "550e8400-e29b-41d4-a716-446655440000", "collection": "xyz.statusphere.status", "did": null, "status": "completed", + "stage": "completed", "total_repos": 42, "processed_repos": 42, "total_records": 1000, @@ -168,3 +221,5 @@ "created_at": "2025-01-01T00:00:00Z" } ] ``` + +The `status` field tracks the overall job state (`running`, `cancelling`, `cancelled`, `completed`, `failed`). The `stage` field tracks the current processing phase (`pending`, `discovering_repos`, `resolving_pds`, `fetching_records`, `completed`, `failed`, `cancelled`). diff --git a/packages/docs/content/docs/getting-started/dashboard.md b/packages/docs/content/docs/getting-started/dashboard.md --- a/packages/docs/content/docs/getting-started/dashboard.md +++ b/packages/docs/content/docs/getting-started/dashboard.md @@ -49,7 +49,7 @@ Navigate to **Records** to browse all indexed atproto records. Records are grouped by collection and searchable. Each record shows its AT URI, author DID, and the raw record JSON. ### Backfill -Navigate to **Backfill** to view and manage backfill jobs. You can start a new backfill for any record-type lexicon to import historical records from the network. The page shows job status, progress (repos processed / total), and record counts. See [Backfill](../guides/backfill.md) for how the process works. +Navigate to **Backfill** to view and manage backfill jobs. You can start a new backfill for any record-type lexicon to import historical records from the network. The table shows each job's collection, DID scope, current stage, and start time. Click a row to open a detail sheet with full metadata and a stage-by-stage progress log that updates in real time. Running jobs can be cancelled from the detail sheet — the job transitions to "cancelling" while the worker finishes its current batch, then to "cancelled". See [Backfill](../guides/backfill.md) for how the process works. ### Dead Letters diff --git a/packages/docs/content/docs/guides/backfill.md b/packages/docs/content/docs/guides/backfill.md --- a/packages/docs/content/docs/guides/backfill.md +++ b/packages/docs/content/docs/guides/backfill.md @@ -13,17 +13,49 @@ See the [admin API](../api-reference/admin/backfill.md) for endpoint details. ## How it works -1. **Determine target collections**: uses the specified collection, or all record lexicons with `backfill: true` -2. **Discover DIDs**: HappyView calls the relay's `com.atproto.sync.listReposByCollection` to find repos that contain records for each target collection (paginated) -3. **Resolve each PDS**: for each discovered DID, HappyView resolves the DID document via PLC to find the user's PDS endpoint -4. **Fetch records**: HappyView calls `com.atproto.repo.listRecords` on each PDS for the target collection (paginated) and upserts each record into the local database -5. **Track progress**: counters for `processed_repos` and `total_records` are updated as the job runs +A backfill job runs through three sequential phases: + +1. **Discovering repos** — HappyView calls the relay's `com.atproto.sync.listReposByCollection` to find repos that contain records for each target collection. Discovered DIDs are stored in a tracking table so progress can be resumed. +2. **Resolving PDS** — For each discovered DID, HappyView resolves the DID document (via PLC directory or `did:web`) to find the user's PDS endpoint. +3. **Fetching records** — HappyView calls `com.atproto.repo.listRecords` on each PDS for the target collection(s), upserting each record into the local database. PDS endpoints are processed concurrently (up to 10 PDS hosts, 3 DIDs per host). + +Progress counters (`total_repos`, `processed_repos`, `total_records`) and the current `stage` are updated in real time. The dashboard's Backfill page shows live progress, and clicking a job opens a detail sheet with a stage-by-stage progress log. + +### Rate limiting + +All three phases handle HTTP 429 responses. HappyView reads the `RateLimit-Reset` header (a Unix timestamp, the AT Protocol convention) to determine how long to wait, falling back to the `retry-after` header, then defaulting to 5 seconds. ## Job lifecycle -A backfill job moves through `pending → running → completed` (or `failed`). Unlike earlier versions of HappyView that relied on Tap, the job is only marked `completed` once every discovered repo has been fully processed — there is no separate downstream queue. Progress is visible in real time on the dashboard's Backfill page. +A backfill job has both a `status` (overall state) and a `stage` (current phase): -If a job fails midway, the `error` field contains the failure reason. Re-running the backfill resumes from scratch but is idempotent (records are upserted by URI). +| Status | Description | +| ------------ | ---------------------------------------------------- | +| `running` | Job is actively processing | +| `cancelling` | Cancel requested, waiting for the worker to stop | +| `cancelled` | Worker has stopped and cleaned up | +| `completed` | Worker finished processing all resolvable repos | +| `failed` | An error occurred | + +The `stage` field tracks which phase the job is in: `pending`, `discovering_repos`, `resolving_pds`, `fetching_records`, `completed`, `failed`, or `cancelled`. + +## Cancelling a job + +Running jobs can be cancelled via `POST /admin/backfill/{id}/cancel` or the Cancel button in the dashboard. Cancellation is two-phase: + +1. The endpoint sets the job status to `cancelling`. +2. The worker checks for cancellation at natural checkpoints (between relay pages, every 100 DIDs during resolution, every 10 repos during fetching). When it detects the `cancelling` status, it stops work and sets the final status to `cancelled`. + +This means there may be a short delay between clicking Cancel and the job fully stopping, depending on what the worker is doing at that moment. + +## Resuming after restart + +Backfill jobs survive server restarts. On startup, HappyView checks for jobs that were running when the server last stopped: + +- **Running** jobs are re-spawned and resume from where they left off. Each phase is idempotent — discovery skips already-known DIDs, resolution skips already-resolved endpoints, and fetching skips already-completed repos. +- **Cancelling** jobs (where the cancel was requested but the worker hadn't stopped yet) are immediately finalised as `cancelled`. + +Per-DID progress is tracked in the database, so a job that was halfway through fetching records will pick up from the next unprocessed repo, not start over. ## Re-running backfills diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -1,9 +1,9 @@ use std::collections::HashMap; use std::sync::Arc; -use std::sync::atomic::{AtomicI32, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicI32, Ordering}; use axum::Json; -use axum::extract::State; +use axum::extract::{Path, State}; use axum::http::StatusCode; use futures_util::stream::{self, StreamExt}; use serde::Deserialize; @@ -14,6 +14,7 @@ use crate::AppState; use crate::db::{adapt_sql, now_rfc3339}; use crate::error::AppError; use crate::event_log::{EventLog, Severity, log_event}; +use crate::http_retry::parse_retry_after; use crate::profile; use crate::record_handler::{self, RecordEvent}; @@ -22,7 +23,7 @@ use super::permissions::Permission; use super::types::{BackfillJob, CreateBackfillBody}; // --------------------------------------------------------------------------- -// Relay discovery (reused from old backfill module) +// Response types // --------------------------------------------------------------------------- #[derive(Deserialize)] @@ -36,10 +37,6 @@ struct RepoEntry { did: String, } -// --------------------------------------------------------------------------- -// PDS record types -// --------------------------------------------------------------------------- - #[derive(Deserialize)] struct ListRecordsResponse { records: Vec, @@ -53,16 +50,183 @@ cid: String, value: serde_json::Value, } -/// Discover all DIDs that have records in `collection` via the relay's -/// `com.atproto.sync.listReposByCollection` endpoint. Paginates until done. -async fn list_repos_by_collection( - http: &reqwest::Client, - relay_url: &str, +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +async fn set_stage(state: &AppState, job_id: &str, stage: &str) { + let sql = adapt_sql( + "UPDATE backfill_jobs SET stage = ? WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(stage) + .bind(job_id) + .execute(&state.db) + .await; +} + +async fn update_job_counter(state: &AppState, job_id: &str, column: &str, value: i32) { + let query = match column { + "total_repos" => "UPDATE backfill_jobs SET total_repos = ? WHERE id = ?", + "processed_repos" => "UPDATE backfill_jobs SET processed_repos = ? WHERE id = ?", + "total_records" => "UPDATE backfill_jobs SET total_records = ? WHERE id = ?", + other => { + tracing::error!( + column = other, + "update_job_counter called with unknown column" + ); + return; + } + }; + let sql = adapt_sql(query, state.db_backend); + let _ = sqlx::query(&sql) + .bind(value) + .bind(job_id) + .execute(&state.db) + .await; +} + +async fn count_repos(state: &AppState, job_id: &str) -> i32 { + let sql = adapt_sql( + "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ?", + state.db_backend, + ); + sqlx::query_as::<_, (i32,)>(&sql) + .bind(job_id) + .fetch_one(&state.db) + .await + .map(|(c,)| c) + .unwrap_or(0) +} + +async fn cleanup_repos(state: &AppState, job_id: &str) { + let sql = adapt_sql( + "DELETE FROM backfill_repos WHERE job_id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql).bind(job_id).execute(&state.db).await; +} + +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', completed_at = ?, error = ? WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(&now) + .bind(error) + .bind(job_id) + .execute(&state.db) + .await; + cleanup_repos(state, job_id).await; +} + +async fn is_cancelled(state: &AppState, job_id: &str) -> bool { + let sql = adapt_sql( + "SELECT status FROM backfill_jobs WHERE id = ?", + state.db_backend, + ); + sqlx::query_as::<_, (String,)>(&sql) + .bind(job_id) + .fetch_optional(&state.db) + .await + .ok() + .flatten() + .is_some_and(|(status,)| status == "cancelling") +} + +async fn request_cancel(state: &AppState, job_id: &str) { + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'cancelling' WHERE id = ? AND status = 'running'", + state.db_backend, + ); + let _ = sqlx::query(&sql).bind(job_id).execute(&state.db).await; +} + +async fn finalise_cancel(state: &AppState, job_id: &str) { + let now = now_rfc3339(); + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'cancelled', completed_at = ?, error = 'cancelled by user' WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(&now) + .bind(job_id) + .execute(&state.db) + .await; + cleanup_repos(state, job_id).await; +} + +async fn complete_job( + state: &AppState, + job_id: &str, + processed_repos: i32, + total_records: i32, + error: Option<&str>, +) { + let now = now_rfc3339(); + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'completed', stage = 'completed', completed_at = ?, processed_repos = ?, total_records = ?, error = ? WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(&now) + .bind(processed_repos) + .bind(total_records) + .bind(error) + .bind(job_id) + .execute(&state.db) + .await; + cleanup_repos(state, job_id).await; +} + +// --------------------------------------------------------------------------- +// Phase 1: Discover repos via relay +// --------------------------------------------------------------------------- + +async fn run_discovery_phase( + state: &AppState, + job_id: &str, + collections: &[String], + specific_did: Option<&str>, +) { + set_stage(state, job_id, "discovering_repos").await; + + if let Some(did) = specific_did { + let sql = adapt_sql( + "INSERT INTO backfill_repos (job_id, did) VALUES (?, ?) ON CONFLICT DO NOTHING", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(job_id) + .bind(did) + .execute(&state.db) + .await; + } else { + for collection in collections { + if is_cancelled(state, job_id).await { + return; + } + if let Err(e) = discover_repos_from_relay(state, job_id, collection).await { + tracing::warn!(collection, error = %e, "failed to discover repos, skipping"); + } + } + } + + let total = count_repos(state, job_id).await; + update_job_counter(state, job_id, "total_repos", total).await; +} + +async fn discover_repos_from_relay( + state: &AppState, + job_id: &str, collection: &str, -) -> Result, String> { - let base = relay_url.trim_end_matches('/'); - let mut dids = Vec::new(); +) -> Result<(), String> { + let base = state.config.relay_url.trim_end_matches('/'); let mut cursor: Option = None; + let mut running_total: i32 = count_repos(state, job_id).await; loop { let mut url = format!( @@ -72,11 +236,23 @@ if let Some(ref c) = cursor { url.push_str(&format!("&cursor={c}")); } - let resp = http - .get(&url) - .send() - .await - .map_err(|e| format!("relay request failed: {e}"))?; + let resp = loop { + let r = state + .http + .get(&url) + .send() + .await + .map_err(|e| format!("relay request failed: {e}"))?; + + if r.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { + let wait = parse_retry_after(r.headers()); + tracing::warn!(collection, wait, "rate limited by relay, sleeping"); + tokio::time::sleep(tokio::time::Duration::from_secs(wait)).await; + continue; + } + + break r; + }; if !resp.status().is_success() { return Err(format!("relay returned {}", resp.status())); @@ -88,8 +264,47 @@ .await .map_err(|e| format!("invalid relay response: {e}"))?; let page_count = body.repos.len(); - for repo in body.repos { - dids.push(repo.did); + + if !body.repos.is_empty() { + // SQLite has a 999 bound-parameter limit; each row uses 2 params + let chunk_size = if state.db_backend == crate::db::DatabaseBackend::Sqlite { + 499 + } else { + 1000 + }; + + for chunk in body.repos.chunks(chunk_size) { + let base_sql = "INSERT INTO backfill_repos (job_id, did) VALUES "; + let placeholders: Vec = chunk + .iter() + .enumerate() + .map(|(i, _)| { + if state.db_backend == crate::db::DatabaseBackend::Postgres { + format!("(${}, ${})", i * 2 + 1, i * 2 + 2) + } else { + "(?, ?)".to_string() + } + }) + .collect(); + let sql = format!( + "{base_sql}{} ON CONFLICT DO NOTHING", + placeholders.join(", ") + ); + + let mut query = sqlx::query(&sql); + for repo in chunk { + query = query.bind(job_id).bind(&repo.did); + } + if let Ok(result) = query.execute(&state.db).await { + running_total += result.rows_affected() as i32; + } + } + } + + update_job_counter(state, job_id, "total_repos", running_total).await; + + if is_cancelled(state, job_id).await { + return Ok(()); } match body.cursor { @@ -98,13 +313,238 @@ _ => break, } } - Ok(dids) + Ok(()) } // --------------------------------------------------------------------------- -// PDS record fetching +// Phase 2: Resolve PDS endpoints // --------------------------------------------------------------------------- +async fn run_resolution_phase(state: &AppState, job_id: &str) { + set_stage(state, job_id, "resolving_pds").await; + + let sql = adapt_sql( + "SELECT did FROM backfill_repos WHERE job_id = ? AND pds_endpoint IS NULL", + state.db_backend, + ); + let unresolved: Vec<(String,)> = sqlx::query_as(&sql) + .bind(job_id) + .fetch_all(&state.db) + .await + .unwrap_or_default(); + + let sql = adapt_sql( + "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ? AND pds_endpoint IS NOT NULL", + state.db_backend, + ); + let already_resolved: i32 = sqlx::query_as::<_, (i32,)>(&sql) + .bind(job_id) + .fetch_one(&state.db) + .await + .map(|(c,)| c) + .unwrap_or(0); + + let mut resolved_count = already_resolved; + + let mut attempted = already_resolved; + for (did,) in &unresolved { + match profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did).await { + Ok(pds) => { + let sql = adapt_sql( + "UPDATE backfill_repos SET pds_endpoint = ? WHERE job_id = ? AND did = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(&pds) + .bind(job_id) + .bind(did) + .execute(&state.db) + .await; + resolved_count += 1; + } + Err(e) => { + tracing::warn!(did, error = %e, "failed to resolve PDS endpoint, skipping DID"); + } + } + attempted += 1; + if attempted % 100 == 0 { + update_job_counter(state, job_id, "processed_repos", resolved_count).await; + if is_cancelled(state, job_id).await { + return; + } + } + } + + update_job_counter(state, job_id, "processed_repos", resolved_count).await; +} + +// --------------------------------------------------------------------------- +// Phase 3: Fetch records from PDS instances +// --------------------------------------------------------------------------- + +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 + let sql = adapt_sql( + "SELECT did, pds_endpoint FROM backfill_repos WHERE job_id = ? AND status = 'pending' AND pds_endpoint IS NOT NULL", + state.db_backend, + ); + let rows: Vec<(String, String)> = sqlx::query_as(&sql) + .bind(job_id) + .fetch_all(&state.db) + .await + .unwrap_or_default(); + + let mut pds_to_dids: HashMap> = HashMap::new(); + for (did, pds) in rows { + pds_to_dids.entry(pds).or_default().push(did); + } + + // Count already-completed repos for accurate progress + let sql = adapt_sql( + "SELECT COUNT(*) FROM backfill_repos WHERE job_id = ? AND status = 'completed'", + state.db_backend, + ); + let already_completed: i32 = sqlx::query_as::<_, (i32,)>(&sql) + .bind(job_id) + .fetch_one(&state.db) + .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; + + // Seed total_records from DB so a resumed job doesn't lose its prior count + let existing_records: i32 = { + let sql = adapt_sql( + "SELECT total_records FROM backfill_jobs WHERE id = ?", + state.db_backend, + ); + sqlx::query_as::<_, (Option,)>(&sql) + .bind(job_id) + .fetch_one(&state.db) + .await + .map(|(c,)| c.unwrap_or(0)) + .unwrap_or(0) + }; + + let processed_repos = Arc::new(AtomicI32::new(already_completed)); + let total_records = Arc::new(AtomicI32::new(existing_records)); + let cancelled = Arc::new(AtomicBool::new(false)); + let state = Arc::new(state.clone()); + let collections = Arc::new(collections.to_vec()); + let job_id_arc = Arc::new(job_id.to_string()); + + let pds_entries: Vec<(String, Vec)> = pds_to_dids.into_iter().collect(); + + stream::iter(pds_entries) + .for_each_concurrent(10, |(pds_endpoint, dids)| { + let state = Arc::clone(&state); + let collections = Arc::clone(&collections); + let processed_repos = Arc::clone(&processed_repos); + let total_records = Arc::clone(&total_records); + let cancelled = Arc::clone(&cancelled); + let job_id = Arc::clone(&job_id_arc); + + async move { + stream::iter(dids) + .for_each_concurrent(3, |did| { + let state = Arc::clone(&state); + let collections = Arc::clone(&collections); + let processed_repos = Arc::clone(&processed_repos); + let total_records = Arc::clone(&total_records); + let cancelled = Arc::clone(&cancelled); + let pds_endpoint = pds_endpoint.clone(); + let job_id = Arc::clone(&job_id); + + async move { + if cancelled.load(Ordering::Relaxed) { + return; + } + + for collection in collections.iter() { + match fetch_records_from_pds( + &state, + &pds_endpoint, + &did, + collection, + ) + .await + { + Ok(count) => { + total_records + .fetch_add(count as i32, Ordering::Relaxed); + } + Err(e) => { + tracing::warn!( + did, + collection, + pds = %pds_endpoint, + error = %e, + "failed to fetch records from PDS" + ); + } + } + } + + // Mark DID as completed + let sql = adapt_sql( + "UPDATE backfill_repos SET status = 'completed' WHERE job_id = ? AND did = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(job_id.as_str()) + .bind(&did) + .execute(&state.db) + .await; + + let repos = processed_repos.fetch_add(1, Ordering::Relaxed) + 1; + + if repos % 10 == 0 { + let records = total_records.load(Ordering::Relaxed); + let backend = state.db_backend; + let sql = adapt_sql( + "UPDATE backfill_jobs SET processed_repos = ?, total_records = ? WHERE id = ?", + backend, + ); + let _ = sqlx::query(&sql) + .bind(repos) + .bind(records) + .bind(job_id.as_str()) + .execute(&state.db) + .await; + + if is_cancelled(&state, job_id.as_str()).await { + cancelled.store(true, Ordering::Relaxed); + } + } + } + }) + .await; + } + }) + .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 /// `com.atproto.repo.listRecords`, paginating and handling rate limits. async fn fetch_records_from_pds( @@ -132,22 +572,11 @@ .send() .await .map_err(|e| format!("PDS request failed: {e}"))?; - // Handle rate limiting if resp.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { - let retry_after = resp - .headers() - .get("retry-after") - .and_then(|v| v.to_str().ok()) - .and_then(|v| v.parse::().ok()) - .unwrap_or(5); - tracing::warn!( - did, - collection, - retry_after, - "rate limited by PDS, sleeping" - ); - tokio::time::sleep(tokio::time::Duration::from_secs(retry_after)).await; - continue; // retry same page + let wait = parse_retry_after(resp.headers()); + tracing::warn!(did, collection, wait, "rate limited by PDS, sleeping"); + tokio::time::sleep(tokio::time::Duration::from_secs(wait)).await; + continue; } if !resp.status().is_success() { @@ -187,77 +616,31 @@ Ok(count) } // --------------------------------------------------------------------------- -// Admin handlers +// Background backfill worker // --------------------------------------------------------------------------- -/// POST /admin/backfill — create a backfill job and spawn background work. -pub(super) async fn create_backfill( - State(state): State, - admin: UserAuth, - Json(body): Json, -) -> Result<(StatusCode, Json), AppError> { - admin.require(Permission::BackfillCreate).await?; +async fn run_backfill_job(state: AppState, job_id: String) { let backend = state.db_backend; - let now = now_rfc3339(); - let job_id = Uuid::new_v4().to_string(); + // Load job metadata let sql = adapt_sql( - "INSERT INTO backfill_jobs (id, collection, did, status, started_at, created_at) VALUES (?, ?, ?, 'running', ?, ?) RETURNING id", + "SELECT collection, did, stage FROM backfill_jobs WHERE id = ?", backend, ); - let row: (String,) = sqlx::query_as(&sql) + let job: Option<(Option, Option, String)> = sqlx::query_as(&sql) .bind(&job_id) - .bind(&body.collection) - .bind(&body.did) - .bind(&now) - .bind(&now) - .fetch_one(&state.db) + .fetch_optional(&state.db) .await - .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?; - - let job_id = row.0.clone(); - - log_event( - &state.db, - EventLog { - event_type: "backfill.started".to_string(), - severity: Severity::Info, - actor_did: Some(admin.did.clone()), - subject: body.collection.clone(), - detail: serde_json::json!({ - "job_id": job_id.clone(), - }), - }, - backend, - ) - .await; - - // Clone what we need and spawn the background job - let spawn_state = state.clone(); - let spawn_job_id = job_id.clone(); - let spawn_body = body.clone(); - tokio::spawn(async move { - run_backfill_job(spawn_state, spawn_job_id, spawn_body).await; - }); + .ok() + .flatten(); - Ok(( - StatusCode::CREATED, - Json(serde_json::json!({ - "id": job_id, - "status": "running", - })), - )) -} - -// --------------------------------------------------------------------------- -// Background backfill worker -// --------------------------------------------------------------------------- - -async fn run_backfill_job(state: AppState, job_id: String, body: CreateBackfillBody) { - let backend = state.db_backend; + let Some((collection, did, stage)) = job else { + tracing::error!(job_id, "backfill job not found"); + return; + }; // Determine target collections - let collections: Vec = if let Some(ref col) = body.collection { + let collections: Vec = if let Some(ref col) = collection { let lexicon_exists: bool = state .lexicons .get(col) @@ -297,157 +680,59 @@ .await; return; } - // Discover DIDs - let mut all_dids = Vec::new(); - - for collection in &collections { - let dids = if let Some(ref did) = body.did { - vec![did.clone()] - } else { - match list_repos_by_collection(&state.http, &state.config.relay_url, collection).await { - Ok(dids) => dids, - Err(e) => { - tracing::warn!(collection, error = %e, "failed to discover repos, skipping"); - continue; - } - } - }; - - all_dids.extend(dids); - } - - all_dids.sort(); - all_dids.dedup(); - - let total_repos = all_dids.len() as i32; - - // Update total_repos in DB - let sql = adapt_sql( - "UPDATE backfill_jobs SET total_repos = ? WHERE id = ?", - backend, - ); - let _ = sqlx::query(&sql) - .bind(total_repos) - .bind(&job_id) - .execute(&state.db) - .await; + // Run phases, skipping those already completed + if matches!(stage.as_str(), "pending" | "discovering_repos") { + run_discovery_phase(&state, &job_id, &collections, did.as_deref()).await; - if all_dids.is_empty() { - complete_job(&state, &job_id, 0, 0, None).await; + if is_cancelled(&state, &job_id).await { + tracing::info!(job_id, "backfill job cancelled"); + finalise_cancel(&state, &job_id).await; + return; + } - log_event( - &state.db, - EventLog { - event_type: "backfill.completed".to_string(), - severity: Severity::Info, - actor_did: None, - subject: body.collection.clone(), - detail: serde_json::json!({ - "job_id": job_id, - "total_repos": 0, - "total_records": 0, - }), - }, - backend, - ) - .await; - return; + let total = count_repos(&state, &job_id).await; + if total == 0 { + complete_job(&state, &job_id, 0, 0, None).await; + log_event( + &state.db, + EventLog { + event_type: "backfill.completed".to_string(), + severity: Severity::Info, + actor_did: None, + subject: collection.clone(), + detail: serde_json::json!({ + "job_id": job_id, + "total_repos": 0, + "total_records": 0, + }), + }, + backend, + ) + .await; + return; + } } - // Resolve DIDs to PDS endpoints and group by PDS - let mut pds_to_dids: HashMap> = HashMap::new(); + if matches!( + stage.as_str(), + "pending" | "discovering_repos" | "resolving_pds" + ) { + run_resolution_phase(&state, &job_id).await; - for did in &all_dids { - match profile::resolve_pds_endpoint(&state.http, &state.config.plc_url, did).await { - Ok(pds) => { - pds_to_dids.entry(pds).or_default().push(did.clone()); - } - Err(e) => { - tracing::warn!(did, error = %e, "failed to resolve PDS endpoint, skipping DID"); - } + if is_cancelled(&state, &job_id).await { + tracing::info!(job_id, "backfill job cancelled"); + finalise_cancel(&state, &job_id).await; + return; } } - let processed_repos = Arc::new(AtomicI32::new(0)); - let total_records = Arc::new(AtomicI32::new(0)); + let (final_processed, final_records) = run_fetching_phase(&state, &job_id, &collections).await; - let state = Arc::new(state); - let collections = Arc::new(collections); - let job_id_arc = Arc::new(job_id.clone()); - - // Process PDSes with nested concurrency - let pds_entries: Vec<(String, Vec)> = pds_to_dids.into_iter().collect(); - - stream::iter(pds_entries) - .for_each_concurrent(10, |(pds_endpoint, dids)| { - let state = Arc::clone(&state); - let collections = Arc::clone(&collections); - let processed_repos = Arc::clone(&processed_repos); - let total_records = Arc::clone(&total_records); - let job_id = Arc::clone(&job_id_arc); - - async move { - stream::iter(dids) - .for_each_concurrent(3, |did| { - let state = Arc::clone(&state); - let collections = Arc::clone(&collections); - let processed_repos = Arc::clone(&processed_repos); - let total_records = Arc::clone(&total_records); - let pds_endpoint = pds_endpoint.clone(); - let job_id = Arc::clone(&job_id); - - async move { - for collection in collections.iter() { - match fetch_records_from_pds( - &state, - &pds_endpoint, - &did, - collection, - ) - .await - { - Ok(count) => { - total_records - .fetch_add(count as i32, Ordering::Relaxed); - } - Err(e) => { - tracing::warn!( - did, - collection, - pds = %pds_endpoint, - error = %e, - "failed to fetch records from PDS" - ); - } - } - } - - let repos = processed_repos.fetch_add(1, Ordering::Relaxed) + 1; - - // Update DB progress every 100 repos - if repos % 100 == 0 { - let records = total_records.load(Ordering::Relaxed); - let backend = state.db_backend; - let sql = adapt_sql( - "UPDATE backfill_jobs SET processed_repos = ?, total_records = ? WHERE id = ?", - backend, - ); - let _ = sqlx::query(&sql) - .bind(repos) - .bind(records) - .bind(job_id.as_str()) - .execute(&state.db) - .await; - } - } - }) - .await; - } - }) - .await; - - let final_processed = processed_repos.load(Ordering::Relaxed); - let final_records = total_records.load(Ordering::Relaxed); + if is_cancelled(&state, &job_id).await { + tracing::info!(job_id, "backfill job cancelled"); + finalise_cancel(&state, &job_id).await; + return; + } complete_job(&state, &job_id, final_processed, final_records, None).await; @@ -457,7 +742,7 @@ EventLog { event_type: "backfill.completed".to_string(), severity: Severity::Info, actor_did: None, - subject: body.collection.clone(), + subject: collection, detail: serde_json::json!({ "job_id": job_id, "total_repos": final_processed, @@ -470,46 +755,111 @@ .await; } // --------------------------------------------------------------------------- -// Helper functions +// Admin handlers // --------------------------------------------------------------------------- -async fn fail_job(state: &AppState, job_id: &str, error: &str) { - let now = now_rfc3339(); +/// POST /admin/backfill — create a backfill job and spawn background work. +pub(super) async fn create_backfill( + State(state): State, + admin: UserAuth, + Json(body): Json, +) -> Result<(StatusCode, Json), AppError> { + admin.require(Permission::BackfillCreate).await?; let backend = state.db_backend; + + let now = now_rfc3339(); + let job_id = Uuid::new_v4().to_string(); let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'failed', completed_at = ?, error = ? WHERE id = ?", + "INSERT INTO backfill_jobs (id, collection, did, status, stage, started_at, created_at) VALUES (?, ?, ?, 'running', 'pending', ?, ?) RETURNING id", backend, ); - let _ = sqlx::query(&sql) + let row: (String,) = sqlx::query_as(&sql) + .bind(&job_id) + .bind(&body.collection) + .bind(&body.did) .bind(&now) - .bind(error) - .bind(job_id) - .execute(&state.db) - .await; + .bind(&now) + .fetch_one(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to create backfill job: {e}")))?; + + let job_id = row.0.clone(); + + log_event( + &state.db, + EventLog { + event_type: "backfill.started".to_string(), + severity: Severity::Info, + actor_did: Some(admin.did.clone()), + subject: body.collection.clone(), + detail: serde_json::json!({ + "job_id": job_id.clone(), + }), + }, + backend, + ) + .await; + + let spawn_state = state.clone(); + let spawn_job_id = job_id.clone(); + tokio::spawn(async move { + run_backfill_job(spawn_state, spawn_job_id).await; + }); + + Ok(( + StatusCode::CREATED, + Json(serde_json::json!({ + "id": job_id, + "status": "running", + })), + )) } -async fn complete_job( - state: &AppState, - job_id: &str, - processed_repos: i32, - total_records: i32, - error: Option<&str>, -) { - let now = now_rfc3339(); - let backend = state.db_backend; +/// POST /admin/backfill/{id}/cancel — cancel a running backfill job. +pub(super) async fn cancel_backfill( + State(state): State, + admin: UserAuth, + Path(job_id): Path, +) -> Result, AppError> { + admin.require(Permission::BackfillCreate).await?; let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'completed', completed_at = ?, processed_repos = ?, total_records = ?, error = ? WHERE id = ?", - backend, + "SELECT status FROM backfill_jobs WHERE id = ?", + state.db_backend, ); - let _ = sqlx::query(&sql) - .bind(&now) - .bind(processed_repos) - .bind(total_records) - .bind(error) - .bind(job_id) - .execute(&state.db) - .await; + let row: Option<(String,)> = sqlx::query_as(&sql) + .bind(&job_id) + .fetch_optional(&state.db) + .await + .map_err(|e| AppError::Internal(format!("failed to query backfill job: {e}")))?; + + match row { + None => Err(AppError::NotFound("backfill job not found".into())), + Some((ref status,)) if status == "cancelling" || status == "cancelled" => { + Ok(Json(serde_json::json!({ "id": job_id, "status": status }))) + } + Some((status,)) if status != "running" => Err(AppError::BadRequest(format!( + "job is not running (status: {status})" + ))), + Some(_) => { + request_cancel(&state, &job_id).await; + log_event( + &state.db, + EventLog { + event_type: "backfill.cancelling".to_string(), + severity: Severity::Info, + actor_did: Some(admin.did.clone()), + subject: None, + detail: serde_json::json!({ "job_id": job_id }), + }, + state.db_backend, + ) + .await; + Ok(Json( + serde_json::json!({ "id": job_id, "status": "cancelling" }), + )) + } + } } /// GET /admin/backfill/status — list all backfill jobs. @@ -521,7 +871,7 @@ auth.require(Permission::BackfillRead).await?; let backend = state.db_backend; let sql = adapt_sql( - "SELECT id, collection, did, status, total_repos, processed_repos, total_records, error, started_at, completed_at, created_at FROM backfill_jobs ORDER BY created_at DESC", + "SELECT id, collection, did, status, stage, total_repos, processed_repos, total_records, error, started_at, completed_at, created_at FROM backfill_jobs ORDER BY created_at DESC", backend, ); #[allow(clippy::type_complexity)] @@ -529,6 +879,7 @@ let rows: Vec<( String, Option, Option, + String, String, Option, Option, @@ -550,6 +901,7 @@ id, collection, did, status, + stage, total_repos, processed_repos, total_records, @@ -563,6 +915,7 @@ id, collection, did, status, + stage, total_repos, processed_repos, total_records, @@ -577,3 +930,36 @@ .collect(); Ok(Json(jobs)) } + +// --------------------------------------------------------------------------- +// Startup resumption +// --------------------------------------------------------------------------- + +/// Resume any backfill jobs that were running when the server last stopped. +/// Jobs stuck in `cancelling` are finalised immediately. +pub async fn resume_backfill_jobs(state: &AppState) { + let sql = adapt_sql( + "SELECT id, status FROM backfill_jobs WHERE status IN ('running', 'cancelling')", + state.db_backend, + ); + let rows: Vec<(String, String)> = sqlx::query_as(&sql) + .fetch_all(&state.db) + .await + .unwrap_or_default(); + + for (job_id, status) in rows { + if status == "cancelling" { + tracing::info!( + job_id, + "finalising cancelled backfill job from previous run" + ); + finalise_cancel(state, &job_id).await; + } else { + tracing::info!(job_id, "resuming interrupted backfill job"); + let spawn_state = state.clone(); + tokio::spawn(async move { + run_backfill_job(spawn_state, job_id).await; + }); + } + } +} diff --git a/src/admin/mod.rs b/src/admin/mod.rs --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -1,7 +1,7 @@ mod api_clients; mod api_keys; pub(crate) mod auth; -mod backfill; +pub mod backfill; mod dead_letters; mod domains; mod events; @@ -37,6 +37,7 @@ ) .route("/stats", get(stats::stats)) .route("/backfill", post(backfill::create_backfill)) .route("/backfill/status", get(backfill::backfill_status)) + .route("/backfill/{id}/cancel", post(backfill::cancel_backfill)) .route("/events", get(events::list_events)) .route("/users", post(users::create_user).get(users::list_users)) .route("/users/transfer-super", post(users::transfer_super)) diff --git a/src/admin/types.rs b/src/admin/types.rs --- a/src/admin/types.rs +++ b/src/admin/types.rs @@ -69,18 +69,19 @@ pub(super) did: Option, } #[derive(Serialize)] -pub(super) struct BackfillJob { - pub(super) id: String, - pub(super) collection: Option, - pub(super) did: Option, - pub(super) status: String, - pub(super) total_repos: Option, - pub(super) processed_repos: Option, - pub(super) total_records: Option, - pub(super) error: Option, - pub(super) started_at: Option, - pub(super) completed_at: Option, - pub(super) created_at: String, +pub(crate) struct BackfillJob { + pub(crate) id: String, + pub(crate) collection: Option, + pub(crate) did: Option, + pub(crate) status: String, + pub(crate) stage: String, + pub(crate) total_repos: Option, + pub(crate) processed_repos: Option, + pub(crate) total_records: Option, + pub(crate) error: Option, + pub(crate) started_at: Option, + pub(crate) completed_at: Option, + pub(crate) created_at: String, } // --------------------------------------------------------------------------- diff --git a/src/http_retry.rs b/src/http_retry.rs new file mode 100644 --- /dev/null +++ b/src/http_retry.rs @@ -0,0 +1,24 @@ +/// Parse rate-limit sleep duration from response headers. +/// Checks `RateLimit-Reset` (Unix timestamp, used by XRPC servers) first, +/// then `retry-after` (seconds), defaulting to 5s. +pub fn parse_retry_after(headers: &reqwest::header::HeaderMap) -> u64 { + if let Some(reset) = headers + .get("ratelimit-reset") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() as i64; + let wait = (reset - now).max(1) as u64; + return wait.min(120); + } + + headers + .get("retry-after") + .and_then(|v| v.to_str().ok()) + .and_then(|v| v.parse::().ok()) + .unwrap_or(5) + .min(120) +} diff --git a/src/lib.rs b/src/lib.rs --- a/src/lib.rs +++ b/src/lib.rs @@ -12,6 +12,7 @@ pub mod event_log; pub mod external_auth; pub mod feature_flags; pub mod feature_middleware; +pub mod http_retry; pub mod jetstream; pub mod labeler; pub mod lexicon; diff --git a/src/main.rs b/src/main.rs --- a/src/main.rs +++ b/src/main.rs @@ -644,6 +644,8 @@ state.config.event_log_retention_days, state.db_backend, )); + happyview::admin::backfill::resume_backfill_jobs(&state).await; + let app = server::router(state); let addr = config.listen_addr(); diff --git a/src/profile.rs b/src/profile.rs --- a/src/profile.rs +++ b/src/profile.rs @@ -1,6 +1,7 @@ use serde::{Deserialize, Serialize}; use crate::error::AppError; +use crate::http_retry::parse_retry_after; #[derive(Serialize)] pub struct Profile { @@ -140,11 +141,38 @@ } else { format!("{}/{did}", plc_url.trim_end_matches('/')) }; - let resp = http - .get(&url) - .send() - .await - .map_err(|e| AppError::Internal(format!("DID resolution failed: {e}")))?; + let resp = { + let max_retries = 5; + let mut attempts = 0; + loop { + let r = http + .get(&url) + .send() + .await + .map_err(|e| AppError::Internal(format!("DID resolution failed: {e}")))?; + + if r.status() == reqwest::StatusCode::TOO_MANY_REQUESTS { + attempts += 1; + if attempts >= max_retries { + return Err(AppError::Internal(format!( + "DID resolution for {did} rate-limited after {max_retries} retries" + ))); + } + let wait = parse_retry_after(r.headers()); + tracing::warn!( + did, + wait, + attempts, + max_retries, + "rate limited during DID resolution, sleeping" + ); + tokio::time::sleep(tokio::time::Duration::from_secs(wait)).await; + continue; + } + + break r; + } + }; if !resp.status().is_success() { return Err(AppError::NotFound(format!( diff --git a/tests/e2e_admin.rs b/tests/e2e_admin.rs --- a/tests/e2e_admin.rs +++ b/tests/e2e_admin.rs @@ -561,6 +561,137 @@ let json = json_body(resp).await; assert_eq!(json.as_array().unwrap().len(), 1); } +#[tokio::test] +#[serial] +async fn backfill_cancel_running_job() { + common::require_db!(); + let app = TestApp::new().await; + let backend = app.state.db_backend; + + // Insert a running job directly so we don't need a real relay. + let job_id = uuid::Uuid::new_v4().to_string(); + let now = now_rfc3339(); + let sql = adapt_sql( + "INSERT INTO backfill_jobs (id, status, stage, started_at, created_at) VALUES (?, 'running', 'discovering_repos', ?, ?)", + backend, + ); + sqlx::query(&sql) + .bind(&job_id) + .bind(&now) + .bind(&now) + .execute(&app.state.db) + .await + .unwrap(); + + let resp = app + .router + .clone() + .oneshot(admin_post( + &format!("/admin/backfill/{job_id}/cancel"), + app.admin_cookie(), + &json!({}), + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let json = json_body(resp).await; + assert_eq!(json["id"], job_id); + assert_eq!(json["status"], "cancelling"); +} + +#[tokio::test] +#[serial] +async fn backfill_cancel_already_cancelling_is_idempotent() { + common::require_db!(); + let app = TestApp::new().await; + let backend = app.state.db_backend; + + let job_id = uuid::Uuid::new_v4().to_string(); + let now = now_rfc3339(); + let sql = adapt_sql( + "INSERT INTO backfill_jobs (id, status, stage, started_at, created_at) VALUES (?, 'cancelling', 'fetching_records', ?, ?)", + backend, + ); + sqlx::query(&sql) + .bind(&job_id) + .bind(&now) + .bind(&now) + .execute(&app.state.db) + .await + .unwrap(); + + let resp = app + .router + .clone() + .oneshot(admin_post( + &format!("/admin/backfill/{job_id}/cancel"), + app.admin_cookie(), + &json!({}), + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::OK); + let json = json_body(resp).await; + assert_eq!(json["status"], "cancelling"); +} + +#[tokio::test] +#[serial] +async fn backfill_cancel_completed_returns_400() { + common::require_db!(); + let app = TestApp::new().await; + let backend = app.state.db_backend; + + let job_id = uuid::Uuid::new_v4().to_string(); + let now = now_rfc3339(); + let sql = adapt_sql( + "INSERT INTO backfill_jobs (id, status, stage, completed_at, created_at) VALUES (?, 'completed', 'completed', ?, ?)", + backend, + ); + sqlx::query(&sql) + .bind(&job_id) + .bind(&now) + .bind(&now) + .execute(&app.state.db) + .await + .unwrap(); + + let resp = app + .router + .clone() + .oneshot(admin_post( + &format!("/admin/backfill/{job_id}/cancel"), + app.admin_cookie(), + &json!({}), + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +#[serial] +async fn backfill_cancel_not_found_returns_404() { + common::require_db!(); + let app = TestApp::new().await; + + let resp = app + .router + .clone() + .oneshot(admin_post( + "/admin/backfill/nonexistent-id/cancel", + app.admin_cookie(), + &json!({}), + )) + .await + .unwrap(); + + assert_eq!(resp.status(), StatusCode::NOT_FOUND); +} + // --------------------------------------------------------------------------- // Admin management // --------------------------------------------------------------------------- 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 @@ -3,9 +3,16 @@ import { useCallback, useEffect, useState } from "react"; import { useCurrentUser } from "@/hooks/use-current-user"; -import { createBackfillJob, getBackfillJobs, getLexicons } from "@/lib/api"; +import { + cancelBackfillJob, + createBackfillJob, + getBackfillJobs, + getLexicons, +} from "@/lib/api"; import type { BackfillJob } from "@/types/backfill"; +import { CheckCircle2, Circle, Loader2 } from "lucide-react"; import { SiteHeader } from "@/components/site-header"; +import { Badge } from "@/components/ui/badge"; import { Button } from "@/components/ui/button"; import { Combobox, @@ -28,6 +35,13 @@ } from "@/components/ui/responsive-dialog"; import { Input } from "@/components/ui/input"; import { Label } from "@/components/ui/label"; import { + Sheet, + SheetContent, + SheetFooter, + SheetHeader, + SheetTitle, +} from "@/components/ui/sheet"; +import { Table, TableBody, TableCell, @@ -36,10 +50,57 @@ TableHeader, TableRow, } from "@/components/ui/table"; +const PROGRESS_PHASES = [ + "discovering_repos", + "resolving_pds", + "fetching_records", +] as const; + +function statusBadge(job: BackfillJob) { + switch (job.status) { + case "completed": + return ( + + completed + + ); + case "failed": + return failed; + case "cancelled": + return ( + + cancelled + + ); + case "cancelling": + return ( + + cancelling + + ); + case "running": + return ( + + {job.stage === "pending" ? "starting" : job.stage.replace(/_/g, " ")} + + ); + default: + return {job.status}; + } +} + +function phaseIndex(stage: string): number { + const idx = PROGRESS_PHASES.indexOf( + stage as (typeof PROGRESS_PHASES)[number], + ); + return idx; +} + export default function BackfillPage() { const { hasPermission } = useCurrentUser(); const [jobs, setJobs] = useState([]); const [error, setError] = useState(null); + const [selectedJobId, setSelectedJobId] = useState(null); const load = useCallback(() => { getBackfillJobs() @@ -51,11 +112,12 @@ useEffect(() => { load(); }, [load]); - // Auto-refresh every 5 seconds useEffect(() => { const interval = setInterval(load, 5000); return () => clearInterval(interval); }, [load]); + + const selectedJob = jobs.find((j) => j.id === selectedJobId) ?? null; return ( <> @@ -77,8 +139,7 @@ ID Collection DID - Progress - Records + Status Started @@ -86,7 +147,7 @@ {jobs.length === 0 && ( No backfill jobs yet. @@ -94,7 +155,19 @@ )} {jobs.map((job) => ( - + setSelectedJobId(job.id)} + onKeyDown={(e) => { + if (e.key === "Enter" || e.key === " ") { + e.preventDefault(); + setSelectedJobId(job.id); + } + }} + > {job.id.slice(0, 8)} @@ -104,14 +177,7 @@ {job.did ?? "All"} - - {job.processed_repos != null && job.total_repos != null - ? `${job.processed_repos} / ${job.total_repos}` - : "--"} - - - {job.total_records?.toLocaleString() ?? "--"} - + {statusBadge(job)} {job.started_at ? new Date(job.started_at).toLocaleString() @@ -122,16 +188,214 @@ ))} + + { + if (!open) setSelectedJobId(null); + }} + > + + {selectedJob && ( + { + await cancelBackfillJob(selectedJob.id); + load(); + }} + /> + )} + + ); } -function CreateDialog({ - onSuccess, +function JobDetail({ + job, + canCancel, + onCancel, }: { - onSuccess: () => void; + job: BackfillJob; + canCancel: boolean; + onCancel: () => Promise; }) { + const [cancelling, setCancelling] = useState(false); + 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); + try { + await onCancel(); + } finally { + setCancelling(false); + } + } + + return ( + <> + + + Backfill Details + + +
+
+
+ Job ID +

{job.id}

+
+
+ Collection +

{job.collection ?? "All"}

+
+
+ DID +

{job.did ?? "All"}

+
+
+ Created +

+ {new Date(job.created_at).toLocaleString()} +

+
+
+ Started +

+ {job.started_at + ? new Date(job.started_at).toLocaleString() + : "--"} +

+
+ {job.completed_at && ( +
+ Completed +

+ {new Date(job.completed_at).toLocaleString()} +

+
+ )} +
+ + {job.error && ( +
+ Error +
+ {job.error} +
+
+ )} + +
+ Progress +
+ + + +
+
+
+ {canCancel && isActive && ( + + + + )} + + ); +} + +function ProgressRow({ + label, + active, + reached, + value, + suffix, +}: { + label: string; + active: boolean; + reached: boolean; + value?: string; + suffix?: string; +}) { + const done = reached && !active; + + return ( +
+ + {active ? ( + + ) : done ? ( + + ) : ( + + )} + + {label} + {reached && value && ( + + {value} + {suffix ? ` · ${suffix}` : ""} + + )} +
+ ); +} + +function CreateDialog({ onSuccess }: { onSuccess: () => void }) { const [collection, setCollection] = useState(null); const [did, setDid] = useState(""); const [error, setError] = useState(null); @@ -177,7 +441,11 @@ { const target = e.target as HTMLElement; - if (target.closest("[data-slot='combobox-item'], [data-slot='combobox-content']")) { + if ( + target.closest( + "[data-slot='combobox-item'], [data-slot='combobox-content']", + ) + ) { e.preventDefault(); } }} diff --git a/web/src/app/dashboard/settings/users/page.tsx b/web/src/app/dashboard/settings/users/page.tsx --- a/web/src/app/dashboard/settings/users/page.tsx +++ b/web/src/app/dashboard/settings/users/page.tsx @@ -62,7 +62,9 @@ displayName?: string; description?: string; }; -function buildCategories(permissions: PermissionEntry[]): Record { +function buildCategories( + permissions: PermissionEntry[], +): Record { const cats: Record = {}; for (const p of permissions) { if (!cats[p.category]) cats[p.category] = []; @@ -80,25 +82,36 @@ const [selectedUserId, setSelectedUserId] = useState(null); const [pendingPermissions, setPendingPermissions] = useState([]); const [saving, setSaving] = useState(false); const [permSearch, setPermSearch] = useState(""); - const [permissionEntries, setPermissionEntries] = useState([]); + const [permissionEntries, setPermissionEntries] = useState( + [], + ); const [profiles, setProfiles] = useState>({}); const [templates, setTemplates] = useState([]); - const permissionCategories = React.useMemo(() => buildCategories(permissionEntries), [permissionEntries]); + const permissionCategories = React.useMemo( + () => buildCategories(permissionEntries), + [permissionEntries], + ); const filteredCategories = useMemo(() => { if (!permSearch.trim()) return permissionCategories; const terms = permSearch.toLowerCase().split(/\s+/); const result: Record = {}; - for (const [category, permissions] of Object.entries(permissionCategories)) { + for (const [category, permissions] of Object.entries( + permissionCategories, + )) { const matched = permissions.filter((p) => { - const haystack = `${p.name} ${p.description} ${p.category} ${p.key}`.toLowerCase(); + const haystack = + `${p.name} ${p.description} ${p.category} ${p.key}`.toLowerCase(); return terms.every((term) => haystack.includes(term)); }); if (matched.length > 0) result[category] = matched; } return result; }, [permissionCategories, permSearch]); - const allPermissionKeys = React.useMemo(() => permissionEntries.map((p) => p.key), [permissionEntries]); + const allPermissionKeys = React.useMemo( + () => permissionEntries.map((p) => p.key), + [permissionEntries], + ); const templatePermissions = React.useMemo(() => { const map: Record = {}; for (const t of templates) map[t.key] = t.permissions; @@ -156,7 +169,9 @@ useEffect(() => { if (!selectedUserId) return; const user = users.find((u) => u.id === selectedUserId); if (!user || user.did in profiles) return; - fetch(`https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile?actor=${encodeURIComponent(user.did)}`) + fetch( + `https://public.api.bsky.app/xrpc/app.bsky.actor.getProfile?actor=${encodeURIComponent(user.did)}`, + ) .then((res) => (res.ok ? res.json() : null)) .then((data) => { if (!data) return; @@ -184,21 +199,24 @@ function handleTogglePermission( _user: UserSummary, permission: string, - enabled: boolean + enabled: boolean, ) { setPendingPermissions((prev) => { const perms = new Set(prev); const [ns, action] = permission.split(":"); const nsReadPerm = allPermissionKeys.find( - (k) => k.startsWith(`${ns}:`) && (k.endsWith(":read") || k.endsWith(":view")) + (k) => + k.startsWith(`${ns}:`) && + (k.endsWith(":read") || k.endsWith(":view")), ); const isReadAction = action === "read" || action === "view"; if (enabled) { perms.add(permission); if (!isReadAction && nsReadPerm) perms.add(nsReadPerm); - if (permission === "records:delete-collection") perms.add("records:delete"); + if (permission === "records:delete-collection") + perms.add("records:delete"); } else { perms.delete(permission); if (isReadAction) { @@ -206,14 +224,18 @@ for (const p of prev) { if (p.startsWith(`${ns}:`) && p !== permission) perms.delete(p); } } - if (permission === "records:delete") perms.delete("records:delete-collection"); + if (permission === "records:delete") + perms.delete("records:delete-collection"); } return [...perms]; }); } - async function handleSavePermissions(userId: string, originalPermissions: string[]) { + async function handleSavePermissions( + userId: string, + originalPermissions: string[], + ) { const originalSet = new Set(originalPermissions); const pendingSet = new Set(pendingPermissions); @@ -254,8 +276,13 @@ {error &&

{error}

}

Users

- {(isCurrentUserSuper || currentUser?.permissions.includes("users:create")) && ( - + {(isCurrentUserSuper || + currentUser?.permissions.includes("users:create")) && ( + )}
@@ -282,14 +309,22 @@ )} {users.map((user) => ( - setSelectedUserId(user.id)}> + setSelectedUserId(user.id)} + >
{handles[user.did] && ( - @{handles[user.did]} + + @{handles[user.did]} + )} - {user.did} + + {user.did} +
{user.is_super && ( @@ -323,27 +358,41 @@ {(() => { const selectedUser = users.find((u) => u.id === selectedUserId); return ( - { - if (!open) { - const user = users.find((u) => u.id === selectedUserId); - if (user) { - const origSet = new Set(user.permissions); - const pendSet = new Set(pendingPermissions); - const unsaved = pendingPermissions.some((p) => !origSet.has(p)) || user.permissions.some((p) => !pendSet.has(p)); - if (unsaved) { - toast.warning("You have unsaved changes. Save or cancel before closing."); - return; + { + if (!open) { + const user = users.find((u) => u.id === selectedUserId); + if (user) { + const origSet = new Set(user.permissions); + const pendSet = new Set(pendingPermissions); + const unsaved = + pendingPermissions.some((p) => !origSet.has(p)) || + user.permissions.some((p) => !pendSet.has(p)); + if (unsaved) { + toast.warning( + "You have unsaved changes. Save or cancel before closing.", + ); + return; + } } + setSelectedUserId(null); + setPermSearch(""); } - setSelectedUserId(null); - setPermSearch(""); - } - }}> - { - if (e.target instanceof HTMLElement && e.target.closest("[data-sonner-toaster]")) { - e.preventDefault(); - } - }}> + }} + > + { + if ( + e.target instanceof HTMLElement && + e.target.closest("[data-sonner-toaster]") + ) { + e.preventDefault(); + } + }} + > {selectedUser && ( <> @@ -361,10 +410,13 @@ /> )}

- {profiles[selectedUser.did]?.displayName || handles[selectedUser.did] ? ( + {profiles[selectedUser.did]?.displayName || + handles[selectedUser.did] ? ( <> {profiles[selectedUser.did]?.displayName && ( - {profiles[selectedUser.did].displayName} + + {profiles[selectedUser.did].displayName} + )} {handles[selectedUser.did] && ( @@ -373,12 +425,18 @@ )} ) : ( - {selectedUser.did} + + {selectedUser.did} + )}

-

{selectedUser.did}

+

+ {selectedUser.did} +

{profiles[selectedUser.did]?.description && ( -

{profiles[selectedUser.did].description}

+

+ {profiles[selectedUser.did].description} +

)}
@@ -387,112 +445,177 @@ {(() => { const originalSet = new Set(selectedUser.permissions); const pendingSet = new Set(pendingPermissions); - const added = pendingPermissions.filter((p) => !originalSet.has(p)).length; - const removed = selectedUser.permissions.filter((p) => !pendingSet.has(p)).length; + const added = pendingPermissions.filter( + (p) => !originalSet.has(p), + ).length; + const removed = selectedUser.permissions.filter( + (p) => !pendingSet.has(p), + ).length; const hasChanges = added > 0 || removed > 0; return ( - <> -
-
- Role -

- {selectedUser.is_super ? ( - Owner - ) : "Member"} -

-
-
- Permissions -

- {selectedUser.is_super - ? `${allPermissionKeys.length}/${allPermissionKeys.length}` - : `${pendingPermissions.filter((p) => allPermissionKeys.includes(p)).length}/${allPermissionKeys.length}`} - {hasChanges && ( - - {added > 0 && +{added}} - {added > 0 && removed > 0 && " "} - {removed > 0 && -{removed}} - - )} -

-
-
- Created -

{new Date(selectedUser.created_at).toLocaleString()}

-
-
- Last Active -

- {selectedUser.last_used_at - ? new Date(selectedUser.last_used_at).toLocaleString() - : "Never"} -

-
-
+ <> +
+
+ + Role + +

+ {selectedUser.is_super ? ( + + Owner + + ) : ( + "Member" + )} +

+
+
+ + Permissions + +

+ {selectedUser.is_super + ? `${allPermissionKeys.length}/${allPermissionKeys.length}` + : `${pendingPermissions.filter((p) => allPermissionKeys.includes(p)).length}/${allPermissionKeys.length}`} + {hasChanges && ( + + {added > 0 && ( + + +{added} + + )} + {added > 0 && removed > 0 && " "} + {removed > 0 && ( + + -{removed} + + )} + + )} +

+
+
+ + Created + +

+ {new Date( + selectedUser.created_at, + ).toLocaleString()} +

+
+
+ + Last Active + +

+ {selectedUser.last_used_at + ? new Date( + selectedUser.last_used_at, + ).toLocaleString() + : "Never"} +

+
+
-
+
-
- - setPermSearch(e.target.value)} - className="pl-9" - /> -
+
+ + setPermSearch(e.target.value)} + className="pl-9" + /> +
-
- -
+
+ +
- -
- - {isCurrentUserSuper && ( - handleTransferSuper(selectedUser.id)} - /> - )} -
-
- - -
- - + +
+ + {isCurrentUserSuper && ( + + handleTransferSuper(selectedUser.id) + } + /> + )} +
+
+ + +
+ + ); })()} @@ -525,7 +648,8 @@ pendingPermissions: string[]; originalPermissions: string[]; onToggle: (user: UserSummary, permission: string, enabled: boolean) => void; }) { - const canUpdate = isCurrentUserSuper || currentUserPermissions.includes("users:update"); + const canUpdate = + isCurrentUserSuper || currentUserPermissions.includes("users:update"); const originalSet = new Set(originalPermissions); return ( @@ -537,7 +661,8 @@ {category}

{permissions.map((perm) => { - const enabled = user.is_super || pendingPermissions.includes(perm.key); + const enabled = + user.is_super || pendingPermissions.includes(perm.key); const wasEnabled = user.is_super || originalSet.has(perm.key); const isAdded = enabled && !wasEnabled; const isRemoved = !enabled && wasEnabled; @@ -550,7 +675,8 @@ disabled={ user.is_super || isSelf || !canUpdate || - (!isCurrentUserSuper && !currentUserPermissions.includes(perm.key)) + (!isCurrentUserSuper && + !currentUserPermissions.includes(perm.key)) } onCheckedChange={(checked) => onToggle(user, perm.key, checked) @@ -563,10 +689,16 @@ className="flex flex-col items-start cursor-pointer text-xs leading-tight" > {perm.name} - {isAdded && } - {isRemoved && } + {isAdded && ( + + )} + {isRemoved && ( + + )} + + + {perm.description} - {perm.description}
); @@ -597,11 +729,7 @@ return ( - @@ -611,8 +739,8 @@ Transfer Ownership Are you sure you want to transfer ownership to{" "} - {user.did}? You will - lose your owner privileges and cannot undo this action without their + {user.did}? You will lose + your owner privileges and cannot undo this action without their cooperation. diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -170,6 +170,13 @@ body: JSON.stringify(body), }); } +export function cancelBackfillJob(id: string) { + return apiFetch<{ id: string; status: string }>( + `/admin/backfill/${id}/cancel`, + { method: "POST" }, + ); +} + // Users export function getUsers() { return apiFetch("/admin/users"); diff --git a/web/src/types/backfill.ts b/web/src/types/backfill.ts --- a/web/src/types/backfill.ts +++ b/web/src/types/backfill.ts @@ -3,6 +3,7 @@ id: string collection: string | null did: string | null status: string + stage: string total_repos: number | null processed_repos: number | null total_records: number | null