diff --git a/src/control/repos/indexer.rs b/src/control/repos/indexer.rs index 435c491..cb8c10c 100644 --- a/src/control/repos/indexer.rs +++ b/src/control/repos/indexer.rs @@ -4,101 +4,6 @@ use rand::Rng; use super::*; impl ReposControl { - /// iterates through pending repositories, returning their state. - #[allow(dead_code)] - pub(crate) fn iter_pending( - &self, - cursor: Option, - ) -> impl Iterator> { - let start_bound = if let Some(cursor) = cursor { - std::ops::Bound::Excluded(cursor.to_be_bytes().to_vec()) - } else { - std::ops::Bound::Unbounded - }; - - let repos = self.0.db.repos.clone(); - let state = self.0.clone(); - self.0 - .db - .indexer - .pending_range((start_bound, std::ops::Bound::Unbounded)) - .map(move |g| { - let (id_raw, did_key) = g.into_inner().into_diagnostic()?; - let id = u64::from_be_bytes( - id_raw - .as_ref() - .try_into() - .into_diagnostic() - .wrap_err("can't parse pending key")?, - ); - let Some(bytes) = repos.get(&did_key).into_diagnostic()? else { - // stale pending that we forgot to delete? shouldn't happen though - tracing::warn!(id, did = ?did_key, "stale pending???"); - return Ok(None); - }; - let repo_state = crate::db::deser_repo_state(bytes.as_ref())?; - let did = TrimmedDid::try_from(did_key.as_ref())?.to_did(); - let metadata_key = keys::repo_metadata_key(&did); - let metadata = state - .db - .repo_metadata - .get(&metadata_key) - .into_diagnostic()? - .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; - let metadata = crate::db::deser_repo_meta(metadata.as_ref())?; - Ok(Some(( - id, - repo_state_to_info(did, repo_state.into_static(), metadata.tracked), - ))) - }) - .filter_map(|b| b.transpose()) - } - - #[allow(dead_code)] - pub(crate) fn iter_resync( - &self, - cursor: Option<&Did>, - ) -> impl Iterator> { - let start_bound = if let Some(cursor) = cursor { - let did_key = keys::repo_key(cursor); - std::ops::Bound::Excluded(did_key) - } else { - std::ops::Bound::Unbounded - }; - - let repos = self.0.db.repos.clone(); - let state = self.0.clone(); - self.0 - .db - .indexer - .resync - .range((start_bound, std::ops::Bound::Unbounded)) - .map(move |g| { - let did_key = g.key().into_diagnostic()?; - let Some(bytes) = repos.get(&did_key).into_diagnostic()? else { - // stale pending that we forgot to delete? shouldn't happen though - tracing::warn!(did = ?did_key, "stale resync???"); - return Ok(None); - }; - let repo_state = crate::db::deser_repo_state(bytes.as_ref())?; - let did = TrimmedDid::try_from(did_key.as_ref())?.to_did(); - let metadata_key = keys::repo_metadata_key(&did); - let metadata = state - .db - .repo_metadata - .get(&metadata_key) - .into_diagnostic()? - .ok_or_else(|| miette::miette!("repo metadata not found for {}", did))?; - let metadata = crate::db::deser_repo_meta(metadata.as_ref())?; - Ok(Some(repo_state_to_info( - did, - repo_state.into_static(), - metadata.tracked, - ))) - }) - .filter_map(|b| b.transpose()) - } - /// repositories hydrant still has work queued for, a backfill or resync /// entry, in the same did order as [`ReposControl::iter`] pub fn iter_queued(&self, cursor: Option<&Did>) -> impl Iterator> { @@ -958,49 +863,6 @@ mod tests { Ok(()) } - #[tokio::test] - async fn pending_listing_pages_by_queue_id() -> miette::Result<()> { - let tmp = tempfile::tempdir().into_diagnostic()?; - let config = crate::config::Config { - database_path: tmp.path().to_path_buf(), - ..Default::default() - }; - let state = Arc::new(AppState::new(&config)?); - let first = Did::new_static("did:web:first.example").into_diagnostic()?; - let second = Did::new_static("did:web:second.example").into_diagnostic()?; - - let db = &state.db; - let mut batch = db.inner.batch(); - // 256 sorts after 1 only as big-endian bytes - for (id, did) in [(1_u64, &first), (256, &second)] { - batch.insert( - &db.repos, - keys::repo_key(did), - crate::db::ser_repo_state(&RepoState::backfilling())?, - ); - batch.insert( - &db.repo_metadata, - keys::repo_metadata_key(did), - crate::db::ser_repo_meta(&RepoMetadata::backfilling(id))?, - ); - db.indexer - .stage_pending_insert(&mut batch, did, &keys::pending_key(id)); - } - batch.commit().into_diagnostic()?; - - let repos = ReposControl(state.clone()); - let listed = |cursor| { - repos - .iter_pending(cursor) - .map(|row| row.map(|(id, info)| (id, info.did))) - .collect::>>() - }; - assert_eq!(listed(None)?, [(1, first), (256, second.clone())]); - assert_eq!(listed(Some(1))?, [(256, second)]); - assert_eq!(listed(Some(256))?, []); - Ok(()) - } - #[tokio::test] #[cfg(not(feature = "backlinks"))] async fn test_generate_car_ephemeral_rejected() -> miette::Result<()> { diff --git a/src/db/keyspaces.rs b/src/db/keyspaces.rs index a2310d8..ac5da7c 100644 --- a/src/db/keyspaces.rs +++ b/src/db/keyspaces.rs @@ -199,14 +199,6 @@ impl IndexerDb { self.pending.iter() } - pub(crate) fn pending_range(&self, range: R) -> fjall::Iter - where - K: AsRef<[u8]>, - R: std::ops::RangeBounds, - { - self.pending.range(range) - } - /// the queue in did order, keys are `pending_did_key`s pub(crate) fn pending_did_range(&self, range: R) -> fjall::Iter where diff --git a/tests/representation_inventory.tsv b/tests/representation_inventory.tsv index 85ca9bc..80ba734 100644 --- a/tests/representation_inventory.tsv +++ b/tests/representation_inventory.tsv @@ -153,8 +153,6 @@ codec-call:src/control/new_pds_crawl.rs::enabled::encode_cursor::rmp_serde::to_v codec-call:src/control/repos/indexer.rs::RepoHandle::generate_car::CarWriter::new rust:src/control/repos/indexer.rs::generate_car_reports_a_deleted_head_body codec-call:src/control/repos/indexer.rs::RepoHandle::get_record::serde_ipld_dagcbor::from_slice rust:src/control/repos/indexer.rs::get_and_list_hide_redacted_heads codec-call:src/control/repos/indexer.rs::RepoHandle::list_records::serde_ipld_dagcbor::from_slice rust:src/control/repos/indexer.rs::get_and_list_hide_redacted_heads -codec-call:src/control/repos/indexer.rs::ReposControl::iter_pending::.to_be_bytes rust:src/control/repos/indexer.rs::pending_listing_pages_by_queue_id -codec-call:src/control/repos/indexer.rs::ReposControl::iter_pending::u64::from_be_bytes rust:src/control/repos/indexer.rs::pending_listing_pages_by_queue_id codec-call:src/control/repos/mod.rs::RepoHandle::collections::u64::from_be_bytes rust:src/control/repos/mod.rs::collections_decodes_big_endian_counts codec-call:src/control/repos/mod.rs::read_mini_doc_snapshot::rmp_serde::from_slice rust:src/control/repos/mod.rs::identity_reconcile_trigger_matrix_is_narrow codec-call:src/control/repos/mod.rs::read_queues::rmp_serde::from_slice rust:src/control/repos/mod.rs::repo_info_shows_why_a_repo_waits_for_resync