From 51c7ddf23d17c16199b61d38cd878077406fd317 Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Mon, 5 Oct 2026 21:11:33 +0300 Subject: [PATCH] [backfill] read a failed backfill's resync and repo state under the repo lock the error path read the retry count and the repo row before taking the lock, so a status change committed in between was overwritten. it also set the repo active with an error status, which brought back an account that went inactive while the fetch failed. both reads now happen in the locked txn that writes, and a failed fetch leaves an inactive account's status alone. --- src/backfill/worker/task.rs | 177 +++++++++++++++++++++--------------- 1 file changed, 106 insertions(+), 71 deletions(-) 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?; -- 2.51.2