From bd65caf32eafcc42ef65aae68559bae03b9f22a4 Mon Sep 17 00:00:00 2001 From: Trezy Date: Sat, 23 May 2026 01:14:07 +0000 Subject: [PATCH] feat: allow backfill jobs to be paused and resumed Signed-off-by: Trezy --- src/admin/backfill.rs | 296 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------- src/admin/mod.rs | 2 ++ src/admin/settings.rs | 24 ++++++++++++++++++++++++ web/package-lock.json | 31 ++++++------------------------- web/src/app/dashboard/backfill/page.tsx | 107 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--- web/src/app/dashboard/settings/general/page.tsx | 248 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++--------------------------------------------------------------------------------------------------- web/src/lib/api.ts | 14 ++++++++++++++ 7 file(s) changed, 558 insertion(s)(+), 164 deletion(s)(-) diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -179,23 +179,32 @@ }, ); } -async fn is_cancelled(state: &AppState, job_id: &str) -> bool { +async fn should_stop(state: &AppState, job_id: &str) -> Option<&'static str> { let sql = adapt_sql( "SELECT status FROM backfill_jobs WHERE id = ?", state.db_backend, ); - sqlx::query_as::<_, (String,)>(&sql) + let status = sqlx::query_as::<_, (String,)>(&sql) .bind(job_id) .fetch_optional(&state.backfill_db) .await .ok() .flatten() - .is_some_and(|(status,)| status == "cancelling") + .map(|(s,)| s); + match status.as_deref() { + Some("cancelling") => Some("cancelling"), + Some("pausing") => Some("pausing"), + _ => None, + } +} + +async fn should_stop_worker(state: &AppState, job_id: &str) -> bool { + should_stop(state, job_id).await.is_some() } async fn request_cancel(state: &AppState, job_id: &str) { let sql = adapt_sql( - "UPDATE backfill_jobs SET status = 'cancelling' WHERE id = ? AND status = 'running'", + "UPDATE backfill_jobs SET status = 'cancelling' WHERE id = ? AND status IN ('running', 'paused')", state.db_backend, ); let _ = sqlx::query(&sql) @@ -225,6 +234,36 @@ }, ); } +async fn request_pause(state: &AppState, job_id: &str) { + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'pausing' WHERE id = ? AND status = 'running'", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(job_id) + .execute(&state.backfill_db) + .await; +} + +async fn finalise_pause(state: &AppState, job_id: &str) { + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'paused' WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(job_id) + .execute(&state.backfill_db) + .await; + publish_event( + state, + super::types::BackfillEvent::JobCompleted { + job_id: job_id.to_string(), + status: "paused".to_string(), + error: None, + }, + ); +} + async fn complete_job( state: &AppState, job_id: &str, @@ -287,7 +326,7 @@ ); } else { stream::iter(collections.iter()) .for_each_concurrent(collections.len(), |collection| async move { - if is_cancelled(state, job_id).await { + if should_stop_worker(state, job_id).await { return; } if let Err(e) = discover_repos_from_relay(state, job_id, collection).await { @@ -394,7 +433,7 @@ } update_job_counter(state, job_id, "total_repos", running_total).await; - if is_cancelled(state, job_id).await { + if should_stop_worker(state, job_id).await { return Ok(()); } @@ -492,7 +531,7 @@ .unwrap_or_default(); let mut attempted: i32 = 0; let mut next_flush = random_batch_threshold(100); - let mut next_cancel_check = random_batch_threshold(100); + let mut next_cancel_check = random_batch_threshold(10); let stream_state = resolver_state.clone(); let stream_cancelled = Arc::clone(&resolver_cancelled); @@ -572,11 +611,11 @@ } attempted += 1; if attempted >= next_cancel_check { - if is_cancelled(&resolver_state, &resolver_job_id).await { + if should_stop_worker(&resolver_state, &resolver_job_id).await { resolver_cancelled.store(true, Ordering::Relaxed); break; } - next_cancel_check = attempted + random_batch_threshold(100); + next_cancel_check = attempted + random_batch_threshold(10); } } @@ -633,7 +672,31 @@ let mut pds_workers: HashMap> = HashMap::new(); let mut worker_handles = FuturesUnordered::new(); let mut overflow: Vec<(String, String)> = Vec::new(); - while let Some((did, pds_endpoint)) = rx.recv().await { + loop { + if cancelled.load(Ordering::Relaxed) { + break; + } + + let poll_state = Arc::clone(&state); + let poll_job_id = Arc::clone(&job_id_arc); + let poll_cancelled = Arc::clone(&cancelled); + let pair = tokio::select! { + biased; + result = rx.recv() => result, + _ = async move { + loop { + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + if poll_cancelled.load(Ordering::Relaxed) || should_stop_worker(&poll_state, &poll_job_id).await { + poll_cancelled.store(true, Ordering::Relaxed); + return; + } + } + } => None, + }; + let Some((did, pds_endpoint)) = pair else { + break; + }; + // Also drain any overflow from previous iterations overflow.push((did, pds_endpoint)); @@ -831,12 +894,11 @@ .bind(records) .bind(job_id.as_str()) .execute(&state.backfill_db) .await; - - if is_cancelled(&state, job_id.as_str()).await { - cancelled.store(true, Ordering::Relaxed); - break; - } next_flush = repos + random_batch_threshold(10); + } + if cancelled.load(Ordering::Relaxed) || should_stop_worker(&state, job_id.as_str()).await { + cancelled.store(true, Ordering::Relaxed); + break; } publish_event(&state, super::types::BackfillEvent::JobCounters { job_id: job_id.to_string(), @@ -853,15 +915,20 @@ Some(did) if !cancelled.load(Ordering::Relaxed) => { let state = Arc::clone(&state); let collections = collections.clone(); let pds_endpoint = pds_endpoint.clone(); + let cancelled = Arc::clone(&cancelled); fetches.push(async move { let mut count: i32 = 0; for collection in collections.iter() { + if cancelled.load(Ordering::Relaxed) { + break; + } match fetch_records_from_pds( &state, &pds_endpoint, &did, collection, + &cancelled, ) .await { @@ -1019,11 +1086,15 @@ } let mut did_records: i32 = 0; for collection in collections.iter() { + if cancelled.load(Ordering::Relaxed) { + break; + } match fetch_records_from_pds( &state, &pds_endpoint, &did, collection, + &cancelled, ) .await { @@ -1075,7 +1146,7 @@ .bind(job_id.as_str()) .execute(&state.backfill_db) .await; - if is_cancelled(&state, job_id.as_str()).await { + if should_stop_worker(&state, job_id.as_str()).await { cancelled.store(true, Ordering::Relaxed); } } @@ -1208,12 +1279,17 @@ state: &AppState, pds_endpoint: &str, did: &str, collection: &str, + cancelled: &AtomicBool, ) -> Result { let base = pds_endpoint.trim_end_matches('/'); let mut cursor: Option = None; let mut count: u32 = 0; loop { + if cancelled.load(Ordering::Relaxed) { + break; + } + let mut url = format!( "{base}/xrpc/com.atproto.repo.listRecords?repo={did}&collection={collection}&limit=100" ); @@ -1365,10 +1441,18 @@ // 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 is_cancelled(&state, &job_id).await { - tracing::info!(job_id, "backfill job cancelled"); - finalise_cancel(&state, &job_id).await; - return; + match should_stop(&state, &job_id).await { + Some("cancelling") => { + tracing::info!(job_id, "backfill job cancelled"); + finalise_cancel(&state, &job_id).await; + return; + } + Some("pausing") => { + tracing::info!(job_id, "backfill job paused"); + finalise_pause(&state, &job_id).await; + return; + } + _ => {} } let total = count_repos(&state, &job_id).await; @@ -1405,10 +1489,18 @@ // stage == "fetching_records": resolution already done (legacy or resumed) run_fetching_phase(&state, &job_id, &collections, &concurrency).await }; - if is_cancelled(&state, &job_id).await { - tracing::info!(job_id, "backfill job cancelled"); - finalise_cancel(&state, &job_id).await; - return; + match should_stop(&state, &job_id).await { + Some("cancelling") => { + tracing::info!(job_id, "backfill job cancelled"); + finalise_cancel(&state, &job_id).await; + return; + } + Some("pausing") => { + tracing::info!(job_id, "backfill job paused"); + finalise_pause(&state, &job_id).await; + return; + } + _ => {} } complete_job(&state, &job_id, final_processed, final_records, None).await; @@ -1515,6 +1607,24 @@ 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((ref status,)) if status == "paused" => { + finalise_cancel(&state, &job_id).await; + log_event( + &state.db, + EventLog { + event_type: "backfill.cancelled".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": "cancelled" }), + )) + } Some((status,)) if status != "running" => Err(AppError::BadRequest(format!( "job is not running (status: {status})" ))), @@ -1534,6 +1644,111 @@ ) .await; Ok(Json( serde_json::json!({ "id": job_id, "status": "cancelling" }), + )) + } + } +} + +/// POST /admin/backfill/{id}/pause — pause a running backfill job. +pub(super) async fn pause_backfill( + State(state): State, + admin: UserAuth, + Path(job_id): Path, +) -> Result, AppError> { + admin.require(Permission::BackfillCreate).await?; + + let sql = adapt_sql( + "SELECT status FROM backfill_jobs WHERE id = ?", + state.db_backend, + ); + let row: Option<(String,)> = sqlx::query_as(&sql) + .bind(&job_id) + .fetch_optional(&state.backfill_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 == "pausing" || status == "paused" => { + 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_pause(&state, &job_id).await; + log_event( + &state.db, + EventLog { + event_type: "backfill.pausing".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": "pausing" }), + )) + } + } +} + +/// POST /admin/backfill/{id}/resume — resume a paused backfill job. +pub(super) async fn resume_backfill( + State(state): State, + admin: UserAuth, + Path(job_id): Path, +) -> Result, AppError> { + admin.require(Permission::BackfillCreate).await?; + + let sql = adapt_sql( + "SELECT status FROM backfill_jobs WHERE id = ?", + state.db_backend, + ); + let row: Option<(String,)> = sqlx::query_as(&sql) + .bind(&job_id) + .fetch_optional(&state.backfill_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((status,)) if status != "paused" => Err(AppError::BadRequest(format!( + "job is not paused (status: {status})" + ))), + Some(_) => { + let sql = adapt_sql( + "UPDATE backfill_jobs SET status = 'running' WHERE id = ?", + state.db_backend, + ); + let _ = sqlx::query(&sql) + .bind(&job_id) + .execute(&state.backfill_db) + .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; + }); + + log_event( + &state.db, + EventLog { + event_type: "backfill.resumed".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": "running" }), )) } } @@ -1865,7 +2080,7 @@ /// 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')", + "SELECT id, status FROM backfill_jobs WHERE status IN ('running', 'cancelling', 'pausing')", state.db_backend, ); let rows: Vec<(String, String)> = sqlx::query_as(&sql) @@ -1874,18 +2089,25 @@ .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; - }); + match status.as_str() { + "cancelling" => { + tracing::info!( + job_id, + "finalising cancelled backfill job from previous run" + ); + finalise_cancel(state, &job_id).await; + } + "pausing" => { + tracing::info!(job_id, "finalising paused backfill job from previous run"); + finalise_pause(state, &job_id).await; + } + _ => { + 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 @@ -42,6 +42,8 @@ "/backfill/details", delete(backfill::flush_all_backfill_details), ) .route("/backfill/{id}/cancel", post(backfill::cancel_backfill)) + .route("/backfill/{id}/pause", post(backfill::pause_backfill)) + .route("/backfill/{id}/resume", post(backfill::resume_backfill)) .route("/backfill/{id}/events", get(backfill::backfill_events)) .route("/backfill/{id}/repos", get(backfill::backfill_repos)) .route( diff --git a/src/admin/settings.rs b/src/admin/settings.rs --- a/src/admin/settings.rs +++ b/src/admin/settings.rs @@ -203,6 +203,29 @@ let main_pool_size = state.db.options().get_max_connections() as i64; let backfill_pool_size = state.backfill_db.options().get_max_connections() as i64; + let pds: i64 = get_setting(&state.db, "backfill_concurrent_pds", state.db_backend) + .await + .and_then(|v| v.parse().ok()) + .unwrap_or(10); + let dids: i64 = get_setting( + &state.db, + "backfill_concurrent_dids_per_pds", + state.db_backend, + ) + .await + .and_then(|v| v.parse().ok()) + .unwrap_or(3); + let resolution: i64 = get_setting( + &state.db, + "backfill_concurrent_resolution", + state.db_backend, + ) + .await + .and_then(|v| v.parse().ok()) + .unwrap_or(100); + let needed_backfill_pool = (pds * dids) + resolution + 4; + let restart_recommended = needed_backfill_pool > backfill_pool_size; + Ok(Json(serde_json::json!({ "backend": match state.db_backend { DatabaseBackend::Sqlite => "sqlite", @@ -211,6 +234,7 @@ }, "server_max_connections": server_max, "main_pool_size": main_pool_size, "backfill_pool_size": backfill_pool_size, + "restart_recommended": restart_recommended, }))) } diff --git a/web/package-lock.json b/web/package-lock.json --- a/web/package-lock.json +++ b/web/package-lock.json @@ -116,7 +116,6 @@ "resolved": "https://registry.npmjs.org/@babel/core/-/core-7.29.0.tgz", "integrity": "sha512-CGOfOJqWjg2qW/Mb6zNsDm+u5vFQ8DxXfbM09z69p5Z6+mE1ikP2jUXw+j42Pf1XTYED2Rni5f95npYeuwMDQA==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@babel/code-frame": "^7.29.0", "@babel/generator": "^7.29.0", @@ -774,7 +773,6 @@ "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz", "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==", "dev": true, "license": "MIT", - "peer": true, "engines": { "node": ">=12" }, @@ -2045,7 +2043,6 @@ "resolved": "https://registry.npmjs.org/@noble/ciphers/-/ciphers-1.3.0.tgz", "integrity": "sha512-2I0gnIVPtfnMw9ee9h1dJG7tp81+8Ob3OJb3Mv37rx5L40/b0i7djjCVvGOVqc9AEIQyvyu1i6ypKdFw8R8gQw==", "dev": true, "license": "MIT", - "peer": true, "engines": { "node": "^14.21.3 || >=16" }, @@ -4343,7 +4340,6 @@ "resolved": "https://registry.npmjs.org/@types/node/-/node-24.10.13.tgz", "integrity": "sha512-oH72nZRfDv9lADUBSo104Aq7gPHpQZc4BTx38r9xf9pg5LfP6EzSyH2n7qFmmxRQXh7YlUXODcYsg6PuTDSxGg==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "undici-types": "~7.16.0" } @@ -4353,7 +4349,6 @@ "version": "19.2.14", "resolved": "https://registry.npmjs.org/@types/react/-/react-19.2.14.tgz", "integrity": "sha512-ilcTH/UniCkMdtexkoCN0bI7pMcJDvmQFPvuPvmEaYA/NSfFTAgdUSLAoVjaRJm7+6PvcM+q1zYOwS4wTYMF9w==", "license": "MIT", - "peer": true, "dependencies": { "csstype": "^3.2.2" } @@ -4364,7 +4359,6 @@ "resolved": "https://registry.npmjs.org/@types/react-dom/-/react-dom-19.2.3.tgz", "integrity": "sha512-jp2L/eY6fn+KgVVQAOqYItbF0VY/YApe5Mz2F0aykSO8gx31bYCZyvSeYxCHKvzHG5eZjc+zyaS5BrBWya2+kQ==", "devOptional": true, "license": "MIT", - "peer": true, "peerDependencies": { "@types/react": "^19.2.0" } @@ -4388,7 +4382,8 @@ "version": "2.0.7", "resolved": "https://registry.npmjs.org/@types/trusted-types/-/trusted-types-2.0.7.tgz", "integrity": "sha512-ScaPdn1dQczgbl0QFTeTOmVHFULt394XJgOQNoyVhZ6r2vLnMLJfBPd53SB52T/3G36VI1/g2MZaX0cwDuXsfw==", "license": "MIT", - "optional": true + "optional": true, + "peer": true }, "node_modules/@types/unist": { "version": "3.0.3", @@ -4454,7 +4449,6 @@ "resolved": "https://registry.npmjs.org/@typescript-eslint/parser/-/parser-8.56.0.tgz", "integrity": "sha512-IgSWvLobTDOjnaxAfDTIHaECbkNlAlKv2j5SjpB2v7QHKv1FIfjwMy8FsDbVfDX/KjmCmYICcw7uGaXLhtsLNg==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@typescript-eslint/scope-manager": "8.56.0", "@typescript-eslint/types": "8.56.0", @@ -4968,7 +4962,6 @@ "resolved": "https://registry.npmjs.org/acorn/-/acorn-8.16.0.tgz", "integrity": "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw==", "dev": true, "license": "MIT", - "peer": true, "bin": { "acorn": "bin/acorn" }, @@ -5355,7 +5348,6 @@ "resolved": "https://registry.npmjs.org/babel-plugin-react-compiler/-/babel-plugin-react-compiler-1.0.0.tgz", "integrity": "sha512-Ixm8tFfoKKIPYdCCKYTsqv+Fd4IJ0DQqMyEimo+pxUOMUR9cVPlwTrFt9Avu+3cb6Zp3mAzl+t1MrG2fxxKsxw==", "devOptional": true, "license": "MIT", - "peer": true, "dependencies": { "@babel/types": "^7.26.0" } @@ -5460,7 +5452,6 @@ "url": "https://github.com/sponsors/ai" } ], "license": "MIT", - "peer": true, "dependencies": { "baseline-browser-mapping": "^2.9.0", "caniuse-lite": "^1.0.30001759", @@ -6422,6 +6413,7 @@ "version": "3.2.7", "resolved": "https://registry.npmjs.org/dompurify/-/dompurify-3.2.7.tgz", "integrity": "sha512-WhL/YuveyGXJaerVlMYGWhvQswa7myDG17P7Vu65EWC05o8vfeNbvNf4d/BOvH99+ZW+LlQsc1GDKMa1vNK6dw==", "license": "(MPL-2.0 OR Apache-2.0)", + "peer": true, "optionalDependencies": { "@types/trusted-types": "^2.0.7" } @@ -6760,7 +6752,6 @@ "resolved": "https://registry.npmjs.org/eslint/-/eslint-9.39.2.tgz", "integrity": "sha512-LEyamqS7W5HB3ujJyvi0HQK/dtVINZvd5mAAp9eT5S/ujByGjiZLCzPcHVzuXbpJDJF/cxwHlfceVUDZ2lnSTw==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@eslint-community/eslint-utils": "^4.8.0", "@eslint-community/regexpp": "^4.12.1", @@ -6901,7 +6892,6 @@ "resolved": "https://registry.npmjs.org/eslint-plugin-import/-/eslint-plugin-import-2.32.0.tgz", "integrity": "sha512-whOE1HFo/qJDyX4SnXzP4N6zOWn79WhnCUY/iDR0mPfQZO8wcYE4JClzI2oZrhBnnMUCBCHZhO6VQyoBU95mZA==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "@rtsao/scc": "^1.1.0", "array-includes": "^3.1.9", @@ -7380,7 +7370,6 @@ "resolved": "https://registry.npmjs.org/express/-/express-5.2.1.tgz", "integrity": "sha512-hIS4idWWai69NezIdRt2xFVofaF4j+6INOpJlVOLDO8zXGpUVEVzIYk12UUi2JzjEzWL3IOAxcTubgz9Po0yXw==", "dev": true, "license": "MIT", - "peer": true, "dependencies": { "accepts": "^2.0.0", "body-parser": "^2.2.1", @@ -8179,7 +8168,6 @@ "resolved": "https://registry.npmjs.org/hono/-/hono-4.11.9.tgz", "integrity": "sha512-Eaw2YTGM6WOxA6CXbckaEvslr2Ne4NFsKrvc0v97JD5awbmeBLO5w9Ho9L9kmKonrwF9RJlW6BxT1PVv/agBHQ==", "dev": true, "license": "MIT", - "peer": true, "engines": { "node": ">=16.9.0" } @@ -9605,6 +9593,7 @@ "version": "14.0.0", "resolved": "https://registry.npmjs.org/marked/-/marked-14.0.0.tgz", "integrity": "sha512-uIj4+faQ+MgHgwUW1l2PsPglZLOLOT1uErt06dAPtx2kjteLAkbsd/0FiYg/MGS+i7ZKLb7w2WClxHkzOOuryQ==", "license": "MIT", + "peer": true, "bin": { "marked": "bin/marked.js" }, @@ -11761,7 +11750,6 @@ "version": "19.2.4", "resolved": "https://registry.npmjs.org/react/-/react-19.2.4.tgz", "integrity": "sha512-9nfp2hYpCwOjAN+8TZFGhtWEwgvWHXqESH8qT89AT/lWklpLON22Lc8pEtnpsZz7VmawabSU0gCjnj8aC0euHQ==", "license": "MIT", - "peer": true, "engines": { "node": ">=0.10.0" } @@ -11792,7 +11780,6 @@ "version": "19.2.4", "resolved": "https://registry.npmjs.org/react-dom/-/react-dom-19.2.4.tgz", "integrity": "sha512-AXJdLo8kgMbimY95O2aKQqsz2iWi9jMgKJhRBAxECE4IFxfcazB2LmzloIoibJI3C12IlY20+KFaLv+71bUJeQ==", "license": "MIT", - "peer": true, "dependencies": { "scheduler": "^0.27.0" }, @@ -11839,7 +11826,6 @@ "version": "9.2.0", "resolved": "https://registry.npmjs.org/react-redux/-/react-redux-9.2.0.tgz", "integrity": "sha512-ROY9fvHhwOD9ySfrF0wmvu//bKCQ6AeZZq1nJNtbDC+kk5DuSuNX/n6YWYF/SYy7bSba4D4FSz8DJeKY/S/r+g==", "license": "MIT", - "peer": true, "dependencies": { "@types/use-sync-external-store": "^0.0.6", "use-sync-external-store": "^1.4.0" @@ -11978,8 +11964,7 @@ "node_modules/redux": { "version": "5.0.1", "resolved": "https://registry.npmjs.org/redux/-/redux-5.0.1.tgz", "integrity": "sha512-M9/ELqF6fy8FwmkpnF0S3YKOqMyoWJ4+CS5Efg2ct3oY9daQvd/Pc71FpGZsVsbl3Cpb+IIcjBDUnnyBdQbq4w==", - "license": "MIT", - "peer": true + "license": "MIT" }, "node_modules/redux-thunk": { "version": "3.1.0", @@ -13174,8 +13159,7 @@ "node_modules/tailwindcss": { "version": "4.2.0", "resolved": "https://registry.npmjs.org/tailwindcss/-/tailwindcss-4.2.0.tgz", "integrity": "sha512-yYzTZ4++b7fNYxFfpnberEEKu43w44aqDMNM9MHMmcKuCH7lL8jJ4yJ7LGHv7rSwiqM0nkiobF9I6cLlpS2P7Q==", - "license": "MIT", - "peer": true + "license": "MIT" }, "node_modules/tapable": { "version": "2.3.0", @@ -13248,7 +13232,6 @@ "resolved": "https://registry.npmjs.org/picomatch/-/picomatch-4.0.3.tgz", "integrity": "sha512-5gTmgEY/sqK6gFXLIsQNH19lWb4ebPDLA4SdLP7dsWkIXHWlG66oPuVvXSGFPppYZz8ZDZq0dYYrbHfBCVUb1Q==", "dev": true, "license": "MIT", - "peer": true, "engines": { "node": ">=12" }, @@ -13526,7 +13509,6 @@ "resolved": "https://registry.npmjs.org/typescript/-/typescript-5.9.3.tgz", "integrity": "sha512-jl1vZzPDinLr9eUt3J/t7V6FgNEw9QjvBPdysz9KfQDD41fQrC2Y4vKQdiaUpFT4bXlb1RHhLpp8wtm6M5TgSw==", "dev": true, "license": "Apache-2.0", - "peer": true, "bin": { "tsc": "bin/tsc", "tsserver": "bin/tsserver" @@ -14276,7 +14258,6 @@ "version": "4.3.6", "resolved": "https://registry.npmjs.org/zod/-/zod-4.3.6.tgz", "integrity": "sha512-rftlrkhHZOcjDwkGlnUtZZkvaPHCsDATp4pGpuOOMDaTdDDXF91wuVDJoWoPsKX/3YPQ5fHuF3STjcYyKr+Qhg==", "license": "MIT", - "peer": true, "funding": { "url": "https://github.com/sponsors/colinhacks" } 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 @@ -5,6 +5,8 @@ import { useCurrentUser } from "@/hooks/use-current-user"; import { cancelBackfillJob, + pauseBackfillJob, + resumeBackfillJob, createBackfillJob, getBackfillJobs, getBackfillRepos, @@ -20,7 +22,7 @@ PdsSummaryEntry, BackfillEvent, BlueskyProfile, } from "@/types/backfill"; -import { CheckCircle2, ChevronRight, Circle, Loader2 } from "lucide-react"; +import { CheckCircle2, ChevronRight, Circle, Loader2, PauseCircle } from "lucide-react"; import { SiteHeader } from "@/components/site-header"; import { AlertDialog, @@ -102,6 +104,18 @@ case "cancelling": return ( cancelling + + ); + case "pausing": + return ( + + pausing + + ); + case "paused": + return ( + + paused ); case "running": @@ -507,6 +521,14 @@ onCancel={async () => { await cancelBackfillJob(selectedJob.id); load(); }} + onPause={async () => { + await pauseBackfillJob(selectedJob.id); + load(); + }} + onResume={async () => { + await resumeBackfillJob(selectedJob.id); + load(); + }} /> )} @@ -521,16 +543,23 @@ job, canCancel, canFlush, onCancel, + onPause, + onResume, }: { job: BackfillJob; canCancel: boolean; canFlush: boolean; onCancel: () => Promise; + onPause: () => Promise; + onResume: () => Promise; }) { const [cancelling, setCancelling] = useState(false); + const [pausing, setPausing] = useState(false); + const [resuming, setResuming] = useState(false); const current = phaseIndex(job.stage); const allDone = job.status === "completed"; - const isActive = job.status === "running" || job.status === "cancelling"; + const isActive = job.status === "running" || job.status === "cancelling" || job.status === "pausing"; + const isPaused = job.status === "paused" || job.status === "pausing"; // Detail data state const [discoveredRepos, setDiscoveredRepos] = useState([]); @@ -556,6 +585,23 @@ } return current >= phaseIndex(phase); } + function isPhasePaused(phase: (typeof PROGRESS_PHASES)[number]): boolean { + if (!isPaused) return false; + if (job.stage === "resolving_and_fetching") { + return phase === "resolving_pds" || phase === "fetching_records"; + } + if (job.stage === "discovering_repos") { + return phase === "discovering_repos"; + } + if (job.stage === "resolving_pds") { + return phase === "resolving_pds"; + } + if (job.stage === "fetching_records") { + return phase === "fetching_records"; + } + return false; + } + async function handleCancel() { setCancelling(true); try { @@ -565,6 +611,24 @@ setCancelling(false); } } + async function handlePause() { + setPausing(true); + try { + await onPause(); + } finally { + setPausing(false); + } + } + + async function handleResume() { + setResuming(true); + try { + await onResume(); + } finally { + setResuming(false); + } + } + // Auto-load detail data when phases are reached const discoveredReached = hasReached("discovering_repos"); const pdsReached = hasReached("resolving_pds"); @@ -727,6 +791,7 @@ : undefined} suffix="repos found" loading={discoveredReached && !discoveredLoaded} @@ -756,6 +821,7 @@ (job.stage === "resolving_pds" || job.stage === "resolving_and_fetching") } reached={hasReached("resolving_pds")} + paused={isPhasePaused("resolving_pds")} value={ hasReached("resolving_pds") ? <> / @@ -790,6 +856,7 @@ (job.stage === "fetching_records" || job.stage === "resolving_and_fetching") } reached={hasReached("fetching_records")} + paused={isPhasePaused("fetching_records")} value={ hasReached("fetching_records") || job.stage === "resolving_and_fetching" @@ -855,6 +922,26 @@ )} + {canCancel && (job.status === "running" || job.status === "pausing") && ( + + )} + {canCancel && job.status === "paused" && ( + + )} {canCancel && isActive && ( )} + {canCancel && job.status === "paused" && ( + + )} ); @@ -874,6 +971,7 @@ function ProgressRow({ label, active, reached, + paused, value, suffix, loading, @@ -882,13 +980,14 @@ }: { label: string; active: boolean; reached: boolean; + paused?: boolean; value?: React.ReactNode; suffix?: React.ReactNode; loading?: boolean; children?: React.ReactNode; }) { const [open, setOpen] = useState(false); - const done = reached && !active; + const done = reached && !active && !paused; const expandable = reached; return ( @@ -905,6 +1004,8 @@ > {active ? ( + ) : paused ? ( + ) : done ? ( ) : ( diff --git a/web/src/app/dashboard/settings/general/page.tsx b/web/src/app/dashboard/settings/general/page.tsx --- a/web/src/app/dashboard/settings/general/page.tsx +++ b/web/src/app/dashboard/settings/general/page.tsx @@ -1,9 +1,9 @@ -"use client" +"use client"; -import { useCallback, useEffect, useMemo, useRef, useState } from "react" -import { Upload, Trash2 } from "lucide-react" +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { Upload, Trash2 } from "lucide-react"; -import { useCurrentUser } from "@/hooks/use-current-user" +import { useCurrentUser } from "@/hooks/use-current-user"; import { getSettings, getDbInfo, @@ -13,11 +13,11 @@ uploadLogo, deleteLogo, type SettingEntry, type DbInfo, -} from "@/lib/api" -import { SiteHeader } from "@/components/site-header" -import { Button } from "@/components/ui/button" -import { Input } from "@/components/ui/input" -import { Label } from "@/components/ui/label" +} from "@/lib/api"; +import { SiteHeader } from "@/components/site-header"; +import { Button } from "@/components/ui/button"; +import { Input } from "@/components/ui/input"; +import { Label } from "@/components/ui/label"; const SETTING_KEYS = [ "app_name", @@ -29,16 +29,16 @@ "client_uri", "logo_uri", "tos_uri", "policy_uri", -] as const +] as const; -type FieldKey = (typeof SETTING_KEYS)[number] +type FieldKey = (typeof SETTING_KEYS)[number]; type FieldConfig = { - key: FieldKey - label: string - placeholder: string - description: string -} + key: FieldKey; + label: string; + placeholder: string; + description: string; +}; const FIELDS: FieldConfig[] = [ { @@ -74,11 +74,11 @@ label: "Privacy Policy URI", placeholder: "https://example.com/privacy", description: "Link to your privacy policy. Optional.", }, -] +]; export default function GeneralSettingsPage() { - const { hasPermission } = useCurrentUser() - const canManage = hasPermission("settings:manage") + const { hasPermission } = useCurrentUser(); + const canManage = hasPermission("settings:manage"); const [values, setValues] = useState>({ app_name: "", @@ -90,8 +90,10 @@ client_uri: "", logo_uri: "", tos_uri: "", policy_uri: "", - }) - const [sources, setSources] = useState>({ + }); + const [sources, setSources] = useState< + Record + >({ app_name: "unset", backfill_concurrent_dids_per_pds: "unset", backfill_concurrent_pds: "unset", @@ -101,34 +103,46 @@ client_uri: "unset", logo_uri: "unset", tos_uri: "unset", policy_uri: "unset", - }) - const [logoUploaded, setLogoUploaded] = useState(false) - const [dbInfo, setDbInfo] = useState(null) - const [error, setError] = useState(null) - const [saving, setSaving] = useState(false) - const [notice, setNotice] = useState(null) - const fileInputRef = useRef(null) + }); + const [logoUploaded, setLogoUploaded] = useState(false); + const [dbInfo, setDbInfo] = useState(null); + const [error, setError] = useState(null); + const [saving, setSaving] = useState(false); + const [notice, setNotice] = useState(null); + const fileInputRef = useRef(null); const load = useCallback(async () => { try { - const entries = await getSettings() - const byKey = new Map(entries.map((e) => [e.key, e])) - const val = (key: string, fallback: string) => byKey.get(key)?.value ?? fallback - const src = (key: string) => (byKey.get(key)?.source as "database" | "env" | undefined) ?? "unset" + const entries = await getSettings(); + const byKey = new Map( + entries.map((e) => [e.key, e]), + ); + const val = (key: string, fallback: string) => + byKey.get(key)?.value ?? fallback; + const src = (key: string) => + (byKey.get(key)?.source as "database" | "env" | undefined) ?? "unset"; setValues({ app_name: val("app_name", ""), - backfill_concurrent_dids_per_pds: val("backfill_concurrent_dids_per_pds", "3"), + backfill_concurrent_dids_per_pds: val( + "backfill_concurrent_dids_per_pds", + "3", + ), backfill_concurrent_pds: val("backfill_concurrent_pds", "10"), - backfill_concurrent_resolution: val("backfill_concurrent_resolution", "100"), + backfill_concurrent_resolution: val( + "backfill_concurrent_resolution", + "100", + ), backfill_retention_days: val("backfill_retention_days", "28"), client_uri: val("client_uri", ""), logo_uri: val("logo_uri", ""), tos_uri: val("tos_uri", ""), policy_uri: val("policy_uri", ""), - }) + }); setSources({ app_name: src("app_name"), - backfill_concurrent_dids_per_pds: src("backfill_concurrent_dids_per_pds"), + backfill_concurrent_dids_per_pds: src( + "backfill_concurrent_dids_per_pds", + ), backfill_concurrent_pds: src("backfill_concurrent_pds"), backfill_concurrent_resolution: src("backfill_concurrent_resolution"), backfill_retention_days: src("backfill_retention_days"), @@ -136,35 +150,35 @@ client_uri: src("client_uri"), logo_uri: src("logo_uri"), tos_uri: src("tos_uri"), policy_uri: src("policy_uri"), - }) - setLogoUploaded(byKey.has("logo_data")) + }); + setLogoUploaded(byKey.has("logo_data")); try { - setDbInfo(await getDbInfo()) + setDbInfo(await getDbInfo()); } catch { // non-critical } } catch (e: unknown) { - setError(e instanceof Error ? e.message : String(e)) + setError(e instanceof Error ? e.message : String(e)); } - }, []) + }, []); useEffect(() => { - load() - }, [load]) + load(); + }, [load]); async function handleSave() { - setError(null) - setNotice(null) - setSaving(true) + setError(null); + setNotice(null); + setSaving(true); try { for (const field of FIELDS) { - const value = values[field.key] + const value = values[field.key]; if (value === "") { if (sources[field.key] === "database") { - await deleteSetting(field.key) + await deleteSetting(field.key); } } else { - await upsertSetting(field.key, value) + await upsertSetting(field.key, value); } } const backfillKeys = [ @@ -172,77 +186,79 @@ "backfill_concurrent_dids_per_pds", "backfill_concurrent_pds", "backfill_concurrent_resolution", "backfill_retention_days", - ] as const + ] as const; for (const key of backfillKeys) { - const value = values[key] + const value = values[key]; if (value === "") { if (sources[key] === "database") { - await deleteSetting(key) + await deleteSetting(key); } } else { - await upsertSetting(key, value) + await upsertSetting(key, value); } } - setNotice("Settings saved.") - await load() + setNotice("Settings saved."); + await load(); } catch (e: unknown) { - setError(e instanceof Error ? e.message : String(e)) + setError(e instanceof Error ? e.message : String(e)); } finally { - setSaving(false) + setSaving(false); } } async function handleLogoUpload(e: React.ChangeEvent) { - const file = e.target.files?.[0] - if (!file) return - setError(null) + const file = e.target.files?.[0]; + if (!file) return; + setError(null); try { - await uploadLogo(file) - setNotice("Logo uploaded.") - await load() + await uploadLogo(file); + setNotice("Logo uploaded."); + await load(); } catch (err: unknown) { - setError(err instanceof Error ? err.message : String(err)) + setError(err instanceof Error ? err.message : String(err)); } finally { - if (fileInputRef.current) fileInputRef.current.value = "" + if (fileInputRef.current) fileInputRef.current.value = ""; } } async function handleLogoDelete() { - setError(null) + setError(null); try { - await deleteLogo() - setNotice("Logo removed.") - await load() + await deleteLogo(); + setNotice("Logo removed."); + await load(); } catch (err: unknown) { - setError(err instanceof Error ? err.message : String(err)) + setError(err instanceof Error ? err.message : String(err)); } } const connectionEstimate = useMemo(() => { - const pds = parseInt(values.backfill_concurrent_pds) || 10 - const dids = parseInt(values.backfill_concurrent_dids_per_pds) || 3 - const resolution = parseInt(values.backfill_concurrent_resolution) || 100 - const needed = pds * dids + resolution + 4 - const mainPool = dbInfo?.main_pool_size ?? 32 - const total = needed + mainPool - const serverMax = dbInfo?.server_max_connections ?? null - return { needed, mainPool, total, serverMax } - }, [values, dbInfo]) + const pds = parseInt(values.backfill_concurrent_pds) || 10; + const dids = parseInt(values.backfill_concurrent_dids_per_pds) || 3; + const resolution = parseInt(values.backfill_concurrent_resolution) || 100; + const needed = pds * dids + resolution + 4; + const mainPool = dbInfo?.main_pool_size ?? 32; + const total = needed + mainPool; + const serverMax = dbInfo?.server_max_connections ?? null; + return { needed, mainPool, total, serverMax }; + }, [values, dbInfo]); const connectionWarning = useMemo(() => { - if (!connectionEstimate.serverMax) return null + if (!connectionEstimate.serverMax) return null; if (connectionEstimate.total > connectionEstimate.serverMax) { - return `These settings need ~${connectionEstimate.total} connections (${connectionEstimate.needed} backfill + ${connectionEstimate.mainPool} main), but the database allows ${connectionEstimate.serverMax}. Reduce concurrency or increase the database's max_connections.` + return `These settings need ~${connectionEstimate.total} connections (${connectionEstimate.needed} backfill + ${connectionEstimate.mainPool} main), but the database allows ${connectionEstimate.serverMax}. Reduce concurrency or increase the database's max_connections.`; } - return null - }, [connectionEstimate]) + return null; + }, [connectionEstimate]); return ( <>
{error &&

{error}

} - {notice &&

{notice}

} + {notice && ( +

{notice}

+ )}

Instance Identity

@@ -320,13 +336,16 @@

Data Retention

- Configure how long HappyView retains detailed data from completed backfill jobs. + Configure how long HappyView retains detailed data from completed + backfill jobs.

- + {sources["backfill_retention_days"] === "env" && ( from env var @@ -340,25 +359,31 @@ min={0} step={1} value={values["backfill_retention_days"]} onChange={(e) => - setValues((v) => ({ ...v, backfill_retention_days: e.target.value })) + setValues((v) => ({ + ...v, + backfill_retention_days: e.target.value, + })) } placeholder="28" disabled={!canManage} />

- How long to keep per-repo detail data from completed backfill jobs. Set to 0 to keep indefinitely. + How long to keep per-repo detail data from completed backfill jobs. + Set to 0 to keep indefinitely.

Backfill Performance

- Tune concurrency limits for backfill jobs. Changes apply to the next job started. - The backfill connection pool is auto-sized on startup based on these values. + Tune concurrency limits for backfill jobs. Changes only apply to new + or resumed jobs.

{dbInfo?.server_max_connections && (

- Database limit: {dbInfo.server_max_connections} connections · Main pool: {connectionEstimate.mainPool} · Backfill estimate: {connectionEstimate.needed} + Database limit: {dbInfo.server_max_connections} connections · Main + pool: {connectionEstimate.mainPool} · Backfill estimate:{" "} + {connectionEstimate.needed}

)} {connectionWarning && ( @@ -366,16 +391,39 @@

{connectionWarning}

)}
- {([ - { key: "backfill_concurrent_resolution" as const, id: "backfill_concurrent_resolution", label: "Concurrent PLC Resolutions", placeholder: "100", description: "How many DID document lookups to run in parallel during PDS resolution." }, - { key: "backfill_concurrent_pds" as const, id: "backfill_concurrent_pds", label: "Concurrent PDS Hosts", placeholder: "10", description: "How many PDS servers to fetch records from simultaneously." }, - { key: "backfill_concurrent_dids_per_pds" as const, id: "backfill_concurrent_dids_per_pds", label: "Concurrent DIDs per PDS", placeholder: "3", description: "How many repos to fetch concurrently from each PDS host." }, - ]).map((field) => ( + {[ + { + key: "backfill_concurrent_resolution" as const, + id: "backfill_concurrent_resolution", + label: "Concurrent PLC Resolutions", + placeholder: "100", + description: + "How many DID document lookups to run in parallel during PDS resolution.", + }, + { + key: "backfill_concurrent_pds" as const, + id: "backfill_concurrent_pds", + label: "Concurrent PDS Hosts", + placeholder: "10", + description: + "How many PDS servers to fetch records from simultaneously.", + }, + { + key: "backfill_concurrent_dids_per_pds" as const, + id: "backfill_concurrent_dids_per_pds", + label: "Concurrent DIDs per PDS", + placeholder: "3", + description: + "How many repos to fetch concurrently from each PDS host.", + }, + ].map((field) => (
{sources[field.key] === "env" && ( - from env var + + from env var + )}
setValues((v) => ({ ...v, [field.key]: e.target.value }))} + onChange={(e) => + setValues((v) => ({ ...v, [field.key]: e.target.value })) + } placeholder={field.placeholder} disabled={!canManage} /> @@ -402,5 +452,5 @@
- ) + ); } 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 @@ -189,6 +189,20 @@ { method: "POST" }, ); } +export function pauseBackfillJob(id: string) { + return apiFetch<{ id: string; status: string }>( + `/admin/backfill/${id}/pause`, + { method: "POST" }, + ); +} + +export function resumeBackfillJob(id: string) { + return apiFetch<{ id: string; status: string }>( + `/admin/backfill/${id}/resume`, + { method: "POST" }, + ); +} + export function getBackfillRepos( jobId: string, params: { phase?: string; cursor?: string; limit?: number } = {}, -- tangled.sh