use miette::Result;
use smol_str::SmolStr;
use std::path::PathBuf;
use std::time::Duration;
use url::Url;
use super::Config;
use super::types::{
BackfillStrategy, Compression, CrawlerMode, CrawlerSource, FirehoseSource, RateTier,
SignatureVerification, StreamKind,
};
use crate::pds_meta::{TierPolicy, TierRule};
/// this is for internal use only, please don't use this macro.
#[doc(hidden)]
#[macro_export]
macro_rules! __cfg {
($lookup:ident, $key:expr, $default:expr) => {
$crate::config::env::read(&$lookup, concat!("HYDRANT_", $key), $default.to_owned())?
};
}
use crate::__cfg as cfg;
/// a type one `HYDRANT_*` variable can hold.
#[doc(hidden)]
pub trait EnvValue: Sized {
/// what a valid value looks like, for the error
const EXPECTED: &'static str;
fn parse_env(s: &str) -> Option<Self>;
}
impl EnvValue for bool {
const EXPECTED: &'static str = "true, false, 1 or 0";
fn parse_env(s: &str) -> Option<Self> {
match s {
"true" | "1" => Some(true),
"false" | "0" => Some(false),
_ => None,
}
}
}
macro_rules! whole_number {
($($t:ty),*) => {$(
impl EnvValue for $t {
const EXPECTED: &'static str = "a whole number";
fn parse_env(s: &str) -> Option<Self> {
s.parse().ok()
}
}
)*};
}
whole_number!(u16, u32, u64, usize, i64);
impl EnvValue for Duration {
const EXPECTED: &'static str = "a duration with a unit, like 30s or 5min";
fn parse_env(s: &str) -> Option<Self> {
crate::util::parse_duration(s)
}
}
impl EnvValue for PathBuf {
const EXPECTED: &'static str = "a path";
fn parse_env(s: &str) -> Option<Self> {
Some(PathBuf::from(s))
}
}
impl EnvValue for Compression {
const EXPECTED: &'static str = "zstd, lz4 or none";
fn parse_env(s: &str) -> Option<Self> {
s.parse().ok()
}
}
impl EnvValue for SignatureVerification {
const EXPECTED: &'static str = "full, backfill-only or none";
fn parse_env(s: &str) -> Option<Self> {
s.parse().ok()
}
}
impl EnvValue for BackfillStrategy {
const EXPECTED: &'static str = "auto, full or sparse-filter";
fn parse_env(s: &str) -> Option<Self> {
s.parse().ok()
}
}
fn invalid(key: &str, raw: &str, expected: &str) -> miette::Report {
miette::miette!("invalid {key} `{raw}`: want {expected}")
}
/// reads `key`, `None` when it's unset or empty. a value that doesn't parse is an error,
/// since quietly running on the default hides the typo until something goes wrong
fn read_optional<T: EnvValue>(
lookup: &impl Fn(&str) -> Option<String>,
key: &str,
) -> Result<Option<T>> {
let Some(raw) = lookup(key) else {
return Ok(None);
};
let value = raw.trim();
if value.is_empty() {
return Ok(None);
}
T::parse_env(value)
.map(Some)
.ok_or_else(|| invalid(key, &raw, T::EXPECTED))
}
#[doc(hidden)]
pub fn read<T: EnvValue>(
lookup: &impl Fn(&str) -> Option<String>,
key: &str,
default: T,
) -> Result<T> {
Ok(read_optional(lookup, key)?.unwrap_or(default))
}
/// like [`read`], but `none` (any case) turns it off
fn read_or_none<T: EnvValue>(
lookup: &impl Fn(&str) -> Option<String>,
key: &str,
default: Option<T>,
) -> Result<Option<T>> {
match lookup(key) {
Some(raw) if raw.trim().eq_ignore_ascii_case("none") => Ok(None),
_ => Ok(read_optional(lookup, key)?.or(default)),
}
}
/// a comma separated list of streams. set but empty means none, unlike other knobs where
/// empty is unset, because turning every cap off is a real choice
fn parse_streams(raw: &str) -> Result<Vec<StreamKind>> {
let mut streams = Vec::new();
for name in raw
.split(',')
.map(str::trim)
.filter(|name| !name.is_empty())
{
let stream = name.parse().map_err(|_| {
invalid(
"HYDRANT_THROTTLED_STREAMS",
raw,
"a comma separated list of indexer, jetstream and relay",
)
})?;
if !streams.contains(&stream) {
streams.push(stream);
}
}
Ok(streams)
}
fn dotenv_assignments(contents: &str) -> impl Iterator<Item = (&str, &str)> {
contents.lines().filter_map(|line| {
let line = line.trim();
if line.is_empty() || line.starts_with('#') {
return None;
}
let (key, val) = line.split_once('=')?;
let val = val.trim();
let val = val
.strip_prefix('"')
.and_then(|v| v.strip_suffix('"'))
.or_else(|| val.strip_prefix('\'').and_then(|v| v.strip_suffix('\'')))
.unwrap_or(val);
Some((key.trim(), val))
})
}
/// loads `.env` from the current directory, setting any variables not already in the
/// environment.
pub fn load_dotenv() {
let Ok(contents) = std::fs::read_to_string(".env") else {
return;
};
for (key, val) in dotenv_assignments(&contents) {
if std::env::var(key).is_err() {
// SAFETY: single-threaded at startup; no other threads are reading env yet.
unsafe { std::env::set_var(key, val) };
}
}
}
impl Config {
/// reads and builds the config from environment variables, loading `.env` first if present.
pub fn from_env() -> Result<Self> {
load_dotenv();
Self::from_lookup(|key| std::env::var(key).ok())
}
/// builds the config from `HYDRANT_*` keys resolved by `lookup` instead of the process
/// environment, so an embedder can configure hydrant without touching its own env or a `.env`.
pub fn from_lookup(lookup: impl Fn(&str) -> Option<String>) -> Result<Self> {
// full_network is read first since it determines which defaults to use.
// relay mode defaults to true so that the network is indexed by default.
#[cfg(feature = "relay")]
let default_full_network = true;
#[cfg(not(feature = "relay"))]
let default_full_network = false;
let full_network: bool = cfg!(lookup, "FULL_NETWORK", default_full_network);
let defaults = full_network
.then(Self::full_network)
.unwrap_or_else(Self::default);
let relay_hosts = match lookup("HYDRANT_RELAY_HOSTS") {
Some(hosts) if !hosts.trim().is_empty() => hosts
.split(',')
.filter_map(|s| {
let s = s.trim();
(!s.is_empty())
.then(|| {
FirehoseSource::parse(s).or_else(|| {
tracing::warn!("invalid relay host URL: {s}");
None
})
})
.flatten()
})
.collect(),
// HYDRANT_RELAY_HOSTS explicitly set to ""
Some(_) => vec![],
// not set at all, fall back to RELAY_HOST
None => match lookup("HYDRANT_RELAY_HOST") {
Some(s) if !s.trim().is_empty() => {
let source = FirehoseSource::parse(s.trim()).ok_or_else(|| {
invalid(
"HYDRANT_RELAY_HOST",
&s,
"a url, optionally with a pds:: prefix",
)
})?;
vec![source]
}
_ => defaults.relays.clone(),
},
};
let plc_urls: Vec<Url> = lookup("HYDRANT_PLC_URL")
.map(|s| {
s.split(',')
.map(|s| {
Url::parse(s.trim())
.map_err(|e| invalid("HYDRANT_PLC_URL", s, &format!("a url ({e})")))
})
.collect::<Result<Vec<_>>>()
})
.unwrap_or_else(|| Ok(defaults.plc_urls.clone()))?;
let cursor_save_interval = cfg!(
lookup,
"CURSOR_SAVE_INTERVAL",
defaults.cursor_save_interval
);
let repo_fetch_timeout = cfg!(lookup, "REPO_FETCH_TIMEOUT", defaults.repo_fetch_timeout);
let maintenance_timeout = cfg!(lookup, "MAINTENANCE_TIMEOUT", defaults.maintenance_timeout);
let max_car_body_bytes = cfg!(lookup, "MAX_CAR_BODY_BYTES", defaults.max_car_body_bytes);
let backfill_memory_bytes = cfg!(
lookup,
"BACKFILL_MEMORY_BYTES",
defaults.backfill_memory_bytes
);
let backfill_chunk_records = cfg!(
lookup,
"BACKFILL_CHUNK_RECORDS",
defaults.backfill_chunk_records
);
let backfill_spill_bytes = cfg!(
lookup,
"BACKFILL_SPILL_BYTES",
defaults.backfill_spill_bytes
);
let backfill_spill_dir = read_optional(&lookup, "HYDRANT_BACKFILL_SPILL_DIR")?;
let backfill_spills = cfg!(lookup, "BACKFILL_SPILLS", defaults.backfill_spills);
let ephemeral: bool = cfg!(lookup, "EPHEMERAL", defaults.ephemeral);
let ephemeral_ttl = cfg!(lookup, "EPHEMERAL_TTL", defaults.ephemeral_ttl);
let history_ttl: Option<Duration> = read_optional(&lookup, "HYDRANT_HISTORY_TTL")?;
let database_path = cfg!(lookup, "DATABASE_PATH", defaults.database_path);
let cache_size = cfg!(lookup, "CACHE_SIZE", defaults.cache_size);
let data_compression = cfg!(lookup, "DATA_COMPRESSION", defaults.data_compression);
let journal_compression = cfg!(lookup, "JOURNAL_COMPRESSION", defaults.journal_compression);
let verify_signatures = cfg!(lookup, "VERIFY_SIGNATURES", defaults.verify_signatures);
let identity_cache_size = cfg!(lookup, "IDENTITY_CACHE_SIZE", defaults.identity_cache_size);
let verify_mst: bool = cfg!(lookup, "VERIFY_MST", defaults.verify_mst);
let rev_clock_skew_secs: i64 = cfg!(lookup, "REV_CLOCK_SKEW", defaults.rev_clock_skew_secs);
let enable_firehose = cfg!(lookup, "ENABLE_FIREHOSE", defaults.enable_firehose);
let enable_backfill = cfg!(lookup, "ENABLE_BACKFILL", defaults.enable_backfill);
let enable_crawler = read_optional(&lookup, "HYDRANT_ENABLE_CRAWLER")?;
let backfill_concurrency_limit = cfg!(
lookup,
"BACKFILL_CONCURRENCY_LIMIT",
defaults.backfill_concurrency_limit
);
let backfill_strategy = cfg!(lookup, "BACKFILL_STRATEGY", defaults.backfill_strategy);
// comma-separated proxy URLs to spread backfill fetches across egress IPs.
let backfill_proxies: Vec<Url> = lookup("HYDRANT_BACKFILL_PROXIES")
.map(|s| {
s.split(',')
.filter_map(|u| {
let u = u.trim();
if u.is_empty() {
return None;
}
Url::parse(u)
.or_else(|_| Url::parse(&format!("http://{u}")))
.ok()
.or_else(|| {
tracing::warn!("invalid backfill proxy URL: {u}");
None
})
})
.collect()
})
.unwrap_or_else(|| defaults.backfill_proxies.clone());
let per_pds_concurrency = cfg!(lookup, "PER_PDS_CONCURRENCY", defaults.per_pds_concurrency);
let firehose_workers = cfg!(lookup, "FIREHOSE_WORKERS", defaults.firehose_workers);
let firehose_max_failures = cfg!(
lookup,
"FIREHOSE_MAX_FAILURES",
defaults.firehose_max_failures
);
let db_worker_threads = cfg!(lookup, "DB_WORKER_THREADS", defaults.db_worker_threads);
let db_max_journaling_size_mb = cfg!(
lookup,
"DB_MAX_JOURNALING_SIZE_MB",
defaults.db_max_journaling_size_mb
);
let db_events_memtable_size_mb = cfg!(
lookup,
"DB_EVENTS_MEMTABLE_SIZE_MB",
defaults.db_events_memtable_size_mb
);
let db_records_memtable_size_mb = cfg!(
lookup,
"DB_RECORDS_MEMTABLE_SIZE_MB",
defaults.db_records_memtable_size_mb
);
let db_records_bloom_filters: bool = cfg!(
lookup,
"DB_RECORDS_BLOOM_FILTERS",
defaults.db_records_bloom_filters
);
let db_repos_memtable_size_mb = cfg!(
lookup,
"DB_REPOS_MEMTABLE_SIZE_MB",
defaults.db_repos_memtable_size_mb
);
let stream_replay_chunk_size = cfg!(
lookup,
"STREAM_REPLAY_CHUNK_SIZE",
defaults.stream_replay_chunk_size
);
let stream_chunk_span = cfg!(lookup, "STREAM_CHUNK_SPAN", defaults.stream_chunk_span);
let throttled_streams = match lookup("HYDRANT_THROTTLED_STREAMS") {
Some(raw) => parse_streams(&raw)?,
None => defaults.throttled_streams.clone(),
};
let stream_replay_speed = cfg!(lookup, "STREAM_REPLAY_SPEED", defaults.stream_replay_speed);
let stream_replay_floor_bytes = cfg!(
lookup,
"STREAM_REPLAY_FLOOR_BYTES",
defaults.stream_replay_floor_bytes
);
let stream_pending_event_limit = cfg!(
lookup,
"STREAM_PENDING_EVENT_LIMIT",
defaults.stream_pending_event_limit
);
let stream_send_timeout = cfg!(lookup, "STREAM_SEND_TIMEOUT", defaults.stream_send_timeout);
let crawler_max_pending_repos = cfg!(
lookup,
"CRAWLER_MAX_PENDING_REPOS",
defaults.crawler_max_pending_repos
);
let crawler_resume_pending_repos = cfg!(
lookup,
"CRAWLER_RESUME_PENDING_REPOS",
defaults.crawler_resume_pending_repos
);
let filter_signals = lookup("HYDRANT_FILTER_SIGNALS").map(|s| {
s.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect()
});
let filter_collections = lookup("HYDRANT_FILTER_COLLECTIONS").map(|s| {
s.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect()
});
let filter_excludes = lookup("HYDRANT_FILTER_EXCLUDES").map(|s| {
s.split(',')
.map(|s| s.trim().to_string())
.filter(|s| !s.is_empty())
.collect()
});
let enable_backlinks: bool = cfg!(lookup, "ENABLE_BACKLINKS", defaults.enable_backlinks);
let get_repo_concurrency_limit = cfg!(
lookup,
"GET_REPO_CONCURRENCY_LIMIT",
defaults.get_repo_concurrency_limit
);
let verify_cids: bool = cfg!(lookup, "VERIFY_CIDS", defaults.verify_cids);
let allow_private_networks: bool = cfg!(
lookup,
"ALLOW_PRIVATE_NETWORKS",
defaults.allow_private_networks
);
let only_index_links: bool = cfg!(lookup, "ONLY_INDEX_LINKS", defaults.only_index_links);
let max_pds_added_per_day =
read_or_none(&lookup, "HYDRANT_NEW_HOST_LIMIT", defaults.new_host_limit)?;
let repo_refetch_limit = read_or_none(
&lookup,
"HYDRANT_REPO_REFETCH_LIMIT",
defaults.repo_refetch_limit,
)?;
let host_refetch_limit = read_or_none(
&lookup,
"HYDRANT_HOST_REFETCH_LIMIT",
defaults.host_refetch_limit,
)?;
let repo_refetch_window = cfg!(lookup, "REPO_REFETCH_WINDOW", defaults.repo_refetch_window);
let rewalk_window = cfg!(lookup, "REWALK_WINDOW", defaults.rewalk_window);
let rewalk_mb = cfg!(lookup, "REWALK_MB", defaults.rewalk_mb);
let seed_poll_interval = read_or_none(
&lookup,
"HYDRANT_SEED_POLL_INTERVAL",
defaults.seed_poll_interval,
)?;
let offline_retry_interval = read_or_none(
&lookup,
"HYDRANT_OFFLINE_HOST_RETRY_INTERVAL",
defaults.offline_host_retry_interval,
)?;
// start with built-in tier definitions, then layer in any env-defined overrides.
// format: HYDRANT_RATE_TIERS=name:base/mul/hourly/daily,...
let mut tiers = defaults.tier_policy.tiers.clone();
if let Some(s) = lookup("HYDRANT_RATE_TIERS") {
for entry in s.split(',') {
let entry = entry.trim();
if entry.is_empty() {
continue;
}
match entry
.split_once(':')
.and_then(|(name, spec)| Some((name, RateTier::parse(spec)?)))
{
Some((name, tier)) => {
tiers.insert(SmolStr::new(name.trim()), tier);
}
None => tracing::warn!(
"ignoring invalid rate tier '{entry}': expected name:base/mul/hourly/daily format"
),
}
}
}
let seed_hosts: Vec<Url> = lookup("HYDRANT_SEED_HOSTS")
.map(|s| {
s.split(',')
.filter_map(|u| {
let u = u.trim();
if u.is_empty() {
return None;
}
Url::parse(u).ok().or_else(|| {
tracing::warn!("invalid seed host URL: {u}");
None
})
})
.collect()
})
.unwrap_or_else(|| defaults.seed_hosts.clone());
// build ordered glob rules from HYDRANT_TIER_RULES
let mut rules: Vec<TierRule> = vec![];
let mut tier_rules: Vec<(String, String)> = vec![];
if let Some(s) = lookup("HYDRANT_TIER_RULES") {
for entry in s.split(',') {
let entry = entry.trim();
if entry.is_empty() {
continue;
}
let Some((pattern_str, tier_name)) = entry.split_once(':') else {
tracing::warn!(
"ignoring invalid tier rule '{entry}': expected pattern:tier format"
);
continue;
};
let pattern_str = pattern_str.trim();
let tier_name = tier_name.trim();
if pattern_str.is_empty() || tier_name.is_empty() {
tracing::warn!("ignoring invalid tier rule '{entry}': empty pattern or tier");
continue;
}
rules.push(TierRule {
pattern: SmolStr::new(pattern_str),
tier_name: SmolStr::new(tier_name),
});
tier_rules.push((pattern_str.to_string(), tier_name.to_string()));
}
}
let tier_policy = TierPolicy { tiers, rules };
let default_mode = CrawlerMode::default_for(full_network);
let crawler_sources = match lookup("HYDRANT_CRAWLER_URLS") {
Some(s) => s
.split(',')
.map(|s| s.trim())
.filter(|s| !s.is_empty())
.filter_map(|s| {
CrawlerSource::parse(s, default_mode).or_else(|| {
tracing::warn!("ignoring invalid crawler url: {s}");
None
})
})
.collect(),
None => match default_mode {
CrawlerMode::ListRepos => relay_hosts
.iter()
.map(|source| CrawlerSource {
url: source.url.clone(),
mode: CrawlerMode::ListRepos,
})
.collect(),
CrawlerMode::RelayFirst => unreachable!("relay_first is never a default mode"),
CrawlerMode::ByCollection => defaults.crawler_sources.clone(),
},
};
let config = Self {
database_path,
full_network,
ephemeral,
seed_hosts,
seed_poll_interval,
ephemeral_ttl,
history_ttl,
relays: relay_hosts,
plc_urls,
enable_firehose,
firehose_workers,
firehose_max_failures,
cursor_save_interval,
enable_backfill,
repo_fetch_timeout,
max_car_body_bytes,
backfill_memory_bytes,
backfill_chunk_records,
backfill_spill_bytes,
backfill_spill_dir,
backfill_spills,
backfill_concurrency_limit,
backfill_strategy,
backfill_proxies,
per_pds_concurrency,
enable_crawler,
crawler_max_pending_repos,
crawler_resume_pending_repos,
crawler_sources,
verify_signatures,
identity_cache_size,
verify_mst,
rev_clock_skew_secs,
filter_signals,
filter_collections,
filter_excludes,
enable_backlinks,
get_repo_concurrency_limit,
verify_cids,
allow_private_networks,
only_index_links,
new_host_limit: max_pds_added_per_day,
repo_refetch_limit,
host_refetch_limit,
repo_refetch_window,
rewalk_window,
rewalk_mb,
offline_host_retry_interval: offline_retry_interval,
tier_policy,
tier_rules,
cache_size,
data_compression,
journal_compression,
db_worker_threads,
db_max_journaling_size_mb,
db_repos_memtable_size_mb,
db_events_memtable_size_mb,
db_records_memtable_size_mb,
db_records_bloom_filters,
stream_replay_chunk_size,
stream_chunk_span,
throttled_streams,
stream_replay_speed,
stream_replay_floor_bytes,
stream_pending_event_limit,
stream_send_timeout,
maintenance_timeout,
};
config.validate()?;
for warning in dead_knobs(&lookup, &config) {
tracing::warn!("{warning}");
}
Ok(config)
}
}
/// knobs that are read, so env files setting them keep working, but do nothing
fn dead_knobs(lookup: &impl Fn(&str) -> Option<String>, config: &Config) -> Vec<&'static str> {
let is_set = |key: &str| lookup(key).is_some_and(|v| !v.trim().is_empty());
let mut dead = Vec::new();
if is_set("HYDRANT_ENABLE_BACKLINKS") {
dead.push(
"HYDRANT_ENABLE_BACKLINKS does nothing, backlinks are on exactly when hydrant is built with the backlinks feature",
);
}
if is_set("HYDRANT_VERIFY_MST") && config.verify_signatures == SignatureVerification::Full {
dead.push(
"HYDRANT_VERIFY_MST does nothing with HYDRANT_VERIFY_SIGNATURES=full, which already checks every live commit's tree",
);
}
// indexer builds refuse any proxy in validate, so only relay builds get here with one
if !config.backfill_proxies.is_empty() {
dead.push("HYDRANT_BACKFILL_PROXIES does nothing, relay builds have no backfill");
}
if is_set("HYDRANT_STREAM_REPLAY_CHUNK_PAUSE") {
dead.push(
"HYDRANT_STREAM_REPLAY_CHUNK_PAUSE does nothing, replays are paced by HYDRANT_THROTTLED_STREAMS, HYDRANT_STREAM_REPLAY_SPEED and HYDRANT_STREAM_REPLAY_FLOOR_BYTES",
);
}
dead
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn env_text_parser_contracts() {
let parsed = dotenv_assignments(
r#"
# ignored
BARE = value
DOUBLE="two words"
SINGLE='three words'
INVALID
"#,
)
.collect::<Vec<_>>();
assert_eq!(
parsed,
[
("BARE", "value"),
("DOUBLE", "two words"),
("SINGLE", "three words"),
]
);
}
#[test]
fn from_lookup_reads_keys_from_the_lookup() {
let keys = std::collections::HashMap::from([
("HYDRANT_DATABASE_PATH", "/tmp/hydrant-lookup-test"),
(
"HYDRANT_FILTER_SIGNALS",
"sh.tangled.repo, sh.tangled.spindle.member",
),
("HYDRANT_RELAY_HOSTS", ""),
// relay mode defaults to ephemeral, which rejects a history ttl
("HYDRANT_EPHEMERAL", "false"),
("HYDRANT_HISTORY_TTL", "2h"),
("HYDRANT_ENABLE_CRAWLER", "false"),
]);
let config = Config::from_lookup(|key| keys.get(key).map(|v| v.to_string())).unwrap();
assert_eq!(
config.database_path,
std::path::PathBuf::from("/tmp/hydrant-lookup-test")
);
assert_eq!(
config.filter_signals,
Some(vec![
"sh.tangled.repo".to_string(),
"sh.tangled.spindle.member".to_string()
])
);
assert!(config.relays.is_empty());
assert!(!config.ephemeral);
assert_eq!(config.history_ttl, Some(Duration::from_secs(2 * 60 * 60)));
assert_eq!(config.enable_crawler, Some(false));
}
#[test]
fn private_networks_are_refused_unless_allowed() {
let allowed = |value: Option<&str>| {
Config::from_lookup(|key| {
value
.filter(|_| key == "HYDRANT_ALLOW_PRIVATE_NETWORKS")
.map(str::to_string)
})
.unwrap()
.allow_private_networks
};
assert!(!allowed(None));
assert!(allowed(Some("true")));
assert!(!allowed(Some("0")));
}
#[test]
fn from_lookup_rejects_invalid_values_it_cannot_default() {
let err = Config::from_lookup(|key| {
(key == "HYDRANT_HISTORY_TTL").then(|| "not a duration".to_string())
})
.unwrap_err();
assert!(err.to_string().contains("HYDRANT_HISTORY_TTL"), "{err}");
}
#[test]
fn dead_knobs_are_named() {
let dead = |pairs: &[(&str, &str)]| {
let lookup = |key: &str| {
pairs
.iter()
.find(|(k, _)| *k == key)
.map(|(_, v)| v.to_string())
};
let config = Config::from_lookup(lookup).unwrap();
dead_knobs(&lookup, &config)
};
assert!(dead(&[]).is_empty());
assert!(
dead(&[
("HYDRANT_VERIFY_MST", "true"),
("HYDRANT_VERIFY_SIGNATURES", "none")
])
.is_empty()
);
let backlinks = dead(&[("HYDRANT_ENABLE_BACKLINKS", "false")]);
assert!(
backlinks[0].contains("HYDRANT_ENABLE_BACKLINKS"),
"{backlinks:?}"
);
let mst = dead(&[("HYDRANT_VERIFY_MST", "false")]);
assert!(mst[0].contains("HYDRANT_VERIFY_MST"), "{mst:?}");
#[cfg(feature = "relay")]
{
let proxies = dead(&[("HYDRANT_BACKFILL_PROXIES", "http://proxy.example.com")]);
assert!(
proxies[0].contains("HYDRANT_BACKFILL_PROXIES"),
"{proxies:?}"
);
}
// an env file from before the replay cap keeps working, whatever it set the pause to
let pause = dead(&[("HYDRANT_STREAM_REPLAY_CHUNK_PAUSE", "not even a duration")]);
assert!(
pause[0].contains("HYDRANT_STREAM_REPLAY_CHUNK_PAUSE"),
"{pause:?}"
);
}
#[test]
fn replay_cap_knobs_parse_strictly() {
let config = from_pairs(&[]).unwrap();
assert_eq!(
config.throttled_streams,
[StreamKind::Jetstream, StreamKind::Relay]
);
assert_eq!(
(config.stream_replay_speed, config.stream_replay_floor_bytes),
(10, 10_000_000)
);
let config = from_pairs(&[
("HYDRANT_THROTTLED_STREAMS", " indexer ,relay,indexer,"),
("HYDRANT_STREAM_REPLAY_SPEED", "3"),
("HYDRANT_STREAM_REPLAY_FLOOR_BYTES", "500000"),
])
.unwrap();
assert_eq!(
config.throttled_streams,
[StreamKind::Indexer, StreamKind::Relay]
);
assert_eq!(
(config.stream_replay_speed, config.stream_replay_floor_bytes),
(3, 500_000)
);
// the names read back the way they're set
assert!(config.to_string().contains("indexer, relay"), "{config}");
// set but empty turns every cap off, where unset keeps the default
let none = from_pairs(&[("HYDRANT_THROTTLED_STREAMS", "")]).unwrap();
assert!(none.throttled_streams.is_empty());
for (key, bad) in [
("HYDRANT_THROTTLED_STREAMS", "jetstream,firehose"),
("HYDRANT_THROTTLED_STREAMS", "Jetstream"),
("HYDRANT_STREAM_REPLAY_SPEED", "10x"),
("HYDRANT_STREAM_REPLAY_SPEED", "1.5"),
("HYDRANT_STREAM_REPLAY_SPEED", "-1"),
("HYDRANT_STREAM_REPLAY_FLOOR_BYTES", "10MB"),
("HYDRANT_STREAM_REPLAY_FLOOR_BYTES", "-5"),
// a cap that never lets a replay move, so the list is the only off switch
("HYDRANT_STREAM_REPLAY_SPEED", "0"),
("HYDRANT_STREAM_REPLAY_FLOOR_BYTES", "0"),
] {
let err = from_pairs(&[(key, bad)]).unwrap_err().to_string();
assert!(err.contains(key) && err.contains(bad), "{key}={bad}: {err}");
}
}
fn from_pairs(pairs: &[(&str, &str)]) -> Result<Config> {
Config::from_lookup(|key| {
pairs
.iter()
.find(|(k, _)| *k == key)
.map(|(_, v)| v.to_string())
})
}
#[test]
fn from_lookup_accepts_good_values() {
for (key, value) in [
("HYDRANT_FULL_NETWORK", "true"),
("HYDRANT_FULL_NETWORK", "0"),
("HYDRANT_ENABLE_FIREHOSE", "false"),
("HYDRANT_ENABLE_FIREHOSE", "1"),
("HYDRANT_ENABLE_BACKFILL", " true "),
("HYDRANT_ENABLE_CRAWLER", "1"),
("HYDRANT_VERIFY_CIDS", "false"),
("HYDRANT_FIREHOSE_WORKERS", "24"),
("HYDRANT_FIREHOSE_WORKERS", ""),
("HYDRANT_FIREHOSE_MAX_FAILURES", "1"),
("HYDRANT_CRAWLER_MAX_PENDING_REPOS", "2"),
("HYDRANT_CRAWLER_RESUME_PENDING_REPOS", "1"),
("HYDRANT_REV_CLOCK_SKEW", "-5"),
("HYDRANT_CACHE_SIZE", "128"),
("HYDRANT_CURSOR_SAVE_INTERVAL", "3sec"),
("HYDRANT_CURSOR_SAVE_INTERVAL", "1s"),
("HYDRANT_CURSOR_SAVE_INTERVAL", "500ms"),
("HYDRANT_EPHEMERAL_TTL", "60min"),
("HYDRANT_EPHEMERAL_TTL", "3d"),
("HYDRANT_MAINTENANCE_TIMEOUT", "0"),
("HYDRANT_MAINTENANCE_TIMEOUT", "0s"),
("HYDRANT_MAINTENANCE_TIMEOUT", "6h"),
("HYDRANT_STREAM_REPLAY_CHUNK_PAUSE", "2ms"),
("HYDRANT_STREAM_SEND_TIMEOUT", "0s"),
("HYDRANT_SEED_POLL_INTERVAL", "none"),
("HYDRANT_SEED_POLL_INTERVAL", "30min"),
("HYDRANT_OFFLINE_HOST_RETRY_INTERVAL", "None"),
("HYDRANT_OFFLINE_HOST_RETRY_INTERVAL", "30min"),
("HYDRANT_NEW_HOST_LIMIT", "NoNe"),
("HYDRANT_NEW_HOST_LIMIT", " 42 "),
("HYDRANT_REPO_REFETCH_LIMIT", "none"),
("HYDRANT_HOST_REFETCH_LIMIT", "none"),
("HYDRANT_HOST_REFETCH_LIMIT", "500"),
("HYDRANT_REPO_REFETCH_LIMIT", "2"),
("HYDRANT_REPO_REFETCH_WINDOW", "10s"),
("HYDRANT_REWALK_WINDOW", "10s"),
("HYDRANT_REWALK_MB", "1"),
("HYDRANT_DATA_COMPRESSION", "lz4"),
("HYDRANT_JOURNAL_COMPRESSION", "none"),
("HYDRANT_VERIFY_SIGNATURES", "backfill-only"),
("HYDRANT_BACKFILL_STRATEGY", "sparse-filter"),
("HYDRANT_DB_WORKER_THREADS", "1"),
("HYDRANT_DB_MAX_JOURNALING_SIZE_MB", "64"),
("HYDRANT_DB_REPOS_MEMTABLE_SIZE_MB", "2"),
("HYDRANT_DB_EVENTS_MEMTABLE_SIZE_MB", "1"),
("HYDRANT_RELAY_HOST", "wss://relay.example.com/"),
("HYDRANT_RELAY_HOST", "pds::wss://pds.example.com"),
("HYDRANT_RATE_TIERS", "custom:1/1/1/1/0"),
("HYDRANT_TIER_RULES", "127.0.0.*:custom"),
] {
if let Err(e) = from_pairs(&[(key, value)]) {
panic!("{key}={value:?} was refused: {e}");
}
}
let config = from_pairs(&[
("HYDRANT_ENABLE_FIREHOSE", "0"),
("HYDRANT_ENABLE_CRAWLER", "1"),
("HYDRANT_FIREHOSE_WORKERS", ""),
("HYDRANT_NEW_HOST_LIMIT", " 42 "),
("HYDRANT_REPO_REFETCH_LIMIT", "2"),
("HYDRANT_HOST_REFETCH_LIMIT", "9"),
("HYDRANT_REPO_REFETCH_WINDOW", "10s"),
("HYDRANT_REWALK_WINDOW", "10s"),
("HYDRANT_REWALK_MB", "3"),
("HYDRANT_SEED_POLL_INTERVAL", "NONE"),
("HYDRANT_MAINTENANCE_TIMEOUT", "0"),
])
.unwrap();
assert!(!config.enable_firehose);
assert_eq!(config.enable_crawler, Some(true));
assert_eq!(
config.firehose_workers,
from_pairs(&[]).unwrap().firehose_workers
);
assert_eq!(config.new_host_limit, Some(42));
assert_eq!(config.repo_refetch_limit, Some(2));
assert_eq!(config.host_refetch_limit, Some(9));
assert_eq!(config.repo_refetch_window, Duration::from_secs(10));
let defaults = from_pairs(&[]).unwrap();
assert_eq!(defaults.repo_refetch_limit, Some(6));
assert_eq!(defaults.host_refetch_limit, Some(60));
assert_eq!(defaults.repo_refetch_window, Duration::from_secs(3600));
assert_eq!(config.rewalk_window, Duration::from_secs(10));
assert_eq!(defaults.rewalk_window, Duration::from_secs(30 * 60));
assert_eq!(config.rewalk_mb, 3);
assert_eq!(defaults.rewalk_mb, 8);
assert_eq!(config.seed_poll_interval, None);
assert_eq!(config.maintenance_timeout, Duration::ZERO);
}
#[test]
fn from_lookup_refuses_bad_values() {
let mut bad = vec![
("HYDRANT_FULL_NETWORK", "yes"),
("HYDRANT_ENABLE_FIREHOSE", "no"),
("HYDRANT_ENABLE_FIREHOSE", "TRUE"),
("HYDRANT_ENABLE_CRAWLER", "on"),
("HYDRANT_ENABLE_BACKLINKS", "y"),
("HYDRANT_FIREHOSE_WORKERS", "4GiB"),
("HYDRANT_FIREHOSE_WORKERS", "1_000"),
("HYDRANT_FIREHOSE_WORKERS", "-1"),
("HYDRANT_FIREHOSE_WORKERS", "0"),
("HYDRANT_CACHE_SIZE", "1e9"),
("HYDRANT_REV_CLOCK_SKEW", "5m"),
("HYDRANT_CURSOR_SAVE_INTERVAL", "300"),
("HYDRANT_CURSOR_SAVE_INTERVAL", "0s"),
("HYDRANT_CURSOR_SAVE_INTERVAL", "0"),
("HYDRANT_MAINTENANCE_TIMEOUT", "300"),
("HYDRANT_SEED_POLL_INTERVAL", "0s"),
("HYDRANT_SEED_POLL_INTERVAL", "never"),
("HYDRANT_OFFLINE_HOST_RETRY_INTERVAL", "0s"),
("HYDRANT_OFFLINE_HOST_RETRY_INTERVAL", "soon"),
("HYDRANT_NEW_HOST_LIMIT", "lots"),
("HYDRANT_REPO_REFETCH_LIMIT", "lots"),
("HYDRANT_REPO_REFETCH_LIMIT", "-1"),
("HYDRANT_REPO_REFETCH_LIMIT", "0"),
("HYDRANT_HOST_REFETCH_LIMIT", "0"),
("HYDRANT_HOST_REFETCH_LIMIT", "lots"),
("HYDRANT_REPO_REFETCH_WINDOW", "0s"),
("HYDRANT_REPO_REFETCH_WINDOW", "never"),
("HYDRANT_REWALK_WINDOW", "0s"),
("HYDRANT_REWALK_WINDOW", "never"),
("HYDRANT_REWALK_MB", "0"),
("HYDRANT_REWALK_MB", "8MiB"),
("HYDRANT_DATA_COMPRESSION", "gzip"),
("HYDRANT_JOURNAL_COMPRESSION", "LZ4"),
("HYDRANT_VERIFY_SIGNATURES", "yes"),
("HYDRANT_BACKFILL_STRATEGY", "fast"),
("HYDRANT_DB_WORKER_THREADS", "0"),
("HYDRANT_DB_MAX_JOURNALING_SIZE_MB", "63"),
("HYDRANT_DB_REPOS_MEMTABLE_SIZE_MB", "1"),
("HYDRANT_DB_EVENTS_MEMTABLE_SIZE_MB", "0"),
("HYDRANT_DB_RECORDS_MEMTABLE_SIZE_MB", "0"),
("HYDRANT_RELAY_HOST", "not a url"),
];
// relay builds have no backfill, so its knobs aren't checked there
if std::cfg!(feature = "indexer") {
bad.push(("HYDRANT_BACKFILL_CONCURRENCY_LIMIT", "0"));
bad.push(("HYDRANT_BACKFILL_SPILLS", "0"));
}
for (key, value) in bad {
let Err(err) = from_pairs(&[(key, value)]) else {
panic!("{key}={value:?} was accepted");
};
let err = err.to_string();
assert!(
err.contains(key) && err.contains(value),
"{key}={value:?}: the error should name both, got: {err}"
);
}
}
}