files cleanup

src/disk_drive.rs

···
-use futures::Stream;
-use futures::TryStreamExt;
-use std::collections::VecDeque;
-use std::error::Error;
-
-use crate::disk_walk::{RkeyError, Trip, Walker};
-use crate::mst::{Commit, Node};
-
-use ipld_core::cid::Cid;
-use serde::{Deserialize, Serialize, de::DeserializeOwned};
-
-/// Errors that can happen while consuming and emitting blocks and records
-#[derive(Debug, thiserror::Error)]
-pub enum DriveError {
-    #[error("Failed to initialize CarReader: {0}")]
-    CarReader(#[from] iroh_car::Error),
-    #[error("Car block stream error: {0}")]
-    CarBlockError(Box<dyn Error>),
-    #[error("Failed to decode commit block: {0}")]
-    BadCommit(Box<dyn Error>),
-    #[error("The Commit block reference by the root was not found")]
-    MissingCommit,
-    #[error("The MST block {0} could not be found")]
-    MissingBlock(Cid),
-    #[error("Failed to walk the mst tree: {0}")]
-    Tripped(#[from] Trip),
-    #[error("whatever: {0}")]
-    WalkingProblem(#[from] WalkError),
-    #[error("whatever: {0}")]
-    Boooooo(String),
-    #[error("Error while encoding: {0}")]
-    EncodingError(#[from] bincode::error::EncodeError),
-    #[error("Error while decoding: {0}")]
-    DecodingError(#[from] bincode::error::DecodeError),
-}
-
-/// Limited subset of errors that can happen while walking
-#[derive(Debug, thiserror::Error)]
-pub enum WalkError {
-    #[error("The MST block {0} could not be found")]
-    MissingBlock(Cid),
-    #[error("Failed to walk the mst tree: {0}")]
-    Tripped(#[from] Trip),
-}
-
-#[derive(Debug, thiserror::Error)]
-pub enum BlockStoreError {
-    #[error("Error from the storage backend: {0}")]
-    StorageBackend(Box<dyn Error + Send>),
-
-    #[error(transparent)]
-    RkeyError(#[from] RkeyError),
-
-    // this should probably not be up here
-    #[error("Failed to join tokio task: {0}")]
-    JoinError(tokio::task::JoinError),
-
-    #[error("Could not find block: {0}")]
-    MissingBlock(Cid),
-}
-
-#[derive(Serialize, Deserialize)]
-pub enum MaybeProcessedBlock<T: Serialize> {
-    Raw(Vec<u8>),
-    Processed(T),
-}
-
-pub type Records = Vec<(String, Vec<u8>)>;
-
-/// Storage backend for caching large-repo blocks
-///
-/// Since
-pub trait BlockStore {
-    fn put_batch(
-        &self,
-        blocks: Vec<(Cid, Vec<u8>)>,
-    ) -> impl Future<Output = Result<(), BlockStoreError>>; // unwraps for now
-    fn walk_batch(
-        &self,
-        walker: Walker,
-        n: usize,
-    ) -> impl Future<Output = Result<(Walker, Records), BlockStoreError>>; // boo string error for now because
-}
-
-type CarBlock<E> = Result<(Cid, Vec<u8>), E>;
-
-/// The core driver between the block stream and MST walker
-pub struct Vehicle<SE, S, T, BS, P>
-where
-    SE: Error + 'static,
-    S: Stream<Item = CarBlock<SE>>,
-    T: Clone + Serialize + DeserializeOwned,
-    BS: BlockStore,
-    P: Fn(&[u8]) -> T,
-{
-    #[allow(dead_code)]
-    block_stream: Option<S>,
-    block_store: BS,
-    walker: Walker,
-    process: P,
-    out_cache: VecDeque<(String, T)>,
-}
-
-impl<SE, S, T, BS, P> Vehicle<SE, S, T, BS, P>
-where
-    SE: Error + 'static,
-    S: Stream<Item = CarBlock<SE>> + Unpin + Send,
-    T: Clone + Serialize + DeserializeOwned + Send,
-    BS: BlockStore + Send,
-    P: Fn(&[u8]) -> T,
-{
-    /// Set up the stream
-    ///
-    /// This will eagerly consume blocks until the `Commit` object is found.
-    /// *Usually* the it's the first block, but there is no guarantee.
-    ///
-    /// ### Parameters
-    ///
-    /// `root`: CID of the commit object that is the root of the MST
-    ///
-    /// `block_stream`: Input stream of raw CAR blocks
-    ///
-    /// `process`: record-transforming callback:
-    ///
-    /// For tasks where records can be quickly processed into a *smaller*
-    /// useful representation, you can do that eagerly as blocks come in by
-    /// passing the processor as a callback here. This can reduce overall
-    /// memory usage.
-    pub async fn init(
-        root: Cid,
-        block_stream: S,
-        block_store: BS,
-        process: P,
-    ) -> Result<(Commit, Self), DriveError> {
-        let mut commit = None;
-
-        log::warn!("init: load blocks");
-
-        let mut chunked = block_stream.try_chunks(256);
-
-        // go ahead and put all blocks in the block store
-        while let Some(chunk) = chunked
-            .try_next()
-            .await
-            .map_err(|e| DriveError::CarBlockError(e.into()))?
-        {
-            let mut to_insert = Vec::with_capacity(chunk.len());
-            for (cid, data) in chunk {
-                if cid == root {
-                    let c: Commit = serde_ipld_dagcbor::from_slice(&data)
-                        .map_err(|e| DriveError::BadCommit(e.into()))?;
-                    commit = Some(c);
-                } else {
-                    let wrapped = if Node::could_be(&data) {
-                        MaybeProcessedBlock::Raw(data)
-                    } else {
-                        MaybeProcessedBlock::Processed(process(&data))
-                    };
-                    let bytes =
-                        bincode::serde::encode_to_vec(wrapped, bincode::config::standard())?;
-
-                    to_insert.push((cid, bytes));
-                }
-            }
-            block_store
-                .put_batch(to_insert)
-                .await
-                .map_err(|e| DriveError::Boooooo(format!("boooOOOOO! {e}")))?; // TODO
-        }
-
-        log::warn!("init: got commit?");
-
-        // we either broke out or read all the blocks without finding the commit...
-        let commit = commit.ok_or(DriveError::MissingCommit)?;
-
-        let walker = Walker::new(commit.data);
-
-        log::warn!("init: wrapping up");
-
-        let me = Self {
-            block_stream: None,
-            block_store,
-            walker,
-            process,
-            out_cache: VecDeque::new(),
-        };
-        Ok((commit, me))
-    }
-
-    async fn load_chunk(&mut self, n: usize) -> Result<(), DriveError> {
-        let walker = std::mem::take(&mut self.walker);
-        let (walker, batch) = self
-            .block_store
-            .walk_batch(walker, n)
-            .await
-            .map_err(|e| DriveError::Boooooo(format!("booo! (here right?) {e}")))?; // TODO
-        self.walker = walker;
-
-        let processed = batch
-            .into_iter()
-            .map(|(k, encoded)| {
-                let (decoded, n): (MaybeProcessedBlock<T>, usize) =
-                    bincode::serde::decode_from_slice(&encoded, bincode::config::standard())?;
-                assert_eq!(n, encoded.len());
-                let processed = match decoded {
-                    MaybeProcessedBlock::Processed(t) => t,
-                    MaybeProcessedBlock::Raw(block) => (self.process)(&block),
-                };
-                Ok((k, processed))
-            })
-            .collect::<Result<Vec<_>, DriveError>>()?;
-
-        self.out_cache.extend(processed);
-        Ok(())
-    }
-
-    /// Get a chunk of records at a time
-    ///
-    /// the number of returned records may be smaller or larger than requested
-    /// (but non-zero), even if it's not the last chunk.
-    ///
-    /// an empty vec will be returned to signal the end.
-    pub async fn next_chunk(&mut self, n: usize) -> Result<Vec<(String, T)>, DriveError> {
-        if self.out_cache.is_empty() {
-            self.load_chunk(n).await?;
-        }
-        Ok(std::mem::take(&mut self.out_cache).into())
-    }
-
-    /// Manually step through the record outputs
-    pub async fn next_record(&mut self) -> Result<Option<(String, T)>, DriveError> {
-        if self.out_cache.is_empty() {
-            self.load_chunk(128).await?; // TODO
-        }
-        Ok(self.out_cache.pop_front())
-    }
-
-    /// Convert to a futures::stream of record outputs
-    pub fn stream(self) -> impl Stream<Item = Result<(String, T), DriveError>> {
-        futures::stream::try_unfold(self, |mut this| async move {
-            let maybe_record = this.next_record().await?;
-            Ok(maybe_record.map(|b| (b, this)))
-        })
-    }
-}

src/disk_redb.rs

···
-use crate::disk_drive::{BlockStore, BlockStoreError, MaybeProcessedBlock, Records};
-use crate::disk_walk::{Need, Walker};
-use ipld_core::cid::Cid;
-use redb::{Database, Durability, Error, ReadableDatabase, TableDefinition};
-use std::path::Path;
-use std::sync::Arc;
-
-const TABLE: TableDefinition<&[u8], &[u8]> = TableDefinition::new("blocks");
-
-pub struct RedbStore {
-    #[allow(dead_code)]
-    db: Arc<Database>,
-}
-
-impl RedbStore {
-    pub async fn new(path: impl AsRef<Path> + 'static + Send) -> Result<Self, Error> {
-        log::warn!("redb new");
-        let db = tokio::task::spawn_blocking(|| Database::create(path))
-            .await
-            .unwrap()?;
-        log::warn!("db created");
-        Ok(Self { db: db.into() })
-    }
-}
-
-// TODO: ship off to a blocking thread
-impl Drop for RedbStore {
-    fn drop(&mut self) {
-        let mut tx = self.db.begin_write().unwrap();
-        tx.set_durability(Durability::None).unwrap();
-        tx.delete_table(TABLE).unwrap();
-        tx.commit().unwrap();
-    }
-}
-
-impl<E: Into<Error>> From<E> for BlockStoreError {
-    fn from(e: E) -> BlockStoreError {
-        let e = Into::<Error>::into(e);
-        BlockStoreError::StorageBackend(Box::new(e))
-    }
-}
-
-impl BlockStore for RedbStore {
-    async fn put_batch(&self, blocks: Vec<(Cid, Vec<u8>)>) -> Result<(), BlockStoreError> {
-        let db = self.db.clone();
-        tokio::task::spawn_blocking(move || -> Result<(), BlockStoreError> {
-            let mut tx = db.begin_write()?;
-            tx.set_durability(Durability::None)?;
-
-            {
-                let mut table = tx.open_table(TABLE)?;
-                for (cid, t) in blocks {
-                    let key_bytes = cid.to_bytes();
-                    table.insert(&*key_bytes, &*t)?;
-                }
-            }
-
-            Ok(tx.commit()?)
-        })
-        .await
-        .map_err(BlockStoreError::JoinError)?
-    }
-
-    async fn walk_batch(
-        &self,
-        mut walker: Walker,
-        n: usize,
-    ) -> Result<(Walker, Records), BlockStoreError> {
-        let db = self.db.clone();
-        tokio::task::spawn_blocking(move || -> Result<_, BlockStoreError> {
-            let tx = db.begin_read()?;
-            let table = tx.open_table(TABLE)?;
-
-            let mut out = Vec::with_capacity(n);
-            loop {
-                let Some(need) = walker.next_needed()? else {
-                    break;
-                };
-                let cid = need.cid();
-                let Some(res) = table.get(&*cid.to_bytes())? else {
-                    return Err(BlockStoreError::MissingBlock(cid));
-                };
-                let block = res.value();
-
-                match need {
-                    Need::Node(_) => {
-                        let (mpb, n) =
-                            bincode::serde::decode_from_slice(block, bincode::config::standard())
-                                .unwrap();
-                        assert_eq!(n, block.len());
-                        // DANGER: we're throwing in unit () as a placeholder here and assuming bincode will still work since Raw is the first variant
-                        let MaybeProcessedBlock::Raw(bytes): MaybeProcessedBlock<()> = mpb else {
-                            panic!("should have not been processed"); // tODO
-                        };
-                        walker.handle_node(&bytes)?
-                    }
-                    Need::Record { rkey, .. } => {
-                        out.push((rkey, block.to_vec()));
-                        if out.len() >= n {
-                            break;
-                        }
-                    }
-                }
-            }
-            Ok((walker, out))
-        })
-        .await
-        .map_err(BlockStoreError::JoinError)?
-    }
-}

src/disk_sqlite.rs

···
-// use crate::disk_drive::BlockStore;
-// use ipld_core::cid::Cid;
-// use rusqlite::{Connection, OptionalExtension, Result};
-// use serde::{Serialize, de::DeserializeOwned};
-// use std::path::Path;
-
-// pub struct SqliteStore {
-//     conn: Connection,
-// }
-
-// impl SqliteStore {
-//     pub fn new(path: impl AsRef<Path>) -> Result<Self> {
-//         let conn = Connection::open(path)?;
-//         conn.pragma_update(None, "journal_mode", "WAL")?;
-//         conn.pragma_update(None, "synchronous", "OFF")?;
-//         conn.pragma_update(None, "cache_size", (-32 * 2_i64.pow(10)).to_string())?;
-//         conn.execute(
-//             "CREATE TABLE blocks (
-//                 key  BLOB PRIMARY KEY NOT NULL,
-//                 val  BLOB NOT NULL
-//             ) WITHOUT ROWID",
-//             (),
-//         )?;
-
-//         Ok(Self { conn })
-//     }
-// }
-
-// impl Drop for SqliteStore {
-//     fn drop(&mut self) {
-//         self.conn.execute("DROP TABLE blocks", ()).unwrap();
-//     }
-// }
-
-// impl<MPB: Serialize + DeserializeOwned> BlockStore<MPB> for SqliteStore {
-//     fn put(&self, c: Cid, t: MPB) {
-//         let key_bytes = c.to_bytes();
-//         let val_bytes = bincode::serde::encode_to_vec(t, bincode::config::standard()).unwrap();
-
-//         self.conn
-//             .execute(
-//                 "INSERT INTO blocks (key, val) VALUES (?1, ?2)",
-//                 (&key_bytes, &val_bytes),
-//             )
-//             .unwrap();
-//     }
-//     fn get(&self, c: Cid) -> Option<MPB> {
-//         let key_bytes = c.to_bytes();
-
-//         let val_bytes: Vec<u8> = self
-//             .conn
-//             .query_one(
-//                 "SELECT val FROM blocks WHERE key = ?1",
-//                 (&key_bytes,),
-//                 |row| row.get(0),
-//             )
-//             .optional()
-//             .unwrap()?;
-
-//         let (t, n): (MPB, usize) =
-//             bincode::serde::decode_from_slice(&val_bytes, bincode::config::standard()).unwrap();
-//         assert_eq!(val_bytes.len(), n);
-//         Some(t)
-//     }
-// }

src/disk_walk.rs

···
-//! Depth-first MST traversal
-
-use crate::mst::Node;
-use std::convert::Infallible;
-
-use ipld_core::cid::Cid;
-use serde::{Serialize, de::DeserializeOwned};
-
-/// Errors that can happen while walking
-#[derive(Debug, thiserror::Error)]
-pub enum Trip {
-    #[error("empty mst nodes are not allowed")]
-    NodeEmpty,
-    #[error("Failed to decode commit block: {0}")]
-    BadCommit(serde_ipld_dagcbor::DecodeError<Infallible>),
-    #[error("Action node error: {0}")]
-    RkeyError(#[from] RkeyError),
-    #[error("Process failed: {0}")]
-    ProcessFailed(String),
-}
-
-/// Errors from invalid Rkeys
-#[derive(Debug, thiserror::Error)]
-pub enum RkeyError {
-    #[error("Failed to compute an rkey due to invalid prefix_len")]
-    EntryPrefixOutOfbounds,
-    #[error("RKey was not utf-8")]
-    EntryRkeyNotUtf8(#[from] std::string::FromUtf8Error),
-    #[error("Encountered an rkey out of order while walking the MST")]
-    RkeyOutOfOrder,
-    #[error("Failed to decode node block: {0}")]
-    NodeDecodeError(#[from] serde_ipld_dagcbor::DecodeError<Infallible>),
-}
-
-/// Walker outputs
-#[derive(Debug)]
-pub enum Step<T: Serialize + DeserializeOwned> {
-    /// We need a CID but it's not in the block store
-    ///
-    /// Give the needed CID to the driver so it can load blocks until it's found
-    Rest(Cid),
-    /// Reached the end of the MST! yay!
-    Finish,
-    /// A record was found!
-    Step { rkey: String, data: T },
-}
-
-#[derive(Debug, Clone, PartialEq)]
-pub enum Need {
-    Node(Cid),
-    Record { rkey: String, cid: Cid },
-}
-
-impl Need {
-    pub fn cid(&self) -> Cid {
-        match self {
-            Need::Node(cid) => *cid,
-            Need::Record { cid, .. } => *cid,
-        }
-    }
-}
-
-fn push_from_node(stack: &mut Vec<Need>, node: &Node) -> Result<(), RkeyError> {
-    let mut entries = Vec::with_capacity(node.entries.len());
-
-    let mut prefix = vec![];
-    for entry in &node.entries {
-        let mut rkey = vec![];
-        let pre_checked = prefix
-            .get(..entry.prefix_len)
-            .ok_or(RkeyError::EntryPrefixOutOfbounds)?;
-        rkey.extend_from_slice(pre_checked);
-        rkey.extend_from_slice(&entry.keysuffix);
-        prefix = rkey.clone();
-
-        entries.push(Need::Record {
-            rkey: String::from_utf8(rkey)?,
-            cid: entry.value,
-        });
-        if let Some(ref tree) = entry.tree {
-            entries.push(Need::Node(*tree));
-        }
-    }
-
-    entries.reverse();
-    stack.append(&mut entries);
-
-    if let Some(tree) = node.left {
-        stack.push(Need::Node(tree));
-    }
-    Ok(())
-}
-
-/// Traverser of an atproto MST
-///
-/// Walks the tree from left-to-right in depth-first order
-///
-/// (turning into more of a navigator)
-/// it doesn't quite feel like the divisions of responsibility are right around
-/// here yet.
-#[derive(Debug, Default)]
-pub struct Walker {
-    stack: Vec<Need>,
-    prev: String,
-}
-
-impl Walker {
-    pub fn new(tree_root_cid: Cid) -> Self {
-        Self {
-            stack: vec![Need::Node(tree_root_cid)],
-            prev: "".to_string(),
-        }
-    }
-
-    pub fn next_needed(&mut self) -> Result<Option<Need>, RkeyError> {
-        let Some(need) = self.stack.pop() else {
-            return Ok(None);
-        };
-        if let Need::Record { ref rkey, .. } = need {
-            // rkeys *must* be in order or else the tree is invalid (or
-            // we have a bug)
-            if *rkey <= self.prev {
-                return Err(RkeyError::RkeyOutOfOrder);
-            }
-            self.prev = rkey.clone();
-        }
-        Ok(Some(need))
-    }
-
-    /// hacky: this must be called after next_needed if it was a node
-    pub fn handle_node(&mut self, block: &[u8]) -> Result<(), RkeyError> {
-        let node = serde_ipld_dagcbor::from_slice::<Node>(block)?;
-        push_from_node(&mut self.stack, &node)?;
-        Ok(())
-    }
-}
-
-#[cfg(test)]
-mod test {
-    use super::*;
-    // use crate::mst::Entry;
-
-    fn cid1() -> Cid {
-        "bafyreihixenvk3ahqbytas4hk4a26w43bh6eo3w6usjqtxkpzsvi655a3m"
-            .parse()
-            .unwrap()
-    }
-    //     fn cid2() -> Cid {
-    //         "QmY7Yh4UquoXHLPFo2XbhXkhBvFoPwmQUSa92pxnxjQuPU"
-    //             .parse()
-    //             .unwrap()
-    //     }
-    //     fn cid3() -> Cid {
-    //         "bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi"
-    //             .parse()
-    //             .unwrap()
-    //     }
-    //     fn cid4() -> Cid {
-    //         "QmbWqxBEKC3P8tqsKc98xmWNzrzDtRLMiMPL8wBuTGsMnR"
-    //             .parse()
-    //             .unwrap()
-    //     }
-    //     fn cid5() -> Cid {
-    //         "QmSnuWmxptJZdLJpKRarxBMS2Ju2oANVrgbr2xWbie9b2D"
-    //             .parse()
-    //             .unwrap()
-    //     }
-    //     fn cid6() -> Cid {
-    //         "QmdmQXB2mzChmMeKY47C43LxUdg1NDJ5MWcKMKxDu7RgQm"
-    //             .parse()
-    //             .unwrap()
-    //     }
-    //     fn cid7() -> Cid {
-    //         "bafybeiaysi4s6lnjev27ln5icwm6tueaw2vdykrtjkwiphwekaywqhcjze"
-    //             .parse()
-    //             .unwrap()
-    //     }
-    //     fn cid8() -> Cid {
-    //         "bafyreif3tfdpr5n4jdrbielmcapwvbpcthepfkwq2vwonmlhirbjmotedi"
-    //             .parse()
-    //             .unwrap()
-    //     }
-    //     fn cid9() -> Cid {
-    //         "bafyreicnokmhmrnlp2wjhyk2haep4tqxiptwfrp2rrs7rzq7uk766chqvq"
-    //             .parse()
-    //             .unwrap()
-    //     }
-
-    #[test]
-    fn test_next_from_node_empty() {
-        let node = Node {
-            left: None,
-            entries: vec![],
-        };
-        let mut stack = vec![];
-        push_from_node(&mut stack, &node).unwrap();
-        assert_eq!(stack.last(), None);
-    }
-
-    #[test]
-    fn test_needs_from_node_just_left() {
-        let node = Node {
-            left: Some(cid1()),
-            entries: vec![],
-        };
-        let mut stack = vec![];
-        push_from_node(&mut stack, &node).unwrap();
-        assert_eq!(stack.last(), Some(Need::Node(cid1())).as_ref());
-    }
-
-    //     #[test]
-    //     fn test_needs_from_node_just_one_record() {
-    //         let node = Node {
-    //             left: None,
-    //             entries: vec![Entry {
-    //                 keysuffix: "asdf".into(),
-    //                 prefix_len: 0,
-    //                 value: cid1(),
-    //                 tree: None,
-    //             }],
-    //         };
-    //         assert_eq!(
-    //             needs_from_node(node).unwrap(),
-    //             vec![Need::Record {
-    //                 rkey: "asdf".into(),
-    //                 cid: cid1(),
-    //             },]
-    //         );
-    //     }
-
-    //     #[test]
-    //     fn test_needs_from_node_two_records() {
-    //         let node = Node {
-    //             left: None,
-    //             entries: vec![
-    //                 Entry {
-    //                     keysuffix: "asdf".into(),
-    //                     prefix_len: 0,
-    //                     value: cid1(),
-    //                     tree: None,
-    //                 },
-    //                 Entry {
-    //                     keysuffix: "gh".into(),
-    //                     prefix_len: 2,
-    //                     value: cid2(),
-    //                     tree: None,
-    //                 },
-    //             ],
-    //         };
-    //         assert_eq!(
-    //             needs_from_node(node).unwrap(),
-    //             vec![
-    //                 Need::Record {
-    //                     rkey: "asdf".into(),
-    //                     cid: cid1(),
-    //                 },
-    //                 Need::Record {
-    //                     rkey: "asgh".into(),
-    //                     cid: cid2(),
-    //                 },
-    //             ]
-    //         );
-    //     }
-
-    //     #[test]
-    //     fn test_needs_from_node_with_both() {
-    //         let node = Node {
-    //             left: None,
-    //             entries: vec![Entry {
-    //                 keysuffix: "asdf".into(),
-    //                 prefix_len: 0,
-    //                 value: cid1(),
-    //                 tree: Some(cid2()),
-    //             }],
-    //         };
-    //         assert_eq!(
-    //             needs_from_node(node).unwrap(),
-    //             vec![
-    //                 Need::Record {
-    //                     rkey: "asdf".into(),
-    //                     cid: cid1(),
-    //                 },
-    //                 Need::Node(cid2()),
-    //             ]
-    //         );
-    //     }
-
-    //     #[test]
-    //     fn test_needs_from_node_left_and_record() {
-    //         let node = Node {
-    //             left: Some(cid1()),
-    //             entries: vec![Entry {
-    //                 keysuffix: "asdf".into(),
-    //                 prefix_len: 0,
-    //                 value: cid2(),
-    //                 tree: None,
-    //             }],
-    //         };
-    //         assert_eq!(
-    //             needs_from_node(node).unwrap(),
-    //             vec![
-    //                 Need::Node(cid1()),
-    //                 Need::Record {
-    //                     rkey: "asdf".into(),
-    //                     cid: cid2(),
-    //                 },
-    //             ]
-    //         );
-    //     }
-
-    //     #[test]
-    //     fn test_needs_from_full_node() {
-    //         let node = Node {
-    //             left: Some(cid1()),
-    //             entries: vec![
-    //                 Entry {
-    //                     keysuffix: "asdf".into(),
-    //                     prefix_len: 0,
-    //                     value: cid2(),
-    //                     tree: Some(cid3()),
-    //                 },
-    //                 Entry {
-    //                     keysuffix: "ghi".into(),
-    //                     prefix_len: 1,
-    //                     value: cid4(),
-    //                     tree: Some(cid5()),
-    //                 },
-    //                 Entry {
-    //                     keysuffix: "jkl".into(),
-    //                     prefix_len: 2,
-    //                     value: cid6(),
-    //                     tree: Some(cid7()),
-    //                 },
-    //                 Entry {
-    //                     keysuffix: "mno".into(),
-    //                     prefix_len: 4,
-    //                     value: cid8(),
-    //                     tree: Some(cid9()),
-    //                 },
-    //             ],
-    //         };
-    //         assert_eq!(
-    //             needs_from_node(node).unwrap(),
-    //             vec![
-    //                 Need::Node(cid1()),
-    //                 Need::Record {
-    //                     rkey: "asdf".into(),
-    //                     cid: cid2(),
-    //                 },
-    //                 Need::Node(cid3()),
-    //                 Need::Record {
-    //                     rkey: "aghi".into(),
-    //                     cid: cid4(),
-    //                 },
-    //                 Need::Node(cid5()),
-    //                 Need::Record {
-    //                     rkey: "agjkl".into(),
-    //                     cid: cid6(),
-    //                 },
-    //                 Need::Node(cid7()),
-    //                 Need::Record {
-    //                     rkey: "agjkmno".into(),
-    //                     cid: cid8(),
-    //                 },
-    //                 Need::Node(cid9()),
-    //             ]
-    //         );
-    //     }
-}

src/lib.rs

···
 //! For now see the [examples](https://tangled.org/@microcosm.blue/repo-stream/tree/main/examples)
 
 pub mod disk;
-pub mod disk_drive;
-pub mod disk_redb;
-pub mod disk_sqlite;
-pub mod disk_walk;
 pub mod drive;
 pub mod mst;
 pub mod walk;