Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201use crate::backfill::admission::BackfillAdmission;use crate::backfill::client::ThrottledHttpClient;use crate::backfill::error::BackfillError;use crate::config::BackfillStrategy;use crate::db::{self, types::TrimmedDid};use crate::ingest::indexer::IndexerTx;use crate::net::public_http_with_timeout;use crate::state::AppState;use crate::util::WatchEnabledExt;use jacquard_common::IntoStatic;use jacquard_common::types::did::Did;use miette::Result;use std::sync::Arc;use std::time::Duration;use tokio::sync::Semaphore;use tracing::{Instrument, debug, error, info};
pub mod process;pub mod task;
pub struct BackfillWorker { state: Arc<AppState>, buffer_tx: IndexerTx, http: ThrottledHttpClient, semaphore: Arc<Semaphore>, admission: BackfillAdmission, verify_signatures: bool, strategy: BackfillStrategy, in_flight: Arc<scc::HashSet<Did>>, deferred: Arc<scc::HashSet<Did>>, enabled: tokio::sync::watch::Receiver<bool>,}
impl BackfillWorker { pub fn new( state: Arc<AppState>, buffer_tx: IndexerTx, timeout: Duration, concurrency_limit: usize, verify_signatures: bool, strategy: BackfillStrategy, proxies: Vec<url::Url>, enabled: tokio::sync::watch::Receiver<bool>, ) -> Result<Self> { // A CONNECT proxy resolves the origin itself, so it cannot carry the // socket-coupled public-PDS guarantee. Reject it at startup instead of // silently changing the address policy or pacing budget. if !proxies.is_empty() { return Err(miette::miette!( "backfill proxies are unsupported for public-PDS egress; remove HYDRANT_BACKFILL_PROXIES" )); } let client = public_http_with_timeout(timeout); info!("direct-only backfill client ready");
let admission = BackfillAdmission::new(concurrency_limit, state.throttler.per_pds_concurrency());
Ok(Self { state: state.clone(), buffer_tx, http: ThrottledHttpClient::new_with_body_limit( client, state.throttler.clone(), state.max_car_body_bytes, ), semaphore: Arc::new(Semaphore::new(concurrency_limit)), admission, verify_signatures, strategy, in_flight: Arc::new(scc::HashSet::new()), deferred: Arc::new(scc::HashSet::new()), enabled, }) }}
struct InFlightGuard { did: Did, set: Arc<scc::HashSet<Did>>,}
impl Drop for InFlightGuard { fn drop(&mut self) { let _ = self.set.remove_sync(&self.did); }}
impl BackfillWorker { pub async fn run(mut self) { info!("backfill worker started");
loop { self.enabled.wait_enabled("backfill").await; let mut spawned = 0;
for guard in self.state.db.indexer.pending.iter() { let (key, value) = match guard.into_inner() { Ok(kv) => kv, Err(e) => { error!(err = %e, "failed to read pending entry"); db::check_poisoned(&e); continue; } };
let did = match TrimmedDid::try_from(value.as_ref()) { Ok(d) => d.to_did(), Err(e) => { error!(err = %e, "invalid did in pending value"); continue; } };
// check before trying to acquire a permit so we dont acquire a permit // for no reason, the read will be cheap anyhow if self.in_flight.contains_sync(&did) || self.deferred.contains_sync(&did) { continue; }
let permit = match self.semaphore.clone().try_acquire_owned() { Ok(p) => p, Err(_) => break, };
// only mark as in flight if we can acquire a permit if self .in_flight .insert_sync(did.clone().into_static()) .is_err() { // a task is already running, weh // so we don't need this one anymore... break; }
let guard = InFlightGuard { did: did.clone().into_static(), set: self.in_flight.clone(), };
let state = self.state.clone(); let http = self.http.clone(); let did = did.clone(); let buffer_tx = self.buffer_tx.clone(); let verify = self.verify_signatures; let strategy = self.strategy; let admission = self.admission.clone(); let deferred = self.deferred.clone();
let span = tracing::info_span!("backfill", did = %did); tokio::spawn( async move { let _guard = guard; let res = task::did_task( &state, http, buffer_tx, &did, key, admission, permit, verify, strategy, ) .await;
match &res { Ok(task::TaskDisposition::Deferred) => { let _ = deferred.insert_sync(did.clone().into_static()); } Ok(task::TaskDisposition::Finished) | Err(_) => { deferred.clear_sync(); } }
if let Err(e) = res { match &e { BackfillError::Ratelimited | BackfillError::PreemptivelyThrottled => { debug!(err = %e, "process failed"); } _ => { error!(err = %e, "process failed"); } } if let BackfillError::Generic(report) = &e { db::check_poisoned_report(report); } }
// wake worker to pick up more (in case we were sleeping at limit) state.backfill_notify.notify_one(); } .instrument(span), );
spawned += 1; }
if spawned == 0 { // wait for new tasks self.state.backfill_notify.notified().await; } // loop immediately since we might have more tasks } }}