use miette::Result;
use smol_str::SmolStr;
use std::collections::HashMap;
use std::fmt;
use std::path::PathBuf;
use std::time::Duration;
use url::Url;
use crate::pds_meta::TierPolicy;
pub mod env;
pub mod types;
pub use types::{
BackfillStrategy, Compression, CrawlerMode, CrawlerSource, FirehoseSource, RateTier,
SignatureVerification, StreamKind,
};
/// the streams whose network readers replay under `stream_replay_speed` unless
/// `HYDRANT_THROTTLED_STREAMS` says otherwise.
pub const DEFAULT_THROTTLED_STREAMS: &[StreamKind] = &[StreamKind::Jetstream, StreamKind::Relay];
pub const DEFAULT_RATETIER: RateTier = if cfg!(feature = "indexer") {
RateTier::trusted()
} else {
RateTier::default_tier()
};
/// per-mode configuration defaults.
///
/// the indexer and relay modes are parallel modules with an identical
/// interface, selected by a single cfg pair. this is the only place the
/// mode choice bends `Config::default`; the struct initializer below is
/// unconditional, so adding a field can never silently break one mode.
#[cfg(not(feature = "relay"))]
mod mode_defaults {
use super::{Duration, FirehoseSource, Url};
pub(super) const EPHEMERAL: bool = false;
pub(super) const EPHEMERAL_TTL: Duration = Duration::from_secs(3600); // 1 hour
pub(super) const FULL_NETWORK: bool = false;
pub(super) fn relays() -> Vec<FirehoseSource> {
vec![FirehoseSource {
url: Url::parse("wss://relay.fire.hose.cam/").unwrap(),
is_pds: false,
}]
}
pub(super) fn seed_hosts() -> Vec<Url> {
vec![]
}
}
#[cfg(feature = "relay")]
mod mode_defaults {
use super::{Duration, FirehoseSource, Url};
pub(super) const EPHEMERAL: bool = true;
pub(super) const EPHEMERAL_TTL: Duration = Duration::from_secs(3600 * 24 * 3); // 3 days
pub(super) const FULL_NETWORK: bool = true;
pub(super) fn relays() -> Vec<FirehoseSource> {
vec![]
}
pub(super) fn seed_hosts() -> Vec<Url> {
vec![Url::parse("https://bsky.network").unwrap()]
}
}
#[derive(Debug, Clone)]
pub struct Config {
/// path to the database folder. set via `HYDRANT_DATABASE_PATH`.
pub database_path: PathBuf,
/// if `true`, discovers and indexes all repositories in the network.
/// set via `HYDRANT_FULL_NETWORK`.
pub full_network: bool,
/// if `true`, record bodies live inline in events and are retained only
/// for `ephemeral_ttl`; no record heads or history are stored. set via
/// `HYDRANT_EPHEMERAL`. this is part of the database's persisted storage
/// identity and changing it requires an explicit offline migration or a
/// fresh database path.
pub ephemeral: bool,
/// how long records and events are retained in ephemeral mode before
/// deletion. set via `HYDRANT_EPHEMERAL_TTL` (humantime duration, e.g.
/// `60min`).
pub ephemeral_ttl: Duration,
/// how long superseded record bodies are retained in the `history` keyspace.
/// unset (the default) keeps permanent history, so every stream event
/// resolves forever. setting it bounds history: replay of events older
/// than the window may then fail to resolve a body. ephemeral mode stores
/// no history, so this setting applies only to permanent databases.
/// set via `HYDRANT_HISTORY_TTL` (humantime duration, e.g. `30d`).
pub history_ttl: Option<Duration>,
/// firehose sources for ingestion. set via `HYDRANT_RELAY_HOST` (single)
/// or `HYDRANT_RELAY_HOSTS` (comma-separated; takes precedence).
/// prefix a URL with `pds::` to mark it as a direct PDS connection.
pub relays: Vec<FirehoseSource>,
/// base URL(s) of the PLC directory (comma-separated for multiple).
/// defaults to `https://plc.wtf`, or `https://plc.directory` in full-network mode.
/// set via `HYDRANT_PLC_URL`.
pub plc_urls: Vec<Url>,
/// whether to ingest events from relay firehose subscriptions.
/// set via `HYDRANT_ENABLE_FIREHOSE`.
pub enable_firehose: bool,
/// whether to process queued repository backfills.
/// set via `HYDRANT_ENABLE_BACKFILL`.
pub enable_backfill: bool,
/// number of firehose shards, each its own thread (two per shard in the indexer).
/// set via `HYDRANT_FIREHOSE_WORKERS`.
pub firehose_workers: usize,
/// number of consecutive firehose connection failures before a PDS is marked offline.
/// set via `HYDRANT_FIREHOSE_MAX_FAILURES`.
pub firehose_max_failures: usize,
/// how often the firehose cursors and count checkpoints are saved and the journal is
/// written out to the os (not fsynced).
/// set via `HYDRANT_CURSOR_SAVE_INTERVAL` (humantime duration, e.g. `3sec`).
pub cursor_save_interval: Duration,
/// timeout for fetching a full repository CAR during backfill.
/// set via `HYDRANT_REPO_FETCH_TIMEOUT` (humantime duration, e.g. `5min`).
pub repo_fetch_timeout: Duration,
/// maximum size in bytes of a full-repository `getRepo` CAR response accepted during
/// backfill. the body is streamed and rejected before it is fully buffered once it would
/// exceed this ceiling, bounding per-task memory regardless of what a PDS returns (the same
/// ceiling also bounds sparse `getBlocks` fetches). legitimate repositories are far smaller;
/// lower this on memory-constrained hosts running high backfill concurrency.
/// set via `HYDRANT_MAX_CAR_BODY_BYTES`.
pub max_car_body_bytes: usize,
/// memory that whole-car backfills may hold at once, counted as each car's expected peak
/// (a few times its size). a backfill whose car would go past it waits for others to
/// finish, except the oldest one running, which always goes ahead so big repos can't
/// block each other. set via `HYDRANT_BACKFILL_MEMORY_BYTES`.
pub backfill_memory_bytes: usize,
/// records a backfill writes per step on its repo's indexer shard, so a big repo doesn't
/// hold the shard or the commit sequencer for its whole length. 0 writes each repo in one
/// step. set via `HYDRANT_BACKFILL_CHUNK_RECORDS`.
pub backfill_chunk_records: usize,
/// how big a whole car may get in memory before its backfill moves it to a file, so a big
/// repo holds its index and one step instead of every block. 0 spills every car.
/// set via `HYDRANT_BACKFILL_SPILL_BYTES`.
pub backfill_spill_bytes: usize,
/// where spilled cars go, `spill` in the database directory unless set. spills a crash left
/// there are removed at start.
/// set via `HYDRANT_BACKFILL_SPILL_DIR`.
pub backfill_spill_dir: Option<PathBuf>,
/// how many backfills may spill their car at once, each to at most `max_car_body_bytes`
/// of disk. set via `HYDRANT_BACKFILL_SPILLS`.
pub backfill_spills: usize,
/// maximum number of concurrent backfill tasks.
/// set via `HYDRANT_BACKFILL_CONCURRENCY_LIMIT`.
pub backfill_concurrency_limit: usize,
/// backfill strategy. `full` preserves existing full-repo backfill behavior.
/// `sparse-filter` attempts authenticated sparse collection backfill first and falls back
/// to full repo backfill. `auto` probes filtered repos and falls back to full repo backfill
/// for tiny MST roots.
/// set via `HYDRANT_BACKFILL_STRATEGY`.
pub backfill_strategy: BackfillStrategy,
/// unsupported, has to stay empty: indexer builds refuse any proxy, since backfill fetches
/// need direct sockets, and relay builds have no backfill. set via
/// `HYDRANT_BACKFILL_PROXIES` (comma-separated).
pub backfill_proxies: Vec<Url>,
/// max requests in flight to one PDS, shared by backfill, crawler checks and PDS listings.
/// it also caps backfill tasks per PDS. set via `HYDRANT_PER_PDS_CONCURRENCY`.
pub per_pds_concurrency: usize,
/// whether to run the network crawler. `None` defers to the default for the current mode.
/// set via `HYDRANT_ENABLE_CRAWLER`.
pub enable_crawler: Option<bool>,
/// maximum number of discovered repos retained as persistent backfill lookahead before the
/// crawler pauses. backfill task concurrency remains bounded separately.
/// set via `HYDRANT_CRAWLER_MAX_PENDING_REPOS`.
pub crawler_max_pending_repos: usize,
/// pending queue size at which the crawler resumes after being paused.
/// set via `HYDRANT_CRAWLER_RESUME_PENDING_REPOS`.
pub crawler_resume_pending_repos: usize,
/// crawler sources: each entry pairs a URL with a discovery mode.
///
/// set via `HYDRANT_CRAWLER_URLS` as a comma-separated list of `[mode::]url` entries,
/// e.g. `list_repos::wss://bsky.network,by_collection::https://lightrail.microcosm.blue`.
/// modes are `list_repos`, `relay_first` and `by_collection`. a bare URL uses the default
/// mode, `list_repos` for full-network and `by_collection` otherwise. read from the env,
/// the default is the relay hosts as `list_repos` in full-network mode and lightrail
/// otherwise, while [`Config::full_network`] and [`Config::default`] carry fixed ones. an
/// empty string means no configured sources, ones added through the api still start.
pub crawler_sources: Vec<CrawlerSource>,
/// signature verification level for incoming commits.
/// set via `HYDRANT_VERIFY_SIGNATURES` (`full`, `backfill-only`, or `none`).
pub verify_signatures: SignatureVerification,
/// number of resolved identities to keep in the in-memory LRU cache.
/// set via `HYDRANT_IDENTITY_CACHE_SIZE`.
pub identity_cache_size: u64,
/// run MST inversion on live commits. `SignatureVerification::Full` already does that for every
/// commit, so this only matters with `backfill-only` or `none`. set via `HYDRANT_VERIFY_MST`.
pub verify_mst: bool,
/// clock drift window for future-rev rejection, in seconds.
/// a rev further ahead of our clock than this gets rejected in a commit and doesn't count as
/// newer in a listing.
/// set via `HYDRANT_REV_CLOCK_SKEW`. default: 300 (5 minutes).
pub rev_clock_skew_secs: i64,
/// NSID patterns that trigger auto-discovery in filter mode (e.g. `app.bsky.feed.post`).
/// set via `HYDRANT_FILTER_SIGNALS` as a comma-separated list.
pub filter_signals: Option<Vec<String>>,
/// NSID patterns used to filter which record collections are stored.
/// if `None`, all collections are stored. set via `HYDRANT_FILTER_COLLECTIONS`.
pub filter_collections: Option<Vec<String>>,
/// DIDs that are always skipped, regardless of mode.
/// set via `HYDRANT_FILTER_EXCLUDES` as a comma-separated list.
pub filter_excludes: Option<Vec<String>>,
/// does nothing: backlinks are on exactly when hydrant is built with the `backlinks`
/// feature. read so env files that set it keep working. set via `HYDRANT_ENABLE_BACKLINKS`.
pub enable_backlinks: bool,
/// maximum number of concurrent public getRepo CAR generations.
/// set via `HYDRANT_GET_REPO_CONCURRENCY_LIMIT`; defaults to 2.
pub get_repo_concurrency_limit: usize,
/// if `true`, rejects CAR blocks whose content does not hash to their claimed CID. without
/// it, signature verification does not cover the MST: a PDS or relay can forge nodes that
/// map a key to a different record. record bodies are hashed before they are stored either
/// way. set via `HYDRANT_VERIFY_CIDS`. default: true.
pub verify_cids: bool,
/// if `true`, hydrant fetches from loopback, private and other non-public addresses, which it
/// refuses otherwise. only for a local dev stack where every name resolves to loopback. it
/// applies to the whole process, so every instance in it shares the setting.
/// set via `HYDRANT_ALLOW_PRIVATE_NETWORKS`. default: false.
pub allow_private_networks: bool,
/// if `true`, record blocks are not stored; only the index (records, counts, events) is kept.
/// `getRecord`, `listRecords`, and `getRepo` will return errors when this is enabled.
/// event stream still functions but create/update events will not include record values.
/// only valid in indexer mode (not relay).
/// set via `HYDRANT_ONLY_INDEX_LINKS=true`.
pub only_index_links: bool,
/// maximum number of new PDS sources `com.atproto.sync.requestCrawl` may add in a single
/// UTC calendar day. seeded hosts and ones added through the management api don't count.
/// `None` means unlimited.
/// set via `HYDRANT_NEW_HOST_LIMIT`.
pub new_host_limit: Option<u64>,
/// per `repo_refetch_window`. an operator's resync doesn't count. `None` is unlimited.
/// set via `HYDRANT_REPO_REFETCH_LIMIT`.
pub repo_refetch_limit: Option<u32>,
/// shared by all repos on one pds. `trusted` hosts skip it but their repos keep
/// `repo_refetch_limit`. `None` is unlimited.
/// set via `HYDRANT_HOST_REFETCH_LIMIT`.
pub host_refetch_limit: Option<u32>,
/// the window starts at the first refetch, it doesn't slide.
/// set via `HYDRANT_REPO_REFETCH_WINDOW` (humantime duration, e.g. `1h`).
pub repo_refetch_window: Duration,
/// the window a followed host's `listRepos` rewalks are counted in, from the first page one
/// reads. the window starts there, it doesn't slide.
/// set via `HYDRANT_REWALK_WINDOW` (humantime duration, e.g. `30min`).
pub rewalk_window: Duration,
/// how many bytes of `listRepos` responses a followed host's rewalks may read per
/// `rewalk_window`, in MiB. a walk that spends it stops and resumes when the window ends.
/// set via `HYDRANT_REWALK_MB`.
pub rewalk_mb: u32,
/// how often offline firehose sources are automatically retried.
/// set via `HYDRANT_OFFLINE_HOST_RETRY_INTERVAL` (humantime duration, e.g. `30min`).
/// set to `none` to disable automatic retries.
pub offline_host_retry_interval: Option<Duration>,
/// base URL(s) of relay or aggregator services to seed firehose PDS sources from at startup and periodically.
///
/// hydrant calls `com.atproto.sync.listHosts` on each URL and adds the returned PDSes
/// as firehose sources (with `is_pds = true`). account counts from the response are
/// applied to newly-seen hosts to initialise rate-limiting immediately.
///
/// set via `HYDRANT_SEED_HOSTS` as a comma-separated list of base URLs.
pub seed_hosts: Vec<Url>,
/// how often configured seed hosts are polled for new PDS sources. `None` disables
/// polling after the startup seed. set via `HYDRANT_SEED_POLL_INTERVAL` (humantime
/// duration, e.g. `30min` or `none`).
pub seed_poll_interval: Option<Duration>,
/// named rate tier definitions for PDS rate limiting.
///
/// built-in tiers ("default" and "trusted") are always present and may be overridden.
/// set via `HYDRANT_RATE_TIERS` as a comma-separated list of `name:base/mul/hourly/daily` entries,
/// e.g. `trusted:5000/10.0/18000000/432000000,custom:100/1.0/7200000/172800000`.
///
/// built from `HYDRANT_TIER_RULES` and `HYDRANT_RATE_TIERS` at startup.
pub tier_policy: TierPolicy,
/// glob rules mapping host patterns to named rate tiers.
///
/// set via `HYDRANT_TIER_RULES` as a comma-separated list of `pattern:tiername` entries,
/// e.g. `*.bsky.network:trusted,pds.example.com:custom`. rules are evaluated in order;
/// api-assigned per-host overrides always take priority over these rules.
pub tier_rules: Vec<(String, String)>,
/// db internals, tune only if you know what you're doing.
///
/// size of the fjall block cache in MB. set via `HYDRANT_CACHE_SIZE`.
pub cache_size: u64,
/// db internals, tune only if you know what you're doing.
///
/// compression algorithm for data keyspaces (record heads, history, repos, events,
/// backlinks). saved when a keyspace is created, so it only applies to new keyspaces.
/// set via `HYDRANT_DATA_COMPRESSION` (`lz4`, `zstd`, or `none`).
pub data_compression: Compression,
/// db internals, tune only if you know what you're doing.
///
/// compression algorithm for the fjall journal.
/// set via `HYDRANT_JOURNAL_COMPRESSION` (`lz4`, `zstd`, or `none`).
pub journal_compression: Compression,
/// db internals, tune only if you know what you're doing.
///
/// number of background threads used by the fjall storage engine.
/// set via `HYDRANT_DB_WORKER_THREADS`.
pub db_worker_threads: usize,
/// db internals, tune only if you know what you're doing.
///
/// maximum total size of the fjall journal in MB before a flush is forced.
/// set via `HYDRANT_DB_MAX_JOURNALING_SIZE_MB`.
pub db_max_journaling_size_mb: u64,
/// db internals, tune only if you know what you're doing.
///
/// in-memory write buffer (memtable) size for the repos keyspace in MB.
/// set via `HYDRANT_DB_REPOS_MEMTABLE_SIZE_MB`.
pub db_repos_memtable_size_mb: u64,
/// db internals, tune only if you know what you're doing.
///
/// in-memory write buffer (memtable) size for the events keyspace in MB.
/// set via `HYDRANT_DB_EVENTS_MEMTABLE_SIZE_MB`.
pub db_events_memtable_size_mb: u64,
/// db internals, tune only if you know what you're doing.
///
/// in-memory write buffer (memtable) size for the records keyspace in MB.
/// set via `HYDRANT_DB_RECORDS_MEMTABLE_SIZE_MB`.
pub db_records_memtable_size_mb: u64,
/// db internals, tune only if you know what you're doing.
///
/// whether the bottom level of the record heads keyspace gets a bloom filter too, the
/// upper levels always have one. set via `HYDRANT_DB_RECORDS_BLOOM_FILTERS`.
pub db_records_bloom_filters: bool,
/// replay batch size.
///
/// `0` (the default) means auto: use about half the subscriber channel capacity
/// (`STREAM_CHANNEL_CAPACITY / 2`), capped to the channel's current available
/// capacity. this prevents DB reads when the output buffer is already saturated.
///
/// set to a non-zero value to fix the batch size manually.
///
/// set via `HYDRANT_STREAM_REPLAY_CHUNK_SIZE`.
pub stream_replay_chunk_size: usize,
/// set via `HYDRANT_STREAM_CHUNK_SPAN`.
pub stream_chunk_span: u64,
/// the streams whose websocket readers have their cursor replay capped. in-process
/// subscribers are never capped, and neither is anyone's live tail.
/// set via `HYDRANT_THROTTLED_STREAMS`, comma separated, empty for none.
pub throttled_streams: Vec<StreamKind>,
/// how many times what keeping up with the live stream would cost a reader its capped
/// replay may use, in bytes sent (rows its filters drop count a few bytes each).
/// set via `HYDRANT_STREAM_REPLAY_SPEED`.
pub stream_replay_speed: u32,
/// bytes a second a capped replay may always use, however quiet the live stream is.
/// set via `HYDRANT_STREAM_REPLAY_FLOOR_BYTES`.
pub stream_replay_floor_bytes: u64,
/// maximum number of live stream events buffered in memory per subscriber while it waits
/// to send, before it falls back to catching up from the database.
/// set via `HYDRANT_STREAM_PENDING_EVENT_LIMIT`.
pub stream_pending_event_limit: usize,
/// maximum time a subscriber may block stream delivery before being disconnected.
/// set via `HYDRANT_STREAM_SEND_TIMEOUT` (humantime duration, e.g. `30sec`).
pub stream_send_timeout: Duration,
/// timeout for database maintenance operations (compaction, training).
/// set via `HYDRANT_MAINTENANCE_TIMEOUT` (humantime duration, e.g. `6h`, `0s` or `0` to disable).
pub maintenance_timeout: Duration,
}
impl Default for Config {
fn default() -> Self {
const BASE_MEMTABLE_MB: u64 = 32;
Self {
database_path: PathBuf::from("./hydrant.db"),
ephemeral: mode_defaults::EPHEMERAL,
ephemeral_ttl: mode_defaults::EPHEMERAL_TTL,
history_ttl: None,
full_network: mode_defaults::FULL_NETWORK,
relays: mode_defaults::relays(),
seed_hosts: mode_defaults::seed_hosts(),
seed_poll_interval: Some(Duration::from_secs(30 * 60)),
plc_urls: vec![Url::parse("https://plc.wtf").unwrap()],
enable_firehose: true,
enable_backfill: true,
firehose_workers: 8,
firehose_max_failures: 15,
cursor_save_interval: Duration::from_secs(3),
repo_fetch_timeout: Duration::from_secs(300),
// 2 GiB safety ceiling: never rejects a legitimate repository, only a runaway or
// malicious response. memory-constrained/high-concurrency hosts should lower it.
max_car_body_bytes: 2 * 1024 * 1024 * 1024,
// a few repos the size of the biggest accounts at once, while the usual repo of a
// few hundred kilobytes never notices it, and one oversized repo still runs alone
backfill_memory_bytes: 4 * 1024 * 1024 * 1024,
// a step of about 4000 batch items, which takes ~10-30 ms on one slow core, so live
// events on the same shard wait at most that long behind a big repo
backfill_chunk_records: 1000,
// a whole car peaks at about 6x its size in memory, so a 32 MiB car is ~200 MiB,
// the most one backfill should hold on a 512 MiB box. spilled, the same car takes
// ~25 MiB, and nearly every repo is far smaller and never spills
backfill_spill_bytes: 32 * 1024 * 1024,
backfill_spill_dir: None,
backfill_spills: 4,
backfill_concurrency_limit: 16,
backfill_strategy: BackfillStrategy::Auto,
backfill_proxies: vec![],
per_pds_concurrency: crate::util::throttle::DEFAULT_PER_PDS_CONCURRENCY,
enable_crawler: None,
crawler_max_pending_repos: 2000,
crawler_resume_pending_repos: 1000,
crawler_sources: vec![CrawlerSource {
url: Url::parse("https://lightrail.microcosm.blue").unwrap(),
mode: CrawlerMode::ByCollection,
}],
verify_signatures: SignatureVerification::Full,
identity_cache_size: 1_000_000,
verify_mst: false,
rev_clock_skew_secs: 300,
filter_signals: None,
filter_collections: None,
filter_excludes: None,
enable_backlinks: false,
get_repo_concurrency_limit: 2,
verify_cids: true,
allow_private_networks: false,
only_index_links: false,
new_host_limit: Some(50),
repo_refetch_limit: Some(6),
host_refetch_limit: Some(60),
repo_refetch_window: Duration::from_secs(60 * 60),
rewalk_window: Duration::from_secs(30 * 60),
rewalk_mb: 8,
offline_host_retry_interval: Some(Duration::from_secs(30 * 60)),
tier_rules: vec![],
tier_policy: {
let mut tiers = HashMap::new();
tiers.insert(SmolStr::new("default"), DEFAULT_RATETIER);
tiers.insert(SmolStr::new("trusted"), RateTier::trusted());
TierPolicy {
tiers,
rules: vec![],
}
},
cache_size: 128,
data_compression: Compression::Zstd,
journal_compression: Compression::Lz4,
db_worker_threads: 4,
db_max_journaling_size_mb: 400,
db_repos_memtable_size_mb: BASE_MEMTABLE_MB / 2,
db_events_memtable_size_mb: BASE_MEMTABLE_MB,
// smaller and a sync spends a lot more cpu compacting, see docs/configuration.md
db_records_memtable_size_mb: BASE_MEMTABLE_MB,
db_records_bloom_filters: false,
stream_replay_chunk_size: 0,
stream_chunk_span: 0,
throttled_streams: DEFAULT_THROTTLED_STREAMS.to_vec(),
stream_replay_speed: 10,
// about 80 Mbit/s, or some 15k jetstream events a second
stream_replay_floor_bytes: 10_000_000,
stream_pending_event_limit: 4096,
stream_send_timeout: Duration::from_secs(30),
maintenance_timeout: Duration::from_secs(6 * 60 * 60),
}
}
}
impl Config {
/// returns the default config for full network usage.
pub fn full_network() -> Self {
const BASE_MEMTABLE_MB: u64 = 192;
Self {
full_network: true,
plc_urls: vec![Url::parse("https://plc.directory").unwrap()],
firehose_workers: 24,
backfill_concurrency_limit: 64,
crawler_max_pending_repos: 25_000,
crawler_resume_pending_repos: 12_500,
crawler_sources: vec![CrawlerSource {
url: Url::parse("wss://relay.fire.hose.cam/").unwrap(),
mode: CrawlerMode::ListRepos,
}],
db_worker_threads: 8,
db_max_journaling_size_mb: 1024,
db_repos_memtable_size_mb: BASE_MEMTABLE_MB / 4,
db_events_memtable_size_mb: BASE_MEMTABLE_MB,
db_records_memtable_size_mb: BASE_MEMTABLE_MB / 3 * 2,
..Self::default()
}
}
/// validates cross-field invariants for this configuration.
pub fn validate(&self) -> Result<()> {
self.validate_ranges()?;
#[cfg(all(feature = "indexer", not(feature = "indexer_stream")))]
if self.ephemeral {
return Err(miette::miette!(
"HYDRANT_EPHEMERAL requires the indexer_stream feature because events are its only record-body storage"
));
}
if self.ephemeral && self.only_index_links {
return Err(miette::miette!(
"HYDRANT_EPHEMERAL and HYDRANT_ONLY_INDEX_LINKS are incompatible: ephemeral events must own their body bytes"
));
}
if self.ephemeral && self.history_ttl.is_some() {
return Err(miette::miette!(
"HYDRANT_HISTORY_TTL is not supported with HYDRANT_EPHEMERAL because ephemeral databases store no history"
));
}
// backlinks hooks are compiled directly into every record mutation;
// `enable_backlinks` does not disable them at runtime. without its own
// retention machinery an ephemeral window would leak stale entries.
#[cfg(feature = "backlinks")]
if self.ephemeral {
return Err(miette::miette!(
"the backlinks feature is not supported with HYDRANT_EPHEMERAL / ephemeral"
));
}
#[cfg(feature = "indexer")]
if !self.backfill_proxies.is_empty() {
return Err(miette::miette!(
"HYDRANT_BACKFILL_PROXIES is unsupported: public backfill egress requires direct sockets"
));
}
#[cfg(feature = "relay")]
if self.only_index_links {
return Err(miette::miette!(
"HYDRANT_ONLY_INDEX_LINKS is not supported in relay mode"
));
}
Ok(())
}
/// values that parse but would crash hydrant or spin a loop
fn validate_ranges(&self) -> Result<()> {
if self.firehose_workers == 0 {
miette::bail!("invalid HYDRANT_FIREHOSE_WORKERS `0`: want at least 1 firehose shard");
}
if self.cursor_save_interval.is_zero() {
miette::bail!(
"invalid HYDRANT_CURSOR_SAVE_INTERVAL `0s`: the save thread would spin, want a duration like 3s"
);
}
if self.seed_poll_interval.is_some_and(|d| d.is_zero()) {
miette::bail!(
help = "set it to none to only seed at startup",
"invalid HYDRANT_SEED_POLL_INTERVAL `0s`: want a duration like 30min"
);
}
if self
.offline_host_retry_interval
.is_some_and(|d| d.is_zero())
{
miette::bail!(
help = "set it to none to never retry offline hosts",
"invalid HYDRANT_OFFLINE_HOST_RETRY_INTERVAL `0s`: the retry loop would spin, want a duration like 30min"
);
}
if self.repo_refetch_limit == Some(0) {
miette::bail!(
help = "set it to none to refetch without limit",
"invalid HYDRANT_REPO_REFETCH_LIMIT `0`: no repo could ever be refetched again, want at least 1"
);
}
if self.host_refetch_limit == Some(0) {
miette::bail!(
help = "set it to none to refetch without limit",
"invalid HYDRANT_HOST_REFETCH_LIMIT `0`: no repo on a host could ever be refetched again, want at least 1"
);
}
if self.repo_refetch_window.is_zero() {
miette::bail!(
"invalid HYDRANT_REPO_REFETCH_WINDOW `0s`: no refetch could be counted, want a duration like 1h"
);
}
if self.rewalk_window.is_zero() {
miette::bail!(
"invalid HYDRANT_REWALK_WINDOW `0s`: no rewalk could be counted, want a duration like 30min"
);
}
if self.rewalk_mb == 0 {
miette::bail!(
"invalid HYDRANT_REWALK_MB `0`: no rewalk could read a page, want at least 1"
);
}
if self.db_worker_threads == 0 {
miette::bail!(
"invalid HYDRANT_DB_WORKER_THREADS `0`: the database needs at least 1 thread to flush"
);
}
if self.db_max_journaling_size_mb < 64 {
miette::bail!(
"invalid HYDRANT_DB_MAX_JOURNALING_SIZE_MB `{}`: the storage engine wants at least 64",
self.db_max_journaling_size_mb
);
}
if self.db_repos_memtable_size_mb < 2 {
miette::bail!(
"invalid HYDRANT_DB_REPOS_MEMTABLE_SIZE_MB `{}`: want at least 2, repo_metadata gets half of it",
self.db_repos_memtable_size_mb
);
}
for (key, mb) in [
(
"HYDRANT_DB_EVENTS_MEMTABLE_SIZE_MB",
self.db_events_memtable_size_mb,
),
(
"HYDRANT_DB_RECORDS_MEMTABLE_SIZE_MB",
self.db_records_memtable_size_mb,
),
] {
if mb == 0 {
miette::bail!("invalid {key} `0`: want at least 1");
}
}
#[cfg(feature = "indexer")]
if self.backfill_concurrency_limit == 0 {
miette::bail!(
help = "set HYDRANT_ENABLE_BACKFILL=false to turn backfill off",
"invalid HYDRANT_BACKFILL_CONCURRENCY_LIMIT `0`: want at least 1"
);
}
if self.stream_replay_speed == 0 {
miette::bail!(
help = "leave a stream out of HYDRANT_THROTTLED_STREAMS to not cap its replays",
"invalid HYDRANT_STREAM_REPLAY_SPEED `0`: want at least 1"
);
}
if self.stream_replay_floor_bytes == 0 {
miette::bail!(
help = "leave a stream out of HYDRANT_THROTTLED_STREAMS to not cap its replays",
"invalid HYDRANT_STREAM_REPLAY_FLOOR_BYTES `0`: a replay on a quiet box would never move, want at least 1"
);
}
#[cfg(feature = "indexer")]
if self.backfill_spills == 0 {
miette::bail!(
help = "raise HYDRANT_BACKFILL_SPILL_BYTES to keep bigger cars in memory instead",
"invalid HYDRANT_BACKFILL_SPILLS `0`: a car past the spill size would wait for a spill slot forever, want at least 1"
);
}
Ok(())
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::time::Duration;
#[test]
fn full_network_keeps_a_large_bounded_discovery_window() {
let config = Config::full_network();
assert_eq!(config.crawler_max_pending_repos, 25_000);
assert_eq!(config.crawler_resume_pending_repos, 12_500);
assert!(config.crawler_resume_pending_repos < config.crawler_max_pending_repos);
}
#[test]
fn default_config_is_valid() {
let config = Config::default();
assert!(config.validate().is_ok());
}
#[test]
#[cfg(all(feature = "backlinks", feature = "indexer_stream"))]
fn ephemeral_rejected_when_backlinks_feature_enabled() {
let mut config = Config::default();
config.ephemeral = true;
config.enable_backlinks = false;
let res = config.validate();
assert!(res.is_err());
let err = res.unwrap_err().to_string();
assert!(err.contains("HYDRANT_EPHEMERAL") || err.contains("ephemeral"));
assert!(err.contains("backlinks feature"));
config.enable_backlinks = true;
assert!(config.validate().is_err());
config.ephemeral = false;
assert!(config.validate().is_ok());
}
#[test]
#[cfg(all(feature = "indexer_stream", not(feature = "backlinks")))]
fn ephemeral_allowed_when_backlinks_feature_disabled() {
let mut config = Config::default();
config.ephemeral = true;
assert!(config.validate().is_ok());
}
#[test]
#[cfg(all(feature = "indexer", not(feature = "indexer_stream")))]
fn ephemeral_requires_stream_storage() {
let mut config = Config::default();
config.ephemeral = true;
let err = config.validate().unwrap_err().to_string();
assert!(err.contains("indexer_stream"));
}
#[test]
#[cfg(feature = "indexer_stream")]
fn ephemeral_cannot_discard_its_only_body_storage() {
let mut config = Config::default();
config.ephemeral = true;
config.only_index_links = true;
let err = config.validate().unwrap_err().to_string();
assert!(err.contains("HYDRANT_ONLY_INDEX_LINKS"));
}
#[test]
#[cfg(feature = "indexer_stream")]
fn ephemeral_rejects_meaningless_history_retention() {
let mut config = Config::default();
config.ephemeral = true;
config.history_ttl = Some(Duration::from_secs(60));
let err = config.validate().unwrap_err().to_string();
assert!(err.contains("HYDRANT_HISTORY_TTL"));
}
#[test]
fn display_reports_history_ttl() {
let mut config = Config::default();
config.history_ttl = Some(Duration::from_secs(900));
let display_str = format!("{config}");
assert!(display_str.contains("history ttl"));
assert!(display_str.contains("15m"));
config.history_ttl = None;
let display_str_non_ephemeral = format!("{config}");
assert!(!display_str_non_ephemeral.contains("history ttl"));
}
}
macro_rules! config_line {
($f:expr, $label:expr, $value:expr) => {
writeln!($f, " {:<width$}{}", $label, $value, width = LABEL_WIDTH)
};
}
impl fmt::Display for Config {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
const LABEL_WIDTH: usize = 27;
writeln!(f, "hydrant configuration:")?;
config_line!(
f,
"relay hosts",
format_args!(
"{:?}",
self.relays
.iter()
.map(|s| if s.is_pds {
format!("pds::{}", s.url)
} else {
s.url.to_string()
})
.collect::<Vec<_>>()
)
)?;
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, "verify cids", self.verify_cids)?;
if self.allow_private_networks {
config_line!(f, "allow private networks", "true")?;
}
config_line!(f, "per-pds concurrency", self.per_pds_concurrency)?;
config_line!(f, "backfill enabled", self.enable_backfill)?;
config_line!(f, "firehose enabled", self.enable_firehose)?;
config_line!(f, "backfill concurrency", self.backfill_concurrency_limit)?;
config_line!(f, "backfill strategy", self.backfill_strategy)?;
config_line!(f, "backfill proxies", self.backfill_proxies.len())?;
config_line!(f, "identity cache size", self.identity_cache_size)?;
config_line!(
f,
"cursor save interval",
format_args!("{}sec", self.cursor_save_interval.as_secs())
)?;
config_line!(
f,
"repo fetch timeout",
format_args!("{}sec", self.repo_fetch_timeout.as_secs())
)?;
config_line!(
f,
"max car body",
format_args!("{} mb", self.max_car_body_bytes / (1024 * 1024))
)?;
config_line!(
f,
"backfill memory",
format_args!("{} mb", self.backfill_memory_bytes / (1024 * 1024))
)?;
config_line!(f, "backfill chunk records", self.backfill_chunk_records)?;
config_line!(
f,
"backfill spill",
format_args!(
"{} mb, {} at once",
self.backfill_spill_bytes / (1024 * 1024),
self.backfill_spills
)
)?;
config_line!(f, "ephemeral", self.ephemeral)?;
if let Some(ttl) = self.history_ttl {
config_line!(f, "history ttl", crate::util::format_duration(ttl))?;
}
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, "data compression", self.data_compression)?;
config_line!(f, "journal compression", self.journal_compression)?;
config_line!(f, "firehose workers", self.firehose_workers)?;
config_line!(f, "firehose max failures", self.firehose_max_failures)?;
config_line!(f, "db worker threads", self.db_worker_threads)?;
config_line!(
f,
"db journal size",
format_args!("{} mb", self.db_max_journaling_size_mb)
)?;
config_line!(
f,
"db repos memtable",
format_args!("{} mb", self.db_repos_memtable_size_mb)
)?;
config_line!(
f,
"db events memtable",
format_args!("{} mb", self.db_events_memtable_size_mb)
)?;
config_line!(
f,
"db records memtable",
format_args!("{} mb", self.db_records_memtable_size_mb)
)?;
config_line!(f, "db records bloom filters", self.db_records_bloom_filters)?;
let replay_chunk = if self.stream_replay_chunk_size == 0 {
"auto".to_owned()
} else {
self.stream_replay_chunk_size.to_string()
};
config_line!(f, "stream replay chunk", replay_chunk)?;
if self.stream_chunk_span == 0 {
config_line!(f, "stream chunks", "off")?;
} else {
config_line!(f, "stream chunk span", self.stream_chunk_span)?;
}
if self.throttled_streams.is_empty() {
config_line!(f, "throttled streams", "none")?;
} else {
let streams: Vec<_> = self
.throttled_streams
.iter()
.map(|s| s.to_string())
.collect();
config_line!(f, "throttled streams", streams.join(", "))?;
}
config_line!(
f,
"stream replay speed",
format_args!(
"{}x live, at least {} bytes/sec",
self.stream_replay_speed, self.stream_replay_floor_bytes
)
)?;
config_line!(f, "stream pending limit", self.stream_pending_event_limit)?;
config_line!(
f,
"stream send timeout",
format_args!("{}sec", self.stream_send_timeout.as_secs())
)?;
config_line!(
f,
"maintenance timeout",
format_args!("{}sec", self.maintenance_timeout.as_secs())
)?;
config_line!(f, "crawler max pending", self.crawler_max_pending_repos)?;
config_line!(
f,
"crawler resume pending",
self.crawler_resume_pending_repos
)?;
if !self.crawler_sources.is_empty() {
let sources: Vec<_> = self
.crawler_sources
.iter()
.map(|s| format!("{}::{}", s.mode, s.url))
.collect();
config_line!(f, "crawler sources", sources.join(", "))?;
}
if let Some(signals) = &self.filter_signals {
config_line!(f, "filter signals", format_args!("{:?}", signals))?;
}
if let Some(collections) = &self.filter_collections {
config_line!(f, "filter collections", format_args!("{:?}", collections))?;
}
if let Some(excludes) = &self.filter_excludes {
config_line!(f, "filter excludes", format_args!("{:?}", excludes))?;
}
if self.only_index_links {
config_line!(f, "only index links", "true")?;
}
if !self.seed_hosts.is_empty() {
config_line!(
f,
"seed hosts",
format_args!(
"{:?}",
self.seed_hosts
.iter()
.map(|u| u.as_str())
.collect::<Vec<_>>()
)
)?;
}
match self.seed_poll_interval {
Some(d) => config_line!(f, "seed poll interval", format_args!("{}sec", d.as_secs()))?,
None => config_line!(f, "seed poll interval", "disabled")?,
}
if let Some(limit) = self.new_host_limit {
config_line!(f, "max pds/day", limit)?;
}
match self.repo_refetch_limit {
Some(limit) => config_line!(
f,
"repo refetches",
format_args!("{limit}/{}sec", self.repo_refetch_window.as_secs())
)?,
None => config_line!(f, "repo refetches", "unlimited")?,
}
match self.host_refetch_limit {
Some(limit) => config_line!(
f,
"host refetches",
format_args!("{limit}/{}sec", self.repo_refetch_window.as_secs())
)?,
None => config_line!(f, "host refetches", "unlimited")?,
}
match self.offline_host_retry_interval {
Some(d) => config_line!(
f,
"offline retry interval",
format_args!("{}sec", d.as_secs())
)?,
None => config_line!(f, "offline retry interval", "disabled")?,
}
Ok(())
}
}