Something went wrong. Try again.
unified development activity
Something went wrong. Try again.
11 kB · 316 lines
Rust
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317//! A small HTTP client with per-host pacing.//!//! Tangled's mirror and knots answer a burst with a bare `403` and no//! `Retry-After`. The crawler is a long, polite background job, so the client//! spaces requests per host and retries the statuses that mean "later" rather//! than treating a single 403 as a dead repository.
use std::collections::HashMap;use std::sync::Arc;use std::time::Duration;
use anyhow::{Context, Result, bail};use reqwest::header::{HeaderMap, HeaderValue, USER_AGENT};use reqwest::{Client, StatusCode};use tokio::sync::Mutex;use tokio::time::{Instant, sleep};
#[derive(Debug, Clone)]pub struct HttpClient { client: Client, gates: Arc<Mutex<HashMap<String, Arc<HostGate>>>>, interval: Duration, attempts: u32, circuit_failures: u32, circuit_cooldown: Duration,}
#[derive(Debug)]struct HostGate { state: Mutex<GateState>,}
#[derive(Debug)]struct GateState { next: Instant, failures: u32, open_until: Option<Instant>, /// The failure that opened the circuit, so a request it refuses says /// what is wrong with the host rather than only that it was skipped. opened_by: Option<String>,}
#[derive(Debug)]pub struct HttpResponse { pub headers: HeaderMap, pub body: Vec<u8>,}
impl HttpClient { pub fn new(timeout: Duration, per_host_rps: f64) -> Result<Self> { let rps = if per_host_rps > 0.0 { per_host_rps } else { 1.0 }; let client = Client::builder() .timeout(timeout) .user_agent(format!( "tally/{} (+https://dev.permadeath.com)", env!("CARGO_PKG_VERSION") )) .build() .context("building HTTP client")?; Ok(Self { client, gates: Arc::new(Mutex::new(HashMap::new())), interval: Duration::from_secs_f64(1.0 / rps), attempts: 5, circuit_failures: 5, circuit_cooldown: Duration::from_secs(60), }) }
/// The global crawl can use fewer attempts and a shorter circuit than an /// interactive run, because a broken mirror is better skipped than retried /// for every repository. pub fn with_policy( timeout: Duration, per_host_rps: f64, attempts: u32, circuit_failures: u32, circuit_cooldown: Duration, ) -> Result<Self> { let mut client = Self::new(timeout, per_host_rps)?; client.attempts = attempts.max(1); client.circuit_failures = circuit_failures.max(1); client.circuit_cooldown = circuit_cooldown; Ok(client) }
/// GET a URL, pacing by host and retrying transient failures. pub async fn get(&self, url: &str, bearer: Option<&str>) -> Result<HttpResponse> { let host = url .split("://") .nth(1) .and_then(|rest| rest.split('/').next()) .unwrap_or(url) .to_ascii_lowercase();
let gate = { let mut gates = self.gates.lock().await; gates .entry(host.clone()) .or_insert_with(|| { Arc::new(HostGate { state: Mutex::new(GateState { next: Instant::now(), failures: 0, open_until: None, opened_by: None, }), }) }) .clone() };
let attempts = self.attempts; for attempt in 0..attempts { self.wait_for_turn(&host, &gate).await?;
let mut request = self.client.get(url).header( USER_AGENT, HeaderValue::from_static(concat!( "tally/", env!("CARGO_PKG_VERSION"), " (+https://dev.permadeath.com)" )), ); if let Some(token) = bearer { request = request.bearer_auth(token); }
let response = match request.send().await { Ok(response) => response, Err(err) => { let opened = self .record_failure(&host, &gate, || { format!("GET {url}: {}", error_chain(&err)) }) .await; if opened || attempt + 1 >= attempts { return Err(err).with_context(|| format!("GET {url}")); } tracing::warn!(url, attempt, error = %err, "HTTP request failed; retrying"); sleep(backoff(attempt)).await; continue; } };
let status = response.status(); let headers = response.headers().clone();
if status.is_success() { self.record_success(&gate).await; let body = response.bytes().await?.to_vec(); return Ok(HttpResponse { headers, body }); }
// GitHub tells us exactly when the window resets. Waiting here // would stall a scheduled run; returning lets the caller record // the partial scan and resume on the next tick. if status == StatusCode::FORBIDDEN && headers .get("x-ratelimit-remaining") .and_then(|value| value.to_str().ok()) == Some("0") { let reset = headers .get("x-ratelimit-reset") .and_then(|value| value.to_str().ok()) .unwrap_or("unknown"); bail!("rate limited by {host}; reset at {reset}"); }
let transient = status == StatusCode::TOO_MANY_REQUESTS || status == StatusCode::FORBIDDEN || status == StatusCode::BAD_GATEWAY || status == StatusCode::SERVICE_UNAVAILABLE || status == StatusCode::GATEWAY_TIMEOUT || status.is_server_error();
if transient { let opened = self .record_failure(&host, &gate, || { format!("GET {url} returned {}", status.as_u16()) }) .await; if opened || attempt + 1 >= attempts { let body = response.text().await.unwrap_or_default(); bail!( "GET {url} returned {}: {}", status.as_u16(), body.chars().take(300).collect::<String>() ); } let delay = backoff(attempt); tracing::warn!( url, attempt, status = status.as_u16(), delay_ms = delay.as_millis() as u64, "transient HTTP response; backing off" ); sleep(delay).await; continue; }
let body = response.text().await.unwrap_or_default(); bail!( "GET {url} returned {}: {}", status.as_u16(), body.chars().take(300).collect::<String>() ); }
unreachable!("the retry loop always returns") }
pub async fn get_json<T: serde::de::DeserializeOwned>( &self, url: &str, bearer: Option<&str>, ) -> Result<T> { let response = self.get(url, bearer).await?; serde_json::from_slice(&response.body).with_context(|| format!("decoding JSON from {url}")) }
async fn wait_for_turn(&self, host: &str, gate: &HostGate) -> Result<()> { let mut state = gate.state.lock().await; let now = Instant::now(); if let Some(open_until) = state.open_until { if open_until > now { bail!( "circuit open for {host} for {}s after: {}", (open_until - now).as_secs().max(1), state.opened_by.as_deref().unwrap_or("repeated failures") ); } state.open_until = None; state.opened_by = None; } let wait = if state.next > now { state.next - now } else { Duration::ZERO }; state.next = state.next.max(now) + self.interval; drop(state); if !wait.is_zero() { sleep(wait).await; } Ok(()) }
/// Returns true when this failure opened the circuit. `failure` says /// what failed, for the requests the open circuit refuses. async fn record_failure( &self, host: &str, gate: &HostGate, failure: impl FnOnce() -> String, ) -> bool { let mut state = gate.state.lock().await; state.failures += 1; if state.failures < self.circuit_failures { return false; } state.open_until = Some(Instant::now() + self.circuit_cooldown); state.opened_by = Some(failure()); state.failures = 0; tracing::warn!( host, cooldown_secs = self.circuit_cooldown.as_secs(), "HTTP circuit opened after repeated failures" ); true }
async fn record_success(&self, gate: &HostGate) { let mut state = gate.state.lock().await; state.failures = 0; state.open_until = None; state.opened_by = None; }}
/// An error and its causes on one line, as `{:#}` prints an `anyhow` error:/// a connection failure's cause ("dns error", "connection refused") is in/// the chain, not the top-level message.fn error_chain(err: &dyn std::error::Error) -> String { let mut text = err.to_string(); let mut source = err.source(); while let Some(cause) = source { text.push_str(": "); text.push_str(&cause.to_string()); source = cause.source(); } text}
fn backoff(attempt: u32) -> Duration { // 300ms, 600ms, 1.2s, 2.4s ... with a small deterministic stagger so two // concurrent requests do not wake together. let base = 300u64.saturating_mul(1u64 << attempt.min(6)); Duration::from_millis(base + (u64::from(attempt) * 73))}
#[cfg(test)]mod tests { use super::*;
#[test] fn backoff_grows() { assert!(backoff(3) > backoff(1)); }}