fix(wintermute): bound the repo backfill queue

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 <him@rudyfraser.com>"]
 description = "Monolithic indexer combining ingester, backfiller, and indexer"

rsky-wintermute/src/config.rs

@@ -28,6 +28,31 @@ pub const FSYNC_MS: Option<u16> = 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<usize> = 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));
+    }
 }

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<Storage>,
     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 {

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"
+        );
+    }
 }

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
                     {