From ec860a528ce455f3c743913dbbf0a1d01b1f9c95 Mon Sep 17 00:00:00 2001 From: Trezy Date: Tue, 26 May 2026 16:36:42 +0000 Subject: [PATCH] fix: reduce backfill db contention and sse event flooding for pds discovery and record fetching during backfill Signed-off-by: Trezy --- src/main.rs | 2 +- src/record_handler.rs | 2 ++ src/admin/backfill.rs | 38 ++++++++++++++++++++++++++------------ src/admin/dead_letters.rs | 1 + src/lua/db_api.rs | 195 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/lua/execute.rs | 43 ++++++++++++++++++++++++++++++++++++++++++- src/lua/mod.rs | 3 ++- web/src/app/dashboard/backfill/page.tsx | 167 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++------------------------------------------------------------------------------------------------------------- 8 file(s) changed, 327 insertion(s)(+), 124 deletion(s)(-) diff --git a/src/main.rs b/src/main.rs --- a/src/main.rs +++ b/src/main.rs @@ -616,7 +616,7 @@ std::sync::Arc::new(arc_swap::ArcSwap::new(std::sync::Arc::new(config))) }; - let (backfill_events_tx, _) = tokio::sync::broadcast::channel(1024); + let (backfill_events_tx, _) = tokio::sync::broadcast::channel(16384); let verbose_event_logging = { let enabled = diff --git a/src/record_handler.rs b/src/record_handler.rs --- a/src/record_handler.rs +++ b/src/record_handler.rs @@ -71,6 +71,7 @@ collection: &record.collection, rkey: &record.rkey, record: Some(rec), + cached_env_vars: None, }) .await; @@ -198,6 +199,7 @@ collection: &record.collection, rkey: &record.rkey, record: None, + cached_env_vars: None, }) .await; diff --git a/src/admin/backfill.rs b/src/admin/backfill.rs --- a/src/admin/backfill.rs +++ b/src/admin/backfill.rs @@ -338,6 +338,16 @@ let total = count_repos(state, job_id).await; update_job_counter(state, job_id, "total_repos", total).await; + publish_event( + state, + super::types::BackfillEvent::JobCounters { + job_id: job_id.to_string(), + total_repos: Some(total), + resolved_repos: None, + processed_repos: None, + total_records: None, + }, + ); } async fn discover_repos_from_relay( @@ -418,15 +428,6 @@ } if let Ok(result) = query.execute(&state.backfill_db).await { running_total += result.rows_affected() as i32; - } - for repo in chunk { - publish_event( - state, - super::types::BackfillEvent::RepoDiscovered { - job_id: job_id.to_string(), - did: repo.did.clone(), - }, - ); } } } @@ -1275,9 +1276,20 @@ let _ = ref_query.execute(&state.backfill_db).await; } - // Queue label backfill for each record - for rec in batch { - crate::labeler::backfill_labels_for_uri(Arc::new(state.clone()), rec.uri.clone()); + // Queue label backfill only if there are active labeler subscriptions. + // Check once per batch instead of spawning a task per record. + let has_subscriptions: bool = sqlx::query_as::<_, (i64,)>( + "SELECT COUNT(*) FROM labeler_subscriptions WHERE status = 'active'", + ) + .fetch_one(&state.db) + .await + .map(|(c,)| c > 0) + .unwrap_or(false); + + if has_subscriptions { + for rec in batch { + crate::labeler::backfill_labels_for_uri(Arc::new(state.clone()), rec.uri.clone()); + } } } @@ -1294,6 +1306,7 @@ let mut cursor: Option = None; let mut count: u32 = 0; let index_hook = state.lexicons.get_index_hook(collection).await; + let env_vars = crate::lua::load_env_vars_cached(&state.db, state.db_backend).await; loop { if cancelled.load(Ordering::Relaxed) { @@ -1348,6 +1361,7 @@ collection, rkey: &rkey, record: Some(&entry.value), + cached_env_vars: Some(&env_vars), }) .await; diff --git a/src/admin/dead_letters.rs b/src/admin/dead_letters.rs --- a/src/admin/dead_letters.rs +++ b/src/admin/dead_letters.rs @@ -527,6 +527,7 @@ collection: &dl.collection, rkey: &dl.rkey, record: record.as_ref(), + cached_env_vars: None, }; match run_hook_once(&event).await { diff --git a/src/lua/db_api.rs b/src/lua/db_api.rs --- a/src/lua/db_api.rs +++ b/src/lua/db_api.rs @@ -42,6 +42,7 @@ true } +#[derive(Debug)] enum FilterNode { Condition { field: String, @@ -944,6 +945,200 @@ let encoded = BASE64.encode("no-pipe-here"); assert!(super::decode_cursor(&encoded).is_none()); } + + // ----------------------------------------------------------------------- + // parse_filter_node / build_filter_sql + // ----------------------------------------------------------------------- + + fn make_condition_table(lua: &Lua, field: &str, op: &str, value: &str) -> mlua::Table { + let t = lua.create_table().unwrap(); + t.set("field", field).unwrap(); + t.set("op", op).unwrap(); + t.set("value", value).unwrap(); + t + } + + #[test] + fn filter_simple_condition() { + let lua = Lua::new(); + let t = make_condition_table(&lua, "name", "=", "alice"); + let node = parse_filter_node(&t, 0).unwrap(); + let mut binds = Vec::new(); + let sql = build_filter_sql(&node, &mut binds); + assert_eq!(sql, "json_extract(record, '$.value.name') = ?"); + assert_eq!(binds, vec!["alice"]); + } + + #[test] + fn filter_defaults_op_to_equals() { + let lua = Lua::new(); + let t = lua.create_table().unwrap(); + t.set("field", "status").unwrap(); + t.set("value", "active").unwrap(); + let node = parse_filter_node(&t, 0).unwrap(); + let mut binds = Vec::new(); + let sql = build_filter_sql(&node, &mut binds); + assert_eq!(sql, "json_extract(record, '$.value.status') = ?"); + } + + #[test] + fn filter_rejects_invalid_op() { + let lua = Lua::new(); + let t = make_condition_table(&lua, "name", "DROP", "x"); + let err = parse_filter_node(&t, 0).unwrap_err(); + assert!(err.to_string().contains("invalid filter op")); + } + + #[test] + fn filter_rejects_invalid_field() { + let lua = Lua::new(); + let t = make_condition_table(&lua, "name; DROP TABLE", "=", "x"); + let err = parse_filter_node(&t, 0).unwrap_err(); + assert!(err.to_string().contains("invalid filter field")); + } + + #[test] + fn filter_and_group() { + let lua = Lua::new(); + let group = lua.create_table().unwrap(); + group.set("combine", "AND").unwrap(); + let c1 = make_condition_table(&lua, "status", "=", "active"); + let c2 = make_condition_table(&lua, "age", ">", "18"); + group.set(1, c1).unwrap(); + group.set(2, c2).unwrap(); + let node = parse_filter_node(&group, 0).unwrap(); + let mut binds = Vec::new(); + let sql = build_filter_sql(&node, &mut binds); + assert_eq!( + sql, + "(json_extract(record, '$.value.status') = ? AND json_extract(record, '$.value.age') > ?)" + ); + assert_eq!(binds, vec!["active", "18"]); + } + + #[test] + fn filter_or_group() { + let lua = Lua::new(); + let group = lua.create_table().unwrap(); + group.set("combine", "OR").unwrap(); + let c1 = make_condition_table(&lua, "role", "=", "admin"); + let c2 = make_condition_table(&lua, "role", "=", "mod"); + group.set(1, c1).unwrap(); + group.set(2, c2).unwrap(); + let node = parse_filter_node(&group, 0).unwrap(); + let mut binds = Vec::new(); + let sql = build_filter_sql(&node, &mut binds); + assert_eq!( + sql, + "(json_extract(record, '$.value.role') = ? OR json_extract(record, '$.value.role') = ?)" + ); + assert_eq!(binds, vec!["admin", "mod"]); + } + + #[test] + fn filter_single_child_group_unwraps() { + let lua = Lua::new(); + let group = lua.create_table().unwrap(); + group.set("combine", "AND").unwrap(); + let c1 = make_condition_table(&lua, "x", "=", "1"); + group.set(1, c1).unwrap(); + let node = parse_filter_node(&group, 0).unwrap(); + let mut binds = Vec::new(); + let sql = build_filter_sql(&node, &mut binds); + assert_eq!(sql, "json_extract(record, '$.value.x') = ?"); + } + + #[test] + fn filter_rejects_invalid_combine() { + let lua = Lua::new(); + let group = lua.create_table().unwrap(); + group.set("combine", "XOR").unwrap(); + let c1 = make_condition_table(&lua, "x", "=", "1"); + group.set(1, c1).unwrap(); + let err = parse_filter_node(&group, 0).unwrap_err(); + assert!(err.to_string().contains("invalid filter combine")); + } + + #[test] + fn filter_rejects_empty_group() { + let lua = Lua::new(); + let group = lua.create_table().unwrap(); + group.set("combine", "AND").unwrap(); + let err = parse_filter_node(&group, 0).unwrap_err(); + assert!(err.to_string().contains("filter group has no conditions")); + } + + #[test] + fn filter_rejects_excessive_depth() { + let lua = Lua::new(); + let c = make_condition_table(&lua, "x", "=", "1"); + let err = parse_filter_node(&c, MAX_FILTER_DEPTH).unwrap_err(); + assert!(err.to_string().contains("filter nesting too deep")); + } + + #[test] + fn filter_accepts_all_ops() { + let lua = Lua::new(); + for op in ALLOWED_OPS { + let t = make_condition_table(&lua, "field", op, "val"); + assert!( + parse_filter_node(&t, 0).is_ok(), + "op '{op}' should be accepted" + ); + } + } + + #[test] + fn filter_op_case_insensitive() { + let lua = Lua::new(); + let t = make_condition_table(&lua, "name", "like", "alice%"); + let node = parse_filter_node(&t, 0).unwrap(); + let mut binds = Vec::new(); + let sql = build_filter_sql(&node, &mut binds); + assert_eq!(sql, "json_extract(record, '$.value.name') LIKE ?"); + } + + #[test] + fn filter_integer_value() { + let lua = Lua::new(); + let t = lua.create_table().unwrap(); + t.set("field", "count").unwrap(); + t.set("op", ">").unwrap(); + t.set("value", 42).unwrap(); + let node = parse_filter_node(&t, 0).unwrap(); + let mut binds = Vec::new(); + build_filter_sql(&node, &mut binds); + assert_eq!(binds, vec!["42"]); + } + + #[test] + fn filter_boolean_value() { + let lua = Lua::new(); + let t = lua.create_table().unwrap(); + t.set("field", "active").unwrap(); + t.set("value", true).unwrap(); + let node = parse_filter_node(&t, 0).unwrap(); + let mut binds = Vec::new(); + build_filter_sql(&node, &mut binds); + assert_eq!(binds, vec!["true"]); + } + + #[test] + fn filter_nested_field_path() { + let lua = Lua::new(); + let t = make_condition_table(&lua, "author.websites[0].url", "=", "https://example.com"); + let node = parse_filter_node(&t, 0).unwrap(); + let mut binds = Vec::new(); + let sql = build_filter_sql(&node, &mut binds); + assert_eq!( + sql, + "json_extract(record, '$.value.author.websites[0].url') = ?" + ); + } + + // ----------------------------------------------------------------------- + // query sort direction + // ----------------------------------------------------------------------- #[tokio::test] async fn query_accepts_valid_sort_direction() { diff --git a/src/lua/execute.rs b/src/lua/execute.rs --- a/src/lua/execute.rs +++ b/src/lua/execute.rs @@ -33,6 +33,38 @@ .collect() } +/// Load env vars, reusing a cached result if less than 30 seconds old. +/// Avoids per-record DB queries during backfill. +pub(crate) async fn load_env_vars_cached( + db: &sqlx::AnyPool, + backend: DatabaseBackend, +) -> HashMap { + use std::sync::Mutex; + + static CACHE: std::sync::OnceLock)>> = + std::sync::OnceLock::new(); + + let cache = CACHE.get_or_init(|| { + Mutex::new(( + Instant::now() - std::time::Duration::from_secs(60), + HashMap::new(), + )) + }); + { + let guard = cache.lock().unwrap(); + if guard.0.elapsed() < std::time::Duration::from_secs(30) { + return guard.1.clone(); + } + } + + let vars = load_env_vars(db, backend).await; + { + let mut guard = cache.lock().unwrap(); + *guard = (Instant::now(), vars.clone()); + } + vars +} + /// Execute a Lua script for a procedure endpoint. #[allow(clippy::too_many_arguments)] pub async fn execute_procedure_script( @@ -889,6 +921,7 @@ pub collection: &'a str, pub rkey: &'a str, pub record: Option<&'a Value>, + pub cached_env_vars: Option<&'a HashMap>, } /// Execute a Lua hook script triggered by a record index event. @@ -1040,7 +1073,14 @@ ) .map_err(|e| format!("failed to set hook context: {e}"))?; - context::set_env_context(&lua, &load_env_vars(&event.state.db, backend).await) + let owned_env_vars; + let env_vars = if let Some(cached) = event.cached_env_vars { + cached + } else { + owned_env_vars = load_env_vars(&event.state.db, backend).await; + &owned_env_vars + }; + context::set_env_context(&lua, env_vars) .map_err(|e| format!("failed to set env context: {e}"))?; lua.load(event.script) @@ -1205,6 +1245,7 @@ collection: "test.collection", rkey: "rkey1", record, + cached_env_vars: None, } } diff --git a/src/lua/mod.rs b/src/lua/mod.rs --- a/src/lua/mod.rs +++ b/src/lua/mod.rs @@ -11,6 +11,7 @@ #[allow(unused_imports)] pub(crate) use context::SpaceContext; pub(crate) use execute::{ - HookEvent, execute_hook_script, execute_procedure_script, execute_query_script, run_hook_once, + HookEvent, execute_hook_script, execute_procedure_script, execute_query_script, + load_env_vars_cached, run_hook_once, }; pub(crate) use sandbox::validate_script; 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 @@ -576,8 +576,6 @@ const [fetchedLoaded, setFetchedLoaded] = useState(false); // Refs for open state and callbacks so the SSE callback doesn't need to re-bind on toggle - const pdsOpenRef = useRef(false); - const fetchedOpenRef = useRef(false); const onJobUpdateRef = useRef(onJobUpdate); onJobUpdateRef.current = onJobUpdate; @@ -641,8 +639,6 @@ const [discoveredOpen, setDiscoveredOpen] = useState(false); const [pdsOpen, setPdsOpen] = useState(false); const [fetchedOpen, setFetchedOpen] = useState(false); - pdsOpenRef.current = pdsOpen; - fetchedOpenRef.current = fetchedOpen; // Lazy-load detail data only when sections are expanded useEffect(() => { @@ -691,77 +687,18 @@ // Uses refs for open state so the callback identity is stable and doesn't // cause the worker to reconnect when sections are toggled. const handleSSEBatch = useCallback((events: BackfillEvent[]) => { - const pOpen = pdsOpenRef.current; - const fOpen = fetchedOpenRef.current; - - if (pOpen) { - const pdsEvents = events.filter( - (e) => (e.type === "repo_resolved" || e.type === "repo_fetched") && e.pds_endpoint, - ); - if (pdsEvents.length > 0) { - setPdsSummary((prev) => { - const byEndpoint = new Map(prev.map((p, i) => [p.pds_endpoint, i])); - const next = [...prev]; - for (const e of pdsEvents) { - const idx = byEndpoint.get(e.pds_endpoint!); - if (e.type === "repo_resolved") { - if (idx != null) { - next[idx] = { ...next[idx], total_repos: next[idx].total_repos + 1 }; - } else { - const newIdx = next.length; - next.push({ pds_endpoint: e.pds_endpoint!, total_repos: 1, completed_repos: 0, total_records: 0 }); - byEndpoint.set(e.pds_endpoint!, newIdx); - } - } else if (e.type === "repo_fetched" && idx != null) { - next[idx] = { - ...next[idx], - completed_repos: next[idx].completed_repos + 1, - total_records: next[idx].total_records + (e.records_fetched ?? 0), - }; - } - } - return next; - }); - } - } - - if (fOpen) { - const fetched = events.filter((e) => e.type === "repo_fetched" && e.did); - if (fetched.length > 0) { - setFetchedRepos((prev) => { - const existing = new Set(prev.map((r) => r.did)); - const newItems = fetched - .filter((e) => !existing.has(e.did!)) - .map((e) => ({ did: e.did!, pds_endpoint: e.pds_endpoint ?? null, status: "completed" as const, records_fetched: e.records_fetched ?? 0 })); - - if (newItems.length === 0) return prev; - return [...newItems, ...prev]; - }); - } - } - - // Update job counters, stage, and status from SSE const update = onJobUpdateRef.current; - // Increment total_repos from repo_discovered events - const discoveredCount = events.filter((e) => e.type === "repo_discovered").length; - const resolvedCount = events.filter((e) => e.type === "repo_resolved").length; - const fetchedEvents = events.filter((e) => e.type === "repo_fetched"); - const fetchedCount = fetchedEvents.length; - const fetchedRecords = fetchedEvents.reduce((sum, e) => sum + (e.records_fetched ?? 0), 0); - - if (discoveredCount > 0 || resolvedCount > 0 || fetchedCount > 0) { - update((j) => ({ - ...j, - total_repos: (j.total_repos ?? 0) + discoveredCount, - resolved_repos: (j.resolved_repos ?? 0) + resolvedCount, - processed_repos: (j.processed_repos ?? 0) + fetchedCount, - total_records: (j.total_records ?? 0) + fetchedRecords, - })); - } - for (const e of events) { - if (e.type === "job_stage_changed" && e.stage) { + if (e.type === "job_counters") { + update((j) => ({ + ...j, + ...(e.total_repos != null && { total_repos: e.total_repos }), + ...(e.resolved_repos != null && { resolved_repos: e.resolved_repos }), + ...(e.processed_repos != null && { processed_repos: e.processed_repos }), + ...(e.total_records != null && { total_records: e.total_records }), + })); + } else if (e.type === "job_stage_changed" && e.stage) { update((j) => ({ ...j, stage: e.stage! })); } else if (e.type === "job_completed" && e.status) { update((j) => ({ ...j, status: e.status!, error: e.error ?? null })); @@ -783,6 +720,11 @@ }, [visibleDiscoveredDids, visibleFetchedDids]); const profiles = useBlueskyProfiles(allVisibleDids); + + const sortedPdsSummary = useMemo( + () => [...pdsSummary].sort((a, b) => b.total_repos - a.total_repos), + [pdsSummary], + ); const fetchedWithRecords = useMemo( () => fetchedRepos.filter((r) => r.records_fetched > 0), @@ -858,9 +800,10 @@ onOpenChange={setDiscoveredOpen} > {discoveredRepos.length > 0 ? ( - r.did} + onVisibleKeysChange={setVisibleDiscoveredDids} hasMore={!!discoveredCursor} onLoadMore={loadMoreDiscovered} rowHeight={28} @@ -894,20 +837,23 @@ open={pdsOpen} onOpenChange={setPdsOpen} > - {pdsSummary.length > 0 ? ( -
- {pdsSummary - .sort((a, b) => b.total_repos - a.total_repos) - .map((pds) => ( -
- - {new URL(pds.pds_endpoint).hostname} - - / repos · records - -
- ))} -
+ {sortedPdsSummary.length > 0 ? ( + p.pds_endpoint} + hasMore={false} + onLoadMore={() => {}} + rowHeight={32} + renderRow={(pds) => ( +
+ + {new URL(pds.pds_endpoint).hostname} + + / repos · records + +
+ )} + /> ) : pdsLoaded ? (

No PDS data yet.

) : null} @@ -938,9 +884,10 @@ onOpenChange={setFetchedOpen} > {fetchedWithRecords.length > 0 ? ( - r.did} + onVisibleKeysChange={setVisibleFetchedDids} hasMore={!!fetchedCursor} onLoadMore={loadMoreFetched} rowHeight={40} @@ -1034,26 +981,28 @@ ); } -function VirtualRepoList({ - repos, - onVisibleDidsChange, +function VirtualList({ + items, + getKey, hasMore, onLoadMore, rowHeight, renderRow, + onVisibleKeysChange, }: { - repos: BackfillRepoEntry[]; - onVisibleDidsChange: (dids: string[]) => void; + items: T[]; + getKey: (item: T) => string; hasMore: boolean; onLoadMore: () => void; rowHeight: number; - renderRow: (repo: BackfillRepoEntry) => React.ReactNode; + renderRow: (item: T) => React.ReactNode; + onVisibleKeysChange?: (keys: string[]) => void; }) { const parentRef = useRef(null); const loadMoreTriggered = useRef(false); const virtualizer = useVirtualizer({ - count: repos.length, + count: items.length, getScrollElement: () => parentRef.current, estimateSize: () => rowHeight, overscan: 5, @@ -1061,22 +1010,22 @@ const virtualItems = virtualizer.getVirtualItems(); - const visibleDidsKey = virtualItems.map((item) => repos[item.index]?.did).filter(Boolean).join(","); + const visibleKeysStr = virtualItems.map((item) => getKey(items[item.index])).filter(Boolean).join(","); useEffect(() => { - onVisibleDidsChange(visibleDidsKey.split(",").filter(Boolean)); - }, [visibleDidsKey, onVisibleDidsChange]); + onVisibleKeysChange?.(visibleKeysStr.split(",").filter(Boolean)); + }, [visibleKeysStr, onVisibleKeysChange]); useEffect(() => { if (!hasMore) return; const lastItem = virtualItems[virtualItems.length - 1]; - if (lastItem && lastItem.index >= repos.length - 5 && !loadMoreTriggered.current) { + if (lastItem && lastItem.index >= items.length - 5 && !loadMoreTriggered.current) { loadMoreTriggered.current = true; onLoadMore(); } - if (lastItem && lastItem.index < repos.length - 5) { + if (lastItem && lastItem.index < items.length - 5) { loadMoreTriggered.current = false; } - }, [virtualItems, repos.length, hasMore, onLoadMore]); + }, [virtualItems, items.length, hasMore, onLoadMore]); return (
@@ -1084,11 +1033,11 @@ style={{ height: virtualizer.getTotalSize(), width: "100%", position: "relative" }} > {virtualItems.map((virtualRow) => { - const repo = repos[virtualRow.index]; - if (!repo) return null; + const item = items[virtualRow.index]; + if (!item) return null; return (
- {renderRow(repo)} + {renderRow(item)}
); })} -- tangled.sh