[ingest] limit how often a peer can make us refetch a repo
560 unmodified linesExpand all561562563564565566567568569570571572573574359 unmodified linesExpand all560 unmodified linesExpand all help = "set it to none to never retry offline hosts", "invalid HYDRANT_OFFLINE_HOST_RETRY_INTERVAL `0s`: the retry loop would spin, want a duration like 30min" ); } if self.repo_refetch_limit == Some(0) { miette::bail!( 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.repo_refetch_window.is_zero() { miette::bail!( "invalid HYDRANT_REPO_REFETCH_WINDOW `0s`: no refetch could be counted, want a duration like 1h" );359 unmodified linesExpand all149 unmodified linesExpand all150151152153154154155156157158159160161162163164165166167168169170171172245 unmodified linesExpand all149 unmodified linesExpand all let metadata_key = keys::repo_metadata_key(did);
let metadata_bytes = db.repo_metadata.get(&metadata_key).into_diagnostic()?; if let Some(metadata_bytes) = metadata_bytes { let metadata = db::deser_repo_meta(&metadata_bytes)?; let mut metadata = db::deser_repo_meta(&metadata_bytes)?; let pending_key = keys::pending_key(metadata.index_id); // the retry worker only requeues these, any other status would strand a deferred refetch let retries = repo_state.active && matches!( new_status, RepoStatus::Error(_) | RepoStatus::Desynchronized | RepoStatus::Throttled ); if !retries && std::mem::take(&mut metadata.refetch_deferred) { batch.insert( &db.repo_metadata, &metadata_key, db::ser_repo_meta(&metadata)?, ); }
// manage queues match &new_status { RepoStatus::Synced => {245 unmodified linesExpand all84 unmodified lines858687888990919293949596141 unmodified linesExpand all238239240241242243244245246247248249410 unmodified linesExpand all84 unmodified lines pub(crate) rewalk_window: Duration, #[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")] pub(crate) refetch_denied: std::sync::atomic::AtomicU64, #[cfg(feature = "indexer")] pub crawler_enabled: watch::Sender<bool>, #[cfg(feature = "indexer")] pub backfill_enabled: watch::Sender<bool>, #[cfg(feature = "indexer")]141 unmodified linesExpand all .unwrap_or(crate::control::new_pds_crawl::DEFAULT_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), #[cfg(feature = "indexer")] refetch_denied: std::sync::atomic::AtomicU64::new(0), #[cfg(feature = "indexer")] crawler_enabled, firehose_enabled, #[cfg(feature = "indexer")] backfill_enabled,410 unmodified linesExpand all150 unmodified linesExpand all151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183781 unmodified linesExpand all150 unmodified linesExpand all .is_some_and(|car| DbTid::from(rev) <= car) } }
impl Allowance { /// `Err` is the unix time the window ends, not how long to wait pub fn take( allowance: &mut Option<Self>, now: i64, limit: u32, window: i64, ) -> Result<(), 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, ..current }); Ok(()) } }
impl ResyncState { pub fn next_backoff(retry_count: u32) -> i64 { // the caller counts the failure before asking, so the first retry waits 2m, then 4m, // 8m... up to 1h781 unmodified linesExpand all311 unmodified linesExpand all312313314315316317318319320586 unmodified linesExpand all907908909910910911912913914107 unmodified linesExpand all311 unmodified linesExpand allcodec-call:src/db/types.rs::DbTid::from_s32::.to_be_bytes rust:src/db/types.rs::dbtid_display_parses_back_for_any_bytescodec-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/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_identity586 unmodified linesExpand allserde-type:src/control/firehose.rs::FirehoseDiagnosticsInfo nu:tests/api_firehose_sources.nuserde-type:src/control/firehose.rs::FirehoseFailureInfo nu:tests/api_firehose_sources.nuserde-type:src/control/firehose.rs::FirehosePdsInfo nu:tests/api_firehose_sources.nuserde-type:src/control/firehose.rs::FirehoseSourceInfo nu:tests/api_firehose_sources.nuserde-type:src/control/new_pds_crawl.rs::enabled::Allowance rust:src/control/new_pds_crawl.rs::the_allowance_and_a_queued_rewalk_survive_a_restartserde-type:src/control/pds.rs::PdsTierAssignment nu:tests/pds_host_status_transitions.nuserde-type:src/control/pds.rs::PdsTierDefinition nu:tests/pds_host_status_transitions.nuserde-type:src/control/repos/mod.rs::FailureKind rust:src/control/repos/mod.rs::repo_info_shows_why_a_repo_waits_for_resyncserde-type:src/control/repos/mod.rs::RepoInfo rust:src/control/repos/mod.rs::repo_info_json_contract107 unmodified linesExpand all1233456788 unmodified lines171819202021222324252613 unmodified lines404142434445464748495051525354555611 unmodified lines68697071617273747576175 unmodified linesExpand all252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324156 unmodified linesExpand alluse crate::db::types::TrimmedDid;use crate::db::{self, keys};use crate::ingest::indexer::repos::write_repo_blocking;use crate::ingest::indexer::refetch::{self, Charged};use crate::ingest::indexer::repos::on_owner_blocking;use crate::state::AppState;use crate::types::{GaugeState, ResyncState};use jacquard_common::types::did::Did;use miette::{IntoDiagnostic, Result};8 unmodified lines eligible: impl FnOnce(&ResyncState) -> bool + Send + Sync + 'static,) -> Result<bool> { let owner = did.clone(); let did = did.clone(); write_repo_blocking(state, &owner, move |db| { on_owner_blocking(state, &owner, move |state, _| { let db = &state.db; let mut txn = db::Txn::new(db); if txn.is_excluded(&did)? { return Ok(false); }13 unmodified lines .get(&metadata_key) .into_diagnostic()? .ok_or_else(|| miette::miette!("repo metadata not found"))?; let mut metadata = crate::db::deser_repo_meta(&metadata_bytes)?; // only a denied refetch gets charged on retry, an admitted one already waited its backoff if metadata.refetch_deferred && let Some(mut repo) = txn.read_repo(&did)? && refetch::charge(state, &mut txn, &did, &mut repo, &mut metadata, "deferred")? == Charged::Deferred { txn.commit()?; return Ok(false); } let old_pending = keys::pending_key(metadata.index_id); metadata.index_id = rand::random(); // a database from before v14 keeps the count only on the entry if let ResyncState::Error { retry_count, .. } = resync_state {11 unmodified lines ); txn.transition_lifecycle(&did, GaugeState::Pending)?; txn.commit()?; Ok(true) }) })?}
pub fn queue_gone_backfills(state: &Arc<AppState>) -> Result<()> { debug!("scanning for deactivated/takendown repos to retry...");175 unmodified linesExpand all } batch.commit().unwrap(); }
fn waiting(state: &AppState, entry: ResyncState, used: Option<(i64, u32)>, deferred: bool) { use crate::ingest::indexer::refetch::tests as refetch; let mut metadata = refetch::seed(state, true, used); metadata.refetch_deferred = deferred; refetch::put_metadata(state, &refetch::did(), &metadata); refetch::put_entry(state, &refetch::did(), &entry); }
fn error(kind: crate::types::ResyncErrorKind, next_retry: i64) -> ResyncState { ResyncState::Error { kind, retry_count: 2, next_retry, } }
#[test] fn retrying_a_failed_fetch_costs_no_refetch() { use crate::ingest::indexer::refetch::tests as refetch; let (_tmp, state) = refetch::state(Some(1)); let now = chrono::Utc::now().timestamp(); waiting( &state, error(crate::types::ResyncErrorKind::Transport, 0), Some((now, 1)), false, );
assert!(queue_resync_if(&state, &refetch::did(), |_| true).unwrap()); assert_eq!(state.db.indexer.pending_iter().count(), 1); assert_eq!( refetch::metadata_of(&state, &refetch::did()).refetch, Some(crate::types::Allowance { since: now, used: 1 }) ); }
#[test] fn retrying_a_denied_refetch_asks_for_one_again() { use crate::ingest::indexer::refetch::tests as refetch; let (_tmp, state) = refetch::state(Some(1)); let now = chrono::Utc::now().timestamp(); let denied = error(crate::types::ResyncErrorKind::Ratelimited, 0);
waiting(&state, denied.clone(), Some((now, 1)), true); assert!(!queue_resync_if(&state, &refetch::did(), |_| true).unwrap()); assert_eq!(state.db.indexer.pending_iter().count(), 0); assert!(matches!( refetch::entry_of(&state, &refetch::did()), Some(ResyncState::Error { next_retry, .. }) if next_retry == now + 3600 ));
waiting(&state, denied, Some((now - 7200, 1)), true); assert!(queue_resync_if(&state, &refetch::did(), |_| true).unwrap()); assert_eq!(state.db.indexer.pending_iter().count(), 1); let metadata = refetch::metadata_of(&state, &refetch::did()); assert!(!metadata.refetch_deferred); assert_eq!(metadata.refetch.map(|allowance| allowance.used), Some(1)); assert!(refetch::entry_of(&state, &refetch::did()).is_none()); assert_eq!(metadata.retry_count, 3, "waiting isn't failing"); }
#[test] fn excluded_repo_is_not_requeued() { let (_tmp, state) = state(); stage_retry(&state, true);
156 unmodified linesExpand all