From c14a168d5a7d531a965997ac724dddbaea1ac7af Mon Sep 17 00:00:00 2001 From: Trezy Date: Thu, 21 May 2026 14:43:52 -0500 Subject: [PATCH] fix: separate backfill db connection pool from main app pool Signed-off-by: Trezy --- src/admin/backfill.rs | 80 +++++++++++-------- src/admin/mod.rs | 1 + src/admin/settings.rs | 32 ++++++++ src/db.rs | 57 +++++++++++++ src/lib.rs | 1 + src/lua/atproto_api.rs | 1 + src/lua/db_api.rs | 1 + src/lua/execute.rs | 1 + src/lua/http_api.rs | 1 + src/lua/xrpc_api.rs | 1 + src/main.rs | 2 + tests/common/app.rs | 1 + tests/lua_atproto_api.rs | 1 + tests/lua_db_api.rs | 1 + .../app/dashboard/settings/general/page.tsx | 40 +++++++++- web/src/lib/api.ts | 11 +++ 16 files changed, 195 insertions(+), 37 deletions(-) diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs index 62f5ec4..97b4a61 100644 --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -66,7 +66,7 @@ async fn set_stage(state: &AppState, job_id: &str, stage: &str) { let _ = sqlx::query(&sql) .bind(stage) .bind(job_id) - .execute(&state.db) + .execute(&state.backfill_db) .await; publish_event( state, @@ -95,7 +95,7 @@ async fn update_job_counter(state: &AppState, job_id: &str, column: &str, value: let _ = sqlx::query(&sql) .bind(value) .bind(job_id) - .execute(&state.db) + .execute(&state.backfill_db) .await; } @@ -106,7 +106,7 @@ async fn count_repos(state: &AppState, job_id: &str) -> i32 { ); sqlx::query_as::<_, (i32,)>(&sql) .bind(job_id) - .fetch_one(&state.db) + .fetch_one(&state.backfill_db) .await .map(|(c,)| c) .unwrap_or(0) @@ -165,7 +165,7 @@ async fn fail_job(state: &AppState, job_id: &str, error: &str) { .bind(&now) .bind(error) .bind(job_id) - .execute(&state.db) + .execute(&state.backfill_db) .await; publish_event( state, @@ -184,7 +184,7 @@ async fn is_cancelled(state: &AppState, job_id: &str) -> bool { ); sqlx::query_as::<_, (String,)>(&sql) .bind(job_id) - .fetch_optional(&state.db) + .fetch_optional(&state.backfill_db) .await .ok() .flatten() @@ -196,7 +196,10 @@ async fn request_cancel(state: &AppState, job_id: &str) { "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; + let _ = sqlx::query(&sql) + .bind(job_id) + .execute(&state.backfill_db) + .await; } async fn finalise_cancel(state: &AppState, job_id: &str) { @@ -208,7 +211,7 @@ async fn finalise_cancel(state: &AppState, job_id: &str) { let _ = sqlx::query(&sql) .bind(&now) .bind(job_id) - .execute(&state.db) + .execute(&state.backfill_db) .await; publish_event( state, @@ -238,7 +241,7 @@ async fn complete_job( .bind(total_records) .bind(error) .bind(job_id) - .execute(&state.db) + .execute(&state.backfill_db) .await; publish_event( state, @@ -270,7 +273,7 @@ async fn run_discovery_phase( let _ = sqlx::query(&sql) .bind(job_id) .bind(did) - .execute(&state.db) + .execute(&state.backfill_db) .await; publish_event( state, @@ -370,7 +373,7 @@ async fn discover_repos_from_relay( for repo in chunk { query = query.bind(job_id).bind(&repo.did); } - if let Ok(result) = query.execute(&state.db).await { + if let Ok(result) = query.execute(&state.backfill_db).await { running_total += result.rows_affected() as i32; } for repo in chunk { @@ -420,7 +423,7 @@ async fn run_pipelined_resolve_and_fetch( ); sqlx::query_as::<_, (i32,)>(&sql) .bind(job_id) - .fetch_one(&state.db) + .fetch_one(&state.backfill_db) .await .map(|(c,)| c) .unwrap_or(0) @@ -433,7 +436,7 @@ async fn run_pipelined_resolve_and_fetch( ); sqlx::query_as::<_, (i32,)>(&sql) .bind(job_id) - .fetch_one(&state.db) + .fetch_one(&state.backfill_db) .await .map(|(c,)| c) .unwrap_or(0) @@ -449,7 +452,7 @@ async fn run_pipelined_resolve_and_fetch( ); sqlx::query_as::<_, (Option,)>(&sql) .bind(job_id) - .fetch_one(&state.db) + .fetch_one(&state.backfill_db) .await .map(|(c,)| c.unwrap_or(0)) .unwrap_or(0) @@ -479,7 +482,7 @@ async fn run_pipelined_resolve_and_fetch( ); let unresolved: Vec<(String,)> = sqlx::query_as(&sql) .bind(&resolver_job_id) - .fetch_all(&resolver_state.db) + .fetch_all(&resolver_state.backfill_db) .await .unwrap_or_default(); @@ -520,7 +523,7 @@ async fn run_pipelined_resolve_and_fetch( .bind(&pds) .bind(&resolver_job_id) .bind(&did) - .execute(&resolver_state.db) + .execute(&resolver_state.backfill_db) .await; publish_event( @@ -592,7 +595,7 @@ async fn run_pipelined_resolve_and_fetch( ); let pending_rows: Vec<(String, String)> = sqlx::query_as(&pending_sql) .bind(job_id) - .fetch_all(&state.db) + .fetch_all(&state.backfill_db) .await .unwrap_or_default(); @@ -754,7 +757,7 @@ async fn run_pipelined_resolve_and_fetch( .bind(final_repos) .bind(final_records) .bind(job_id) - .execute(&state.db) + .execute(&state.backfill_db) .await; (final_repos, final_records) @@ -801,7 +804,7 @@ async fn run_pds_worker(ctx: FetchContext, pds_endpoint: String, mut rx: mpsc::R .bind(records) .bind(job_id.as_str()) .bind(&did) - .execute(&state.db) + .execute(&state.backfill_db) .await; publish_event(&state, super::types::BackfillEvent::RepoFetched { @@ -822,7 +825,7 @@ async fn run_pds_worker(ctx: FetchContext, pds_endpoint: String, mut rx: mpsc::R .bind(repos) .bind(records) .bind(job_id.as_str()) - .execute(&state.db) + .execute(&state.backfill_db) .await; if is_cancelled(&state, job_id.as_str()).await { @@ -896,7 +899,7 @@ async fn run_pds_worker(ctx: FetchContext, pds_endpoint: String, mut rx: mpsc::R .bind(records) .bind(job_id.as_str()) .bind(&did) - .execute(&state.db) + .execute(&state.backfill_db) .await; publish_event( @@ -932,7 +935,7 @@ async fn run_fetching_phase( ); let rows: Vec<(String, String)> = sqlx::query_as(&sql) .bind(job_id) - .fetch_all(&state.db) + .fetch_all(&state.backfill_db) .await .unwrap_or_default(); @@ -948,7 +951,7 @@ async fn run_fetching_phase( ); let already_completed: i32 = sqlx::query_as::<_, (i32,)>(&sql) .bind(job_id) - .fetch_one(&state.db) + .fetch_one(&state.backfill_db) .await .map(|(c,)| c) .unwrap_or(0); @@ -964,7 +967,7 @@ async fn run_fetching_phase( ); sqlx::query_as::<_, (Option,)>(&sql) .bind(job_id) - .fetch_one(&state.db) + .fetch_one(&state.backfill_db) .await .map(|(c,)| c.unwrap_or(0)) .unwrap_or(0) @@ -1046,7 +1049,7 @@ async fn run_fetching_phase( .bind(did_records) .bind(job_id.as_str()) .bind(&did) - .execute(&state.db) + .execute(&state.backfill_db) .await; let repos = processed_repos.fetch_add(1, Ordering::Relaxed) + 1; @@ -1065,7 +1068,7 @@ async fn run_fetching_phase( .bind(repos) .bind(records) .bind(job_id.as_str()) - .execute(&state.db) + .execute(&state.backfill_db) .await; if is_cancelled(&state, job_id.as_str()).await { @@ -1099,7 +1102,7 @@ async fn run_fetching_phase( .bind(final_repos) .bind(final_records) .bind(job_id) - .execute(&state.db) + .execute(&state.backfill_db) .await; (final_repos, final_records) @@ -1189,7 +1192,7 @@ async fn run_backfill_job(state: AppState, job_id: String) { ); let job: Option<(Option, Option, String)> = sqlx::query_as(&sql) .bind(&job_id) - .fetch_optional(&state.db) + .fetch_optional(&state.backfill_db) .await .ok() .flatten(); @@ -1217,7 +1220,7 @@ async fn run_backfill_job(state: AppState, job_id: String) { "SELECT id FROM lexicons WHERE json_extract(lexicon_json, '$.defs.main.type') = 'record'", backend, ); - let rows: Vec<(String,)> = match sqlx::query_as(&sql).fetch_all(&state.db).await { + let rows: Vec<(String,)> = match sqlx::query_as(&sql).fetch_all(&state.backfill_db).await { Ok(rows) => rows, Err(e) => { let error = format!("failed to query backfill-eligible lexicons: {e}"); @@ -1692,11 +1695,14 @@ pub async fn run_backfill_retention_cleanup(state: &AppState) { loop { interval.tick().await; - let retention_days: i64 = - get_setting(&state.db, "backfill_retention_days", state.db_backend) - .await - .and_then(|v| v.parse().ok()) - .unwrap_or(28); + let retention_days: i64 = get_setting( + &state.backfill_db, + "backfill_retention_days", + state.db_backend, + ) + .await + .and_then(|v| v.parse().ok()) + .unwrap_or(28); if retention_days == 0 { continue; @@ -1709,7 +1715,11 @@ pub async fn run_backfill_retention_cleanup(state: &AppState) { "DELETE FROM backfill_repos WHERE job_id IN (SELECT id FROM backfill_jobs WHERE completed_at IS NOT NULL AND completed_at < ?)", state.db_backend, ); - match sqlx::query(&sql).bind(&cutoff_str).execute(&state.db).await { + match sqlx::query(&sql) + .bind(&cutoff_str) + .execute(&state.backfill_db) + .await + { Ok(result) => { let deleted = result.rows_affected(); if deleted > 0 { @@ -1739,7 +1749,7 @@ pub async fn resume_backfill_jobs(state: &AppState) { state.db_backend, ); let rows: Vec<(String, String)> = sqlx::query_as(&sql) - .fetch_all(&state.db) + .fetch_all(&state.backfill_db) .await .unwrap_or_default(); diff --git a/src/admin/mod.rs b/src/admin/mod.rs index 5cba8b4..0bb7d4f 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -94,6 +94,7 @@ pub fn admin_routes(_state: AppState) -> Router { ) .route("/feature-flags", get(feature_flags::list)) .route("/settings", get(settings::list)) + .route("/settings/db-info", get(settings::db_info)) .route( "/settings/logo", put(settings::upload_logo).delete(settings::delete_logo), diff --git a/src/admin/settings.rs b/src/admin/settings.rs index 0993997..de001a3 100644 --- a/src/admin/settings.rs +++ b/src/admin/settings.rs @@ -182,6 +182,38 @@ pub(super) async fn delete( Ok(StatusCode::NO_CONTENT) } +/// GET /admin/settings/db-info — return database connection pool info. +pub(super) async fn db_info( + State(state): State, + auth: UserAuth, +) -> Result, AppError> { + auth.require(Permission::SettingsManage).await?; + + let server_max: Option = if state.db_backend == DatabaseBackend::Postgres { + sqlx::query_as::<_, (String,)>("SHOW max_connections") + .fetch_optional(&state.db) + .await + .ok() + .flatten() + .and_then(|(v,)| v.parse().ok()) + } else { + None + }; + + let main_pool_size = state.db.size() as i64; + let backfill_pool_size = state.backfill_db.size() as i64; + + Ok(Json(serde_json::json!({ + "backend": match state.db_backend { + DatabaseBackend::Sqlite => "sqlite", + DatabaseBackend::Postgres => "postgres", + }, + "server_max_connections": server_max, + "main_pool_size": main_pool_size, + "backfill_pool_size": backfill_pool_size, + }))) +} + /// PUT /admin/settings/logo — upload a logo image (max 5MB). pub(super) async fn upload_logo( State(state): State, diff --git a/src/db.rs b/src/db.rs index 114ced1..638fc02 100644 --- a/src/db.rs +++ b/src/db.rs @@ -283,6 +283,63 @@ pub async fn connect(url: &str, backend: DatabaseBackend) -> AnyPool { pool } +pub async fn connect_backfill_pool(url: &str, backend: DatabaseBackend) -> AnyPool { + let max_connections: u32 = std::env::var("BACKFILL_DATABASE_MAX_CONNECTIONS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or_else(|| { + let pds: u32 = std::env::var("BACKFILL_CONCURRENT_PDS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(10); + let dids: u32 = std::env::var("BACKFILL_CONCURRENT_DIDS_PER_PDS") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(3); + let resolution: u32 = std::env::var("BACKFILL_CONCURRENT_RESOLUTION") + .ok() + .and_then(|v| v.parse().ok()) + .unwrap_or(100); + // Each concurrent worker may need a connection: PDS×DIDs for fetching + + // resolution concurrency + a few for bookkeeping queries. + let needed = (pds * dids) + resolution + 4; + let ceiling = match backend { + DatabaseBackend::Sqlite => 64, + DatabaseBackend::Postgres => 256, + }; + needed.min(ceiling) + }); + + tracing::info!(max_connections, "backfill pool sized"); + + let pool = PoolOptions::::new() + .max_connections(max_connections) + .acquire_timeout(std::time::Duration::from_secs(30)) + .idle_timeout(std::time::Duration::from_secs(300)) + .connect(url) + .await + .expect("Failed to connect backfill database pool"); + + if backend == DatabaseBackend::Sqlite { + sqlx::query("PRAGMA foreign_keys = ON") + .execute(&pool) + .await + .expect("Failed to enable foreign keys on backfill pool"); + + sqlx::query("PRAGMA journal_mode = WAL") + .execute(&pool) + .await + .expect("Failed to enable WAL mode on backfill pool"); + + sqlx::query("PRAGMA busy_timeout = 5000") + .execute(&pool) + .await + .expect("Failed to set busy timeout on backfill pool"); + } + + pool +} + #[cfg(test)] mod tests { use super::*; diff --git a/src/lib.rs b/src/lib.rs index 9ea8e48..3b3a061 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -63,6 +63,7 @@ pub struct AppState { pub config: Config, pub http: reqwest::Client, pub db: sqlx::AnyPool, + pub backfill_db: sqlx::AnyPool, pub db_backend: DatabaseBackend, pub domain_cache: domain::DomainCache, pub lexicons: LexiconRegistry, diff --git a/src/lua/atproto_api.rs b/src/lua/atproto_api.rs index 560e638..40af1b0 100644 --- a/src/lua/atproto_api.rs +++ b/src/lua/atproto_api.rs @@ -547,6 +547,7 @@ mod tests { config, http: reqwest::Client::new(), db: test_db.clone(), + backfill_db: test_db.clone(), db_backend: DatabaseBackend::Sqlite, domain_cache: crate::domain::DomainCache::new(), lexicons: LexiconRegistry::new(), diff --git a/src/lua/db_api.rs b/src/lua/db_api.rs index f22725b..e8d95b8 100644 --- a/src/lua/db_api.rs +++ b/src/lua/db_api.rs @@ -755,6 +755,7 @@ mod tests { config, http: reqwest::Client::new(), db: test_db.clone(), + backfill_db: test_db.clone(), db_backend: DatabaseBackend::Sqlite, domain_cache: crate::domain::DomainCache::new(), lexicons: LexiconRegistry::new(), diff --git a/src/lua/execute.rs b/src/lua/execute.rs index 97f5e6b..bf50cf3 100644 --- a/src/lua/execute.rs +++ b/src/lua/execute.rs @@ -1145,6 +1145,7 @@ mod tests { config, http: reqwest::Client::new(), db: test_db.clone(), + backfill_db: test_db.clone(), db_backend: DatabaseBackend::Sqlite, domain_cache: crate::domain::DomainCache::new(), lexicons: LexiconRegistry::new(), diff --git a/src/lua/http_api.rs b/src/lua/http_api.rs index 2c20331..12b3716 100644 --- a/src/lua/http_api.rs +++ b/src/lua/http_api.rs @@ -153,6 +153,7 @@ mod tests { config, http: reqwest::Client::new(), db: test_db.clone(), + backfill_db: test_db.clone(), db_backend: crate::db::DatabaseBackend::Sqlite, domain_cache: crate::domain::DomainCache::new(), lexicons: LexiconRegistry::new(), diff --git a/src/lua/xrpc_api.rs b/src/lua/xrpc_api.rs index 6cb7860..82c6ed7 100644 --- a/src/lua/xrpc_api.rs +++ b/src/lua/xrpc_api.rs @@ -257,6 +257,7 @@ mod tests { config, http: reqwest::Client::new(), db: test_db.clone(), + backfill_db: test_db.clone(), db_backend: DatabaseBackend::Sqlite, domain_cache: crate::domain::DomainCache::new(), lexicons: LexiconRegistry::new(), diff --git a/src/main.rs b/src/main.rs index 595bb0a..9a3bf7c 100644 --- a/src/main.rs +++ b/src/main.rs @@ -38,6 +38,7 @@ async fn main() { // Connect to database and run migrations. let db_pool = db::connect(&config.database_url, db_backend).await; + let backfill_db_pool = db::connect_backfill_pool(&config.database_url, db_backend).await; info!( backend = ?db_backend, @@ -618,6 +619,7 @@ async fn main() { config: config.clone(), http, db: db_pool, + backfill_db: backfill_db_pool, db_backend, domain_cache: domain_cache.clone(), lexicons, diff --git a/tests/common/app.rs b/tests/common/app.rs index 2331947..bac7ff8 100644 --- a/tests/common/app.rs +++ b/tests/common/app.rs @@ -167,6 +167,7 @@ impl TestApp { proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( happyview::proxy_config::ProxyConfig::default(), ))), + backfill_db: pool.clone(), backfill_events_tx: tokio::sync::broadcast::channel(16).0, }; diff --git a/tests/lua_atproto_api.rs b/tests/lua_atproto_api.rs index 459a9b6..f0a24c1 100644 --- a/tests/lua_atproto_api.rs +++ b/tests/lua_atproto_api.rs @@ -101,6 +101,7 @@ async fn test_state_with_pool(pool: sqlx::AnyPool, backend: DatabaseBackend) -> proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( happyview::proxy_config::ProxyConfig::default(), ))), + backfill_db: pool.clone(), backfill_events_tx: tokio::sync::broadcast::channel(16).0, } } diff --git a/tests/lua_db_api.rs b/tests/lua_db_api.rs index cd6a621..bbf17f2 100644 --- a/tests/lua_db_api.rs +++ b/tests/lua_db_api.rs @@ -104,6 +104,7 @@ async fn test_state_with_pool(pool: sqlx::AnyPool, backend: DatabaseBackend) -> proxy_config: std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new( happyview::proxy_config::ProxyConfig::default(), ))), + backfill_db: pool.clone(), backfill_events_tx: tokio::sync::broadcast::channel(16).0, } } diff --git a/web/src/app/dashboard/settings/general/page.tsx b/web/src/app/dashboard/settings/general/page.tsx index 6376fe0..c7924a8 100644 --- a/web/src/app/dashboard/settings/general/page.tsx +++ b/web/src/app/dashboard/settings/general/page.tsx @@ -1,16 +1,18 @@ "use client" -import { useCallback, useEffect, useRef, useState } from "react" +import { useCallback, useEffect, useMemo, useRef, useState } from "react" import { Upload, Trash2 } from "lucide-react" import { useCurrentUser } from "@/hooks/use-current-user" import { getSettings, + getDbInfo, upsertSetting, deleteSetting, uploadLogo, deleteLogo, type SettingEntry, + type DbInfo, } from "@/lib/api" import { SiteHeader } from "@/components/site-header" import { Button } from "@/components/ui/button" @@ -101,6 +103,7 @@ export default function GeneralSettingsPage() { 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) @@ -135,6 +138,11 @@ export default function GeneralSettingsPage() { policy_uri: src("policy_uri"), }) setLogoUploaded(byKey.has("logo_data")) + try { + setDbInfo(await getDbInfo()) + } catch { + // non-critical + } } catch (e: unknown) { setError(e instanceof Error ? e.message : String(e)) } @@ -210,6 +218,25 @@ export default function GeneralSettingsPage() { } } + 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 connectionWarning = useMemo(() => { + 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 null + }, [connectionEstimate]) + return ( <> @@ -325,7 +352,16 @@ export default function GeneralSettingsPage() {

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.

+ {dbInfo?.server_max_connections && ( +

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

+ )} + {connectionWarning && ( +

{connectionWarning}

+ )} {([ @@ -355,7 +391,7 @@ export default function GeneralSettingsPage() {
diff --git a/web/src/lib/api.ts b/web/src/lib/api.ts index 4438ace..1c43284 100644 --- a/web/src/lib/api.ts +++ b/web/src/lib/api.ts @@ -365,6 +365,17 @@ export function getSettings() { return apiFetch("/admin/settings"); } +export type DbInfo = { + backend: "sqlite" | "postgres"; + server_max_connections: number | null; + main_pool_size: number; + backfill_pool_size: number; +}; + +export function getDbInfo() { + return apiFetch("/admin/settings/db-info"); +} + export function upsertSetting(key: string, value: string) { return apiFetch(`/admin/settings/${encodeURIComponent(key)}`, { method: "PUT", -- 2.51.2