1060 unmodified lines
Expand all
);
Ok(())
}
/// 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<()> {
let dir = tempfile::tempdir().into_diagnostic()?;
let hydrant = Hydrant::new(test_config(&dir)).await?;
startable(&hydrant);
hydrant.firehose.disable_spawns_for_test();
let url = Url::parse("wss://retrying.example/").into_diagnostic()?;
let failing = || hydrant.state.throttler.snapshot(&url).is_failing();
hydrant.firehose.add_source(url.clone(), true).await?;
hydrant
.state
.throttler
.get_handle(&url)
.await
.record_failure();
let cancel = CancellationToken::new();
let _ = hydrant
.firehose
.tasks
.upsert_async(
url.clone(),
FirehoseIngestorHandle {
id: 0,
cancel: cancel.clone(),
},
)
.await;
assert_eq!(
hydrant.firehose.add_source_validated(url.clone()).await?,
Ok(PdsAdmission::AlreadyRunning)
);
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;
assert_eq!(
hydrant.firehose.add_source_validated(url.clone()).await?,
Ok(PdsAdmission::RestartedKnown)
);
assert!(!failing(), "a source that gave up starts over");
Ok(())
}
#[tokio::test]
async fn validated_admission_rejects_invalid_and_private_hosts() -> Result<()> {
let dir = tempfile::tempdir().into_diagnostic()?;
let hydrant = Hydrant::new(test_config(&dir)).await?;
270 unmodified lines
Expand all