[ingest] cap refetches per host too
248 unmodified linesExpand all249250251252253253254255256257258259260261262169 unmodified linesExpand all432433434435436437438439440132 unmodified linesExpand all573574575576577578579580581582583584585586344 unmodified linesExpand all9319329339349359369379389399409419429439449459467 unmodified lines248 unmodified linesExpand all /// 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>,
/// how long a refetch allowance lasts, counted from the first refetch in it. /// 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,
/// how often offline firehose sources are automatically retried.169 unmodified linesExpand all 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), offline_host_retry_interval: Some(Duration::from_secs(30 * 60)), tier_rules: vec![], tier_policy: {132 unmodified linesExpand all 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" );344 unmodified linesExpand all 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",7 unmodified lines134 unmodified linesExpand all13513613713813914014114214314414514614714814915015115215315415515633 unmodified lines19019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722818 unmodified lines134 unmodified linesExpand all pub rules: Vec<TierRule>,}
impl TierPolicy { /// an unknown override gives "default" without trying the host's rule pub fn resolve_name<'a>(&'a self, host: &str, override_name: Option<&'a SmolStr>) -> &'a str { let rule = || { self.rules .iter() .find(|r| glob_match(r.pattern.as_bytes(), host.as_bytes())) .map(|r| &r.tier_name) }; override_name .or_else(rule) .filter(|name| self.tiers.contains_key(*name)) .map_or("default", SmolStr::as_str) }
/// resolves the effective `RateTier` for `host`. /// /// `override_name` is the api-assigned tier name from `HostDesc.tier`, if any. pub fn resolve(&self, host: &str, override_name: Option<&SmolStr>) -> RateTier {33 unmodified lines use super::*; use crate::util::hex;
#[test] fn tier_names_resolve_in_the_order_tiers_do() { let tier = |name: &str| (SmolStr::new(name), crate::config::RateTier::trusted()); let policy = TierPolicy { tiers: [tier("default"), tier("trusted"), tier("strict")].into(), rules: vec![ TierRule { pattern: "*.strict.example".into(), tier_name: "strict".into(), }, TierRule { pattern: "gone.example".into(), tier_name: "unknown".into(), }, ], }; let trusted = SmolStr::new("trusted"); let nonesuch = SmolStr::new("nonesuch"); assert_eq!(policy.resolve_name("a.strict.example", None), "strict"); assert_eq!( policy.resolve_name("a.strict.example", Some(&trusted)), "trusted" ); assert_eq!(policy.resolve_name("elsewhere.example", None), "default"); assert_eq!(policy.resolve_name("gone.example", None), "default"); assert_eq!( policy.resolve_name("a.strict.example", Some(&nonesuch)), "default" ); }
#[test] fn host_status_persisted_golden_fixtures() { for (status, msgpack, json) in [ (HostStatus::Active, "a6416374697665", "\"Active\""), (HostStatus::Idle, "a449646c65", "\"Idle\""),18 unmodified lines529 unmodified linesExpand all530531532533534535536537538539540541542543544545546121 unmodified linesExpand all529 unmodified linesExpand all .account_limit .is_some_and(|l| count >= l) }
/// checks the tier name, because in an indexer build the default tier has trusted limits #[cfg(feature = "indexer")] pub(crate) fn is_trusted_host(&self, host: &str) -> bool { let meta = self.pds_meta.load(); let override_name = meta.hosts.get(host).and_then(|h| h.tier.as_ref()); self.tier_policy.resolve_name(host, override_name) == "trusted" }
#[cfg(feature = "indexer")] pub(crate) fn resolve_pds_tier(&self, host: &str) -> crate::config::RateTier { let meta = self.pds_meta.load(); let override_name = meta.hosts.get(host).and_then(|h| h.tier.as_ref()); self.tier_policy.resolve(host, override_name)121 unmodified linesExpand all312 unmodified linesExpand all31331431531631731831932032132288 unmodified lines411412413414415416417418419605 unmodified linesExpand all312 unmodified linesExpand allcodec-call:src/db/types.rs::DbTid::to_smolstr::u64::from_be_bytes rust:src/db/types.rs::dbrkey_msgpack_binary_marker_preserves_tidcodec-call:src/db/types.rs::DidKey::encode::multibase::encode rust:src/db/types.rs::db_action_and_did_key_representation_contractscodec-call:src/db/types.rs::DidKey::from_did_key::multibase::decode rust:src/db/types.rs::db_action_and_did_key_representation_contractscodec-call:src/ingest/indexer/refetch.rs::defer::rmp_serde::to_vec rust:src/ingest/indexer/refetch.rs::a_repo_refetches_up_to_its_limit_and_then_waits_for_the_window_to_endcodec-call:src/ingest/indexer/refetch.rs::take_from_host::rmp_serde::from_slice rust:src/ingest/indexer/refetch.rs::the_repos_of_one_host_share_its_refetchescodec-call:src/ingest/indexer/refetch.rs::take_from_host::rmp_serde::to_vec rust:src/ingest/indexer/refetch.rs::the_repos_of_one_host_share_its_refetchescodec-call:src/ingest/indexer/shard.rs::FirehoseWorker::apply_buffered_commit::rmp_serde::from_slice rust:src/ingest/indexer/shard.rs::resync_buffer_commit_msgpack_round_trip_preserves_pathcodec-call:src/ingest/relay.rs::probe_repo_status::serde_json::from_slice rust:src/ingest/relay.rs::active_repo_status_probe_maps_to_synced_repo_statecodec-call:src/ingest/stream/codec.rs::decode_frame::Deserialize::deserialize rust:src/ingest/stream/codec.rs::firehose_frame_fixture_decodes_identitycodec-call:src/ingest/stream/codec.rs::decode_frame::ErrorFrame::deserialize rust:src/ingest/stream/codec.rs::firehose_frame_fixture_decodes_identity88 unmodified linesdb-key-codec:src/db/keys/indexer.rs::relay_first_pass_key rust:src/db/keys/indexer.rs::indexer_key_codec_inventory_contractsdb-key-codec:src/db/keys/indexer.rs::relay_first_pds_cursor_key rust:src/db/keys/indexer.rs::indexer_key_codec_inventory_contractsdb-key-codec:src/db/keys/indexer.rs::resync_buffer_key rust:src/db/keys/indexer.rs::indexer_key_codec_inventory_contractsdb-key-codec:src/db/keys/indexer.rs::resync_buffer_prefix rust:src/db/keys/indexer.rs::indexer_key_codec_inventory_contractsdb-key-codec:src/db/keys/indexer.rs::refetch_allowance_key rust:src/db/keys/indexer.rs::indexer_key_codec_inventory_contractsdb-key-codec:src/db/keys/indexer.rs::rewalk_allowance_key rust:src/db/keys/indexer.rs::indexer_key_codec_inventory_contractsdb-key-codec:src/db/keys/indexer.rs::rewalk_key rust:src/db/keys/indexer.rs::indexer_key_codec_inventory_contractsdb-key-codec:src/db/keys/indexer.rs::seed_listed_key rust:src/db/keys/indexer.rs::indexer_key_codec_inventory_contractsdb-key-codec:src/db/keys/indexer.rs::split_pending_did_key rust:src/control/repos/indexer.rs::queued_listing_is_the_full_listing_minus_whats_not_queued605 unmodified linesExpand all316 unmodified linesExpand all317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352156 unmodified linesExpand all316 unmodified linesExpand all assert_eq!(metadata.retry_count, 3, "waiting isn't failing"); }
#[test] fn a_repo_the_host_cap_denied_is_requeued_once_the_hosts_window_ends() { use crate::ingest::indexer::refetch::tests as refetch; let (_tmp, state) = refetch::host_state_in(Some(1), None, Duration::from_secs(1)); let repos = refetch::dids(2); for did in &repos { refetch::on_a_host(&state, did, true); } assert_eq!( refetch::charge_did(&state, &repos[0]), Charged::Allowed, "this one spends the host's refetch" ); assert_eq!(refetch::charge_did(&state, &repos[1]), Charged::Deferred); let queue = |did| queue_resync_if(&state, did, |_| true).unwrap();
assert!(!queue(&repos[1]), "the host's window is still open"); assert!(refetch::metadata_of(&state, &repos[1]).refetch_deferred);
std::thread::sleep(Duration::from_millis(1200)); assert!(queue(&repos[1])); let metadata = refetch::metadata_of(&state, &repos[1]); assert!(!metadata.refetch_deferred); assert_eq!(state.db.indexer.pending_iter().count(), 1); assert!(refetch::entry_of(&state, &repos[1]).is_none()); state.db.indexer.assert_pending_dids_match(); }
#[test] fn excluded_repo_is_not_requeued() { let (_tmp, state) = state(); stage_retry(&state, true);
156 unmodified linesExpand all425 unmodified linesExpand all426427428429430431432433434435436437438145 unmodified linesExpand all584585586587588589590591592267 unmodified linesExpand all86086186286386486586686786886918 unmodified lines8888898908918928938948958965 unmodified lines90290390490590690790890991091191291391423 unmodified lines93893994094194294394494594694726 unmodified lines425 unmodified linesExpand all &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 seed_poll_interval = read_or_none( &lookup, "HYDRANT_SEED_POLL_INTERVAL",145 unmodified linesExpand all allow_private_networks, only_index_links, new_host_limit: max_pds_added_per_day, repo_refetch_limit, host_refetch_limit, repo_refetch_window, offline_host_retry_interval: offline_retry_interval, tier_policy, tier_rules,267 unmodified linesExpand all ("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_DATA_COMPRESSION", "lz4"), ("HYDRANT_JOURNAL_COMPRESSION", "none"),18 unmodified lines ("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_SEED_POLL_INTERVAL", "NONE"), ("HYDRANT_MAINTENANCE_TIMEOUT", "0"), ])5 unmodified lines 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.seed_poll_interval, None); assert_eq!(config.maintenance_timeout, Duration::ZERO); }23 unmodified lines ("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_DATA_COMPRESSION", "gzip"), ("HYDRANT_JOURNAL_COMPRESSION", "LZ4"),26 unmodified lines