diff --git a/docs/concepts/relay.md b/docs/concepts/relay.md index 7394712..1d47b14 100644 --- a/docs/concepts/relay.md +++ b/docs/concepts/relay.md @@ -24,7 +24,7 @@ only when a source is marked as a direct PDS (`is_pds: true`), hydrant enforces ## firehose seeding -in relay mode, `RELAY_HOSTS` defaults to empty. set `SEED_HOSTS` to one or more relay base URLs and hydrant will call `com.atproto.sync.listHosts` on each at startup, adding every returned PDS as a firehose source: +in relay mode, `RELAY_HOSTS` defaults to empty. set `SEED_HOSTS` to one or more relay base URLs and hydrant will call `com.atproto.sync.listHosts` on each at startup and every 30 minutes thereafter, adding every returned PDS as a firehose source: ``` HYDRANT_SEED_HOSTS=https://bsky.network diff --git a/docs/configuration.md b/docs/configuration.md index c571ccf..a85aa6c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -37,7 +37,8 @@ hydrant is configured via environment variables, all prefixed with `HYDRANT_` (e | :--- | :--- | :--- | | `RELAY_HOST` | `wss://relay.fire.hose.cam/` (indexer), empty (relay) | URL of a single firehose source | | `RELAY_HOSTS` | | comma-separated list of firehose sources. if unset, falls back to `RELAY_HOST`. prefix a URL with `pds::` to mark it as a direct PDS connection (e.g. `pds::wss://pds.example.com`). bare URLs are treated as relays. defaults to empty in relay mode; PDS' are expected to be seeded via `SEED_HOSTS` or the firehose management API | -| `SEED_HOSTS` | `https://bsky.network` (relay) | comma-separated list of base URLs to call `com.atproto.sync.listHosts` on at startup. hydrant adds every non-banned host as a PDS firehose source | +| `SEED_HOSTS` | `https://bsky.network` (relay) | comma-separated list of base URLs to call `com.atproto.sync.listHosts` on at startup and periodically. hydrant adds every non-banned host as a PDS firehose source | +| `SEED_POLL_INTERVAL` | `30min` | humantime interval between completed `listHosts` polls; set to `none` to seed once at startup | | `ENABLE_FIREHOSE` | `true` | whether to ingest relay subscriptions | | `FIREHOSE_WORKERS` | `8` (`24` full network) | number of concurrent workers for firehose events | | `CURSOR_SAVE_INTERVAL` | `3sec` | how often to persist the firehose cursor | diff --git a/src/config.rs b/src/config.rs index b10c4fe..99477f0 100644 --- a/src/config.rs +++ b/src/config.rs @@ -218,7 +218,7 @@ pub struct Config { /// set to `none` to disable automatic retries. pub offline_host_retry_interval: Option, - /// base URL(s) of relay or aggregator services to seed firehose PDS sources from at startup. + /// 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 @@ -226,6 +226,11 @@ pub struct Config { /// /// set via `HYDRANT_SEED_HOSTS` as a comma-separated list of base URLs. pub seed_hosts: Vec, + + /// 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, /// named rate tier definitions for PDS rate limiting. /// /// built-in tiers ("default" and "trusted") are always present and may be overridden. @@ -327,6 +332,7 @@ impl Default for Config { 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, @@ -692,6 +698,10 @@ impl fmt::Display for Config { ) )?; } + 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)?; } diff --git a/src/config/env.rs b/src/config/env.rs index 60210b6..722b4aa 100644 --- a/src/config/env.rs +++ b/src/config/env.rs @@ -60,6 +60,27 @@ fn load_dotenv() { } } +fn parse_seed_poll_interval(default: Option) -> Option { + parse_seed_poll_interval_value( + std::env::var("HYDRANT_SEED_POLL_INTERVAL").ok().as_deref(), + default, + ) +} + +fn parse_seed_poll_interval_value( + value: Option<&str>, + default: Option, +) -> Option { + match value.map(str::trim) { + Some(value) if value.eq_ignore_ascii_case("none") => None, + Some(value) => humantime::parse_duration(value) + .ok() + .filter(|duration| !duration.is_zero()) + .or(default), + None => default, + } +} + fn parse_new_host_limit(default: Option) -> Option { parse_new_host_limit_value( std::env::var("HYDRANT_NEW_HOST_LIMIT").ok().as_deref(), @@ -262,6 +283,7 @@ impl Config { let verify_cids: bool = cfg!("VERIFY_CIDS", defaults.verify_cids); let only_index_links: bool = cfg!("ONLY_INDEX_LINKS", defaults.only_index_links); let max_pds_added_per_day = parse_new_host_limit(defaults.new_host_limit); + let seed_poll_interval = parse_seed_poll_interval(defaults.seed_poll_interval); let offline_retry_interval: Option = match std::env::var("HYDRANT_OFFLINE_HOST_RETRY_INTERVAL") @@ -368,6 +390,7 @@ impl Config { full_network, ephemeral, seed_hosts, + seed_poll_interval, ephemeral_ttl, history_ttl, relays: relay_hosts, @@ -455,5 +478,26 @@ mod tests { Some(10) ); assert_eq!(parse_new_host_limit_value(Some("invalid"), None), None); + let default_interval = Some(Duration::from_secs(1800)); + assert_eq!( + parse_seed_poll_interval_value(None, default_interval), + default_interval + ); + assert_eq!( + parse_seed_poll_interval_value(Some("30min"), None), + default_interval + ); + assert_eq!( + parse_seed_poll_interval_value(Some("none"), default_interval), + None + ); + assert_eq!( + parse_seed_poll_interval_value(Some("0s"), default_interval), + default_interval + ); + assert_eq!( + parse_seed_poll_interval_value(Some("invalid"), default_interval), + default_interval + ); } } diff --git a/src/control/hydrant/run.rs b/src/control/hydrant/run.rs index 6343afb..e7595a2 100644 --- a/src/control/hydrant/run.rs +++ b/src/control/hydrant/run.rs @@ -317,10 +317,19 @@ impl Hydrant { // 10c. seed firehose PDS sources from listHosts on configured seed URLs if !config.seed_hosts.is_empty() { let seed_urls = config.seed_hosts.clone(); + let seed_poll_interval = config + .seed_poll_interval + .filter(|interval| !interval.is_zero()); let firehose = firehose.clone(); let state = state.clone(); tokio::spawn(async move { - seed::seed_from_list_hosts(&seed_urls, &firehose, &state).await; + loop { + seed::seed_from_list_hosts(&seed_urls, &firehose, &state).await; + let Some(interval) = seed_poll_interval else { + break; + }; + tokio::time::sleep(interval).await; + } }); }