diff --git a/src/backfill/worker/task.rs b/src/backfill/worker/task.rs index c08e139..4f44916 100644 --- a/src/backfill/worker/task.rs +++ b/src/backfill/worker/task.rs @@ -22,6 +22,14 @@ pub(super) enum TaskDisposition { Deferred, } +/// what writing down a failed backfill came to +enum Recorded { + /// the repo already sits in resync as gone, so only its queue entry goes + Gone, + /// `applied` when the failure moved the repo from the queue into resync + Retrying { applied: bool }, +} + fn stage_excluded_pending_cleanup( txn: &mut DbTxn<'_>, db: &crate::db::Db, @@ -87,8 +95,6 @@ pub(super) async fn did_task( verify_signatures: bool, strategy: BackfillStrategy, ) -> Result { - let db = &state.db; - // Drop stale/excluded work before resolver, admission, or network I/O. A // repository drop may race task spawning, and orphan pending keys are not // necessarily the one named by current metadata. @@ -246,75 +252,55 @@ pub(super) async fn did_task( BackfillError::UnsafeEndpoint(_) => unreachable!("handled above"), }; - let did_key = keys::repo_key(did); - - // 1. get current retry count - let did_key_clone = did_key.clone(); - let existing_state = db - .run(move |db| { - db.indexer - .resync - .get(&did_key_clone) - .into_diagnostic() - .and_then(|b| { - b.map(|b| rmp_serde::from_slice::(&b).into_diagnostic()) - .transpose() - }) - }) - .await?; - - let mut retry_count = match existing_state { - Some(ResyncState::Error { retry_count, .. }) => retry_count, - Some(ResyncState::Gone { .. }) => { - remove_pending(state, did, pending_key).await?; - return Ok(TaskDisposition::Finished); - } - None => 0, - }; - - // Calculate new stats - let next_retry = if matches!(e, BackfillError::PreemptivelyThrottled) { - chrono::Utc::now().timestamp() + 60 - } else { - retry_count += 1; - ResyncState::next_backoff(retry_count) - }; - - let resync_state = ResyncState::Error { - kind: error_kind, - retry_count, - next_retry, - }; let error_string = e.to_string(); - - let applied = state + let throttled = matches!(e, BackfillError::PreemptivelyThrottled); + let recorded = state .db .run({ - let did_key = did_key.into_static(); + let did_key = keys::repo_key(did).into_static(); let did = did.clone(); let pending_key = pending_key.clone(); move |db| { - // 3. save to resync - let serialized_resync_state = - rmp_serde::to_vec(&resync_state).into_diagnostic()?; - - // 4. and update the main repo state - let serialized_repo_state = - if let Some(state_bytes) = db.repos.get(&did_key).into_diagnostic()? { - let mut state: RepoState = - rmp_serde::from_slice(&state_bytes).into_diagnostic()?; - state.active = true; - state.status = RepoStatus::Error(error_string.into()); - Some(rmp_serde::to_vec(&state).into_diagnostic()?) - } else { - None - }; + // everything here is read under the repo lock, so a retry or a status + // change that lands while the fetch fails is built on, not overwritten let mut txn = DbTxn::new(db); + txn.hold_repo_write_lock(&did); + let existing = db + .indexer + .resync + .get(&did_key) + .into_diagnostic()? + .map(|bytes| { + rmp_serde::from_slice::(&bytes).into_diagnostic() + }) + .transpose()?; + let mut retry_count = match existing { + Some(ResyncState::Error { retry_count, .. }) => retry_count, + Some(ResyncState::Gone { .. }) => { + db.indexer + .stage_pending_remove(&mut txn.batch, &did, &pending_key); + txn.commit()?; + return Ok::<_, miette::Report>(Recorded::Gone); + } + None => 0, + }; if stage_excluded_pending_cleanup(&mut txn, db, &did, pending_key.as_ref())? { txn.commit()?; - return Ok::<_, miette::Report>(false); + return Ok(Recorded::Retrying { applied: false }); } + + let next_retry = if throttled { + chrono::Utc::now().timestamp() + 60 + } else { + retry_count += 1; + ResyncState::next_backoff(retry_count) + }; + let resync_state = ResyncState::Error { + kind: error_kind, + retry_count, + next_retry, + }; let applied = txn.transition_pending_key( &did, pending_key.as_ref(), @@ -323,25 +309,43 @@ pub(super) async fn did_task( db.indexer .stage_pending_remove(&mut txn.batch, &did, &pending_key); if applied { - txn.batch - .insert(&db.indexer.resync, &did_key, serialized_resync_state); - if let Some(state_bytes) = serialized_repo_state { - txn.batch.insert(&db.repos, &did_key, state_bytes); + txn.batch.insert( + &db.indexer.resync, + &did_key, + rmp_serde::to_vec(&resync_state).into_diagnostic()?, + ); + if let Some(bytes) = db.repos.get(&did_key).into_diagnostic()? { + let mut repo: RepoState = + rmp_serde::from_slice(&bytes).into_diagnostic()?; + // a failed fetch says nothing about the account, so one that + // went inactive keeps its status + if repo.active { + repo.status = RepoStatus::Error(error_string.into()); + } + txn.batch.insert( + &db.repos, + &did_key, + rmp_serde::to_vec(&repo).into_diagnostic()?, + ); } } txn.commit()?; - Ok::<_, miette::Report>(applied) + Ok(Recorded::Retrying { applied }) } }) .await?; - if applied { - state - .backfill_drained - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + match recorded { + Recorded::Gone => Ok(TaskDisposition::Finished), + Recorded::Retrying { applied } => { + if applied { + state + .backfill_drained + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + } + Err(e) + } } - - Err(e) } } } @@ -1490,6 +1494,37 @@ mod tests { Ok(()) } + #[tokio::test] + async fn a_failed_fetch_leaves_a_repo_that_went_inactive_inactive() -> miette::Result<()> { + let fixture = Fixture::new().await?; + let seeded = rooted("3jzfcijpj2z2a")?; + fixture.seed(&seeded, 7, &[])?; + + // the stub answers getRepo with RepoNotFound, which the task retries + let result = fixture + .run_changing(keys::pending_key(7), async { + let mut row = fixture.repo()?; + row.active = false; + row.status = RepoStatus::Deactivated; + fixture.put_repo(&row) + }) + .await?; + + assert!( + matches!(result, Err(BackfillError::RepoNotFound)), + "{result:?}" + ); + let after = fixture.repo()?; + assert_eq!(after.status, RepoStatus::Deactivated); + assert!(!after.active); + assert!(matches!( + fixture.resync()?, + Some(ResyncState::Error { retry_count: 1, .. }) + )); + assert!(fixture.pending_keys()?.is_empty()); + Ok(()) + } + #[tokio::test] async fn a_later_valid_car_replaces_the_error_after_requeue() -> miette::Result<()> { let fixture = Fixture::new().await?;