diff --git a/src/control/seed.rs b/src/control/seed.rs --- a/src/control/seed.rs +++ b/src/control/seed.rs @@ -1,5 +1,4 @@ use std::sync::Arc; -use std::sync::atomic::{AtomicI64, Ordering}; use std::time::Duration; use futures::StreamExt; @@ -10,7 +9,6 @@ use url::Url; use super::firehose::FirehoseHandle; -use crate::db::keys; use crate::state::AppState; const MAX_CONCURRENT_SEEDS: usize = 4; @@ -39,10 +37,10 @@ while futs.next().await.is_some() {} } -/// refresh seed relay host status/cursor snapshots without adding sources. +/// refresh seed relay host status snapshots without adding sources. /// /// this runs before persisted sources are spawned so existing PDS tasks start -/// from the seed relay's current cursor instead of racing ahead with no cursor. +/// with updated host status metadata while preserving local firehose cursors. pub(crate) async fn refresh_seed_snapshots(seed_urls: &[Url], state: &Arc) { info!("will refresh seed snapshots..."); @@ -245,7 +243,6 @@ fn apply_seed_snapshot(state: &Arc, hosts: &[Host<'_>]) -> miette::Result<()> { let mut batch = state.db.inner.batch(); - let mut cursor_updates = Vec::with_capacity(hosts.len()); let mut status_updates = Vec::with_capacity(hosts.len()); for host in hosts { @@ -254,42 +251,12 @@ crate::db::pds_meta::set_status(&mut batch, &state.db.filter, hostname, status)?; status_updates.push((hostname.to_string(), status)); - - let Some(seq) = host - .seq - .and_then(|seq| i64::try_from(seq).ok()) - .filter(|seq| *seq > 0) - else { - continue; - }; - - let cursor_key = keys::firehose_cursor_key(hostname); - let existing_seq = state - .db - .cursors - .get(&cursor_key) - .into_diagnostic()? - .map(|bytes| { - bytes - .as_ref() - .try_into() - .into_diagnostic() - .wrap_err("cursor value is not 8 bytes") - .map(i64::from_be_bytes) - }) - .transpose()? - .unwrap_or(0); - if seq > existing_seq { - batch.insert(&state.db.cursors, cursor_key, seq.to_be_bytes()); - cursor_updates.push((hostname.to_string(), seq)); - } } batch.commit().into_diagnostic()?; debug!( hosts = hosts.len(), status_updates = status_updates.len(), - cursor_updates = cursor_updates.len(), "applied listHosts seed snapshot" ); @@ -300,19 +267,6 @@ } next }); - for (hostname, seq) in cursor_updates { - let Ok(url) = Url::parse(&format!("wss://{hostname}/")) else { - continue; - }; - let _ = state - .firehose_cursors - .insert_sync(url.clone(), AtomicI64::new(seq)); - state.firehose_cursors.peek_with(&url, |_, cursor| { - if seq > cursor.load(Ordering::SeqCst) { - cursor.store(seq, Ordering::SeqCst); - } - }); - } Ok(()) } @@ -332,7 +286,9 @@ mod tests { use super::*; use crate::config::Config; + use crate::db::keys; use jacquard_common::CowStr; + use std::sync::atomic::{AtomicI64, Ordering}; use tempfile::tempdir; fn persisted_cursor(state: &AppState, hostname: &str) -> miette::Result> { @@ -352,7 +308,7 @@ } #[test] - fn apply_seed_snapshot_persists_statuses_and_cursors() -> miette::Result<()> { + fn apply_seed_snapshot_persists_statuses_and_preserves_absent_cursor() -> miette::Result<()> { let tmp = tempdir().into_diagnostic()?; let cfg = Config { database_path: tmp.path().to_path_buf(), @@ -404,20 +360,20 @@ assert!(meta.hosts.contains_key("active.example")); assert!(meta.hosts.contains_key("offline.example")); - assert_eq!(persisted_cursor(&state, "active.example")?, Some(100)); - assert_eq!(persisted_cursor(&state, "offline.example")?, Some(5)); + assert_eq!(persisted_cursor(&state, "active.example")?, None); + assert_eq!(persisted_cursor(&state, "offline.example")?, None); let active_url = Url::parse("wss://active.example/").into_diagnostic()?; let in_memory = state .firehose_cursors .peek_with(&active_url, |_, cursor| cursor.load(Ordering::SeqCst)); - assert_eq!(in_memory, Some(100)); + assert_eq!(in_memory, None); Ok(()) } #[test] - fn apply_seed_snapshot_does_not_lower_existing_cursor() -> miette::Result<()> { + fn apply_seed_snapshot_preserves_existing_cursor() -> miette::Result<()> { let tmp = tempdir().into_diagnostic()?; let cfg = Config { database_path: tmp.path().to_path_buf(), @@ -434,7 +390,7 @@ let hosts = vec![Host { hostname: CowStr::Borrowed("active.example"), account_count: Some(42), - seq: Some(100), + seq: Some(500), status: Some(HostStatus::Active), extra_data: None, }]; diff --git a/src/ingest/firehose.rs b/src/ingest/firehose.rs --- a/src/ingest/firehose.rs +++ b/src/ingest/firehose.rs @@ -138,6 +138,11 @@ } impl AddJitter for R {} +fn select_start_cursor(local_cursor: Option, is_pds: bool) -> Option { + local_cursor + .filter(|cursor| *cursor > 0) + .or_else(|| is_pds.then_some(0)) +} pub struct FirehoseIngestor { state: Arc, buffer_tx: BufferTx, @@ -192,14 +197,11 @@ self.enabled.wait_enabled("firehose").await; // get cursor - let start_cursor = self + let local_cursor = self .state .firehose_cursors - .peek_with(&self.relay_host, |_, c| { - let val = c.load(Ordering::SeqCst); - (val > 0).then_some(val) - }) - .flatten(); + .peek_with(&self.relay_host, |_, c| c.load(Ordering::SeqCst)); + let start_cursor = select_start_cursor(local_cursor, self.is_pds); match start_cursor { Some(c) => info!(cursor = %c, "resuming from cursor"), None => info!("no cursor found, live tailing"), @@ -662,5 +664,16 @@ assert_eq!(in_memory, 0); Ok(()) + } + #[test] + fn select_start_cursor_cases() { + assert_eq!(select_start_cursor(Some(1234), true), Some(1234)); + assert_eq!(select_start_cursor(Some(1234), false), Some(1234)); + assert_eq!(select_start_cursor(Some(0), true), Some(0)); + assert_eq!(select_start_cursor(None, true), Some(0)); + assert_eq!(select_start_cursor(Some(0), false), None); + assert_eq!(select_start_cursor(None, false), None); + assert_eq!(select_start_cursor(Some(-5), true), Some(0)); + assert_eq!(select_start_cursor(Some(-5), false), None); } }