use crate::db::types::TrimmedDid;
use crate::db::{self, keys};
use crate::ingest::indexer::refetch::{self, Charged};
use crate::ingest::indexer::repos::on_owner_blocking;
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};
pub(crate) fn queue_resync_if(
state: &AppState,
did: &Did,
eligible: impl FnOnce(&ResyncState) -> bool + Send + Sync + 'static,
) -> Result<bool> {
let owner = did.clone();
let did = did.clone();
on_owner_blocking(state, &owner, move |state, _| {
let db = &state.db;
let mut txn = db::Txn::new(db);
if txn.is_excluded(&did)? {
return Ok(false);
}
let did_key = keys::repo_key(&did);
let Some(value) = 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 = 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)?;
// only a denied refetch gets charged on retry, an admitted one already waited its backoff
if metadata.refetch_deferred
&& let Some(mut repo) = txn.read_repo(&did)?
&& refetch::charge(state, &mut txn, &did, &mut repo, &mut metadata, "deferred")?
== Charged::Deferred
{
txn.commit()?;
return Ok(false);
}
let old_pending = keys::pending_key(metadata.index_id);
metadata.index_id = rand::random();
// a database from before v14 keeps the count only on the entry
if let ResyncState::Error { retry_count, .. } = resync_state {
metadata.retry_count = metadata.retry_count.max(retry_count);
}
txn.batch.remove(&db.indexer.resync, &did_key);
let indexer = &db.indexer;
indexer.stage_pending_remove(&mut txn.batch, &did, &old_pending);
indexer.stage_pending_insert(&mut txn.batch, &did, &keys::pending_key(metadata.index_id));
txn.batch.insert(
&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");
state.db.poison.check_report(&e);
}
}
}
if transitions > 0 {
state.notify_backfill();
info!(count = transitions, "queued gone backfills");
}
Ok(())
}
pub fn retry_worker(state: Arc<AppState>) {
info!("retry worker started");
while !state.stop.wait(Duration::from_secs(60)) {
let transitions = queue_due_retries(&state, chrono::Utc::now().timestamp());
if transitions > 0 {
state.notify_backfill();
info!(count = transitions, "queued retries");
}
}
}
/// queues every failed backfill whose retry is due at `now`, and says how many
fn queue_due_retries(state: &AppState, now: i64) -> usize {
let mut candidates = Vec::new();
for guard in state.db.indexer.resync.iter() {
let (key, value) = match guard.into_inner() {
Ok(t) => t,
Err(e) => {
error!(err = %e, "failed to get resync state");
state.db.poison.check(&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 {
// shutdown takes the shards away, and a requeue then would write next to one still
// finishing its step
if state.stop.is_set() {
break;
}
let did_key = keys::repo_key(&did);
let is_pds_throttled = state
.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,
move |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");
state.db.poison.check_report(&e);
}
}
}
transitions
}
#[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();
}
fn waiting(state: &AppState, entry: ResyncState, used: Option<(i64, u32)>, deferred: bool) {
use crate::ingest::indexer::refetch::tests as refetch;
let mut metadata = refetch::seed(state, true, used);
metadata.refetch_deferred = deferred;
refetch::put_metadata(state, &refetch::did(), &metadata);
refetch::put_entry(state, &refetch::did(), &entry);
}
fn error(kind: crate::types::ResyncErrorKind, next_retry: i64) -> ResyncState {
ResyncState::Error {
kind,
retry_count: 2,
next_retry,
}
}
#[test]
fn retrying_a_failed_fetch_costs_no_refetch() {
use crate::ingest::indexer::refetch::tests as refetch;
let (_tmp, state) = refetch::state(Some(1));
let now = chrono::Utc::now().timestamp();
waiting(
&state,
error(crate::types::ResyncErrorKind::Transport, 0),
Some((now, 1)),
false,
);
assert!(queue_resync_if(&state, &refetch::did(), |_| true).unwrap());
assert_eq!(state.db.indexer.pending_iter().count(), 1);
assert_eq!(
refetch::metadata_of(&state, &refetch::did()).refetch,
Some(crate::types::Allowance {
since: now,
used: 1
})
);
}
#[test]
fn retrying_a_denied_refetch_asks_for_one_again() {
use crate::ingest::indexer::refetch::tests as refetch;
let (_tmp, state) = refetch::state(Some(1));
let now = chrono::Utc::now().timestamp();
let denied = error(crate::types::ResyncErrorKind::Ratelimited, 0);
waiting(&state, denied.clone(), Some((now, 1)), true);
assert!(!queue_resync_if(&state, &refetch::did(), |_| true).unwrap());
assert_eq!(state.db.indexer.pending_iter().count(), 0);
assert!(matches!(
refetch::entry_of(&state, &refetch::did()),
Some(ResyncState::Error { next_retry, .. }) if next_retry == now + 3600
));
waiting(&state, denied, Some((now - 7200, 1)), true);
assert!(queue_resync_if(&state, &refetch::did(), |_| true).unwrap());
assert_eq!(state.db.indexer.pending_iter().count(), 1);
let metadata = refetch::metadata_of(&state, &refetch::did());
assert!(!metadata.refetch_deferred);
assert_eq!(metadata.refetch.map(|allowance| allowance.used), Some(1));
assert!(refetch::entry_of(&state, &refetch::did()).is_none());
assert_eq!(metadata.retry_count, 3, "waiting isn't failing");
}
#[test]
fn a_repo_the_host_cap_denied_is_requeued_once_the_hosts_window_ends() {
use crate::ingest::indexer::refetch::tests as refetch;
let (_tmp, state) = refetch::host_state_in(Some(1), None, Duration::from_secs(1));
let repos = refetch::dids(2);
for did in &repos {
refetch::on_a_host(&state, did, true);
}
assert_eq!(
refetch::charge_did(&state, &repos[0]),
Charged::Allowed,
"this one spends the host's refetch"
);
assert_eq!(refetch::charge_did(&state, &repos[1]), Charged::Deferred);
let queue = |did| queue_resync_if(&state, did, |_| true).unwrap();
assert!(!queue(&repos[1]), "the host's window is still open");
assert!(refetch::metadata_of(&state, &repos[1]).refetch_deferred);
std::thread::sleep(Duration::from_millis(1200));
assert!(queue(&repos[1]));
let metadata = refetch::metadata_of(&state, &repos[1]);
assert!(!metadata.refetch_deferred);
assert_eq!(state.db.indexer.pending_iter().count(), 1);
assert!(refetch::entry_of(&state, &repos[1]).is_none());
state.db.indexer.assert_pending_dids_match();
}
#[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_eq!(state.db.indexer.pending_iter().count(), 0);
state.db.indexer.assert_pending_dids_match();
}
#[test]
fn requeue_keeps_the_retry_count_an_older_entry_holds() {
let (_tmp, state) = state();
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),
rmp_serde::to_vec(&crate::types::v4::RepoMetadata {
tracked: true,
index_id: 7,
})
.unwrap(),
);
batch.insert(
&state.db.indexer.resync,
keys::repo_key(&did),
rmp_serde::to_vec(&ResyncState::Error {
kind: crate::types::ResyncErrorKind::Generic,
retry_count: 5,
next_retry: 0,
})
.unwrap(),
);
batch.commit().unwrap();
assert!(queue_resync_if(&state, &did, |_| true).unwrap());
let metadata = state
.db
.repo_metadata
.get(keys::repo_metadata_key(&did))
.unwrap()
.unwrap();
assert_eq!(
crate::db::deser_repo_meta(&metadata).unwrap().retry_count,
5
);
state.db.indexer.assert_pending_dids_match();
}
#[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_iter().count(), 1);
state.db.indexer.assert_pending_dids_match();
}
#[tokio::test]
async fn a_retry_is_queued_on_the_repos_indexer_shard() {
let (_tmp, state) = state();
let state = Arc::new(state);
stage_retry(&state, false);
// the repo's shard never runs, so a requeue that goes through it never happens
let _shards =
crate::ingest::indexer::FirehoseWorker::serve_shards(state.clone(), 1, Some(0));
let queued = tokio::task::spawn_blocking({
let state = state.clone();
move || queue_resync_if(&state, &did(), |_| true)
})
.await
.unwrap();
assert!(queued.is_err());
assert!(
state
.db
.indexer
.resync
.contains_key(keys::repo_key(&did()))
.unwrap()
);
assert_eq!(state.db.indexer.pending_iter().count(), 0);
}
fn stage_due_retry(state: &AppState) {
stage_retry(state, false);
let failed = ResyncState::Error {
kind: crate::types::ResyncErrorKind::Generic,
retry_count: 1,
next_retry: 0,
};
state
.db
.indexer
.resync
.insert(keys::repo_key(&did()), rmp_serde::to_vec(&failed).unwrap())
.unwrap();
}
#[test]
fn a_due_retry_is_queued() {
let (_tmp, state) = state();
stage_due_retry(&state);
assert_eq!(queue_due_retries(&state, 1), 1);
assert_eq!(state.db.indexer.pending_iter().count(), 1);
}
#[test]
fn a_stopping_hydrant_queues_no_retries() {
let (_tmp, state) = state();
stage_due_retry(&state);
state.stop.trigger();
// no shards run here, so a requeue would be written right here instead of on one
assert_eq!(queue_due_retries(&state, 1), 0);
assert!(
state
.db
.indexer
.resync
.contains_key(keys::repo_key(&did()))
.unwrap()
);
assert_eq!(state.db.indexer.pending_iter().count(), 0);
}
}