disk kinda sorta works?

examples/disk-read-file/main.rs

···
 extern crate repo_stream;
 use clap::Parser;
-use futures::TryStreamExt;
-use iroh_car::CarReader;
-use std::convert::Infallible;
+use repo_stream::disk::SqliteStore;
+use repo_stream::drive::Processable;
+use serde::{Deserialize, Serialize};
 use std::path::PathBuf;
 
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
···
     tmpfile: PathBuf,
 }
 
+#[derive(Clone, Serialize, Deserialize)]
+struct S(usize);
+
+impl Processable for S {}
+
 #[tokio::main]
 async fn main() -> Result<()> {
     env_logger::init();
···
     let reader = tokio::fs::File::open(car).await?;
     let reader = tokio::io::BufReader::new(reader);
 
-    println!("hello!");
-
-    let reader = CarReader::new(reader).await?;
-
-    let redb_store = repo_stream::disk_redb::RedbStore::new(tmpfile).await?;
-
-    let root = reader
-        .header()
-        .roots()
-        .first()
-        .ok_or("missing root")?
-        .clone();
-    log::debug!("root: {root:?}");
-
-    // let stream = Box::pin(reader.stream());
-    let stream = std::pin::pin!(reader.stream());
-
-    let (commit, v) =
-        repo_stream::disk_drive::Vehicle::init(root, stream, redb_store, |block| block.len())
-            .await?;
-    let mut record_stream = std::pin::pin!(v.stream());
+    let mut driver =
+        match repo_stream::drive::load_car(reader, |block| S(block.len()), 1024).await? {
+            repo_stream::drive::Vehicle::Lil(_, _) => panic!("try this on a bigger car"),
+            repo_stream::drive::Vehicle::Big(big_stuff) => {
+                let disk_store = SqliteStore::new(tmpfile);
+                let (commit, driver) = big_stuff.finish_loading(disk_store).await?;
+                log::warn!("big: {:?}", commit);
+                driver
+            }
+        };
 
-    log::info!("got commit: {commit:?}");
+    println!("hello!");
 
-    while let Some((rkey, _rec)) = record_stream.try_next().await? {
-        log::info!("got {rkey:?}");
+    let mut n = 0;
+    loop {
+        let (d, Some(pairs)) = driver.next_chunk(256).await? else {
+            break;
+        };
+        driver = d;
+        n += pairs.len();
+        // log::info!("got {rkey:?}");
     }
-    log::info!("bye!");
+    log::info!("bye! {n}");
 
     Ok(())
 }

examples/read-file/main.rs

···
 extern crate repo_stream;
 use clap::Parser;
-use futures::TryStreamExt;
-use iroh_car::CarReader;
-use std::convert::Infallible;
 use std::path::PathBuf;
 
 type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
···
     let reader = tokio::fs::File::open(file).await?;
     let reader = tokio::io::BufReader::new(reader);
 
-    println!("hello!");
-
-    let reader = CarReader::new(reader).await?;
-
-    let root = reader
-        .header()
-        .roots()
-        .first()
-        .ok_or("missing root")?
-        .clone();
-    log::debug!("root: {root:?}");
-
-    // let stream = Box::pin(reader.stream());
-    let stream = std::pin::pin!(reader.stream());
-
-    let (commit, v) =
-        repo_stream::drive::Vehicle::init(root, stream, |block| Ok::<_, Infallible>(block.len()))
-            .await?;
-    let mut record_stream = std::pin::pin!(v.stream());
+    let (commit, mut driver) =
+        match repo_stream::drive::load_car(reader, |block| block.len(), 1024 * 1024).await? {
+            repo_stream::drive::Vehicle::Lil(commit, mem_driver) => (commit, mem_driver),
+            repo_stream::drive::Vehicle::Big(_) => panic!("can't handle big cars yet"),
+        };
 
     log::info!("got commit: {commit:?}");
 
-    while let Some((rkey, _rec)) = record_stream.try_next().await? {
-        log::info!("got {rkey:?}");
+    let mut n = 0;
+    while let Some(pairs) = driver.next_chunk(256).await? {
+        n += pairs.len();
+        // log::info!("got {rkey:?}");
     }
-    log::info!("bye!");
+    log::info!("bye! {n}");
 
     Ok(())
 }

src/disk.rs

···
+use rusqlite::OptionalExtension;
+use std::error::Error;
+use std::path::PathBuf;
+
+pub trait StorageErrorBase: Error + Send + 'static {}
+
+/// high level potential storage resource
+///
+/// separating this allows (hopefully) implementing a storage pool that can
+/// async-block when until a member is available to use
+pub trait DiskStore {
+    type StorageError: StorageErrorBase + Send;
+    type Access: DiskAccess<StorageError = Self::StorageError>;
+    fn get_access(&mut self) -> impl Future<Output = Result<Self::Access, Self::StorageError>>;
+}
+
+/// actual concrete access to disk storage
+pub trait DiskAccess: Send {
+    type StorageError: StorageErrorBase;
+
+    fn get_writer(&mut self) -> Result<impl DiskWriter<Self::StorageError>, Self::StorageError>;
+
+    fn get_reader(
+        &self,
+    ) -> Result<impl DiskReader<StorageError = Self::StorageError>, Self::StorageError>;
+
+    // TODO: force a cleanup implementation?
+}
+
+pub trait DiskWriter<E: StorageErrorBase> {
+    fn put(&mut self, key: Vec<u8>, val: Vec<u8>) -> Result<(), E>;
+}
+
+pub trait DiskReader {
+    type StorageError: StorageErrorBase;
+    fn get(&mut self, key: Vec<u8>) -> Result<Option<Vec<u8>>, Self::StorageError>;
+}
+
+/////////////////
+
+pub struct SqliteStore {
+    path: PathBuf,
+}
+
+impl SqliteStore {
+    pub fn new(path: PathBuf) -> Self {
+        Self { path }
+    }
+}
+
+impl StorageErrorBase for rusqlite::Error {}
+
+impl DiskStore for SqliteStore {
+    type StorageError = rusqlite::Error;
+    type Access = SqliteAccess;
+    async fn get_access(&mut self) -> Result<SqliteAccess, rusqlite::Error> {
+        let path = self.path.clone();
+        let conn = tokio::task::spawn_blocking(move || {
+            let conn = rusqlite::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::StorageError>(conn)
+        })
+        .await
+        .expect("join error")?;
+
+        Ok(SqliteAccess { conn })
+    }
+}
+
+pub struct SqliteAccess {
+    conn: rusqlite::Connection,
+}
+
+impl DiskAccess for SqliteAccess {
+    type StorageError = rusqlite::Error;
+    fn get_writer(&mut self) -> Result<impl DiskWriter<rusqlite::Error>, rusqlite::Error> {
+        let insert_stmt = self
+            .conn
+            .prepare("INSERT INTO blocks (key, val) VALUES (?1, ?2)")?;
+        Ok(SqliteWriter { insert_stmt })
+    }
+    fn get_reader(
+        &self,
+    ) -> Result<impl DiskReader<StorageError = rusqlite::Error>, rusqlite::Error> {
+        let select_stmt = self.conn.prepare("SELECT val FROM blocks WHERE key = ?1")?;
+        Ok(SqliteReader { select_stmt })
+    }
+}
+
+pub struct SqliteWriter<'conn> {
+    insert_stmt: rusqlite::Statement<'conn>,
+}
+
+impl DiskWriter<rusqlite::Error> for SqliteWriter<'_> {
+    fn put(&mut self, key: Vec<u8>, val: Vec<u8>) -> rusqlite::Result<()> {
+        self.insert_stmt.execute((key, val))?;
+        Ok(())
+    }
+}
+
+pub struct SqliteReader<'conn> {
+    select_stmt: rusqlite::Statement<'conn>,
+}
+
+impl DiskReader for SqliteReader<'_> {
+    type StorageError = rusqlite::Error;
+    fn get(&mut self, key: Vec<u8>) -> rusqlite::Result<Option<Vec<u8>>> {
+        self.select_stmt
+            .query_one((&key,), |row| row.get(0))
+            .optional()
+    }
+}
+
+// /// The main storage interface for MST blocks
+// ///
+// /// **Note**: `get` and `put` are **synchronous methods that may block**
+// pub trait BlockStore<T: Clone> {
+//     fn get(&self, cid: Cid) -> Option<MaybeProcessedBlock<T>>;
+//     fn put(&mut self, cid: Cid, mpb: MaybeProcessedBlock<T>);
+// }
+
+// ///// wheee
+
+// /// In-memory MST block storage
+// ///
+// /// a thin wrapper around a hashmap
+// pub struct MemoryStore<T: Clone> {
+//     map: HashMap<Cid, MaybeProcessedBlock<T>>,
+// }
+
+// impl<T: Clone> BlockStore<T> for MemoryStore<T> {
+//     fn get(&self, cid: Cid) -> Option<MaybeProcessedBlock<T>> {
+//         self.map.get(&cid).map(|t| t.clone())
+//     }
+//     fn put(&mut self, cid: Cid, mpb: MaybeProcessedBlock<T>) {
+//         self.map.insert(cid, mpb);
+//     }
+// }
+
+// //// the fun bits
+
+// pub struct HybridStore<T: Clone, D: DiskStore> {
+//     mem: MemoryStore<T>,
+//     disk: D,
+// }
+
+// impl<T: Clone, D: DiskStore> BlockStore<T> for HybridStore<T, D> {
+//     fn get(&self, _cid: Cid) -> Option<MaybeProcessedBlock<T>> { todo!() }
+//     fn put(&mut self, _cid: Cid, _mpb: MaybeProcessedBlock<T>) { todo!() }
+// }

src/drive.rs

···
 //! Consume an MST block stream, producing an ordered stream of records
 
-use futures::{Stream, TryStreamExt};
+use crate::disk::{DiskAccess, DiskStore, DiskWriter, StorageErrorBase};
 use ipld_core::cid::Cid;
+use iroh_car::CarReader;
+use serde::de::DeserializeOwned;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use std::error::Error;
+use std::convert::Infallible;
+use tokio::io::AsyncRead;
 
 use crate::mst::{Commit, Node};
-use crate::walk::{Step, Trip, Walker};
+use crate::walk::{DiskTrip, Step, Trip, Walker};
 
 /// Errors that can happen while consuming and emitting blocks and records
 #[derive(Debug, thiserror::Error)]
-pub enum DriveError<E: Error> {
-    #[error("Failed to initialize CarReader: {0}")]
+pub enum DriveError {
+    #[error("Error from iroh_car: {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>),
+    BadBlock(#[from] serde_ipld_dagcbor::DecodeError<Infallible>),
     #[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<E>),
+    Tripped(#[from] Trip),
+    #[error("CAR file had no roots")]
+    MissingRoot,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum DiskDriveError<E: StorageErrorBase> {
+    #[error("Error from iroh_car: {0}")]
+    CarReader(#[from] iroh_car::Error),
+    #[error("Failed to decode commit block: {0}")]
+    BadBlock(#[from] serde_ipld_dagcbor::DecodeError<Infallible>),
+    #[error("Storage error")]
+    StorageError(#[from] E),
+    #[error("The Commit block reference by the root was not found")]
+    MissingCommit,
+    #[error("The MST block {0} could not be found")]
+    MissingBlock(Cid),
+    #[error("Encode error: {0}")]
+    BincodeEncodeError(#[from] bincode::error::EncodeError),
+    #[error("Decode error: {0}")]
+    BincodeDecodeError(#[from] bincode::error::DecodeError),
+    #[error("disk tripped: {0}")]
+    DiskTripped(#[from] DiskTrip<E>),
 }
 
-type CarBlock<E> = Result<(Cid, Vec<u8>), E>;
+// #[derive(Debug, thiserror::Error)]
+// pub enum Boooooo<E: StorageErrorBase> {
+//     #[error("disk tripped: {0}")]
+//     DiskTripped(#[from] DiskTrip<E>),
+//     #[error("dde whatever: {0}")]
+//     DiskDriveError(#[from] DiskDriveError<E>),
+// }
+
+pub trait Processable: Clone + Serialize + DeserializeOwned {}
 
-#[derive(Debug)]
-pub enum MaybeProcessedBlock<T, E> {
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum MaybeProcessedBlock<T> {
     /// A block that's *probably* a Node (but we can't know yet)
     ///
     /// It *can be* a record that suspiciously looks a lot like a node, so we
···
     /// There's an alternative here, which would be to kick unprocessable blocks
     /// back to Raw, or maybe even a new RawUnprocessable variant. Then we could
     /// surface the typed error later if needed by trying to reprocess.
-    Processed(Result<T, E>),
+    Processed(T),
 }
 
-/// The core driver between the block stream and MST walker
-pub struct Vehicle<SE, S, T, P, PE>
-where
-    S: Stream<Item = CarBlock<SE>>,
-    P: Fn(&[u8]) -> Result<T, PE>,
-    PE: Error,
-{
-    block_stream: S,
-    blocks: HashMap<Cid, MaybeProcessedBlock<T, PE>>,
-    walker: Walker,
-    process: P,
+impl<T: Processable> Processable for MaybeProcessedBlock<T> {}
+
+pub enum Vehicle<R: AsyncRead + Unpin, T: Processable> {
+    Lil(Commit, MemDriver<T>),
+    Big(BigCar<R, T>),
 }
 
-impl<SE, S, T: Clone, P, PE> Vehicle<SE, S, T, P, PE>
-where
-    SE: Error + 'static,
-    S: Stream<Item = CarBlock<SE>> + Unpin,
-    P: Fn(&[u8]) -> Result<T, PE>,
-    PE: Error,
-{
-    /// 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,
-        mut block_stream: S,
-        process: P,
-    ) -> Result<(Commit, Self), DriveError<PE>> {
-        let mut blocks = HashMap::new();
+pub async fn load_car<R: AsyncRead + Unpin, T: Processable>(
+    reader: R,
+    process: fn(&[u8]) -> T,
+    max_size: usize,
+) -> Result<Vehicle<R, T>, DriveError> {
+    let mut mem_blocks = HashMap::new();
+
+    let mut car = CarReader::new(reader).await?;
+
+    let root = *car
+        .header()
+        .roots()
+        .first()
+        .ok_or(DriveError::MissingRoot)?;
+    log::debug!("root: {root:?}");
+
+    let mut commit = None;
+
+    // try to load all the blocks into memory
+    while let Some((cid, data)) = car.next_block().await? {
+        // the root commit is a Special Third Kind of block that we need to make
+        // sure not to optimistically send to the processing function
+        if cid == root {
+            let c: Commit = serde_ipld_dagcbor::from_slice(&data)?;
+            commit = Some(c);
+            continue;
+        }
 
-        let mut commit = None;
+        // remaining possible types: node, record, other. optimistically process
+        // TODO: get the actual in-memory size to compute disk spill
+        let maybe_processed = if Node::could_be(&data) {
+            MaybeProcessedBlock::Raw(data)
+        } else {
+            MaybeProcessedBlock::Processed(process(&data))
+        };
 
-        while let Some((cid, data)) = block_stream
-            .try_next()
-            .await
-            .map_err(|e| DriveError::CarBlockError(e.into()))?
-        {
-            if cid == root {
-                let c: Commit = serde_ipld_dagcbor::from_slice(&data)
-                    .map_err(|e| DriveError::BadCommit(e.into()))?;
-                commit = Some(c);
-                break;
-            } else {
-                blocks.insert(
-                    cid,
-                    if Node::could_be(&data) {
-                        MaybeProcessedBlock::Raw(data)
-                    } else {
-                        MaybeProcessedBlock::Processed(process(&data))
-                    },
-                );
-            }
+        // stash (maybe processed) blocks in memory as long as we have room
+        mem_blocks.insert(cid, maybe_processed);
+        if mem_blocks.len() >= max_size {
+            return Ok(Vehicle::Big(BigCar {
+                car,
+                root,
+                process,
+                max_size,
+                mem_blocks,
+                commit,
+            }));
         }
+    }
 
-        // we either broke out or read all the blocks without finding the commit...
-        let commit = commit.ok_or(DriveError::MissingCommit)?;
+    // all blocks loaded and we fit in memory! hopefully we found the commit...
+    let commit = commit.ok_or(DriveError::MissingCommit)?;
 
-        let walker = Walker::new(commit.data);
+    let walker = Walker::new(commit.data);
 
-        let me = Self {
-            block_stream,
-            blocks,
+    Ok(Vehicle::Lil(
+        commit,
+        MemDriver {
+            blocks: mem_blocks,
             walker,
             process,
-        };
-        Ok((commit, me))
-    }
+        },
+    ))
+}
 
-    async fn drive_until(&mut self, cid_needed: Cid) -> Result<(), DriveError<PE>> {
-        while let Some((cid, data)) = self
-            .block_stream
-            .try_next()
-            .await
-            .map_err(|e| DriveError::CarBlockError(e.into()))?
-        {
-            self.blocks.insert(
-                cid,
-                if Node::could_be(&data) {
+/// a paritally memory-loaded car file that needs disk spillover to continue
+pub struct BigCar<R: AsyncRead + Unpin, T: Processable> {
+    car: CarReader<R>,
+    root: Cid,
+    process: fn(&[u8]) -> T,
+    max_size: usize,
+    mem_blocks: HashMap<Cid, MaybeProcessedBlock<T>>,
+    pub commit: Option<Commit>,
+}
+
+fn encode(v: impl Serialize) -> Result<Vec<u8>, bincode::error::EncodeError> {
+    bincode::serde::encode_to_vec(v, bincode::config::standard())
+}
+
+pub fn decode<T: Processable>(bytes: &[u8]) -> Result<T, bincode::error::DecodeError> {
+    let (t, n) = bincode::serde::decode_from_slice(bytes, bincode::config::standard())?;
+    assert_eq!(n, bytes.len(), "expected to decode all bytes"); // TODO
+    Ok(t)
+}
+
+impl<R: AsyncRead + Unpin, T: Processable + Send + 'static> BigCar<R, T> {
+    pub async fn finish_loading<S: DiskStore>(
+        mut self,
+        mut store: S,
+    ) -> Result<(Commit, BigCarReady<T, S::Access>), DiskDriveError<S::StorageError>>
+    where
+        S::Access: Send + 'static,
+        S::StorageError: 'static,
+    {
+        // set up access for real
+        let mut access = store.get_access().await?;
+
+        // move access in and back out so we can manage lifetimes
+        // dump mem blocks into the store
+        access = tokio::task::spawn(async move {
+            let mut writer = access.get_writer()?;
+            for (k, v) in self.mem_blocks {
+                let key_bytes = k.to_bytes();
+                let val_bytes = encode(v)?; // TODO
+                writer.put(key_bytes, val_bytes)?;
+            }
+            drop(writer); // cannot outlive access
+            Ok::<_, DiskDriveError<S::StorageError>>(access)
+        })
+        .await
+        .unwrap()?;
+
+        // dump the rest to disk (in chunks)
+        loop {
+            let mut chunk = vec![];
+            loop {
+                let Some((cid, data)) = self.car.next_block().await? else {
+                    break;
+                };
+                // we still gotta keep checking for the root since we might not have it
+                if cid == self.root {
+                    let c: Commit = serde_ipld_dagcbor::from_slice(&data)?;
+                    self.commit = Some(c);
+                    continue;
+                }
+                // remaining possible types: node, record, other. optimistically process
+                // TODO: get the actual in-memory size to compute disk spill
+                let maybe_processed = if Node::could_be(&data) {
                     MaybeProcessedBlock::Raw(data)
                 } else {
                     MaybeProcessedBlock::Processed((self.process)(&data))
-                },
-            );
-            if cid == cid_needed {
-                return Ok(());
+                };
+                chunk.push((cid, maybe_processed));
+                if chunk.len() >= self.max_size {
+                    // eventually this won't be .len()
+                    break;
+                }
             }
+            if chunk.is_empty() {
+                break;
+            }
+
+            // move access in and back out so we can manage lifetimes
+            // dump mem blocks into the store
+            access = tokio::task::spawn_blocking(move || {
+                let mut writer = access.get_writer()?;
+                for (k, v) in chunk {
+                    let key_bytes = k.to_bytes();
+                    let val_bytes = encode(v)?; // TODO
+                    writer.put(key_bytes, val_bytes)?;
+                }
+                drop(writer); // cannot outlive access
+                Ok::<_, DiskDriveError<S::StorageError>>(access)
+            })
+            .await
+            .unwrap()?; // TODO
         }
 
-        // if we never found the block
-        Err(DriveError::MissingBlock(cid_needed))
+        let commit = self.commit.ok_or(DiskDriveError::MissingCommit)?;
+
+        let walker = Walker::new(commit.data);
+
+        Ok((
+            commit,
+            BigCarReady {
+                process: self.process,
+                access,
+                walker,
+            },
+        ))
     }
+}
 
+pub struct BigCarReady<T: Clone, A: DiskAccess> {
+    process: fn(&[u8]) -> T,
+    access: A,
+    walker: Walker,
+}
+
+impl<T: Processable + Send + 'static, A: DiskAccess + Send + 'static> BigCarReady<T, A> {
+    pub async fn next_chunk(
+        mut self,
+        n: usize,
+    ) -> Result<(Self, Option<Vec<(String, T)>>), DiskDriveError<A::StorageError>>
+    where
+        A::StorageError: Send,
+    {
+        let mut out = Vec::with_capacity(n);
+        (self, out) = tokio::task::spawn_blocking(move || {
+            let access = self.access;
+            let mut reader = access.get_reader()?;
+
+            for _ in 0..n {
+                // walk as far as we can until we run out of blocks or find a record
+                match self.walker.disk_step(&mut reader, self.process)? {
+                    Step::Missing(cid) => return Err(DiskDriveError::MissingBlock(cid)),
+                    Step::Finish => break,
+                    Step::Step { rkey, data } => {
+                        out.push((rkey, data));
+                        continue;
+                    }
+                };
+            }
+
+            drop(reader); // cannot outlive access
+            self.access = access;
+            Ok::<_, DiskDriveError<A::StorageError>>((self, out))
+        })
+        .await
+        .unwrap()?; // TODO
+
+        if out.is_empty() {
+            Ok((self, None))
+        } else {
+            Ok((self, Some(out)))
+        }
+    }
+}
+
+/// The core driver between the block stream and MST walker
+///
+/// In the future, PDSs will export CARs in a stream-friendly order that will
+/// enable processing them with tiny memory overhead. But that future is not
+/// here yet.
+///
+/// CARs are almost always in a stream-unfriendly order, so I'm reverting the
+/// optimistic stream features: we load all block first, then walk the MST.
+///
+/// This makes things much simpler: we only need to worry about spilling to disk
+/// in one place, and we always have a reasonable expecatation about how much
+/// work the init function will do. We can drop the CAR reader before walking,
+/// so the sync/async boundaries become a little easier to work around.
+#[derive(Debug)]
+pub struct MemDriver<T: Processable> {
+    blocks: HashMap<Cid, MaybeProcessedBlock<T>>,
+    walker: Walker,
+    process: fn(&[u8]) -> T,
+}
+
+impl<T: Processable> MemDriver<T> {
     /// Manually step through the record outputs
-    pub async fn next_record(&mut self) -> Result<Option<(String, T)>, DriveError<PE>> {
-        loop {
+    pub async fn next_chunk(&mut self, n: usize) -> Result<Option<Vec<(String, T)>>, DriveError> {
+        let mut out = Vec::with_capacity(n);
+        for _ in 0..n {
             // walk as far as we can until we run out of blocks or find a record
-            let cid_needed = match self.walker.step(&mut self.blocks, &self.process)? {
-                Step::Rest(cid) => cid,
-                Step::Finish => return Ok(None),
-                Step::Step { rkey, data } => return Ok(Some((rkey, data))),
+            match self.walker.step(&mut self.blocks, self.process)? {
+                Step::Missing(cid) => return Err(DriveError::MissingBlock(cid)),
+                Step::Finish => break,
+                Step::Step { rkey, data } => {
+                    out.push((rkey, data));
+                    continue;
+                }
             };
+        }
 
-            // load blocks until we reach that cid
-            self.drive_until(cid_needed).await?;
+        if out.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(out))
         }
-    }
-
-    /// Convert to a futures::stream of record outputs
-    pub fn stream(self) -> impl Stream<Item = Result<(String, T), DriveError<PE>>> {
-        futures::stream::try_unfold(self, |mut this| async move {
-            let maybe_record = this.next_record().await?;
-            Ok(maybe_record.map(|b| (b, this)))
-        })
     }
 }

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;

src/walk.rs

···
 //! Depth-first MST traversal
 
-use crate::drive::MaybeProcessedBlock;
+use crate::disk::{DiskReader, StorageErrorBase};
+use crate::drive::{MaybeProcessedBlock, Processable};
 use crate::mst::Node;
 use ipld_core::cid::Cid;
 use std::collections::HashMap;
-use std::error::Error;
+use std::convert::Infallible;
 
 /// Errors that can happen while walking
 #[derive(Debug, thiserror::Error)]
-pub enum Trip<E: Error> {
+pub enum Trip {
     #[error("empty mst nodes are not allowed")]
     NodeEmpty,
+    #[error("Failed to fingerprint commit block")]
+    BadCommitFingerprint,
     #[error("Failed to decode commit block: {0}")]
-    BadCommit(Box<dyn std::error::Error>),
+    BadCommit(#[from] serde_ipld_dagcbor::DecodeError<Infallible>),
     #[error("Action node error: {0}")]
     RkeyError(#[from] RkeyError),
-    #[error("Process failed: {0}")]
-    ProcessFailed(E),
     #[error("Encountered an rkey out of order while walking the MST")]
     RkeyOutOfOrder,
 }
 
+/// Errors that can happen while walking
+#[derive(Debug, thiserror::Error)]
+pub enum DiskTrip<E: StorageErrorBase> {
+    #[error("tripped: {0}")]
+    Trip(#[from] Trip),
+    #[error("storage error: {0}")]
+    StorageError(#[from] E),
+    #[error("Decode error: {0}")]
+    BincodeDecodeError(#[from] bincode::error::DecodeError),
+}
+
 /// Errors from invalid Rkeys
 #[derive(Debug, thiserror::Error)]
 pub enum RkeyError {
···
 /// Walker outputs
 #[derive(Debug)]
 pub enum Step<T> {
-    /// 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),
+    /// We needed this CID but it's not in the block store
+    Missing(Cid),
     /// Reached the end of the MST! yay!
     Finish,
     /// A record was found!
···
     }
 
     /// Advance through nodes until we find a record or can't go further
-    pub fn step<T: Clone, E: Error>(
+    pub fn step<T: Processable>(
         &mut self,
-        blocks: &mut HashMap<Cid, MaybeProcessedBlock<T, E>>,
-        process: impl Fn(&[u8]) -> Result<T, E>,
-    ) -> Result<Step<T>, Trip<E>> {
+        blocks: &mut HashMap<Cid, MaybeProcessedBlock<T>>,
+        process: impl Fn(&[u8]) -> T,
+    ) -> Result<Step<T>, Trip> {
         loop {
             let Some(mut need) = self.stack.last() else {
                 log::trace!("tried to walk but we're actually done.");
···
                     log::trace!("need node {cid:?}");
                     let Some(block) = blocks.remove(cid) else {
                         log::trace!("node not found, resting");
-                        return Ok(Step::Rest(*cid));
+                        return Ok(Step::Missing(*cid));
                     };
 
                     let MaybeProcessedBlock::Raw(data) = block else {
-                        return Err(Trip::BadCommit("failed commit fingerprint".into()));
+                        return Err(Trip::BadCommitFingerprint);
                     };
-                    let node = serde_ipld_dagcbor::from_slice::<Node>(&data)
-                        .map_err(|e| Trip::BadCommit(e.into()))?;
+                    let node =
+                        serde_ipld_dagcbor::from_slice::<Node>(&data).map_err(Trip::BadCommit)?;
 
                     // found node, make sure we remember
                     self.stack.pop();
···
                     log::trace!("need record {cid:?}");
                     let Some(data) = blocks.get_mut(cid) else {
                         log::trace!("record block not found, resting");
-                        return Ok(Step::Rest(*cid));
+                        return Ok(Step::Missing(*cid));
                     };
                     let rkey = rkey.clone();
                     let data = match data {
                         MaybeProcessedBlock::Raw(data) => process(data),
-                        MaybeProcessedBlock::Processed(Ok(t)) => Ok(t.clone()),
-                        bad => {
-                            // big hack to pull the error out -- this corrupts
-                            // a block, so we should not continue trying to work
-                            let mut steal = MaybeProcessedBlock::Raw(vec![]);
-                            std::mem::swap(&mut steal, bad);
-                            let MaybeProcessedBlock::Processed(Err(e)) = steal else {
-                                unreachable!();
-                            };
-                            return Err(Trip::ProcessFailed(e));
-                        }
+                        MaybeProcessedBlock::Processed(t) => t.clone(),
                     };
 
                     // found node, make sure we remember
                     self.stack.pop();
 
                     log::trace!("emitting a block as a step. depth={}", self.stack.len());
-                    let data = data.map_err(Trip::ProcessFailed)?;
 
                     // rkeys *must* be in order or else the tree is invalid (or
                     // we have a bug)
                     if rkey <= self.prev {
                         return Err(Trip::RkeyOutOfOrder);
+                    }
+                    self.prev = rkey.clone();
+
+                    return Ok(Step::Step { rkey, data });
+                }
+            }
+        }
+    }
+
+    /// blocking!!!!!!
+    pub fn disk_step<T: Processable, R: DiskReader>(
+        &mut self,
+        reader: &mut R,
+        process: impl Fn(&[u8]) -> T,
+    ) -> Result<Step<T>, DiskTrip<R::StorageError>> {
+        loop {
+            let Some(mut need) = self.stack.last() else {
+                log::trace!("tried to walk but we're actually done.");
+                return Ok(Step::Finish);
+            };
+
+            match &mut need {
+                Need::Node(cid) => {
+                    let cid_bytes = cid.to_bytes();
+                    log::trace!("need node {cid:?}");
+                    let Some(block_bytes) = reader.get(cid_bytes)? else {
+                        log::trace!("node not found, resting");
+                        return Ok(Step::Missing(*cid));
+                    };
+
+                    let block: MaybeProcessedBlock<T> = crate::drive::decode(&block_bytes)?;
+
+                    let MaybeProcessedBlock::Raw(data) = block else {
+                        return Err(Trip::BadCommitFingerprint.into());
+                    };
+                    let node =
+                        serde_ipld_dagcbor::from_slice::<Node>(&data).map_err(Trip::BadCommit)?;
+
+                    // found node, make sure we remember
+                    self.stack.pop();
+
+                    // queue up work on the found node next
+                    push_from_node(&mut self.stack, &node).map_err(Trip::RkeyError)?;
+                }
+                Need::Record { rkey, cid } => {
+                    log::trace!("need record {cid:?}");
+                    let cid_bytes = cid.to_bytes();
+                    let Some(data_bytes) = reader.get(cid_bytes)? else {
+                        log::trace!("record block not found, resting");
+                        return Ok(Step::Missing(*cid));
+                    };
+                    let data: MaybeProcessedBlock<T> = crate::drive::decode(&data_bytes)?;
+                    let rkey = rkey.clone();
+                    let data = match data {
+                        MaybeProcessedBlock::Raw(data) => process(&data),
+                        MaybeProcessedBlock::Processed(t) => t.clone(),
+                    };
+
+                    // found node, make sure we remember
+                    self.stack.pop();
+
+                    log::trace!("emitting a block as a step. depth={}", self.stack.len());
+
+                    // rkeys *must* be in order or else the tree is invalid (or
+                    // we have a bug)
+                    if rkey <= self.prev {
+                        return Err(DiskTrip::Trip(Trip::RkeyOutOfOrder));
                     }
                     self.prev = rkey.clone();