[firehose] keep retrying offline hosts instead of giving up
509 unmodified linesExpand all510511512513514514515516517518519520521581 unmodified linesExpand all11031104110511061104110711081109110711101111111211131111111411151116111711185 unmodified lines11241125112611271125112811291130113111321133113411311132113511361137113411381139114011411142111 unmodified linesExpand all125412551256125712541258125912601261126226 unmodified lines12891290129112921289129012931294129512961297129812991300279 unmodified linesExpand all509 unmodified linesExpand all } #[cfg(test)] self.pause_validated_admission_after_ban_check_if_requested() .await; // a request wakes an offline host's task instead of waiting out its backoff if self.is_source_running(url) { if self.is_source_running(url) && self.state.pds_meta.load().status(host) != crate::pds_meta::HostStatus::Offline { return Ok(Some(Ok(PdsAdmission::AlreadyRunning))); } if self.is_source_known(url) { // Do not route this through admin add_source: that trusted override581 unmodified linesExpand all assert!(!hydrant.firehose.is_source_running(&url)); Ok(()) }
/// the crawler waits on a relay's dispatched time, which a relay that gave up never moves /// the crawler waits on a relay's dispatched time, which an offline relay doesn't move #[cfg(feature = "indexer")] #[tokio::test] async fn a_relay_that_gives_up_stops_counting_for_the_crawler() -> Result<()> { async fn an_offline_relay_stops_counting_for_the_crawler() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let hydrant = Hydrant::new(test_config(&dir)).await?; startable(&hydrant); // nothing listens there, and the test firehose gives up after one failure // nothing listens there, and the test firehose goes offline after one failure let url = Url::parse("ws://127.0.0.1:1/").into_diagnostic()?;
let counted = || { hydrant5 unmodified lines
hydrant.firehose.add_source(url.clone(), false).await?; assert!(counted()); for _ in 0..500 { if !hydrant.firehose.is_source_running(&url) { if !counted() { break; } tokio::time::sleep(std::time::Duration::from_millis(10)).await; } assert!(!counted(), "the relay never went offline"); assert!( !hydrant.firehose.is_source_running(&url), "the relay never gave up" hydrant.firehose.is_source_running(&url), "and it keeps retrying" ); assert!(!counted()); hydrant.firehose.tasks.remove_async(&url).await; Ok(()) }
#[tokio::test]111 unmodified linesExpand all }
/// a host that keeps failing could otherwise call requestCrawl to wipe its own backoff #[tokio::test] async fn a_validated_request_restarts_a_source_only_after_it_gave_up() -> Result<()> { async fn a_validated_request_restarts_a_source_only_once_its_offline() -> Result<()> { let dir = tempfile::tempdir().into_diagnostic()?; let hydrant = Hydrant::new(test_config(&dir)).await?; startable(&hydrant); hydrant.firehose.disable_spawns_for_test();26 unmodified lines ); assert!(failing(), "a source still retrying keeps its backoff"); assert!(!cancel.is_cancelled(), "and its task is left running");
// a task removes itself from `tasks` when it gives up hydrant.firehose.tasks.remove_async(&url).await; // past max_failures the task marks the host offline and keeps retrying slowly crate::pds_meta::PdsMeta::update_host(&hydrant.state.pds_meta, "retrying.example", |h| { h.status = crate::pds_meta::HostStatus::Offline }); assert_eq!( hydrant.firehose.add_source_validated(url.clone()).await?, Ok(PdsAdmission::RestartedKnown) );279 unmodified linesExpand all22 unmodified lines232425262728293031215 unmodified linesExpand all247248249250250251252253254255256257258259260261262263264251252253254255100 unmodified lines35635735835937336036136236336436536667 unmodified lines43443543643744945045145245345445545645745845946046146243843944044144211 unmodified lines4544554564574824834844854864874884894904914924934944954584594604614626 unmodified lines4694704714725105115125135145155165175185195205215225235244734744754764778 unmodified lines486487488489541542543490491492493494495496497498552499500501502503504558505506560561562507508509510511565512567513514515516517518519520521522523524164 unmodified linesExpand all689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741311 unmodified linesExpand all22 unmodified linesuse url::Url;
// these match ref relayconst MAX_BACKOFF: Duration = Duration::from_secs(60);const MAX_OFFLINE_BACKOFF: Duration = Duration::from_secs(3600);
#[derive(Debug, Clone)]struct FirehoseFailure { kind: &'static str,215 unmodified linesExpand all Ok(s) => s, Err(e) => { let failure = classify_firehose_error(&e); self.stats.record_connect_error(failure.kind); let secs = match self.on_failure(&failure).await { Some(secs) => secs, None => { error!( err = %e, failure_kind = failure.kind, failure = %failure.detail, failures = self.throttle.consecutive_failures(), max_failures = self.max_failures, "failed to connect to firehose, giving up" ); break Ok(()); } }; let timeout = rng.add_jitter(Duration::from_secs(secs).min(MAX_BACKOFF)); let timeout = self.on_failure(&failure, &mut rng).await; let fmt = crate::util::format_duration(timeout); error!( err = %e, failure_kind = failure.kind,100 unmodified lines marked_active = true; // only reset failure state once the stream has been healthy for a bit // so we dont get in a "connects successfully, sends garbage" situation self.throttle.record_success(); if self.is_pds { if !self.is_pds { self.state.track_dispatched(&self.relay_host).await; } else { let (current_status, tier) = { let meta = self.state.pds_meta.load(); let override_name = meta.hosts.get(host).and_then(|h| h.tier.as_ref()); (meta.status(host), self.state.tier_policy.resolve(host, override_name))67 unmodified lines tokio::time::sleep(Duration::from_secs(1)).await; continue; }
let secs = match self.on_failure(&failure).await { Some(secs) => secs, None => { error!( failure_kind = failure.kind, failure = %failure.detail, failures = self.throttle.consecutive_failures(), max_failures = self.max_failures, "relay reported a future cursor without a requested cursor, giving up" ); break Ok(()); } }; let timeout = rng.add_jitter(Duration::from_secs(secs).min(MAX_BACKOFF)); let timeout = self.on_failure(&failure, &mut rng).await; let fmt = crate::util::format_duration(timeout); error!( failure_kind = failure.kind, failure = %failure.detail,11 unmodified lines let failure = FirehoseFailure::new("relay_error", format!("{error}: {message}")); self.stats.record_stream_error(failure.kind); error!(err = %error, "relay sent error: {message}"); let secs = match self.on_failure(&failure).await { Some(secs) => secs, None => { error!( failure_kind = failure.kind, failure = %failure.detail, failures = self.throttle.consecutive_failures(), max_failures = self.max_failures, "firehose disconnected, giving up" ); break Ok(()); } }; let timeout = rng.add_jitter(Duration::from_secs(secs).min(MAX_BACKOFF)); let timeout = self.on_failure(&failure, &mut rng).await; let fmt = crate::util::format_duration(timeout); error!( failure_kind = failure.kind, failure = %failure.detail,6 unmodified lines } Err(e) => { let failure = classify_firehose_error(&e); self.stats.record_stream_error(failure.kind); let secs = match self.on_failure(&failure).await { Some(secs) => secs, None => { error!( err = %e, failure_kind = failure.kind, failure = %failure.detail, failures = self.throttle.consecutive_failures(), max_failures = self.max_failures, "firehose stream error, giving up" ); break Ok(()); } }; let timeout = rng.add_jitter(Duration::from_secs(secs).min(MAX_BACKOFF)); let timeout = self.on_failure(&failure, &mut rng).await; let fmt = crate::util::format_duration(timeout); error!( err = %e, failure_kind = failure.kind,8 unmodified lines } } }
/// record a failure and return the backoff duration in seconds, /// or `None` if the failure threshold was reached and the subscriber should stop async fn on_failure(&self, failure: &FirehoseFailure) -> Option<u64> { async fn on_failure(&self, failure: &FirehoseFailure, rng: &mut SmallRng) -> Duration { let secs = self .throttle .record_failure_detail(failure.kind, failure.detail.clone()) .unwrap_or_else(|| { let until = self.throttle.throttled_until(); 0.max(until - chrono::Utc::now().timestamp()) as u64 }); let failures = self.throttle.consecutive_failures(); if failures >= self.max_failures { if failures == self.max_failures { warn!( failures, max_failures = self.max_failures, failure_kind = failure.kind, failure = %failure.detail, "too many consecutive failures, giving up on host" "too many consecutive failures, marking host offline" ); if self.is_pds && let Err(e) = self.set_host_status(HostStatus::Offline) { if !self.is_pds { self.state.forget_dispatched(&self.relay_host).await; } else if let Err(e) = self.set_host_status(HostStatus::Offline) { error!(err = %e, "failed to update host status to offline"); } return None; } Some(secs) let Some(past) = failures.checked_sub(self.max_failures) else { return rng.add_jitter(Duration::from_secs(secs).min(MAX_BACKOFF)); }; let backoff = MAX_BACKOFF .saturating_mul(2u32.pow(past.min(6) as u32 + 1)) .min(MAX_OFFLINE_BACKOFF); // hosts that went down together would otherwise all retry at once rng.random_range(backoff / 2..=backoff) }
fn set_host_status(&self, status: HostStatus) -> Result<()> { let Some(host) = self.relay_host.host_str() else {164 unmodified linesExpand all ) }
#[tokio::test] async fn a_host_past_max_failures_goes_offline_and_is_retried_up_to_hourly() -> Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; let hydrant = Hydrant::new(Config { database_path: tmp.path().to_path_buf(), ..Config::default() }) .await?; let host = "flaky.example"; let ingestor = FirehoseIngestor::new( hydrant.state.clone(), BufferTx::channel(1).0, Url::parse("wss://flaky.example/").into_diagnostic()?, true, false, hydrant.state.filter.clone(), hydrant.state.firehose_enabled.subscribe(), false, 2, ) .await; let failure = FirehoseFailure::new("test", "down"); let mut rng: SmallRng = rand::make_rng(); let mut waits = Vec::new(); for _ in 0..20 { waits.push(ingestor.on_failure(&failure, &mut rng).await); }
assert!( waits[0] <= MAX_BACKOFF, "before max_failures the ref relay backoff applies" ); assert_eq!( hydrant.state.pds_meta.load().status(host), HostStatus::Offline ); assert!((MAX_BACKOFF..=2 * MAX_BACKOFF).contains(&waits[1])); assert!(waits.iter().all(|w| *w <= MAX_OFFLINE_BACKOFF)); assert!( waits[19] >= MAX_OFFLINE_BACKOFF / 2, "and it keeps retrying" ); Ok(()) }
#[tokio::test] async fn operational_transition_does_not_clear_an_operator_ban() -> Result<()> { let tmp = tempfile::tempdir().into_diagnostic()?; let cfg = Config { database_path: tmp.path().to_path_buf(),311 unmodified linesExpand all