diff --git a/src/control/repos/mod.rs b/src/control/repos/mod.rs index f278456..acaead1 100644 --- a/src/control/repos/mod.rs +++ b/src/control/repos/mod.rs @@ -158,7 +158,6 @@ fn read_queues(state: &AppState, info: &mut RepoInfo, index_id: Option) -> else { return Ok(()); }; - let resync = rmp_serde::from_slice(&bytes).into_diagnostic()?; // a repo that never backfilled has no pds in its state yet, but resolving // it for the failed attempt left one in the identity cache let pds = info @@ -166,7 +165,15 @@ fn read_queues(state: &AppState, info: &mut RepoInfo, index_id: Option) -> .clone() .or_else(|| state.resolver.cached_pds(&info.did)); let host_failures = pds.map_or(0, |pds| state.throttler.snapshot(&pds).consecutive_failures); - info.resync = Some(ResyncInfo::new(resync, host_failures)?); + // one unreadable entry would fail every listing page it lands on, so the cursor + // could never get past it. the row lists without it and the log says why + match rmp_serde::from_slice(&bytes) + .into_diagnostic() + .and_then(|resync| ResyncInfo::new(resync, host_failures)) + { + Ok(resync) => info.resync = Some(resync), + Err(e) => tracing::warn!(did = %info.did, err = %e, "skipping an unreadable resync entry"), + } Ok(()) } @@ -433,6 +440,52 @@ mod tests { assert!(ResyncInfo::new(state, 0).is_err()); } + #[cfg(feature = "indexer")] + #[tokio::test] + async fn an_unreadable_resync_entry_does_not_stop_the_listing() -> miette::Result<()> { + use crate::types::{ResyncErrorKind, ResyncState}; + + let tmp = tempfile::tempdir().into_diagnostic()?; + let state = Arc::new(AppState::new(&crate::config::Config { + database_path: tmp.path().to_path_buf(), + ..Default::default() + })?); + let bad = Did::new_static("did:web:bad.example").into_diagnostic()?; + let garbled = Did::new_static("did:web:garbled.example").into_diagnostic()?; + let fine = Did::new_static("did:web:fine.example").into_diagnostic()?; + let mut batch = state.db.inner.batch(); + for did in [&bad, &garbled, &fine] { + batch.insert( + &state.db.repos, + keys::repo_key(did), + db::ser_repo_state(&RepoState::backfilling())?, + ); + } + let impossible = ResyncState::Error { + kind: ResyncErrorKind::Generic, + retry_count: 1, + next_retry: i64::MAX, + }; + batch.insert( + &state.db.indexer.resync, + keys::repo_key(&bad), + rmp_serde::to_vec(&impossible).into_diagnostic()?, + ); + batch.insert( + &state.db.indexer.resync, + keys::repo_key(&garbled), + b"not msgpack", + ); + batch.commit().into_diagnostic()?; + + let listed = ReposControl::new(state) + .iter(None) + .collect::>>()?; + assert_eq!(listed.len(), 3); + assert!(listed.iter().all(|info| info.resync.is_none())); + Ok(()) + } + #[cfg(feature = "indexer")] #[tokio::test] async fn repo_info_shows_why_a_repo_waits_for_resync() -> miette::Result<()> {