Something went wrong. Try again.
very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314use std::collections::HashMap;use std::future::Future;use std::sync::atomic::AtomicI64;use std::time::Duration;
use miette::Result;use smol_str::SmolStr;#[cfg(feature = "indexer")]use tokio::sync::Notify;use tokio::sync::{Semaphore, watch};use url::Url;
use crate::ingest::firehose_stats::FirehoseStats;#[cfg(feature = "relay")]use crate::pds_daily_limit::PdsDailyLimit;use crate::{ config::Config, db::Db, filter::{FilterHandle, new_handle as new_filter_handle}, pds_meta::{PdsMeta, PdsMetaHandle, TierPolicy, new_handle as new_pds_handle}, resolver::Resolver, util::throttle::Throttler,};
pub struct AppState { pub db: Db, pub resolver: Resolver, pub throttler: Throttler, pub(crate) filter: FilterHandle, pub(crate) pds_meta: PdsMetaHandle, #[cfg(feature = "relay")] pub(crate) pds_daily_limit: PdsDailyLimit, pub(crate) tier_policy: TierPolicy, pub firehose_cursors: scc::HashIndex<Url, AtomicI64>, pub(crate) firehose_stats: FirehoseStats, pub firehose_enabled: watch::Sender<bool>, #[cfg(feature = "indexer")] pub backfill_notify: Notify, #[cfg(feature = "indexer")] pub crawler_enabled: watch::Sender<bool>, #[cfg(feature = "indexer")] pub backfill_enabled: watch::Sender<bool>, #[cfg(feature = "indexer")] pub backfill_drained: std::sync::atomic::AtomicUsize, pub ephemeral: bool, pub ephemeral_ttl: Duration, pub only_index_links: bool, pub verify_cids: bool, pub max_car_body_bytes: usize, pub get_repo_semaphore: Semaphore, pub maintenance_timeout: Duration,}
impl AppState { pub fn new(config: &Config) -> Result<Self> { let db = Db::open(config)?; let resolver = Resolver::new(config.plc_urls.clone(), config.identity_cache_size); let filter_config = crate::db::filter::load(&db.filter)?;
#[cfg(feature = "indexer")] let crawler_default = match config.enable_crawler { Some(b) => b, // default: enabled if full-network mode, or if crawler sources are configured None => { filter_config.mode == crate::filter::FilterMode::Full || !config.crawler_sources.is_empty() } };
let filter = new_filter_handle(filter_config);
let tiers: HashMap<String, SmolStr> = crate::db::pds_meta::load_tiers(&db.filter) .unwrap_or_default() .into_iter() .map(|(host, tier)| (host.to_string(), tier)) .collect();
let statuses: HashMap<String, crate::pds_meta::HostStatus> = crate::db::pds_meta::load_statuses(&db.filter) .unwrap_or_default() .into_iter() .map(|(host, stat)| (host.to_string(), stat)) .collect();
let mut hosts = HashMap::new(); for (host, tier) in tiers { hosts .entry(host) .or_insert_with(crate::pds_meta::HostDesc::default) .tier = Some(tier); } for (host, stat) in statuses { hosts .entry(host) .or_insert_with(crate::pds_meta::HostDesc::default) .status = stat; }
let pds_meta = new_pds_handle(PdsMeta { hosts });
let relay_cursors = scc::HashIndex::new(); #[cfg(feature = "relay")] let pds_daily_limit = PdsDailyLimit::new(config.new_host_limit, crate::db::load_pds_daily_adds(&db)?);
#[cfg(feature = "indexer")] let (crawler_enabled, _) = watch::channel(crawler_default); let (firehose_enabled, _) = watch::channel(config.enable_firehose); #[cfg(feature = "indexer")] let (backfill_enabled, _) = watch::channel(config.enable_backfill);
Ok(Self { db, resolver, filter, pds_meta, tier_policy: config.tier_policy.clone(), firehose_cursors: relay_cursors, firehose_stats: FirehoseStats::default(), #[cfg(feature = "indexer")] backfill_notify: Notify::new(), #[cfg(feature = "indexer")] crawler_enabled, firehose_enabled, #[cfg(feature = "indexer")] backfill_enabled, #[cfg(feature = "indexer")] backfill_drained: std::sync::atomic::AtomicUsize::new(0), ephemeral: config.ephemeral, ephemeral_ttl: config.ephemeral_ttl, only_index_links: config.only_index_links, verify_cids: config.verify_cids, max_car_body_bytes: config.max_car_body_bytes, get_repo_semaphore: Semaphore::new(config.get_repo_concurrency_limit), maintenance_timeout: config.maintenance_timeout, throttler: Throttler::new( config.backfill_proxies.len() + 1, config.per_pds_concurrency, ), #[cfg(feature = "relay")] pds_daily_limit, }) }
pub fn is_block_storage_enabled(&self) -> bool { !self.ephemeral && !self.only_index_links }
#[cfg(feature = "indexer")] pub fn notify_backfill(&self) { self.backfill_notify.notify_one(); }
/// pauses the crawler, firehose, and backfill worker, runs `f`, then restores their prior state. /// the restore always happens, even if `f` returns an error. pub async fn with_ingestion_paused<F, Fut, T>(&self, f: F) -> T where F: FnOnce() -> Fut, Fut: Future<Output = T>, { #[cfg(feature = "indexer")] let crawler_was = *self.crawler_enabled.borrow(); let firehose_was = *self.firehose_enabled.borrow(); #[cfg(feature = "indexer")] let backfill_was = *self.backfill_enabled.borrow();
#[cfg(feature = "indexer")] self.crawler_enabled.send_replace(false); self.firehose_enabled.send_replace(false); #[cfg(feature = "indexer")] self.backfill_enabled.send_replace(false);
struct IngestionPauseGuard<'a> { state: &'a AppState, #[cfg(feature = "indexer")] crawler_was: bool, firehose_was: bool, #[cfg(feature = "indexer")] backfill_was: bool, }
impl<'a> Drop for IngestionPauseGuard<'a> { fn drop(&mut self) { #[cfg(feature = "indexer")] self.state.crawler_enabled.send_replace(self.crawler_was); self.state.firehose_enabled.send_replace(self.firehose_was); #[cfg(feature = "indexer")] self.state.backfill_enabled.send_replace(self.backfill_was); tracing::info!("ingestion resumed"); } }
let _guard = IngestionPauseGuard { state: self, #[cfg(feature = "indexer")] crawler_was, firehose_was, #[cfg(feature = "indexer")] backfill_was, };
f().await }
/// applies an account limit status transition for `host`, writing to `batch` and updating /// in-memory state. call this after any event that changes the active account count for a PDS. pub(crate) fn apply_host_limit_status( &self, batch: &mut fjall::OwnedWriteBatch, host: &str, count: u64, ) { use crate::db::pds_meta as db_pds; use crate::pds_meta::PdsMeta; use tracing::{debug, error};
let (current_status, limit) = { let meta = self.pds_meta.load(); let override_name = meta.hosts.get(host).and_then(|h| h.tier.as_ref()); let tier = self.tier_policy.resolve(host, override_name); (meta.status(host), tier.account_limit) };
debug!(%host, ?current_status, ?limit, count, "apply_host_limit_status");
let Some(new_status) = current_status.check_limit_transition(count, limit) else { return; };
debug!(%host, count, ?limit, ?new_status, "account count crossed limit, shifting status");
if let Err(e) = db_pds::set_status(batch, &self.db.filter, host, new_status) { error!(%host, err = %e, "failed to write host status"); return; }
PdsMeta::update_host(&self.pds_meta, host, |h| h.status = new_status); }
/// checks whether `host` is at or over its account limit at the given count. /// does not modify any state. pub(crate) fn is_over_account_limit(&self, host: &str, count: u64) -> bool { let meta = self.pds_meta.load(); let override_name = meta.hosts.get(host).and_then(|h| h.tier.as_ref()); self.tier_policy .resolve(host, override_name) .account_limit .is_some_and(|l| count >= l) }
#[cfg(feature = "indexer")] pub(crate) fn resolve_pds_tier(&self, host: &str) -> crate::config::RateTier { let meta = self.pds_meta.load(); let override_name = meta.hosts.get(host).and_then(|h| h.tier.as_ref()); self.tier_policy.resolve(host, override_name) }}
#[cfg(test)]mod tests { use super::*; use crate::config::Config; use std::time::Duration; use tokio::time::timeout;
#[tokio::test] async fn test_with_ingestion_paused_restore_on_cancel() -> Result<()> { let temp = tempfile::tempdir().unwrap(); let mut config = Config::default(); config.database_path = temp.path().to_path_buf(); #[cfg(feature = "indexer")] { config.enable_crawler = Some(true); config.enable_backfill = true; } config.enable_firehose = true;
let state = AppState::new(&config)?;
#[cfg(feature = "indexer")] { assert!(*state.crawler_enabled.borrow()); assert!(*state.backfill_enabled.borrow()); } assert!(*state.firehose_enabled.borrow());
let result = timeout( Duration::from_millis(50), state.with_ingestion_paused(async || { #[cfg(feature = "indexer")] { assert!(!*state.crawler_enabled.borrow()); assert!(!*state.backfill_enabled.borrow()); } assert!(!*state.firehose_enabled.borrow());
tokio::time::sleep(Duration::from_secs(10)).await; }), ) .await;
assert!(result.is_err()); // should be timed out
#[cfg(feature = "indexer")] { assert!(*state.crawler_enabled.borrow()); assert!(*state.backfill_enabled.borrow()); } assert!(*state.firehose_enabled.borrow());
Ok(()) }}