From b53e91e0d6952c0cd5db4e860a7694707b5ad19d Mon Sep 17 00:00:00 2001 From: dawn <90008@klbr.net> Date: Sat, 18 Jul 2026 16:27:30 +0000 Subject: [PATCH] [control] add sequential collection block scan API --- src/control/hydrant.rs | 200 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++---- src/control/mod.rs | 2 ++ src/db/keyspaces.rs | 22 ++++++++++++++++++++++ 3 file(s) changed, 220 insertion(s)(+), 4 deletion(s)(-) diff --git a/src/control/hydrant.rs b/src/control/hydrant.rs --- a/src/control/hydrant.rs +++ b/src/control/hydrant.rs @@ -119,10 +119,7 @@ .await?; // 3. reload the live filter into the hot-path arc-swap - let new_filter = state - .db - .run(move |db| db_filter::load(&db.filter)) - .await?; + let new_filter = state.db.run(move |db| db_filter::load(&db.filter)).await?; state.filter.store(Arc::new(new_filter)); } @@ -193,10 +190,205 @@ let state = self.state.clone(); async move { crate::api::serve_debug(state, port).await } } + + /// iterate sequentially over every block stored for `collection`, ordered by CID. + /// + /// values are raw DAG-CBOR. deleted or updated records may leave orphan blocks + /// behind since the CAS is content-addressed, not record-referenced. + #[cfg(feature = "indexer")] + pub fn scan_collection_blocks( + &self, + collection: &str, + ) -> impl Iterator> + 'static { + let db = self.state.db.indexer.clone(); + let collection = collection.to_string(); + + let mut start = Vec::with_capacity(collection.len() + 1); + start.extend_from_slice(collection.as_bytes()); + start.push(b'|'); + + let mut end = Vec::with_capacity(collection.len() + 1); + end.extend_from_slice(collection.as_bytes()); + end.push(b'|' + 1); + + let iter = db.block_range(( + std::ops::Bound::Included(start), + std::ops::Bound::Excluded(end), + )); + + iter.map(|item| { + let (key, value) = item.into_inner().into_diagnostic()?; + Ok(ScannedBlock { key, value }) + }) + } +} + +#[cfg(feature = "indexer")] +/// a raw block entry yielded by [`Hydrant::scan_collection_blocks`]. +/// +/// key and value are kept as raw slices; use the accessor methods for lazy parsing. +#[derive(Clone, Debug)] +pub struct ScannedBlock { + pub(crate) key: fjall::Slice, + pub(crate) value: fjall::Slice, +} + +#[cfg(feature = "indexer")] +impl ScannedBlock { + /// the collection name part of the block's key. + pub fn collection(&self) -> &str { + let sep_idx = self + .key + .iter() + .position(|&b| b == b'|') + .expect("malformed block key: missing separator"); + std::str::from_utf8(&self.key[..sep_idx]).expect("collection name must be valid UTF-8") + } + + /// the raw CID bytes after the separator. + pub fn cid(&self) -> &[u8] { + let sep_idx = self + .key + .iter() + .position(|&b| b == b'|') + .expect("malformed block key: missing separator"); + &self.key[sep_idx + 1..] + } + + /// the key's CID, parsed. + pub fn parsed_cid(&self) -> Result { + cid::Cid::read_bytes(self.cid()).map_err(|e| miette::miette!("failed to parse CID: {e}")) + } + + /// the key's CID as a jacquard CID. + pub fn jacquard_cid( + &self, + ) -> Result, miette::Report> { + let parsed = cid::Cid::read_bytes(self.cid()) + .map_err(|e| miette::miette!("failed to parse CID: {e}"))?; + Ok(jacquard_common::types::cid::Cid::ipld(parsed)) + } + + /// the raw DAG-CBOR block bytes. + pub fn value(&self) -> &[u8] { + &self.value + } } impl axum::extract::FromRef for Arc { fn from_ref(h: &Hydrant) -> Self { h.state.clone() + } +} + +#[cfg(all(test, feature = "indexer"))] +mod tests { + use super::*; + use crate::config::Config; + use crate::db::keys; + use cid::Cid; + use cid::multihash::Multihash; + use tempfile::tempdir; + + const SHA2_256: u64 = 0x12; + const DAG_CBOR: u64 = 0x71; + + fn test_config(path: &std::path::Path) -> Config { + Config { + database_path: path.to_path_buf(), + ..Default::default() + } + } + + #[tokio::test] + async fn test_scan_collection_blocks() -> Result<()> { + let tmp = tempdir().into_diagnostic()?; + let hydrant = Hydrant::new(test_config(tmp.path())).await?; + let state = hydrant.state.clone(); + + let hash1 = [1u8; 32]; + let mh1 = Multihash::<64>::wrap(SHA2_256, &hash1).unwrap(); + let valid_cid1 = Cid::new_v1(DAG_CBOR, mh1); + let cid_bytes1 = valid_cid1.to_bytes(); + + let hash2 = [2u8; 32]; + let mh2 = Multihash::<64>::wrap(SHA2_256, &hash2).unwrap(); + let valid_cid2 = Cid::new_v1(DAG_CBOR, mh2); + let cid_bytes2 = valid_cid2.to_bytes(); + + let hash3 = [3u8; 32]; + let mh3 = Multihash::<64>::wrap(SHA2_256, &hash3).unwrap(); + let valid_cid3 = Cid::new_v1(DAG_CBOR, mh3); + let cid_bytes3 = valid_cid3.to_bytes(); + + let cid_bytes1_val = cid_bytes1.clone(); + let cid_bytes2_val = cid_bytes2.clone(); + let cid_bytes3_val = cid_bytes3.clone(); + + // Insert a couple of blocks for two collections: "app.bsky.feed.post" and "app.bsky.feed.like" + state + .db + .run(move |db| -> Result<()> { + let mut batch = db.inner.batch(); + + let post_key = keys::indexer::block_key("app.bsky.feed.post", &cid_bytes1_val); + let post_val = b"post_cbor_value"; + db.indexer + .stage_block(&mut batch, post_key, post_val.to_vec()); + + let like_key = keys::indexer::block_key("app.bsky.feed.like", &cid_bytes2_val); + let like_val = b"like_cbor_value"; + db.indexer + .stage_block(&mut batch, like_key, like_val.to_vec()); + + // Let's add one more post to test multiple blocks in same collection + let post_key2 = keys::indexer::block_key("app.bsky.feed.post", &cid_bytes3_val); + let post_val2 = b"post_cbor_value_2"; + db.indexer + .stage_block(&mut batch, post_key2, post_val2.to_vec()); + + batch.commit().into_diagnostic()?; + Ok(()) + }) + .await?; + + // Scan posts + let post_blocks: Vec = hydrant + .scan_collection_blocks("app.bsky.feed.post") + .collect::>>()?; + + assert_eq!(post_blocks.len(), 2); + assert_eq!(post_blocks[0].collection(), "app.bsky.feed.post"); + assert_eq!(post_blocks[0].cid(), cid_bytes1); + assert_eq!(post_blocks[0].parsed_cid().unwrap(), valid_cid1); + assert_eq!( + post_blocks[0].jacquard_cid().unwrap(), + jacquard_common::types::cid::Cid::ipld(valid_cid1) + ); + assert_eq!(post_blocks[0].value(), b"post_cbor_value"); + + assert_eq!(post_blocks[1].collection(), "app.bsky.feed.post"); + assert_eq!(post_blocks[1].cid(), cid_bytes3); + assert_eq!(post_blocks[1].parsed_cid().unwrap(), valid_cid3); + assert_eq!(post_blocks[1].value(), b"post_cbor_value_2"); + + // Scan likes + let like_blocks: Vec = hydrant + .scan_collection_blocks("app.bsky.feed.like") + .collect::>>()?; + + assert_eq!(like_blocks.len(), 1); + assert_eq!(like_blocks[0].collection(), "app.bsky.feed.like"); + assert_eq!(like_blocks[0].cid(), cid_bytes2); + assert_eq!(like_blocks[0].parsed_cid().unwrap(), valid_cid2); + assert_eq!(like_blocks[0].value(), b"like_cbor_value"); + + // Scan non-existent + let empty_blocks: Vec = hydrant + .scan_collection_blocks("app.bsky.feed.repost") + .collect::>>()?; + assert!(empty_blocks.is_empty()); + + Ok(()) } } diff --git a/src/control/mod.rs b/src/control/mod.rs --- a/src/control/mod.rs +++ b/src/control/mod.rs @@ -38,6 +38,8 @@ pub use db::DbControl; pub use hosts::{ApiBinds, Host}; pub use hydrant::Hydrant; +#[cfg(feature = "indexer")] +pub use hydrant::ScannedBlock; pub use stats::StatsResponse; #[cfg(feature = "indexer_stream")] diff --git a/src/db/keyspaces.rs b/src/db/keyspaces.rs --- a/src/db/keyspaces.rs +++ b/src/db/keyspaces.rs @@ -96,6 +96,28 @@ pub(crate) fn block>(&self, key: K) -> fjall::Result> { self.blocks.get(key) } + + pub(crate) fn block_range(&self, range: R) -> fjall::Iter + where + K: AsRef<[u8]>, + R: std::ops::RangeBounds, + { + self.blocks.range(range) + } + + pub(crate) fn approximate_pending_count(&self) -> usize { + self.pending.approximate_len() + } + + #[cfg(test)] + #[allow(dead_code)] + pub(crate) fn stage_block(&self, batch: &mut fjall::OwnedWriteBatch, key: K, value: V) + where + K: Into, + V: Into, + { + batch.insert(&self.blocks, key, value); + } } #[cfg(feature = "indexer_stream")] -- tangled.sh