diff --git a/ufos/src/main.rs b/ufos/src/main.rs
index 5c77ad0..f5f08be 100644
--- a/ufos/src/main.rs
+++ b/ufos/src/main.rs
@@ -36,9 +36,11 @@ struct Args {
     #[arg(long)]
     data: PathBuf,
     /// DEBUG: don't start the jetstream consumer or its write loop
-    /// todo: restore this
     #[arg(long, action)]
     pause_writer: bool,
+    /// Adjust runtime settings like background task intervals for efficient backfill
+    #[arg(long, action)]
+    backfill: bool,
     /// DEBUG: force the rw loop to fall behind  by pausing it
     /// todo: restore this
     #[arg(long, action)]
@@ -46,6 +48,9 @@ struct Args {
     /// DEBUG: use an in-memory store instead of fjall
     #[arg(long, action)]
     in_mem: bool,
+    /// reset the rollup cursor, scrape through missed things in the past (backfill)
+    #[arg(long, action)]
+    reroll: bool,
     /// DEBUG: interpret jetstream as a file fixture
     #[arg(long, action)]
     jetstream_fixture: bool,
@@ -68,6 +73,8 @@ async fn main() -> anyhow::Result<()> {
             args.jetstream,
             args.jetstream_fixture,
             args.pause_writer,
+            args.backfill,
+            args.reroll,
             read_store,
             write_store,
             cursor,
@@ -85,6 +92,8 @@ async fn main() -> anyhow::Result<()> {
             args.jetstream,
             args.jetstream_fixture,
             args.pause_writer,
+            args.backfill,
+            args.reroll,
             read_store,
             write_store,
             cursor,
@@ -96,10 +105,13 @@ async fn main() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[allow(clippy::too_many_arguments)]
 async fn go<B: StoreBackground>(
     jetstream: String,
     jetstream_fixture: bool,
     pause_writer: bool,
+    backfill: bool,
+    reroll: bool,
     read_store: impl StoreReader + 'static,
     mut write_store: impl StoreWriter<B> + 'static,
     cursor: Option<Cursor>,
@@ -125,7 +137,7 @@ async fn go<B: StoreBackground>(
         consumer::consume(&jetstream, cursor, false, sketch_secret).await?
     };
 
-    let rolling = write_store.background_tasks()?.run();
+    let rolling = write_store.background_tasks(reroll)?.run(backfill);
     let storing = write_store.receive_batches(batches);
 
     tokio::select! {
diff --git a/ufos/src/storage.rs b/ufos/src/storage.rs
index 8ab709e..7965a9f 100644
--- a/ufos/src/storage.rs
+++ b/ufos/src/storage.rs
@@ -26,7 +26,7 @@ pub trait StoreWriter<B: StoreBackground>: Send + Sync
 where
     Self: 'static,
 {
-    fn background_tasks(&mut self) -> StorageResult<B>;
+    fn background_tasks(&mut self, reroll: bool) -> StorageResult<B>;
 
     fn receive_batches<const LIMIT: usize>(
         mut self,
@@ -60,7 +60,7 @@ where
 
 #[async_trait]
 pub trait StoreBackground: Send + Sync {
-    async fn run(mut self) -> StorageResult<()>;
+    async fn run(mut self, backfill: bool) -> StorageResult<()>;
 }
 
 #[async_trait]
diff --git a/ufos/src/storage_fjall.rs b/ufos/src/storage_fjall.rs
index 5ef4db1..1291976 100644
--- a/ufos/src/storage_fjall.rs
+++ b/ufos/src/storage_fjall.rs
@@ -187,7 +187,7 @@ impl StorageWhatever<FjallReader, FjallWriter, FjallBackground, FjallConfig> for
                     )?;
                 } else {
                     return Err(StorageError::InitError(format!(
-                        "stored js_endpoint {stored:?} differs from provided {endpoint:?}, refusing to start.")));
+                        "stored js_endpoint {stored:?} differs from provided {endpoint:?}, refusing to start without --jetstream-force.")));
                 }
             }
             stored_secret
@@ -801,10 +801,13 @@ impl FjallWriter {
 }
 
 impl StoreWriter<FjallBackground> for FjallWriter {
-    fn background_tasks(&mut self) -> StorageResult<FjallBackground> {
+    fn background_tasks(&mut self, reroll: bool) -> StorageResult<FjallBackground> {
         if self.bg_taken.swap(true, Ordering::SeqCst) {
             Err(StorageError::BackgroundAlreadyStarted)
         } else {
+            if reroll {
+                insert_static_neu::<NewRollupCursorKey>(&self.global, Cursor::from_start())?;
+            }
             Ok(FjallBackground(self.clone()))
         }
     }
@@ -1048,13 +1051,15 @@ pub struct FjallBackground(FjallWriter);
 
 #[async_trait]
 impl StoreBackground for FjallBackground {
-    async fn run(mut self) -> StorageResult<()> {
+    async fn run(mut self, backfill: bool) -> StorageResult<()> {
         let mut dirty_nsids = HashSet::new();
 
-        let mut rollup = tokio::time::interval(Duration::from_millis(81));
+        let mut rollup =
+            tokio::time::interval(Duration::from_millis(if backfill { 1_300 } else { 140 }));
         rollup.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
 
-        let mut trim = tokio::time::interval(Duration::from_millis(6_000));
+        let mut trim =
+            tokio::time::interval(Duration::from_millis(if backfill { 12_000 } else { 6_000 }));
         trim.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
         loop {
@@ -1065,7 +1070,7 @@ impl StoreBackground for FjallBackground {
                         rollup.reset_after(Duration::from_millis(1_200)); // we're caught up, take a break
                     }
                     dirty_nsids.extend(dirty);
-                    log::info!("rolled up {n} items ({} collections now dirty)", dirty_nsids.len());
+                    log::trace!("rolled up {n} items ({} collections now dirty)", dirty_nsids.len());
                 },
                 _ = trim.tick() => {
                     log::info!("trimming {} nsids: {dirty_nsids:?}", dirty_nsids.len());
@@ -1155,18 +1160,6 @@ pub struct StorageInfo {
 
 ////////// temp stuff to remove:
 
-// fn summarize_batch<const LIMIT: usize>(batch: &EventBatch<LIMIT>) -> String {
-//     format!(
-//         "batch of {: >3} samples from {: >4} records in {: >2} collections from ~{: >4} DIDs, {} acct removes, cursor {: <12?}",
-//         batch.total_records(),
-//         batch.total_seen(),
-//         batch.total_collections(),
-//         batch.estimate_dids(),
-//         batch.account_removes(),
-//         batch.latest_cursor().map(|c| c.elapsed()),
-//     )
-// }
-
 #[cfg(test)]
 mod tests {
     use super::*;
diff --git a/ufos/src/storage_mem.rs b/ufos/src/storage_mem.rs
index 8b26614..b86ca74 100644
--- a/ufos/src/storage_mem.rs
+++ b/ufos/src/storage_mem.rs
@@ -797,7 +797,7 @@ impl MemWriter {
 }
 
 impl StoreWriter<MemBackground> for MemWriter {
-    fn background_tasks(&mut self) -> StorageResult<MemBackground> {
+    fn background_tasks(&mut self, _reroll: bool) -> StorageResult<MemBackground> {
         Ok(MemBackground {})
     }
 
@@ -1040,7 +1040,7 @@ pub struct MemBackground;
 
 #[async_trait]
 impl StoreBackground for MemBackground {
-    async fn run(mut self) -> StorageResult<()> {
+    async fn run(mut self, _backfill: bool) -> StorageResult<()> {
         // noop for mem (is there a nicer way to do this?)
         loop {
             tokio::time::sleep(std::time::Duration::from_secs_f64(10.)).await;
-- 
2.54.0


From bfc36ff75fbd214d5d6a9b5cff23176abe080ecc Mon Sep 17 00:00:00 2001
From: phil <uniphil@gmail.com>
Date: Fri, 23 May 2025 20:26:00 -0400
Subject: [PATCH 2/4] backfill, still sketchy

---
 jetstream/src/lib.rs      |   4 +-
 ufos/src/consumer.rs      |  33 +++++---
 ufos/src/file_consumer.rs |   5 +-
 ufos/src/lib.rs           |   1 +
 ufos/src/main.rs          | 162 +++++++++++++++++++++++++++++++++++++-
 ufos/src/storage.rs       |   3 +-
 ufos/src/storage_fjall.rs |  31 +++++---
 ufos/src/storage_mem.rs   |   9 ++-
 8 files changed, 220 insertions(+), 28 deletions(-)

diff --git a/jetstream/src/lib.rs b/jetstream/src/lib.rs
index 6b8888e..8f1e6b2 100644
--- a/jetstream/src/lib.rs
+++ b/jetstream/src/lib.rs
@@ -439,7 +439,7 @@ async fn websocket_task(
 
                         if let Some(last) = last_cursor {
                             if event_cursor <= *last {
-                                log::warn!("event cursor {event_cursor:?} was older than the last one: {last:?}. dropping event.");
+                                log::warn!("event cursor {event_cursor:?} was not newer than the last one: {last:?}. dropping event.");
                                 continue;
                             }
                         }
@@ -475,7 +475,7 @@ async fn websocket_task(
 
                         if let Some(last) = last_cursor {
                             if event_cursor <= *last {
-                                log::warn!("event cursor {event_cursor:?} was older than the last one: {last:?}. dropping event.");
+                                log::warn!("event cursor {event_cursor:?} was not newer than the last one: {last:?}. dropping event.");
                                 continue;
                             }
                         }
diff --git a/ufos/src/consumer.rs b/ufos/src/consumer.rs
index b93bc33..7acfc1b 100644
--- a/ufos/src/consumer.rs
+++ b/ufos/src/consumer.rs
@@ -8,6 +8,7 @@ use jetstream::{
 use std::mem;
 use std::time::Duration;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
+use tokio::time::timeout;
 
 use crate::error::{BatchInsertError, FirehoseEventError};
 use crate::{DeleteAccount, EventBatch, UFOsCommit};
@@ -84,19 +85,29 @@ impl Batcher {
 
     pub async fn run(&mut self) -> anyhow::Result<()> {
         loop {
-            if let Some(event) = self.jetstream_receiver.recv().await {
-                self.handle_event(event).await?
-            } else {
-                anyhow::bail!("channel closed");
+            match timeout(Duration::from_millis(9_000), self.jetstream_receiver.recv()).await {
+                Err(_elapsed) => self.no_events_step().await?,
+                Ok(Some(event)) => self.handle_event(event).await?,
+                Ok(None) => anyhow::bail!("channel closed"),
             }
         }
     }
 
+    async fn no_events_step(&mut self) -> anyhow::Result<()> {
+        let empty = self.current_batch.batch.is_empty();
+        log::info!("no events received, stepping batcher (empty? {empty})");
+        if !empty {
+            self.send_current_batch_now(true, "no events step").await?;
+        }
+        Ok(())
+    }
+
     async fn handle_event(&mut self, event: JetstreamEvent) -> anyhow::Result<()> {
         if let Some(earliest) = &self.current_batch.initial_cursor {
             if event.cursor.duration_since(earliest)? > Duration::from_secs_f64(MAX_BATCH_SPAN_SECS)
             {
-                self.send_current_batch_now(false).await?;
+                self.send_current_batch_now(false, "time since event")
+                    .await?;
             }
         } else {
             self.current_batch.initial_cursor = Some(event.cursor);
@@ -126,7 +137,7 @@ impl Batcher {
             if event.cursor.duration_since(earliest)?.as_secs_f64() > MIN_BATCH_SPAN_SECS
                 && self.batch_sender.capacity() == BATCH_QUEUE_SIZE
             {
-                self.send_current_batch_now(true).await?;
+                self.send_current_batch_now(true, "available queue").await?;
             }
         }
         Ok(())
@@ -141,7 +152,7 @@ impl Batcher {
         );
 
         if let Err(BatchInsertError::BatchFull(commit)) = optimistic_res {
-            self.send_current_batch_now(false).await?;
+            self.send_current_batch_now(false, "handle commit").await?;
             self.current_batch.batch.insert_commit_by_nsid(
                 &collection,
                 commit,
@@ -157,7 +168,7 @@ impl Batcher {
 
     async fn handle_delete_account(&mut self, did: Did, cursor: Cursor) -> anyhow::Result<()> {
         if self.current_batch.batch.account_removes.len() >= MAX_ACCOUNT_REMOVES {
-            self.send_current_batch_now(false).await?;
+            self.send_current_batch_now(false, "delete account").await?;
         }
         self.current_batch
             .batch
@@ -168,14 +179,14 @@ impl Batcher {
 
     // holds up all consumer progress until it can send to the channel
     // use this when the current batch is too full to add more to it
-    async fn send_current_batch_now(&mut self, small: bool) -> anyhow::Result<()> {
+    async fn send_current_batch_now(&mut self, small: bool, referrer: &str) -> anyhow::Result<()> {
         let beginning = match self.current_batch.initial_cursor.map(|c| c.elapsed()) {
             None => "unknown".to_string(),
             Some(Ok(t)) => format!("{:?}", t),
             Some(Err(e)) => format!("+{:?}", e.duration()),
         };
-        log::info!(
-            "sending batch now from {beginning}, {}, queue capacity: {}",
+        log::trace!(
+            "sending batch now from {beginning}, {}, queue capacity: {}, referrer: {referrer}",
             if small { "small" } else { "full" },
             self.batch_sender.capacity(),
         );
diff --git a/ufos/src/file_consumer.rs b/ufos/src/file_consumer.rs
index b5ecbaa..3d38a5b 100644
--- a/ufos/src/file_consumer.rs
+++ b/ufos/src/file_consumer.rs
@@ -19,7 +19,10 @@ async fn read_jsonl(f: File, sender: Sender<JetstreamEvent>) -> Result<()> {
             return Err(JetstreamEventError::ReceiverClosedError.into());
         }
     }
-    Ok(())
+    log::info!("reached end of jsonl file, looping on noop to keep server alive.");
+    loop {
+        tokio::time::sleep(std::time::Duration::from_secs_f64(10.)).await;
+    }
 }
 
 pub async fn consume(
diff --git a/ufos/src/lib.rs b/ufos/src/lib.rs
index 2af46ec..0f0e48e 100644
--- a/ufos/src/lib.rs
+++ b/ufos/src/lib.rs
@@ -230,6 +230,7 @@ pub enum ConsumerInfo {
         endpoint: String,
         started_at: u64,
         latest_cursor: Option<u64>,
+        rollup_cursor: Option<u64>,
     },
 }
 
diff --git a/ufos/src/main.rs b/ufos/src/main.rs
index f5f08be..938465a 100644
--- a/ufos/src/main.rs
+++ b/ufos/src/main.rs
@@ -1,6 +1,7 @@
 use clap::Parser;
 use jetstream::events::Cursor;
 use std::path::PathBuf;
+use std::time::{Duration, SystemTime};
 use ufos::consumer;
 use ufos::file_consumer;
 use ufos::server;
@@ -8,6 +9,7 @@ use ufos::storage::{StorageWhatever, StoreBackground, StoreReader, StoreWriter};
 use ufos::storage_fjall::FjallStorage;
 use ufos::storage_mem::MemStorage;
 use ufos::store_types::SketchSecretPrefix;
+use ufos::ConsumerInfo;
 
 #[cfg(not(target_env = "msvc"))]
 use tikv_jemallocator::Jemalloc;
@@ -112,13 +114,13 @@ async fn go<B: StoreBackground>(
     pause_writer: bool,
     backfill: bool,
     reroll: bool,
-    read_store: impl StoreReader + 'static,
+    read_store: impl StoreReader + 'static + Clone,
     mut write_store: impl StoreWriter<B> + 'static,
     cursor: Option<Cursor>,
     sketch_secret: SketchSecretPrefix,
 ) -> anyhow::Result<()> {
     println!("starting server with storage...");
-    let serving = server::serve(read_store);
+    let serving = server::serve(read_store.clone());
 
     if pause_writer {
         log::info!("not starting jetstream or the write loop.");
@@ -140,13 +142,169 @@ async fn go<B: StoreBackground>(
     let rolling = write_store.background_tasks(reroll)?.run(backfill);
     let storing = write_store.receive_batches(batches);
 
+    let stating = do_update_stuff(read_store, backfill);
+
     tokio::select! {
         z = serving => log::warn!("serve task ended: {z:?}"),
         z = rolling => log::warn!("rollup task ended: {z:?}"),
         z = storing => log::warn!("storage task ended: {z:?}"),
+        z = stating => log::warn!("status task ended: {z:?}"),
     };
 
     println!("bye!");
 
     Ok(())
 }
+
+async fn do_update_stuff(read_store: impl StoreReader, actually: bool) {
+    if !actually {
+        loop {
+            tokio::time::sleep(std::time::Duration::from_secs_f64(4.)).await;
+        }
+    }
+    let started_at = std::time::SystemTime::now();
+    let mut first_cursor = None;
+    let mut first_rollup = None;
+    let mut last_at = std::time::SystemTime::now();
+    let mut last_cursor = None;
+    let mut last_rollup = None;
+    loop {
+        log::info!("stat thing: sleeping");
+        tokio::time::sleep(std::time::Duration::from_secs_f64(4.)).await;
+        log::info!("stat thing: slept, getting info");
+        match read_store.get_consumer_info().await {
+            Err(e) => log::warn!("failed to get jetstream consumer info: {e:?}"),
+            Ok(ConsumerInfo::Jetstream {
+                latest_cursor,
+                rollup_cursor,
+                ..
+            }) => {
+                log::info!("stat thing: got info, reporting");
+                let now = std::time::SystemTime::now();
+                let latest_cursor = latest_cursor.map(Cursor::from_raw_u64);
+                let rollup_cursor = rollup_cursor.map(Cursor::from_raw_u64);
+                backfill_info(
+                    latest_cursor,
+                    rollup_cursor,
+                    last_cursor,
+                    last_rollup,
+                    last_at,
+                    first_cursor,
+                    first_rollup,
+                    started_at,
+                    now,
+                );
+                first_cursor = first_cursor.or(latest_cursor);
+                first_rollup = first_rollup.or(rollup_cursor);
+                last_cursor = latest_cursor;
+                last_rollup = rollup_cursor;
+                last_at = now;
+            }
+        }
+    }
+}
+
+fn nice_duration(dt: Duration) -> String {
+    let secs = dt.as_secs_f64();
+    if secs < 1. {
+        return format!("{:.0}ms", secs * 1000.);
+    }
+    if secs < 60. {
+        return format!("{secs:.02}s");
+    }
+    let mins = (secs / 60.).floor();
+    let rsecs = secs - (mins * 60.);
+    if mins < 60. {
+        return format!("{mins:.0}m{rsecs:.0}s");
+    }
+    let hrs = (mins / 60.).floor();
+    let rmins = mins - (hrs * 60.);
+    if hrs < 24. {
+        return format!("{hrs:.0}h{rmins:.0}m{rsecs:.0}s");
+    }
+    let days = (hrs / 24.).floor();
+    let rhrs = hrs - (days * 24.);
+    format!("{days:.0}d{rhrs:.0}h{rmins:.0}m{rsecs:.0}s")
+}
+
+fn nice_dt_two_maybes(earlier: Option<Cursor>, later: Option<Cursor>) -> String {
+    match (earlier, later) {
+        (Some(earlier), Some(later)) => match later.duration_since(&earlier) {
+            Ok(dt) => nice_duration(dt),
+            Err(e) => {
+                let rev_dt = e.duration();
+                format!("+{}", nice_duration(rev_dt))
+            }
+        },
+        _ => "unknown".to_string(),
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
+fn backfill_info(
+    latest_cursor: Option<Cursor>,
+    rollup_cursor: Option<Cursor>,
+    last_cursor: Option<Cursor>,
+    last_rollup: Option<Cursor>,
+    last_at: SystemTime,
+    first_cursor: Option<Cursor>,
+    first_rollup: Option<Cursor>,
+    started_at: SystemTime,
+    now: SystemTime,
+) {
+    let dt_real = now
+        .duration_since(last_at)
+        .unwrap_or(Duration::from_millis(1));
+
+    let dt_real_total = now
+        .duration_since(started_at)
+        .unwrap_or(Duration::from_millis(1));
+
+    let cursor_rate = latest_cursor
+        .zip(last_cursor)
+        .map(|(latest, last)| {
+            latest
+                .duration_since(&last)
+                .unwrap_or(Duration::from_millis(1))
+        })
+        .map(|dtc| format!("{:.2}", dtc.as_secs_f64() / dt_real.as_secs_f64()))
+        .unwrap_or("??".into());
+
+    let cursor_avg = latest_cursor
+        .zip(first_cursor)
+        .map(|(latest, first)| {
+            latest
+                .duration_since(&first)
+                .unwrap_or(Duration::from_millis(1))
+        })
+        .map(|dtc| format!("{:.2}", dtc.as_secs_f64() / dt_real_total.as_secs_f64()))
+        .unwrap_or("??".into());
+
+    let rollup_rate = rollup_cursor
+        .zip(last_rollup)
+        .map(|(latest, last)| {
+            latest
+                .duration_since(&last)
+                .unwrap_or(Duration::from_millis(1))
+        })
+        .map(|dtc| format!("{:.2}", dtc.as_secs_f64() / dt_real.as_secs_f64()))
+        .unwrap_or("??".into());
+
+    let rollup_avg = rollup_cursor
+        .zip(first_rollup)
+        .map(|(latest, first)| {
+            latest
+                .duration_since(&first)
+                .unwrap_or(Duration::from_millis(1))
+        })
+        .map(|dtc| format!("{:.2}", dtc.as_secs_f64() / dt_real_total.as_secs_f64()))
+        .unwrap_or("??".into());
+
+    log::info!(
+        "cursor: {} behind (→{}, {cursor_rate}x, {cursor_avg}x avg). rollup: {} behind (→{}, {rollup_rate}x, {rollup_avg}x avg).",
+        latest_cursor.map(|c| c.elapsed().map(nice_duration).unwrap_or("++".to_string())).unwrap_or("?".to_string()),
+        nice_dt_two_maybes(last_cursor, latest_cursor),
+        rollup_cursor.map(|c| c.elapsed().map(nice_duration).unwrap_or("++".to_string())).unwrap_or("?".to_string()),
+        nice_dt_two_maybes(last_rollup, rollup_cursor),
+    );
+}
diff --git a/ufos/src/storage.rs b/ufos/src/storage.rs
index 7965a9f..275b3f8 100644
--- a/ufos/src/storage.rs
+++ b/ufos/src/storage.rs
@@ -53,7 +53,8 @@ where
 
     fn step_rollup(&mut self) -> StorageResult<(usize, HashSet<Nsid>)>;
 
-    fn trim_collection(&mut self, collection: &Nsid, limit: usize) -> StorageResult<()>;
+    fn trim_collection(&mut self, collection: &Nsid, limit: usize)
+        -> StorageResult<(usize, usize)>;
 
     fn delete_account(&mut self, did: &Did) -> StorageResult<usize>;
 }
diff --git a/ufos/src/storage_fjall.rs b/ufos/src/storage_fjall.rs
index 1291976..60f73fa 100644
--- a/ufos/src/storage_fjall.rs
+++ b/ufos/src/storage_fjall.rs
@@ -149,8 +149,8 @@ impl StorageWhatever<FjallReader, FjallWriter, FjallBackground, FjallConfig> for
         let keyspace = {
             let config = Config::new(path);
 
-            #[cfg(not(test))]
-            let config = config.fsync_ms(Some(4_000));
+            // #[cfg(not(test))]
+            // let config = config.fsync_ms(Some(4_000));
 
             config.open()?
         };
@@ -360,10 +360,15 @@ impl FjallReader {
             get_snapshot_static_neu::<JetstreamCursorKey, JetstreamCursorValue>(&global)?
                 .map(|c| c.to_raw_u64());
 
+        let rollup_cursor =
+            get_snapshot_static_neu::<NewRollupCursorKey, NewRollupCursorValue>(&global)?
+                .map(|c| c.to_raw_u64());
+
         Ok(ConsumerInfo::Jetstream {
             endpoint,
             started_at,
             latest_cursor,
+            rollup_cursor,
         })
     }
 
@@ -951,7 +956,11 @@ impl StoreWriter<FjallBackground> for FjallWriter {
         Ok((cursors_stepped, dirty_nsids))
     }
 
-    fn trim_collection(&mut self, collection: &Nsid, limit: usize) -> StorageResult<()> {
+    fn trim_collection(
+        &mut self,
+        collection: &Nsid,
+        limit: usize,
+    ) -> StorageResult<(usize, usize)> {
         let mut dangling_feed_keys_cleaned = 0;
         let mut records_deleted = 0;
 
@@ -1025,8 +1034,8 @@ impl StoreWriter<FjallBackground> for FjallWriter {
 
         batch.commit()?;
 
-        log::info!("trim_collection ({collection:?}) removed {dangling_feed_keys_cleaned} dangling feed entries and {records_deleted} records");
-        Ok(())
+        log::trace!("trim_collection ({collection:?}) removed {dangling_feed_keys_cleaned} dangling feed entries and {records_deleted} records");
+        Ok((dangling_feed_keys_cleaned, records_deleted))
     }
 
     fn delete_account(&mut self, did: &Did) -> Result<usize, StorageError> {
@@ -1055,7 +1064,7 @@ impl StoreBackground for FjallBackground {
         let mut dirty_nsids = HashSet::new();
 
         let mut rollup =
-            tokio::time::interval(Duration::from_millis(if backfill { 1_300 } else { 140 }));
+            tokio::time::interval(Duration::from_millis(if backfill { 1 } else { 81 }));
         rollup.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
 
         let mut trim =
@@ -1073,12 +1082,16 @@ impl StoreBackground for FjallBackground {
                     log::trace!("rolled up {n} items ({} collections now dirty)", dirty_nsids.len());
                 },
                 _ = trim.tick() => {
-                    log::info!("trimming {} nsids: {dirty_nsids:?}", dirty_nsids.len());
+                    let n = dirty_nsids.len();
+                    log::trace!("trimming {n} nsids: {dirty_nsids:?}");
                     let t0 = Instant::now();
+                    let (mut total_danglers, mut total_deleted) = (0, 0);
                     for collection in &dirty_nsids {
-                        self.0.trim_collection(collection, 512).inspect_err(|e| log::error!("trim error: {e:?}"))?;
+                        let (danglers, deleted) = self.0.trim_collection(collection, 512).inspect_err(|e| log::error!("trim error: {e:?}"))?;
+                        total_danglers += danglers;
+                        total_deleted += deleted;
                     }
-                    log::info!("finished trimming in {:?}", t0.elapsed());
+                    log::info!("finished trimming {n} nsids in {:?}: {total_danglers} dangling and {total_deleted} total removed.", t0.elapsed());
                     dirty_nsids.clear();
                 },
             };
diff --git a/ufos/src/storage_mem.rs b/ufos/src/storage_mem.rs
index b86ca74..eb2cb69 100644
--- a/ufos/src/storage_mem.rs
+++ b/ufos/src/storage_mem.rs
@@ -441,10 +441,15 @@ impl MemReader {
             get_snapshot_static_neu::<JetstreamCursorKey, JetstreamCursorValue>(&global)?
                 .map(|c| c.to_raw_u64());
 
+        let rollup_cursor =
+            get_snapshot_static_neu::<NewRollupCursorKey, JetstreamCursorValue>(&global)?
+                .map(|c| c.to_raw_u64());
+
         Ok(ConsumerInfo::Jetstream {
             endpoint,
             started_at,
             latest_cursor,
+            rollup_cursor,
         })
     }
 
@@ -959,7 +964,7 @@ impl StoreWriter<MemBackground> for MemWriter {
         collection: &Nsid,
         limit: usize,
         // TODO: could add a start cursor limit to avoid iterating deleted stuff at the start (/end)
-    ) -> StorageResult<()> {
+    ) -> StorageResult<(usize, usize)> {
         let mut dangling_feed_keys_cleaned = 0;
         let mut records_deleted = 0;
 
@@ -1015,7 +1020,7 @@ impl StoreWriter<MemBackground> for MemWriter {
         batch.commit()?;
 
         log::info!("trim_collection ({collection:?}) removed {dangling_feed_keys_cleaned} dangling feed entries and {records_deleted} records");
-        Ok(())
+        Ok((dangling_feed_keys_cleaned, records_deleted))
     }
 
     fn delete_account(&mut self, did: &Did) -> Result<usize, StorageError> {
-- 
2.54.0


From 2abb7faecba075ff48225ec57c67600d1a00cc1e Mon Sep 17 00:00:00 2001
From: phil <uniphil@gmail.com>
Date: Sat, 24 May 2025 20:08:21 -0400
Subject: [PATCH 3/4] backfill: rate-limit and don't die on parse fail

---
 Cargo.lock                | 15 ++++++---------
 jetstream/Cargo.toml      |  2 +-
 ufos/src/consumer.rs      | 13 +++++++++++--
 ufos/src/file_consumer.rs | 27 ++++++++++++++++++++-------
 ufos/src/main.rs          |  3 ---
 ufos/src/storage_fjall.rs | 12 ++++++++++--
 6 files changed, 48 insertions(+), 24 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index ba290f0..e810f41 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -225,9 +225,8 @@ checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0"
 
 [[package]]
 name = "atrium-api"
-version = "0.25.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0d4eb9b4787aba546015c8ccda1d3924c157cee13d67848997fba74ac8144a07"
+version = "0.25.3"
+source = "git+https://github.com/uniphil/atrium?branch=fix%2Fnsid-allow-nonleading-name-digits#c4364f318d337bbc3e3e3aaf97c9f971e95f5f7e"
 dependencies = [
  "atrium-common",
  "atrium-xrpc",
@@ -245,9 +244,8 @@ dependencies = [
 
 [[package]]
 name = "atrium-common"
-version = "0.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ba30d2f9e1a8b3db8fc97d0a5f91ee5a28f8acdddb771ad74c1b08eda357ca3d"
+version = "0.1.2"
+source = "git+https://github.com/uniphil/atrium?branch=fix%2Fnsid-allow-nonleading-name-digits#c4364f318d337bbc3e3e3aaf97c9f971e95f5f7e"
 dependencies = [
  "dashmap",
  "lru",
@@ -260,9 +258,8 @@ dependencies = [
 
 [[package]]
 name = "atrium-xrpc"
-version = "0.12.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "18a9e526cb2ed3e0a2ca78c3ce2a943d9041a68e067dadf42923b523771e07df"
+version = "0.12.3"
+source = "git+https://github.com/uniphil/atrium?branch=fix%2Fnsid-allow-nonleading-name-digits#c4364f318d337bbc3e3e3aaf97c9f971e95f5f7e"
 dependencies = [
  "http",
  "serde",
diff --git a/jetstream/Cargo.toml b/jetstream/Cargo.toml
index 724d1d9..2f379e8 100644
--- a/jetstream/Cargo.toml
+++ b/jetstream/Cargo.toml
@@ -10,7 +10,7 @@ readme = "README.md"
 
 [dependencies]
 async-trait = "0.1.83"
-atrium-api = { version = "0.25.2", default-features = false, features = [
+atrium-api = { git = "https://github.com/uniphil/atrium", branch = "fix/nsid-allow-nonleading-name-digits", default-features = false, features = [
     "namespace-appbsky",
 ] }
 tokio = { version = "1.44.2", features = ["full", "sync", "time"] }
diff --git a/ufos/src/consumer.rs b/ufos/src/consumer.rs
index 7acfc1b..c2fbba7 100644
--- a/ufos/src/consumer.rs
+++ b/ufos/src/consumer.rs
@@ -8,7 +8,7 @@ use jetstream::{
 use std::mem;
 use std::time::Duration;
 use tokio::sync::mpsc::{channel, Receiver, Sender};
-use tokio::time::timeout;
+use tokio::time::{timeout, Interval};
 
 use crate::error::{BatchInsertError, FirehoseEventError};
 use crate::{DeleteAccount, EventBatch, UFOsCommit};
@@ -35,6 +35,7 @@ pub struct Batcher {
     batch_sender: Sender<LimitedBatch>,
     current_batch: CurrentBatch,
     sketch_secret: SketchSecretPrefix,
+    rate_limit: Interval,
 }
 
 pub async fn consume(
@@ -65,7 +66,10 @@ pub async fn consume(
         .await?;
     let (batch_sender, batch_reciever) = channel::<LimitedBatch>(BATCH_QUEUE_SIZE);
     let mut batcher = Batcher::new(jetstream_receiver, batch_sender, sketch_secret);
-    tokio::task::spawn(async move { batcher.run().await });
+    tokio::task::spawn(async move {
+        let r = batcher.run().await;
+        log::info!("batcher ended: {r:?}");
+    });
     Ok(batch_reciever)
 }
 
@@ -75,15 +79,19 @@ impl Batcher {
         batch_sender: Sender<LimitedBatch>,
         sketch_secret: SketchSecretPrefix,
     ) -> Self {
+        let mut rate_limit = tokio::time::interval(std::time::Duration::from_micros(3_900));
+        rate_limit.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
         Self {
             jetstream_receiver,
             batch_sender,
             current_batch: Default::default(),
             sketch_secret,
+            rate_limit,
         }
     }
 
     pub async fn run(&mut self) -> anyhow::Result<()> {
+        // TODO: report errors *from here* probably, since this gets shipped off into a spawned task that might just vanish
         loop {
             match timeout(Duration::from_millis(9_000), self.jetstream_receiver.recv()).await {
                 Err(_elapsed) => self.no_events_step().await?,
@@ -191,6 +199,7 @@ impl Batcher {
             self.batch_sender.capacity(),
         );
         let current = mem::take(&mut self.current_batch);
+        self.rate_limit.tick().await;
         self.batch_sender
             .send_timeout(current.batch, Duration::from_secs_f64(SEND_TIMEOUT_S))
             .await?;
diff --git a/ufos/src/file_consumer.rs b/ufos/src/file_consumer.rs
index 3d38a5b..a42833c 100644
--- a/ufos/src/file_consumer.rs
+++ b/ufos/src/file_consumer.rs
@@ -12,11 +12,18 @@ use tokio::{
 async fn read_jsonl(f: File, sender: Sender<JetstreamEvent>) -> Result<()> {
     let mut lines = BufReader::new(f).lines();
     while let Some(line) = lines.next_line().await? {
-        let event: JetstreamEvent =
-            serde_json::from_str(&line).map_err(JetstreamEventError::ReceivedMalformedJSON)?;
-        if sender.send(event).await.is_err() {
-            log::warn!("All receivers for the jsonl fixture have been dropped, bye.");
-            return Err(JetstreamEventError::ReceiverClosedError.into());
+        match serde_json::from_str::<JetstreamEvent>(&line) {
+            Ok(event) => match sender.send(event).await {
+                Ok(_) => {}
+                Err(e) => {
+                    log::warn!("All receivers for the jsonl fixture have been dropped, bye: {e:?}");
+                    return Err(JetstreamEventError::ReceiverClosedError.into());
+                }
+            },
+            Err(parse_err) => {
+                log::warn!("failed to parse event: {parse_err:?} from event:\n{line}");
+                continue;
+            }
         }
     }
     log::info!("reached end of jsonl file, looping on noop to keep server alive.");
@@ -33,7 +40,13 @@ pub async fn consume(
     let (jsonl_sender, jsonl_receiver) = channel::<JetstreamEvent>(16);
     let (batch_sender, batch_reciever) = channel::<LimitedBatch>(BATCH_QUEUE_SIZE);
     let mut batcher = Batcher::new(jsonl_receiver, batch_sender, sketch_secret);
-    tokio::task::spawn(async move { read_jsonl(f, jsonl_sender).await });
-    tokio::task::spawn(async move { batcher.run().await });
+    tokio::task::spawn(async move {
+        let r = read_jsonl(f, jsonl_sender).await;
+        log::info!("read_jsonl finished: {r:?}");
+    });
+    tokio::task::spawn(async move {
+        let r = batcher.run().await;
+        log::info!("batcher finished: {r:?}");
+    });
     Ok(batch_reciever)
 }
diff --git a/ufos/src/main.rs b/ufos/src/main.rs
index 938465a..0ad842c 100644
--- a/ufos/src/main.rs
+++ b/ufos/src/main.rs
@@ -169,9 +169,7 @@ async fn do_update_stuff(read_store: impl StoreReader, actually: bool) {
     let mut last_cursor = None;
     let mut last_rollup = None;
     loop {
-        log::info!("stat thing: sleeping");
         tokio::time::sleep(std::time::Duration::from_secs_f64(4.)).await;
-        log::info!("stat thing: slept, getting info");
         match read_store.get_consumer_info().await {
             Err(e) => log::warn!("failed to get jetstream consumer info: {e:?}"),
             Ok(ConsumerInfo::Jetstream {
@@ -179,7 +177,6 @@ async fn do_update_stuff(read_store: impl StoreReader, actually: bool) {
                 rollup_cursor,
                 ..
             }) => {
-                log::info!("stat thing: got info, reporting");
                 let now = std::time::SystemTime::now();
                 let latest_cursor = latest_cursor.map(Cursor::from_raw_u64);
                 let rollup_cursor = rollup_cursor.map(Cursor::from_raw_u64);
diff --git a/ufos/src/storage_fjall.rs b/ufos/src/storage_fjall.rs
index 60f73fa..07c5f61 100644
--- a/ufos/src/storage_fjall.rs
+++ b/ufos/src/storage_fjall.rs
@@ -979,7 +979,11 @@ impl StoreWriter<FjallBackground> for FjallWriter {
         let mut live_records_found = 0;
         let mut latest_expired_feed_cursor = None;
         let mut batch = self.keyspace.batch();
-        for kv in self.feeds.range(live_range).rev() {
+        for (i, kv) in self.feeds.range(live_range).rev().enumerate() {
+            if i > 1_000_000 {
+                log::info!("stopping collection trim early: already scanned 1M elements");
+                break;
+            }
             let (key_bytes, val_bytes) = kv?;
             let feed_key = db_complete::<NsidRecordFeedKey>(&key_bytes)?;
             let feed_val = db_complete::<NsidRecordFeedVal>(&val_bytes)?;
@@ -1068,7 +1072,7 @@ impl StoreBackground for FjallBackground {
         rollup.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
 
         let mut trim =
-            tokio::time::interval(Duration::from_millis(if backfill { 12_000 } else { 6_000 }));
+            tokio::time::interval(Duration::from_millis(if backfill { 3_000 } else { 6_000 }));
         trim.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
 
         loop {
@@ -1090,6 +1094,10 @@ impl StoreBackground for FjallBackground {
                         let (danglers, deleted) = self.0.trim_collection(collection, 512).inspect_err(|e| log::error!("trim error: {e:?}"))?;
                         total_danglers += danglers;
                         total_deleted += deleted;
+                        if total_deleted > 1_000_000 {
+                            log::info!("trim stopped early, more than 1M records already deleted.");
+                            break;
+                        }
                     }
                     log::info!("finished trimming {n} nsids in {:?}: {total_danglers} dangling and {total_deleted} total removed.", t0.elapsed());
                     dirty_nsids.clear();
-- 
2.54.0


From 7693c53e5816e162425b66a449077647fee831e5 Mon Sep 17 00:00:00 2001
From: phil <uniphil@gmail.com>
Date: Sun, 25 May 2025 11:15:29 -0400
Subject: [PATCH 4/4] simplify cursor status updates

---
 ufos/src/consumer.rs |   2 +-
 ufos/src/main.rs     | 134 +++++++++++++++++--------------------------
 2 files changed, 54 insertions(+), 82 deletions(-)

diff --git a/ufos/src/consumer.rs b/ufos/src/consumer.rs
index c2fbba7..27a813c 100644
--- a/ufos/src/consumer.rs
+++ b/ufos/src/consumer.rs
@@ -79,7 +79,7 @@ impl Batcher {
         batch_sender: Sender<LimitedBatch>,
         sketch_secret: SketchSecretPrefix,
     ) -> Self {
-        let mut rate_limit = tokio::time::interval(std::time::Duration::from_micros(3_900));
+        let mut rate_limit = tokio::time::interval(std::time::Duration::from_millis(5));
         rate_limit.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
         Self {
             jetstream_receiver,
diff --git a/ufos/src/main.rs b/ufos/src/main.rs
index 0ad842c..496fd6f 100644
--- a/ufos/src/main.rs
+++ b/ufos/src/main.rs
@@ -142,7 +142,7 @@ async fn go<B: StoreBackground>(
     let rolling = write_store.background_tasks(reroll)?.run(backfill);
     let storing = write_store.receive_batches(batches);
 
-    let stating = do_update_stuff(read_store, backfill);
+    let stating = do_update_stuff(read_store);
 
     tokio::select! {
         z = serving => log::warn!("serve task ended: {z:?}"),
@@ -156,12 +156,7 @@ async fn go<B: StoreBackground>(
     Ok(())
 }
 
-async fn do_update_stuff(read_store: impl StoreReader, actually: bool) {
-    if !actually {
-        loop {
-            tokio::time::sleep(std::time::Duration::from_secs_f64(4.)).await;
-        }
-    }
+async fn do_update_stuff(read_store: impl StoreReader) {
     let started_at = std::time::SystemTime::now();
     let mut first_cursor = None;
     let mut first_rollup = None;
@@ -201,42 +196,6 @@ async fn do_update_stuff(read_store: impl StoreReader, actually: bool) {
     }
 }
 
-fn nice_duration(dt: Duration) -> String {
-    let secs = dt.as_secs_f64();
-    if secs < 1. {
-        return format!("{:.0}ms", secs * 1000.);
-    }
-    if secs < 60. {
-        return format!("{secs:.02}s");
-    }
-    let mins = (secs / 60.).floor();
-    let rsecs = secs - (mins * 60.);
-    if mins < 60. {
-        return format!("{mins:.0}m{rsecs:.0}s");
-    }
-    let hrs = (mins / 60.).floor();
-    let rmins = mins - (hrs * 60.);
-    if hrs < 24. {
-        return format!("{hrs:.0}h{rmins:.0}m{rsecs:.0}s");
-    }
-    let days = (hrs / 24.).floor();
-    let rhrs = hrs - (days * 24.);
-    format!("{days:.0}d{rhrs:.0}h{rmins:.0}m{rsecs:.0}s")
-}
-
-fn nice_dt_two_maybes(earlier: Option<Cursor>, later: Option<Cursor>) -> String {
-    match (earlier, later) {
-        (Some(earlier), Some(later)) => match later.duration_since(&earlier) {
-            Ok(dt) => nice_duration(dt),
-            Err(e) => {
-                let rev_dt = e.duration();
-                format!("+{}", nice_duration(rev_dt))
-            }
-        },
-        _ => "unknown".to_string(),
-    }
-}
-
 #[allow(clippy::too_many_arguments)]
 fn backfill_info(
     latest_cursor: Option<Cursor>,
@@ -249,6 +208,53 @@ fn backfill_info(
     started_at: SystemTime,
     now: SystemTime,
 ) {
+    let nice_duration = |dt: Duration| {
+        let secs = dt.as_secs_f64();
+        if secs < 1. {
+            return format!("{:.0}ms", secs * 1000.);
+        }
+        if secs < 60. {
+            return format!("{secs:.02}s");
+        }
+        let mins = (secs / 60.).floor();
+        let rsecs = secs - (mins * 60.);
+        if mins < 60. {
+            return format!("{mins:.0}m{rsecs:.0}s");
+        }
+        let hrs = (mins / 60.).floor();
+        let rmins = mins - (hrs * 60.);
+        if hrs < 24. {
+            return format!("{hrs:.0}h{rmins:.0}m{rsecs:.0}s");
+        }
+        let days = (hrs / 24.).floor();
+        let rhrs = hrs - (days * 24.);
+        format!("{days:.0}d{rhrs:.0}h{rmins:.0}m{rsecs:.0}s")
+    };
+
+    let nice_dt_two_maybes = |earlier: Option<Cursor>, later: Option<Cursor>| match (earlier, later)
+    {
+        (Some(earlier), Some(later)) => match later.duration_since(&earlier) {
+            Ok(dt) => nice_duration(dt),
+            Err(e) => {
+                let rev_dt = e.duration();
+                format!("+{}", nice_duration(rev_dt))
+            }
+        },
+        _ => "unknown".to_string(),
+    };
+
+    let rate = |mlatest: Option<Cursor>, msince: Option<Cursor>, real: Duration| {
+        mlatest
+            .zip(msince)
+            .map(|(latest, since)| {
+                latest
+                    .duration_since(&since)
+                    .unwrap_or(Duration::from_millis(1))
+            })
+            .map(|dtc| format!("{:.2}", dtc.as_secs_f64() / real.as_secs_f64()))
+            .unwrap_or("??".into())
+    };
+
     let dt_real = now
         .duration_since(last_at)
         .unwrap_or(Duration::from_millis(1));
@@ -257,45 +263,11 @@ fn backfill_info(
         .duration_since(started_at)
         .unwrap_or(Duration::from_millis(1));
 
-    let cursor_rate = latest_cursor
-        .zip(last_cursor)
-        .map(|(latest, last)| {
-            latest
-                .duration_since(&last)
-                .unwrap_or(Duration::from_millis(1))
-        })
-        .map(|dtc| format!("{:.2}", dtc.as_secs_f64() / dt_real.as_secs_f64()))
-        .unwrap_or("??".into());
-
-    let cursor_avg = latest_cursor
-        .zip(first_cursor)
-        .map(|(latest, first)| {
-            latest
-                .duration_since(&first)
-                .unwrap_or(Duration::from_millis(1))
-        })
-        .map(|dtc| format!("{:.2}", dtc.as_secs_f64() / dt_real_total.as_secs_f64()))
-        .unwrap_or("??".into());
-
-    let rollup_rate = rollup_cursor
-        .zip(last_rollup)
-        .map(|(latest, last)| {
-            latest
-                .duration_since(&last)
-                .unwrap_or(Duration::from_millis(1))
-        })
-        .map(|dtc| format!("{:.2}", dtc.as_secs_f64() / dt_real.as_secs_f64()))
-        .unwrap_or("??".into());
+    let cursor_rate = rate(latest_cursor, last_cursor, dt_real);
+    let cursor_avg = rate(latest_cursor, first_cursor, dt_real_total);
 
-    let rollup_avg = rollup_cursor
-        .zip(first_rollup)
-        .map(|(latest, first)| {
-            latest
-                .duration_since(&first)
-                .unwrap_or(Duration::from_millis(1))
-        })
-        .map(|dtc| format!("{:.2}", dtc.as_secs_f64() / dt_real_total.as_secs_f64()))
-        .unwrap_or("??".into());
+    let rollup_rate = rate(rollup_cursor, last_rollup, dt_real);
+    let rollup_avg = rate(rollup_cursor, first_rollup, dt_real_total);
 
     log::info!(
         "cursor: {} behind (→{}, {cursor_rate}x, {cursor_avg}x avg). rollup: {} behind (→{}, {rollup_rate}x, {rollup_avg}x avg).",