[config] add REWALK_WINDOW
16 unmodified lines1718192021212223242556 unmodified lines8283848586868788899069 unmodified lines1601611621631641651661671687 unmodified lines16 unmodified lines- durations use [humantime](https://docs.rs/humantime/latest/humantime/fn.parse_duration.html) and need a unit: `3s`, `5min`, `2h`, `3d`. a bare `300` is an error, only a bare `0` works without one.- the fixed-choice ones (`DATA_COMPRESSION`, `VERIFY_SIGNATURES`, `BACKFILL_STRATEGY`, ...) only take the values listed for them, lowercase.- the ones that take `none` take it in any case.- lists are comma separated, entries trimmed, empty entries dropped. a bad entry in `RELAY_HOSTS`, `SEED_HOSTS`, `CRAWLER_URLS`, `RATE_TIERS` or `TIER_RULES` is dropped with a warning in the log, and the rest of the list is used. `PLC_URL` and `FILTER_EXCLUDES` are stricter, a bad entry there is an error.- values that parse but would crash hydrant or spin a loop are errors too: `FIREHOSE_WORKERS`, `DB_WORKER_THREADS`, `BACKFILL_CONCURRENCY_LIMIT` and `BACKFILL_SPILLS` (indexer), `DB_EVENTS_MEMTABLE_SIZE_MB` and `DB_RECORDS_MEMTABLE_SIZE_MB` of `0`, a zero `CURSOR_SAVE_INTERVAL`, `SEED_POLL_INTERVAL` or `OFFLINE_HOST_RETRY_INTERVAL`, `REPO_REFETCH_LIMIT` or `HOST_REFETCH_LIMIT` of `0`, a zero `REPO_REFETCH_WINDOW`, `DB_MAX_JOURNALING_SIZE_MB` below `64` and `DB_REPOS_MEMTABLE_SIZE_MB` below `2`.- values that parse but would crash hydrant or spin a loop are errors too: `FIREHOSE_WORKERS`, `DB_WORKER_THREADS`, `BACKFILL_CONCURRENCY_LIMIT` and `BACKFILL_SPILLS` (indexer), `DB_EVENTS_MEMTABLE_SIZE_MB` and `DB_RECORDS_MEMTABLE_SIZE_MB` of `0`, a zero `CURSOR_SAVE_INTERVAL`, `SEED_POLL_INTERVAL` or `OFFLINE_HOST_RETRY_INTERVAL`, `REPO_REFETCH_LIMIT` or `HOST_REFETCH_LIMIT` of `0`, a zero `REPO_REFETCH_WINDOW` or `REWALK_WINDOW`, `DB_MAX_JOURNALING_SIZE_MB` below `64` and `DB_REPOS_MEMTABLE_SIZE_MB` below `2`.- `BACKFILL_PROXIES` with anything in it is an error in an indexer.
## fixed when the database is created
56 unmodified lines| `SEED_POLL_INTERVAL` | `30min` | humantime interval between completed `listHosts` polls. set to `none` (any case) to seed once at startup. `0s` is an error || `ENABLE_FIREHOSE` | `true` | whether to ingest relay subscriptions. `false` keeps the sources from connecting, but their tasks, the shard threads, seeding and the offline retry loop all still run. can also be toggled at runtime through `/ingestion`. || `FIREHOSE_WORKERS` | `8` (`24` full network) | number of firehose shards. messages go to a shard by a hash of their repo's did, so one repo's messages stay in order, and each shard is its own thread with a queue of 64 messages. an indexer runs two sets of shards, one to validate and one to write records, so `2n` threads, and a relay runs `n`. a shard doesn't wait on identity lookups: it holds a repo's messages while a shared lookup pool resolves it and goes on with the others. every commit that emits events still goes through one shared lock, so more shards don't add write throughput past that. backfill results also drain through the shards, so they run even with the firehose off. at least `1` || `FIREHOSE_MAX_FAILURES` | `15` | consecutive connect or stream failures before hydrant gives up on a source. the wait between tries doubles from 15 seconds and is capped at a minute, so `15` is about 14 minutes of retrying. failures only reset after a minute connected. a PDS that hits it is marked offline. given-up sources come back through `OFFLINE_HOST_RETRY_INTERVAL`. `0` and `1` both give up after the first failure || `OFFLINE_HOST_RETRY_INTERVAL` | `30min` | how often sources that were given up on get restarted, which also resets their failure count. in an indexer it's also the window a followed host's 3 requested rewalks count in. `none` (any case) disables it. `0s` is an error, it would restart dead hosts in a hot loop || `OFFLINE_HOST_RETRY_INTERVAL` | `30min` | how often sources that were given up on get restarted, which also resets their failure count. `none` (any case) disables it. `0s` is an error, it would restart dead hosts in a hot loop || `CURSOR_SAVE_INTERVAL` | `3sec` | how often a background thread saves every firehose cursor, checkpoints counts and writes the journal buffer out to the os. it's a write to the os, not an fsync, so it protects against a crash of hydrant but not a power loss. after a crash hydrant re-reads up to this much of the firehose, which is cheap since already applied commits are skipped. every source's cursor is rewritten each tick, changed or not, which adds up on a relay with thousands of seeded sources. `0s` is an error, it would save in a hot loop |
## identity & verification
69 unmodified lines| `NEW_HOST_LIMIT` | `50` | how many new hosts can be added via `com.atproto.sync.requestCrawl` per UTC day. `none` (any case) means unlimited. seeded hosts and ones added through the management API don't count || `REPO_REFETCH_LIMIT` | `6` | how many full refetches one repo may use per `REPO_REFETCH_WINDOW`, whatever asks for them (a `#sync`, a chain break, a newer listing, a crawl). `none` (any case) means unlimited. `0` is an error, the repo could never be refetched again. see [refetch budgets](concepts/relay.md#refetch-budgets) || `HOST_REFETCH_LIMIT` | `60` | how many full refetches the repos of one PDS may use together per `REPO_REFETCH_WINDOW`. hosts on the built-in `trusted` tier are exempt. `none` (any case) means unlimited, `0` is an error. a repo with no known PDS only counts against its own limit || `REPO_REFETCH_WINDOW` | `1h` | how long a refetch allowance lasts, counted from the first refetch it covers (a humantime duration, `0s` is an error) || `REWALK_WINDOW` | `30min` | in an indexer, how long a followed host's `listRepos` rewalk allowance lasts, counted from the first page a rewalk reads (a humantime duration, `0s` is an error). see [`requestCrawl`](concepts/relay.md) || `RATE_TIERS` | | comma-separated list of named rate tier definitions in `name:base/mul/hourly/daily[/account_limit]` format (e.g. `trusted:5000/10.0/18000000/432000000/10000000`). the per second limit is `max(base, accounts × mul)`, with hourly and daily caps on top. it paces firehose events from `pds::` sources and backfill requests to that PDS. the optional account limit prevents new accounts from being created on a PDS once reached, and leaving it off (or giving one that doesn't parse) means no account limit at all. built-in tiers (`default`, `trusted`) are always present and can be overridden. in indexer builds `default` has the same limits as `trusted` (5000/s), so pacing rarely kicks in for backfill there. in relay builds it's 50/s with a 100 account limit || `TIER_RULES` | | comma-separated ordered list of glob rules in `pattern:tier_name` format (e.g. `*.bsky.network:trusted`). rules are evaluated in order, first match wins. explicit API assignments via `PUT /pds/tiers` take precedence, and the `default` tier is the final fallback (defaults to `trusted` in indexer mode). a rule naming a tier that doesn't exist falls back to `default`. uses standard glob wildcards (`*`, `?`) matched against the PDS hostname |
## compatibility settings7 unmodified lines257 unmodified linesExpand all258259260261262263264265266267268269270168 unmodified linesExpand all439440441442443444445446447142 unmodified linesExpand all590591592593594595596597598599600601602362 unmodified linesExpand all257 unmodified linesExpand all /// 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 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>,168 unmodified linesExpand all 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), offline_host_retry_interval: Some(Duration::from_secs(30 * 60)), tier_rules: vec![], tier_policy: { let mut tiers = HashMap::new();142 unmodified linesExpand all 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.db_worker_threads == 0 { miette::bail!( "invalid HYDRANT_DB_WORKER_THREADS `0`: the database needs at least 1 thread to flush" );362 unmodified linesExpand all231 unmodified linesExpand all232233234235236237238236237238239240425 unmodified linesExpand all231 unmodified linesExpand all backfill_notify: Notify::new(), #[cfg(feature = "indexer")] new_pds_crawl_notify: Notify::new(), #[cfg(feature = "indexer")] rewalk_window: config .offline_host_retry_interval .unwrap_or(crate::control::new_pds_crawl::DEFAULT_REWALK_WINDOW), rewalk_window: config.rewalk_window, #[cfg(feature = "indexer")] rev_clock_skew_secs: config.rev_clock_skew_secs, #[cfg(feature = "indexer")] refetch_budget: crate::ingest::indexer::refetch::RefetchBudget::new(config),425 unmodified linesExpand all51 unmodified lines5253545556565758596026 unmodified lines51 unmodified lines- a host hydrant doesn't follow is added as a PDS firehose source, the same as `POST /firehose/sources`. it counts against `NEW_HOST_LIMIT`, and once that's used up for the UTC day the request gets `RateLimitExceeded`. in indexer mode it also gets the `listRepos` walk [above](#firehose-seeding).- a host hydrant follows whose firehose is connected or still retrying is left as it is.- a host whose firehose hydrant gave up on is restarted right away, after the usual startup jitter, with its failure count reset, instead of waiting for `OFFLINE_HOST_RETRY_INTERVAL`. a PDS that was down for an upgrade can ask to be picked up again this way.
in indexer mode a host hydrant follows also gets walked again when it asks, connected or not. a rewalk adds the repos the host lists that we've never seen, like the first walk. it also resyncs the ones whose doc points at the host and that we hold desynchronized or with a failed backfill waiting on its retry, and the ones the host lists at a newer rev than ours. the spec doesn't define that last one (the IETF drafts don't cover crawling, and the lexicon only says to "persistently crawl hosted repos"). it's how hydrant lets a host without a firehose say it changed. a newer rev costs the repo one of its refetches, see [below](#refetch-budgets). the stuck ones are the same resync `POST /repos/resync` does, so a repo keeps its retry count. a host gets 3 rewalks per `OFFLINE_HOST_RETRY_INTERVAL` (30 minutes when that's off), counted from its first one and kept across restarts. a request while its walk is queued or running, or once the 3 are used, gets the usual `200` and does nothing. a rewalk doesn't count against `NEW_HOST_LIMIT`, isn't given to a banned host, and leaves the host's backoff alone: it's paced and throttled like any other walk.in indexer mode a host hydrant follows also gets walked again when it asks, connected or not. a rewalk adds the repos the host lists that we've never seen, like the first walk. it also resyncs the ones whose doc points at the host and that we hold desynchronized or with a failed backfill waiting on its retry, and the ones the host lists at a newer rev than ours. the spec doesn't define that last one (the IETF drafts don't cover crawling, and the lexicon only says to "persistently crawl hosted repos"). it's how hydrant lets a host without a firehose say it changed. a newer rev costs the repo one of its refetches, see [below](#refetch-budgets). the stuck ones are the same resync `POST /repos/resync` does, so a repo keeps its retry count. a host gets 3 rewalks per `REWALK_WINDOW` (30 minutes by default), counted from its first one and kept across restarts. a request while its walk is queued or running, or once the 3 are used, gets the usual `200` and does nothing. a rewalk doesn't count against `NEW_HOST_LIMIT`, isn't given to a banned host, and leaves the host's backoff alone: it's paced and throttled like any other walk.
## refetch budgets
a full refetch is the cheapest thing a flaky or hostile peer can ask of hydrant: a `#sync` at a fresh rev, a chain break, a `listRepos` listing at a newer rev, a crawl's resync. each one costs the repo a refetch out of `REPO_REFETCH_LIMIT` per `REPO_REFETCH_WINDOW` (6 an hour), counted from the repo's first refetch in the window. the repos of one host also share `HOST_REFETCH_LIMIT` (60) in the same window. hosts on the built-in `trusted` tier are exempt from that one, because an operator who trusts a host already accepts its volume, but their repos keep their own limit.26 unmodified lines431 unmodified linesExpand all432433434435436437438439440146 unmodified linesExpand all587588589590591592593594595270 unmodified linesExpand all86686786886987087187287387418 unmodified lines89389489589689789889990090110 unmodified lines91291391491591691791891992092126 unmodified lines94894995095195295395495595695724 unmodified lines431 unmodified linesExpand all "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 seed_poll_interval = read_or_none( &lookup, "HYDRANT_SEED_POLL_INTERVAL", defaults.seed_poll_interval,146 unmodified linesExpand all new_host_limit: max_pds_added_per_day, repo_refetch_limit, host_refetch_limit, repo_refetch_window, rewalk_window, offline_host_retry_interval: offline_retry_interval, tier_policy, tier_rules, cache_size,270 unmodified linesExpand all ("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_DATA_COMPRESSION", "lz4"), ("HYDRANT_JOURNAL_COMPRESSION", "none"), ("HYDRANT_VERIFY_SIGNATURES", "backfill-only"), ("HYDRANT_BACKFILL_STRATEGY", "sparse-filter"),18 unmodified lines ("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_SEED_POLL_INTERVAL", "NONE"), ("HYDRANT_MAINTENANCE_TIMEOUT", "0"), ]) .unwrap();10 unmodified lines 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.seed_poll_interval, None); assert_eq!(config.maintenance_timeout, Duration::ZERO); }
26 unmodified lines ("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_DATA_COMPRESSION", "gzip"), ("HYDRANT_JOURNAL_COMPRESSION", "LZ4"), ("HYDRANT_VERIFY_SIGNATURES", "yes"), ("HYDRANT_BACKFILL_STRATEGY", "fast"),24 unmodified lines21 unmodified lines2223242526272826272829117 unmodified linesExpand all14714814915015115215315415515615715815916016116216316416516616767 unmodified lines21 unmodified lines use crate::state::AppState; use crate::types::Allowance;
pub(crate) const REWALKS_PER_WINDOW: u32 = 3; /// the rewalk window when `OFFLINE_HOST_RETRY_INTERVAL` is off, its default pub(crate) const DEFAULT_REWALK_WINDOW: std::time::Duration = std::time::Duration::from_secs(30 * 60);
#[derive(Debug, Clone, Copy, PartialEq, Eq)] pub(crate) enum Rewalk { Queued,117 unmodified linesExpand all .into_diagnostic() }
#[test] fn the_window_is_the_configured_one() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let state = AppState::new(&Config { database_path: dir.path().to_path_buf(), rewalk_window: std::time::Duration::from_secs(7), offline_host_retry_interval: None, ..Default::default() })?; assert_eq!(state.rewalk_window, std::time::Duration::from_secs(7)); Ok(()) }
#[test] fn the_allowance_comes_back_once_its_window_is_over() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let state = AppState::new(&Config { database_path: dir.path().to_path_buf(),67 unmodified lines