Something went wrong. Try again.
fork of hydrant
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276use crate::db::types::TrimmedDid;use crate::db::{self, keys};use crate::state::AppState;use crate::types::{GaugeState, ResyncState};use jacquard_common::types::did::Did;use miette::{IntoDiagnostic, Result};
use std::sync::Arc;use std::time::Duration;use tracing::{debug, error, info};
fn queue_resync_if( state: &AppState, did: &Did, eligible: impl FnOnce(&ResyncState) -> bool,) -> Result<bool> { let mut txn = db::Txn::new(&state.db); if txn.lock_repo_and_is_excluded(did)? { return Ok(false); }
let did_key = keys::repo_key(did); let Some(value) = state.db.indexer.resync.get(&did_key).into_diagnostic()? else { return Ok(false); }; let resync_state: ResyncState = rmp_serde::from_slice(&value).into_diagnostic()?; if !eligible(&resync_state) { return Ok(false); }
let metadata_key = keys::repo_metadata_key(did); let metadata_bytes = state .db .repo_metadata .get(&metadata_key) .into_diagnostic()? .ok_or_else(|| miette::miette!("repo metadata not found"))?; let mut metadata = crate::db::deser_repo_meta(&metadata_bytes)?; let old_pending = keys::pending_key(metadata.index_id); metadata.index_id = rand::random();
txn.batch.remove(&state.db.indexer.resync, &did_key); txn.batch.remove(&state.db.indexer.pending, old_pending); txn.batch.insert( &state.db.indexer.pending, keys::pending_key(metadata.index_id), &did_key, ); txn.batch.insert( &state.db.repo_metadata, metadata_key, crate::db::ser_repo_meta(&metadata)?, ); 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..."); let mut candidates = Vec::new(); for guard in state.db.indexer.resync.iter() { let (key, val) = guard.into_inner().into_diagnostic()?; let did = match TrimmedDid::try_from(key.as_ref()) { Ok(did) => did.to_did(), Err(e) => { error!(err = %e, "invalid did in db, skipping"); continue; } };
if let Ok(resync_state) = rmp_serde::from_slice::<ResyncState>(&val) && matches!(resync_state, ResyncState::Gone { .. }) { candidates.push(did); } }
let mut transitions = 0usize; for did in candidates { debug!(did = %did, "queuing retry for gone repo"); match queue_resync_if(state, &did, |resync| { matches!(resync, ResyncState::Gone { .. }) }) { Ok(true) => transitions += 1, Ok(false) => {} Err(e) => { error!(did = %did, err = %e, "failed to queue gone repo"); db::check_poisoned_report(&e); } } } if transitions > 0 { state.notify_backfill(); info!(count = transitions, "queued gone backfills"); } Ok(())}
pub fn retry_worker(state: Arc<AppState>) { let db = &state.db; info!("retry worker started"); loop { std::thread::sleep(Duration::from_secs(60));
let now = chrono::Utc::now().timestamp(); let mut candidates = Vec::new(); for guard in db.indexer.resync.iter() { let (key, value) = match guard.into_inner() { Ok(t) => t, Err(e) => { error!(err = %e, "failed to get resync state"); db::check_poisoned(&e); continue; } }; let did = match TrimmedDid::try_from(key.as_ref()) { Ok(did) => did.to_did(), Err(e) => { error!(err = %e, "invalid did in db, skipping"); continue; } };
match rmp_serde::from_slice::<ResyncState>(&value) { Ok(ResyncState::Error { next_retry, .. }) if next_retry <= now => { candidates.push(did); } Ok(_) => {} Err(e) => { error!(did = %did, err = %e, "failed to deserialize resync state"); } } }
let mut transitions = 0usize; for did in candidates { let did_key = keys::repo_key(&did); let is_pds_throttled = db .repos .get(&did_key) .ok() .flatten() .and_then(|bytes| { let repo = rmp_serde::from_slice::<crate::types::RepoState>(&bytes).ok()?; repo.pds .as_ref() .and_then(|pds| url::Url::parse(pds.as_ref()).ok()) }) .is_some_and(|pds| { state .throttler .snapshot(&pds) .is_throttled(chrono::Utc::now().timestamp()) }); if is_pds_throttled { continue; }
debug!(did = %did, "retrying backfill"); match queue_resync_if( &state, &did, |resync| matches!(resync, ResyncState::Error { next_retry, .. } if *next_retry <= now), ) { Ok(true) => transitions += 1, Ok(false) => {} Err(e) => { error!(did = %did, err = %e, "failed to queue retry"); db::check_poisoned_report(&e); } } } if transitions > 0 { state.notify_backfill(); info!(count = transitions, "queued retries"); } }}
#[cfg(test)]mod tests { use super::*; use crate::config::Config; use crate::types::{RepoMetadata, RepoState};
fn state() -> (tempfile::TempDir, AppState) { let tmp = tempfile::tempdir().unwrap(); let config = Config { database_path: tmp.path().to_path_buf(), ..Default::default() }; let state = AppState::new(&config).unwrap(); (tmp, state) }
fn did() -> Did { Did::new_static("did:plc:ewvi7nxzyoun6zhxrhs64oiz").unwrap() }
fn stage_retry(state: &AppState, excluded: bool) { let did = did(); let mut batch = state.db.inner.batch(); batch.insert( &state.db.repos, keys::repo_key(&did), crate::db::ser_repo_state(&RepoState::synced()).unwrap(), ); batch.insert( &state.db.repo_metadata, keys::repo_metadata_key(&did), crate::db::ser_repo_meta(&RepoMetadata::backfilling(7)).unwrap(), ); batch.insert( &state.db.indexer.resync, keys::repo_key(&did), rmp_serde::to_vec(&ResyncState::Gone { status: crate::types::RepoStatus::Deactivated, }) .unwrap(), ); if excluded { batch.insert( &state.db.filter, crate::db::filter::exclude_key(did.as_str()).unwrap(), [], ); } batch.commit().unwrap(); }
#[test] fn excluded_repo_is_not_requeued() { let (_tmp, state) = state(); stage_retry(&state, true);
assert!( !queue_resync_if(&state, &did(), |resync| { matches!(resync, ResyncState::Gone { .. }) }) .unwrap() ); assert!( state .db .indexer .resync .contains_key(keys::repo_key(&did())) .unwrap() ); assert!(state.db.indexer.pending.is_empty().unwrap()); }
#[test] fn eligible_repo_moves_from_resync_to_pending() { let (_tmp, state) = state(); stage_retry(&state, false);
assert!( queue_resync_if(&state, &did(), |resync| { matches!(resync, ResyncState::Gone { .. }) }) .unwrap() ); assert!( !state .db .indexer .resync .contains_key(keys::repo_key(&did())) .unwrap() ); assert_eq!(state.db.indexer.pending.len().unwrap(), 1); }}