diff --git a/Cargo.lock b/Cargo.lock index 8251308a..4b0aa295 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8569,7 +8569,7 @@ dependencies = [ [[package]] name = "rsky-wintermute" -version = "0.9.4" +version = "0.9.5" dependencies = [ "base64 0.22.1", "bytes", diff --git a/rsky-wintermute/Cargo.toml b/rsky-wintermute/Cargo.toml index 08498d5e..1d298e46 100644 --- a/rsky-wintermute/Cargo.toml +++ b/rsky-wintermute/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rsky-wintermute" -version = "0.9.4" +version = "0.9.5" edition = "2024" authors = ["Rudy Fraser "] description = "Monolithic indexer combining ingester, backfiller, and indexer" diff --git a/rsky-wintermute/src/config.rs b/rsky-wintermute/src/config.rs index f6f1f9a7..eb26e7b4 100644 --- a/rsky-wintermute/src/config.rs +++ b/rsky-wintermute/src/config.rs @@ -28,6 +28,31 @@ pub const FSYNC_MS: Option = Some(1000); pub const MEMTABLE_SIZE: u32 = 256 * 1024 * 1024; // 256MB (up from 64MB) pub const BLOCK_SIZE: u32 = 64 * 1024; +/// Upper bound on the `repo_backfill` queue, in entries. `0` disables the bound. +/// +/// The relay enumerator is a producer with no consumer whenever +/// `BACKFILLER_WORKERS=0`, which is the recommended setting where repo backfill is +/// handled elsewhere. Unbounded, the queue reaches millions of entries within hours; +/// its Fjall partition grows to gigabytes, and the LSM read and compaction paths +/// allocate block buffers in proportion. The result reads as a memory leak -- resident +/// memory climbs steadily and only a restart reclaims it -- when it is really a queue +/// nothing drains. +pub static REPO_BACKFILL_MAX_QUEUE: LazyLock = LazyLock::new(|| { + std::env::var("REPO_BACKFILL_MAX_QUEUE") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(250_000) +}); + +/// Whether the enumerator may enqueue another repo. +/// +/// Separate from the enumeration loop so the bound is testable without a relay or a +/// storage engine. +#[must_use] +pub const fn repo_backfill_has_room(queue_len: usize, max_queue: usize) -> bool { + max_queue == 0 || queue_len < max_queue +} + pub const FIREHOSE_PING_INTERVAL: Duration = Duration::from_secs(30); // Cursor save interval - like indigo/tap's cursorSaveInterval @@ -542,4 +567,24 @@ mod tests { // deployment supplies -csearch_path. This must never move back there. assert!(!SESSION_SETUP_SQL.contains("search_path")); } + + #[test] + fn repo_backfill_bound_admits_below_the_limit() { + assert!(repo_backfill_has_room(0, 10)); + assert!(repo_backfill_has_room(9, 10)); + } + + #[test] + fn repo_backfill_bound_rejects_at_and_above_the_limit() { + assert!(!repo_backfill_has_room(10, 10)); + assert!(!repo_backfill_has_room(11, 10)); + // A queue already past the bound when the process starts must not grow further. + assert!(!repo_backfill_has_room(5_000_000, 250_000)); + } + + #[test] + fn repo_backfill_bound_of_zero_means_unbounded() { + assert!(repo_backfill_has_room(0, 0)); + assert!(repo_backfill_has_room(usize::MAX, 0)); + } } diff --git a/rsky-wintermute/src/ingester/backfill_queue.rs b/rsky-wintermute/src/ingester/backfill_queue.rs index 34d9bb35..a3255250 100644 --- a/rsky-wintermute/src/ingester/backfill_queue.rs +++ b/rsky-wintermute/src/ingester/backfill_queue.rs @@ -1,4 +1,5 @@ use crate::SHUTDOWN; +use crate::config::repo_backfill_has_room; use crate::storage::Storage; use crate::types::{BackfillJob, WintermuteError}; use deadpool_postgres::{Config, ManagerConfig, Pool, RecyclingMethod, Runtime}; @@ -52,6 +53,7 @@ pub async fn populate_backfill_queue( storage: Arc, relay_host: String, database_url: String, + max_queue: usize, ) -> Result<(), WintermuteError> { use crate::metrics; @@ -127,6 +129,10 @@ pub async fn populate_backfill_queue( let mut total_enumerated = 0u64; let mut last_log_count = 0u64; + // Tracked rather than re-read: Fjall's `len()` scans the partition, so polling it + // per page would cost more as the queue grows. A concurrent drain only makes this + // an overestimate, which errs toward enqueueing less. + let mut queue_len = repo_backfill_len; loop { if SHUTDOWN.load(Ordering::Relaxed) { @@ -163,6 +169,12 @@ pub async fn populate_backfill_queue( let list_response: ListReposResponse = response.json().await?; for repo in &list_response.repos { + if !repo_backfill_has_room(queue_len, max_queue) { + tracing::warn!( + "repo_backfill queue at its bound ({queue_len} entries); stopping enumeration. Raise REPO_BACKFILL_MAX_QUEUE, or run backfiller workers to drain it, to resume." + ); + return Ok(()); + } metrics::INGESTER_BACKFILL_REPOS_FETCHED_TOTAL.inc(); let job = BackfillJob { did: repo.did.clone(), @@ -172,6 +184,7 @@ pub async fn populate_backfill_queue( storage.enqueue_backfill(&job)?; metrics::INGESTER_BACKFILL_REPOS_WRITTEN_TOTAL.inc(); total_enumerated += 1; + queue_len += 1; } if let Some(next_cursor) = list_response.cursor { diff --git a/rsky-wintermute/src/ingester/backfill_queue_tests.rs b/rsky-wintermute/src/ingester/backfill_queue_tests.rs index fca1e937..e2604384 100644 --- a/rsky-wintermute/src/ingester/backfill_queue_tests.rs +++ b/rsky-wintermute/src/ingester/backfill_queue_tests.rs @@ -37,7 +37,7 @@ mod tests { .create_async() .await; - let result = populate_backfill_queue(storage.clone(), server.url(), String::new()).await; + let result = populate_backfill_queue(storage.clone(), server.url(), String::new(), 0).await; drop(server); assert!( @@ -108,7 +108,7 @@ mod tests { .create_async() .await; - let result = populate_backfill_queue(storage.clone(), server.url(), String::new()).await; + let result = populate_backfill_queue(storage.clone(), server.url(), String::new(), 0).await; drop(server); assert!( @@ -134,7 +134,7 @@ mod tests { .create_async() .await; - let result = populate_backfill_queue(storage.clone(), server.url(), String::new()).await; + let result = populate_backfill_queue(storage.clone(), server.url(), String::new(), 0).await; drop(server); assert!(result.is_err(), "should fail with HTTP 500"); @@ -163,7 +163,7 @@ mod tests { .create_async() .await; - let result = populate_backfill_queue(storage.clone(), server.url(), String::new()).await; + let result = populate_backfill_queue(storage.clone(), server.url(), String::new(), 0).await; drop(server); assert!( @@ -218,7 +218,7 @@ mod tests { let server_url = server.url(); let result = - populate_backfill_queue(storage.clone(), server_url.clone(), String::new()).await; + populate_backfill_queue(storage.clone(), server_url.clone(), String::new(), 0).await; drop(server); assert!( @@ -267,7 +267,7 @@ mod tests { .await; let result = - populate_backfill_queue(storage.clone(), server_url.clone(), String::new()).await; + populate_backfill_queue(storage.clone(), server_url.clone(), String::new(), 0).await; drop(server); assert!( @@ -289,4 +289,79 @@ mod tests { "should have re-enumerated repos" ); } + + /// The enumerator is a producer with no consumer whenever backfiller workers are + /// disabled. Without a bound the queue grows for as long as the relay has repos to + /// list, which is what drives resident memory up until a restart. + #[tokio::test] + async fn populate_backfill_queue_stops_at_the_bound() { + let (storage, _dir) = setup_test_storage(); + let mut server = mockito::Server::new_async().await; + + // Two pages of five, and a cursor that never ends: unbounded, this would run + // until the mock stopped answering. + let body = serde_json::json!({ + "repos": [ + {"did": "did:plc:a1"}, {"did": "did:plc:a2"}, {"did": "did:plc:a3"}, + {"did": "did:plc:a4"}, {"did": "did:plc:a5"} + ], + "cursor": "keep-going" + }); + let _mock = server + .mock("GET", mockito::Matcher::Any) + .with_status(200) + .with_header("content-type", "application/json") + .with_body(body.to_string()) + .expect_at_least(1) + .create_async() + .await; + + let result = populate_backfill_queue(storage.clone(), server.url(), String::new(), 3).await; + drop(server); + + assert!(result.is_ok(), "should stop cleanly, not error: {result:?}"); + assert_eq!( + storage.repo_backfill_len().unwrap(), + 3, + "enumeration must stop at the bound, not overshoot it" + ); + } + + /// A queue already past the bound at startup must not grow at all. + #[tokio::test] + async fn populate_backfill_queue_enqueues_nothing_when_already_over_the_bound() { + let (storage, _dir) = setup_test_storage(); + for i in 0..4 { + storage + .enqueue_backfill(&crate::types::BackfillJob { + did: format!("did:plc:pre{i}"), + retry_count: 0, + priority: false, + }) + .unwrap(); + } + + let mut server = mockito::Server::new_async().await; + let body = serde_json::json!({ + "repos": [{"did": "did:plc:new1"}, {"did": "did:plc:new2"}], + "cursor": "keep-going" + }); + let _mock = server + .mock("GET", mockito::Matcher::Any) + .with_status(200) + .with_header("content-type", "application/json") + .with_body(body.to_string()) + .create_async() + .await; + + let result = populate_backfill_queue(storage.clone(), server.url(), String::new(), 3).await; + drop(server); + + assert!(result.is_ok(), "should stop cleanly: {result:?}"); + assert_eq!( + storage.repo_backfill_len().unwrap(), + 4, + "an over-full queue must not grow" + ); + } } diff --git a/rsky-wintermute/src/ingester/mod.rs b/rsky-wintermute/src/ingester/mod.rs index 388b7cfc..b3b3e1d8 100644 --- a/rsky-wintermute/src/ingester/mod.rs +++ b/rsky-wintermute/src/ingester/mod.rs @@ -6,7 +6,10 @@ mod tests; use crate::SHUTDOWN; use crate::backfiller::convert_record_to_ipld; -use crate::config::{CURSOR_SAVE_INTERVAL, DB_POOL_SIZE, FIREHOSE_PING_INTERVAL, WORKERS_INGESTER}; +use crate::config::{ + CURSOR_SAVE_INTERVAL, DB_POOL_SIZE, FIREHOSE_PING_INTERVAL, REPO_BACKFILL_MAX_QUEUE, + WORKERS_INGESTER, +}; use crate::storage::Storage; use crate::types::{CommitData, FirehoseEvent, IndexJob, WintermuteError, WriteAction}; use deadpool_postgres::{Config, ManagerConfig, Pool, RecyclingMethod, Runtime}; @@ -91,6 +94,7 @@ impl IngesterManager { backfill_storage, backfill_host, backfill_db_url, + *REPO_BACKFILL_MAX_QUEUE, ) .await {