[crawler] count rewalks by listRepos bytes, not walks
81 unmodified lines828384858687888990919293143 unmodified linesExpand all237238239240241242243244245246424 unmodified linesExpand all81 unmodified lines pub(crate) new_pds_crawl_notify: Notify, /// how long a host's rewalk allowance lasts, see [`crate::control::new_pds_crawl`] #[cfg(feature = "indexer")] pub(crate) rewalk_window: Duration, /// the `listRepos` bytes a host's rewalks may read per `rewalk_window` #[cfg(feature = "indexer")] pub(crate) rewalk_bytes: u32, #[cfg(feature = "indexer")] pub(crate) rev_clock_skew_secs: i64, #[cfg(feature = "indexer")] pub(crate) refetch_budget: crate::ingest::indexer::refetch::RefetchBudget, #[cfg(feature = "indexer")]143 unmodified linesExpand all new_pds_crawl_notify: Notify::new(), #[cfg(feature = "indexer")] rewalk_window: config.rewalk_window, #[cfg(feature = "indexer")] rewalk_bytes: config.rewalk_mb.saturating_mul(1024 * 1024), #[cfg(feature = "indexer")] rev_clock_skew_secs: config.rev_clock_skew_secs, #[cfg(feature = "indexer")] refetch_budget: crate::ingest::indexer::refetch::RefetchBudget::new(config), #[cfg(feature = "indexer")]424 unmodified linesExpand all152 unmodified linesExpand all153154155156157158157158159160161162163164165166167168169170171172173174175176169170171177173178179180181182183184185186187188189190191192193194785 unmodified linesExpand all152 unmodified linesExpand all }
impl Allowance { /// `Err` is the unix time the window ends, not how long to wait pub fn take( allowance: &mut Option<Self>, pub fn check( allowance: &Option<Self>, now: i64, limit: u32, window: i64, ) -> Result<(), i64> { match allowance.filter(|allowance| now < allowance.since.saturating_add(window)) { Some(current) if current.used >= limit => Err(current.since.saturating_add(window)), _ => Ok(()), } }
/// a charge past the window's end starts the next window pub fn charge(allowance: &mut Option<Self>, now: i64, amount: u32, window: i64) { let current = allowance .filter(|allowance| now < allowance.since.saturating_add(window)) .unwrap_or(Self { since: now, used: 0, }); if current.used >= limit { return Err(current.since.saturating_add(window)); } *allowance = Some(Self { used: current.used + 1, used: current.used.saturating_add(amount), ..current }); }
pub fn take( allowance: &mut Option<Self>, now: i64, limit: u32, window: i64, ) -> Result<(), i64> { Self::check(allowance, now, limit, window)?; Self::charge(allowance, now, 1, window); Ok(()) } }
785 unmodified linesExpand all172 unmodified linesExpand all173174175176177178177178179180181182842 unmodified linesExpand all172 unmodified linesExpand allcodec-call:src/control/indexer.rs::Hydrant::seed_events_for_bench::rmp_serde::to_vec nu:tests/debug_endpoints.nucodec-call:src/control/indexer.rs::Hydrant::seed_events_for_bench::serde_ipld_dagcbor::to_vec nu:tests/debug_endpoints.nucodec-call:src/control/new_pds_crawl.rs::enabled::decode_cursor::rmp_serde::from_slice rust:src/crawler/new_pds.rs::a_broken_walk_keeps_its_cursor_and_the_last_page_clears_itcodec-call:src/control/new_pds_crawl.rs::enabled::encode_cursor::rmp_serde::to_vec rust:src/crawler/new_pds.rs::a_broken_walk_keeps_its_cursor_and_the_last_page_clears_itcodec-call:src/control/new_pds_crawl.rs::enabled::request_rewalk::rmp_serde::from_slice rust:src/control/new_pds_crawl.rs::the_allowance_and_a_queued_rewalk_survive_a_restartcodec-call:src/control/new_pds_crawl.rs::enabled::request_rewalk::rmp_serde::to_vec rust:src/control/new_pds_crawl.rs::the_allowance_and_a_queued_rewalk_survive_a_restartcodec-call:src/control/new_pds_crawl.rs::enabled::Budget::charge::rmp_serde::to_vec rust:src/control/new_pds_crawl.rs::the_spent_bytes_and_a_queued_rewalk_survive_a_restartcodec-call:src/control/new_pds_crawl.rs::enabled::Budget::load::rmp_serde::from_slice rust:src/control/new_pds_crawl.rs::the_spent_bytes_and_a_queued_rewalk_survive_a_restartcodec-call:src/control/repos/indexer.rs::RepoHandle::generate_car::CarWriter::new rust:src/control/repos/indexer.rs::generate_car_reports_a_deleted_head_bodycodec-call:src/control/repos/indexer.rs::RepoHandle::get_record::serde_ipld_dagcbor::from_slice rust:src/control/repos/indexer.rs::get_and_list_hide_redacted_headscodec-call:src/control/repos/indexer.rs::RepoHandle::list_records::serde_ipld_dagcbor::from_slice rust:src/control/repos/indexer.rs::get_and_list_hide_redacted_headscodec-call:src/control/repos/mod.rs::RepoHandle::collections::u64::from_be_bytes rust:src/control/repos/mod.rs::collections_decodes_big_endian_counts842 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 `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.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's rewalks may read `REWALK_MB` (8 by default) of `listRepos` responses per `REWALK_WINDOW` (30 minutes by default), counted from the first page one reads and kept across restarts, so a host can ask as often as it likes while it stays inside that. a walk charges each page's size once it has read it, so it can overshoot by a page, and stops before the next one when the bytes are spent. it keeps its cursor and goes on when the window ends, also after a restart. a request while its walk is queued, running or waiting, or once the bytes are spent, gets the usual `200` and does nothing. every `listRepos` walk, a new host's included, refuses a page over 512 KiB or with more repos than the 1000 we asked for, and ends with an error when a page returns a cursor an earlier page of the walk already returned. 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 lines459 unmodified linesExpand all460461462463464465466467468468469470471472473443 unmodified linesExpand all9179189199209209219229239249259263 unmodified lines9309319329339329349359369379386 unmodified lines945946947948949950951952953954955956957958959960961962963964965608 unmodified linesExpand all459 unmodified linesExpand all #[cfg(feature = "indexer")] async fn request_rewalk(&self, host: &str) -> Result<()> { let now = chrono::Utc::now().timestamp(); let window = self.state.rewalk_window.as_secs() as i64; let limit = self.state.rewalk_bytes; let walked = host.to_owned(); let rewalk = self .state .db .run(move |db| new_pds_crawl::request_rewalk(db, &walked, now, window)) .run(move |db| new_pds_crawl::request_rewalk(db, &walked, now, window, limit)) .await?; debug!(%host, ?rewalk, "requested a rewalk"); if rewalk == new_pds_crawl::Rewalk::Queued { new_pds_crawl::wake(&self.state);443 unmodified linesExpand all }
#[cfg(feature = "indexer")] #[tokio::test] async fn a_known_host_gets_three_rewalks_a_window_outside_the_daily_limit() -> Result<()> { async fn a_known_host_gets_rewalks_until_its_bytes_are_spent_outside_the_daily_limit() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let hydrant = Hydrant::new(test_config_with_limit(&dir, 2)).await?; startable(&hydrant); hydrant.firehose.disable_spawns_for_test();3 unmodified lines Ok(PdsAdmission::StartedNew) ); finish_walk(&hydrant, "known.example")?;
for _ in 0..3 { for _ in 0..4 { assert_eq!( hydrant.firehose.add_source_validated(url.clone()).await?, Ok(PdsAdmission::RestartedKnown) );6 unmodified lines .into_diagnostic()? ); finish_walk(&hydrant, "known.example")?; } hydrant .state .db .cursors .insert( keys::rewalk_allowance_key("known.example"), rmp_serde::to_vec(&Some(crate::types::Allowance { since: chrono::Utc::now().timestamp(), used: hydrant.state.rewalk_bytes, })) .into_diagnostic()?, ) .into_diagnostic()?; assert_eq!( hydrant.firehose.add_source_validated(url).await?, Ok(PdsAdmission::RestartedKnown) );608 unmodified linesExpand all