diff --git a/Cargo.lock b/Cargo.lock index 83ab874..c504c03 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1041,9 +1041,9 @@ checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" [[package]] name = "fjall" -version = "3.0.4" +version = "3.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ebf22b812878dcd767879cb19e03124fd62563dce6410f96538175fba0c132d" +checksum = "40cb1eb0cef3792900897b32c8282f6417bc978f6af46400a2f14bf0e649ae30" dependencies = [ "byteorder-lite", "byteview", @@ -1536,6 +1536,7 @@ dependencies = [ "jacquard-derive", "jacquard-identity", "jacquard-repo", + "lsm-tree", "miette", "mimalloc", "multibase", @@ -2088,9 +2089,9 @@ checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" [[package]] name = "libc" -version = "0.2.182" +version = "0.2.183" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6800badb6cb2082ffd7b6a67e6125bb39f18782f793520caee8cb8846be06112" +checksum = "b5b646652bf6661599e1da8901b3b9522896f01e736bad5f723fe7a3a27f899d" [[package]] name = "libmimalloc-sys" @@ -2150,9 +2151,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] name = "lsm-tree" -version = "3.0.4" +version = "3.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e9bfd2a6ea0c1d430c13643002f35800a87f200fc8ac4827f18a2db9d9fd0644" +checksum = "fc5fa40c207eed45c811085aaa1b0a25fead22e298e286081cd4b98785fe759b" dependencies = [ "byteorder-lite", "byteview", @@ -2673,9 +2674,9 @@ dependencies = [ [[package]] name = "quinn-proto" -version = "0.11.13" +version = "0.11.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f1906b49b0c3bc04b5fe5d86a77925ae6524a19b816ae38ce1e426255f1d8a31" +checksum = "434b42fec591c96ef50e21e886936e66d3cc3f737104fdb9b737c40ffb94c098" dependencies = [ "aws-lc-rs", "bytes", @@ -4690,18 +4691,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.40" +version = "0.8.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a789c6e490b576db9f7e6b6d661bcc9799f7c0ac8352f56ea20193b2681532e5" +checksum = "f2578b716f8a7a858b7f02d5bd870c14bf4ddbbcf3a4c05414ba6503640505e3" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.40" +version = "0.8.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f65c489a7071a749c849713807783f70672b28094011623e200cb86dcb835953" +checksum = "7e6cc098ea4d3bd6246687de65af3f920c430e236bee1e3bf2e441463f08a02f" dependencies = [ "proc-macro2", "quote", diff --git a/Cargo.toml b/Cargo.toml index ccc7af8..6618cfb 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -15,7 +15,7 @@ serde_bytes = "0.11" serde_json = "1.0" rmp-serde = { git = "https://github.com/90-008/msgpack-rust.git" } -fjall = "3.0" +fjall = "3.1" serde_ipld_dagcbor = "0.6" serde_urlencoded = "0.7" @@ -46,6 +46,7 @@ arc-swap = "1.8.2" rustls = { version = "0.23", features = ["aws-lc-rs"] } tokio-tungstenite = { version = "0.28.0", features = ["rustls-tls-native-roots"] } multibase = "0.9.2" +lsm-tree = "3.1.0" [dev-dependencies] tempfile = "3.26.0" diff --git a/README.md b/README.md index 796900b..4cf6d42 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,7 @@ the `WS /stream` (hydrant) and `WS /channel` (tap) endpoints have different desi | `ENABLE_DEBUG` | `false` | enable debug endpoints. | | `DEBUG_PORT` | `3001` | port for debug endpoints (if enabled). | | `NO_LZ4_COMPRESSION` | `false` | disable lz4 compression for storage. | +| `EPHEMERAL` | `false` | if enabled, no records are stored (XRPCs won't be reliable). events are only stored up to an hour for playback. | | `ENABLE_FIREHOSE` | `true` | whether to ingest relay subscriptions. | | `ENABLE_BACKFILL` | `true` | whether to backfill from PDS instances. | | `ENABLE_CRAWLER` | `false` (if Filter), `true` (if Full) | whether to actively query the network for unknown repositories. | diff --git a/src/config.rs b/src/config.rs index 02b0ccb..6024b02 100644 --- a/src/config.rs +++ b/src/config.rs @@ -41,6 +41,7 @@ pub struct Config { pub relay_host: Url, pub plc_urls: Vec, pub full_network: bool, + pub ephemeral: bool, pub cursor_save_interval: Duration, pub repo_fetch_timeout: Duration, pub log_level: SmolStr, @@ -111,6 +112,7 @@ impl Config { let cursor_save_interval = cfg!("CURSOR_SAVE_INTERVAL", 5, sec); let repo_fetch_timeout = cfg!("REPO_FETCH_TIMEOUT", 300, sec); + let ephemeral: bool = cfg!("EPHEMERAL", false); let database_path = cfg!("DATABASE_PATH", "./hydrant.db"); let cache_size = cfg!("CACHE_SIZE", 256u64); let disable_lz4_compression = cfg!("NO_LZ4_COMPRESSION", false); @@ -184,6 +186,7 @@ impl Config { database_path, relay_host, plc_urls, + ephemeral, full_network, cursor_save_interval, repo_fetch_timeout, @@ -216,101 +219,89 @@ impl Config { } } +macro_rules! config_line { + ($f:expr, $label:expr, $value:expr) => { + writeln!($f, " {:) -> fmt::Result { + const LABEL_WIDTH: usize = 27; + writeln!(f, "hydrant configuration:")?; - writeln!(f, " log level: {}", self.log_level)?; - writeln!(f, " relay host: {}", self.relay_host)?; - writeln!(f, " plc urls: {:?}", self.plc_urls)?; - writeln!(f, " full network indexing: {}", self.full_network)?; - writeln!(f, " verify signatures: {}", self.verify_signatures)?; - writeln!( - f, - " backfill concurrency: {}", - self.backfill_concurrency_limit - )?; - writeln!( - f, - " identity cache size: {}", - self.identity_cache_size - )?; - writeln!( + config_line!(f, "log level", self.log_level)?; + config_line!(f, "relay host", self.relay_host)?; + config_line!(f, "plc urls", format_args!("{:?}", self.plc_urls))?; + config_line!(f, "full network indexing", self.full_network)?; + config_line!(f, "verify signatures", self.verify_signatures)?; + config_line!(f, "backfill concurrency", self.backfill_concurrency_limit)?; + config_line!(f, "identity cache size", self.identity_cache_size)?; + config_line!( f, - " cursor save interval: {}sec", - self.cursor_save_interval.as_secs() + "cursor save interval", + format_args!("{}sec", self.cursor_save_interval.as_secs()) )?; - writeln!( + config_line!( f, - " repo fetch timeout: {}sec", - self.repo_fetch_timeout.as_secs() + "repo fetch timeout", + format_args!("{}sec", self.repo_fetch_timeout.as_secs()) )?; - writeln!( + config_line!(f, "ephemeral", self.ephemeral)?; + config_line!(f, "database path", self.database_path.to_string_lossy())?; + config_line!(f, "cache size", format_args!("{} mb", self.cache_size))?; + config_line!(f, "disable lz4 compression", self.disable_lz4_compression)?; + config_line!(f, "api port", self.api_port)?; + config_line!(f, "firehose workers", self.firehose_workers)?; + config_line!(f, "db worker threads", self.db_worker_threads)?; + config_line!( f, - " database path: {}", - self.database_path.to_string_lossy() + "db journal size", + format_args!("{} mb", self.db_max_journaling_size_mb) )?; - writeln!(f, " cache size: {} mb", self.cache_size)?; - writeln!( + config_line!( f, - " disable lz4 compression: {}", - self.disable_lz4_compression + "db pending memtable", + format_args!("{} mb", self.db_pending_memtable_size_mb) )?; - writeln!(f, " api port: {}", self.api_port)?; - writeln!(f, " firehose workers: {}", self.firehose_workers)?; - writeln!(f, " db worker threads: {}", self.db_worker_threads)?; - writeln!( + config_line!( f, - " db journal size: {} mb", - self.db_max_journaling_size_mb + "db blocks memtable", + format_args!("{} mb", self.db_blocks_memtable_size_mb) )?; - writeln!( + config_line!( f, - " db pending memtable: {} mb", - self.db_pending_memtable_size_mb + "db repos memtable", + format_args!("{} mb", self.db_repos_memtable_size_mb) )?; - writeln!( + config_line!( f, - " db blocks memtable: {} mb", - self.db_blocks_memtable_size_mb + "db events memtable", + format_args!("{} mb", self.db_events_memtable_size_mb) )?; - writeln!( - f, - " db repos memtable: {} mb", - self.db_repos_memtable_size_mb - )?; - writeln!( - f, - " db events memtable: {} mb", - self.db_events_memtable_size_mb - )?; - writeln!( - f, - " db records memtable: {} mb", - self.db_records_memtable_size_mb - )?; - - writeln!( + config_line!( f, - " crawler max pending: {}", - self.crawler_max_pending_repos + "db records memtable", + format_args!("{} mb", self.db_records_memtable_size_mb) )?; - writeln!( + config_line!(f, "crawler max pending", self.crawler_max_pending_repos)?; + config_line!( f, - " crawler resume pending: {}", + "crawler resume pending", self.crawler_resume_pending_repos )?; if let Some(signals) = &self.filter_signals { - writeln!(f, " filter signals: {:?}", signals)?; + config_line!(f, "filter signals", format_args!("{:?}", signals))?; } if let Some(collections) = &self.filter_collections { - writeln!(f, " filter collections: {:?}", collections)?; + config_line!(f, "filter collections", format_args!("{:?}", collections))?; } if let Some(excludes) = &self.filter_excludes { - writeln!(f, " filter excludes: {:?}", excludes)?; + config_line!(f, "filter excludes", format_args!("{:?}", excludes))?; } - writeln!(f, " enable debug: {}", self.enable_debug)?; + config_line!(f, "enable debug", self.enable_debug)?; if self.enable_debug { - writeln!(f, " debug port: {}", self.debug_port)?; + config_line!(f, "debug port", self.debug_port)?; } Ok(()) } diff --git a/src/crawler/mod.rs b/src/crawler/mod.rs index 2577098..104de70 100644 --- a/src/crawler/mod.rs +++ b/src/crawler/mod.rs @@ -24,7 +24,7 @@ use tracing::{Instrument, debug, error, info, trace, warn}; use url::Url; const MAX_RETRY_ATTEMPTS: u32 = 5; -const MAX_RETRY_BATCH: usize = 512; +const MAX_RETRY_BATCH: usize = 1000; #[derive(Debug, Serialize, Deserialize)] struct RetryState { @@ -192,8 +192,8 @@ impl Crawler { } } - async fn get_cursor(crawler: &Self) -> Result { - let cursor_bytes = Db::get(crawler.state.db.cursors.clone(), CURSOR_KEY).await?; + async fn get_cursor(&self) -> Result { + let cursor_bytes = Db::get(self.state.db.cursors.clone(), CURSOR_KEY).await?; let cursor: Cursor = cursor_bytes .as_deref() .map(rmp_serde::from_slice) @@ -297,7 +297,7 @@ impl Crawler { let mut rng: SmallRng = rand::make_rng(); let db = &crawler.state.db; - let mut cursor = Self::get_cursor(&crawler).await?; + let mut cursor = crawler.get_cursor().await?; match &cursor { Cursor::Next(Some(c)) => info!(cursor = %c, "resuming"), @@ -401,7 +401,6 @@ impl Crawler { }; let mut batch = db.inner.batch(); - let mut to_queue = Vec::new(); let filter = crawler.state.filter.load(); struct ParseResult { @@ -510,7 +509,6 @@ impl Crawler { let state = RepoState::untracked(rng.next_u64()); batch.insert(&db.repos, &did_key, ser_repo_state(&state)?); batch.insert(&db.pending, keys::pending_key(state.index_id), &did_key); - to_queue.push(did.clone()); } if let Some(new_cursor) = next_cursor { @@ -545,7 +543,7 @@ impl Crawler { }) .ok(); - crawler.account_new_repos(to_queue.len()).await; + crawler.account_new_repos(valid_dids.len()).await; if matches!(cursor, Cursor::Done(_)) { tokio::time::sleep(Duration::from_secs(3600)).await; diff --git a/src/db/compaction.rs b/src/db/compaction.rs new file mode 100644 index 0000000..8bcdbbe --- /dev/null +++ b/src/db/compaction.rs @@ -0,0 +1,67 @@ +use fjall::compaction::filter::Context; +use lsm_tree::compaction::{CompactionFilter, Factory}; +use lsm_tree::compaction::{ItemAccessor, Verdict}; + +mod drop_all { + use super::*; + + pub struct DropAllFilter; + + impl CompactionFilter for DropAllFilter { + fn filter_item(&mut self, _: ItemAccessor<'_>, _: &Context) -> lsm_tree::Result { + Ok(Verdict::Destroy) + } + } + + pub struct DropAllFilterFactory; + + impl Factory for DropAllFilterFactory { + fn name(&self) -> &str { + "drop_all" + } + + fn make_filter(&self, _: &Context) -> Box { + Box::new(DropAllFilter) + } + } +} +pub use drop_all::*; + +mod drop_prefix { + use super::*; + + pub struct DropPrefixFilter { + prefix: &'static [u8], + } + + impl CompactionFilter for DropPrefixFilter { + fn filter_item( + &mut self, + item: ItemAccessor<'_>, + _: &Context, + ) -> lsm_tree::Result { + Ok(item + .key() + .starts_with(&self.prefix) + .then_some(Verdict::Destroy) + .unwrap_or(Verdict::Keep)) + } + } + + pub struct DropPrefixFilterFactory { + pub prefix: &'static [u8], + } + + impl Factory for DropPrefixFilterFactory { + fn name(&self) -> &str { + "drop_prefix" + } + + fn make_filter(&self, _: &Context) -> Box { + Box::new(DropPrefixFilter { + prefix: self.prefix, + }) + } + } +} +pub use drop_prefix::*; diff --git a/src/db/mod.rs b/src/db/mod.rs index 01378bc..3c93ac9 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -1,14 +1,18 @@ +use crate::db::compaction::{DropAllFilterFactory, DropPrefixFilterFactory}; use crate::types::{BroadcastEvent, RepoState}; +use fjall::compaction::{Fifo, Levelled}; use fjall::config::BlockSizePolicy; use fjall::{Database, Keyspace, KeyspaceCreateOptions, OwnedWriteBatch, PersistMode, Slice}; use jacquard_common::IntoStatic; use jacquard_common::types::string::Did; +use lsm_tree::compaction::Factory; use miette::{Context, IntoDiagnostic, Result}; use scc::HashMap; use smol_str::SmolStr; use std::sync::Arc; +pub mod compaction; pub mod filter; pub mod keys; pub mod types; @@ -90,6 +94,9 @@ impl Db { const fn kb(v: u32) -> u32 { v * 1024 } + const fn mb(v: u64) -> u64 { + v * 1024 * 1024 + } let db = Database::builder(&cfg.database_path) .cache_size(cfg.cache_size * 2_u64.pow(20) / 2) @@ -100,7 +107,22 @@ impl Db { .unwrap_or(fjall::CompressionType::Lz4), ) .worker_threads(cfg.db_worker_threads) - .max_journaling_size(cfg.db_max_journaling_size_mb * 1024 * 1024) + .max_journaling_size(mb(cfg.db_max_journaling_size_mb)) + .with_compaction_filter_factories({ + let ephemeral = cfg.ephemeral; + let f = move |ks: &str| match ks { + "records" => { + ephemeral.then(|| -> Arc { Arc::new(DropAllFilterFactory) }) + } + "counts" => ephemeral.then(|| -> Arc { + Arc::new(DropPrefixFilterFactory { + prefix: keys::COUNT_COLLECTION_PREFIX, + }) + }), + _ => None, + }; + Arc::new(f) + }) .open() .into_diagnostic()?; let db = Arc::new(db); @@ -115,7 +137,7 @@ impl Db { opts() // crawler checks if a repo doesn't exist .expect_point_read_hits(false) - .max_memtable_size(cfg.db_repos_memtable_size_mb * 1024 * 1024) + .max_memtable_size(mb(cfg.db_repos_memtable_size_mb)) .data_block_size_policy(BlockSizePolicy::all(kb(4))), )?; let blocks = open_ks( @@ -123,7 +145,7 @@ impl Db { opts() // point reads are used a lot by stream .expect_point_read_hits(true) - .max_memtable_size(cfg.db_blocks_memtable_size_mb * 1024 * 1024) + .max_memtable_size(mb(cfg.db_blocks_memtable_size_mb)) // 32 - 64 kb is probably fine, as the newer blocks will be in the first levels // and any consumers will probably be streaming the newer events... .data_block_size_policy(BlockSizePolicy::new([kb(4), kb(8), kb(32), kb(64)])), @@ -135,7 +157,7 @@ impl Db { // since this keyspace is big, turning off bloom filters will help a lot opts() .expect_point_read_hits(true) - .max_memtable_size(cfg.db_records_memtable_size_mb * 1024 * 1024) + .max_memtable_size(mb(cfg.db_records_memtable_size_mb)) .data_block_size_policy(BlockSizePolicy::all(kb(8))), )?; let cursors = open_ks( @@ -143,6 +165,7 @@ impl Db { opts() // cursor point reads hit almost 100% of the time .expect_point_read_hits(true) + .max_memtable_size(mb(4)) .data_block_size_policy(BlockSizePolicy::all(kb(1))), )?; let pending = open_ks( @@ -150,19 +173,22 @@ impl Db { opts() // iterated over as a queue, no point reads are used so bloom filters are disabled .expect_point_read_hits(true) - .max_memtable_size(cfg.db_pending_memtable_size_mb * 1024 * 1024) + .max_memtable_size(mb(cfg.db_pending_memtable_size_mb)) .data_block_size_policy(BlockSizePolicy::all(kb(4))), )?; // resync point reads often miss (because most repos aren't resyncing), so keeping the bloom filter helps avoid disk hits let resync = open_ks( "resync", - opts().data_block_size_policy(BlockSizePolicy::all(kb(8))), + opts() + .max_memtable_size(mb(cfg.db_pending_memtable_size_mb)) + .data_block_size_policy(BlockSizePolicy::all(kb(8))), )?; let resync_buffer = open_ks( "resync_buffer", opts() // iterated during backfill, no point reads .expect_point_read_hits(true) + .max_memtable_size(mb(16)) .data_block_size_policy(BlockSizePolicy::all(kb(32))), )?; let events = open_ks( @@ -170,14 +196,20 @@ impl Db { opts() // only iterators are used here, no point reads .expect_point_read_hits(true) - .max_memtable_size(cfg.db_events_memtable_size_mb * 1024 * 1024) - .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(32)])), + .max_memtable_size(mb(cfg.db_events_memtable_size_mb)) + .data_block_size_policy(BlockSizePolicy::new([kb(16), kb(32)])) + .compaction_strategy(if cfg.ephemeral { + Arc::new(Fifo::new(mb(512), Some(60 * 60))) + } else { + Arc::new(Levelled::default()) + }), )?; let counts = open_ks( "counts", opts() // count increments hit because counters are mostly pre-initialized .expect_point_read_hits(true) + .max_memtable_size(mb(32)) // the data is very small .data_block_size_policy(BlockSizePolicy::all(kb(1))), )?; @@ -188,14 +220,14 @@ impl Db { "filter", // this can be pretty small since the DIDs wont be compressed that well anyhow opts() - .max_memtable_size((kb(1024) * 16) as u64) + .max_memtable_size(mb(16)) .data_block_size_policy(BlockSizePolicy::all(kb(1))), )?; let crawler = open_ks( "crawler", opts() - .max_memtable_size((kb(1024) * 16) as u64) + .max_memtable_size(mb(16)) .data_block_size_policy(BlockSizePolicy::all(kb(1))), )?;