Merge branch 'disk'

Cargo.lock

···
 ]
 
 [[package]]
+name = "bincode"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "36eaf5d7b090263e8150820482d5d93cd964a81e4019913c972f4edcc6edb740"
+dependencies = [
+ "bincode_derive",
+ "serde",
+ "unty",
+]
+
+[[package]]
+name = "bincode_derive"
+version = "2.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf95709a440f45e986983918d0e8a1f30a9b1df04918fc828670606804ac3c09"
+dependencies = [
+ "virtue",
+]
+
+[[package]]
 name = "bitflags"
 version = "2.9.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2261d10cca569e4643e526d8dc2e62e433cc8aba21ab764233731f8d369bf394"
+
+[[package]]
+name = "block-buffer"
+version = "0.10.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
+dependencies = [
+ "generic-array",
+]
 
 [[package]]
 name = "bumpalo"
···
 ]
 
 [[package]]
+name = "cpufeatures"
+version = "0.2.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "59ed5838eebb26a2bb2e58f6d5b5316989ae9d08bab10e0e6d103e656d1b0280"
+dependencies = [
+ "libc",
+]
+
+[[package]]
 name = "criterion"
 version = "0.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5"
 
 [[package]]
+name = "crypto-common"
+version = "0.1.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
+dependencies = [
+ "generic-array",
+ "typenum",
+]
+
+[[package]]
 name = "data-encoding"
 version = "2.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "digest"
+version = "0.10.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
+dependencies = [
+ "block-buffer",
+ "crypto-common",
+]
+
+[[package]]
 name = "either"
 version = "1.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "errno"
+version = "0.3.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
+dependencies = [
+ "libc",
+ "windows-sys 0.60.2",
+]
+
+[[package]]
+name = "fallible-iterator"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
+
+[[package]]
+name = "fallible-streaming-iterator"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
+
+[[package]]
+name = "fastrand"
+version = "2.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be"
+
+[[package]]
+name = "foldhash"
+version = "0.1.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2"
+
+[[package]]
 name = "futures"
 version = "0.3.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "generic-array"
+version = "0.14.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2"
+dependencies = [
+ "typenum",
+ "version_check",
+]
+
+[[package]]
+name = "getrandom"
+version = "0.3.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "26145e563e54f2cadc477553f1ec5ee650b00862f0a58bcd12cbdc5f0ea2d2f4"
+dependencies = [
+ "cfg-if",
+ "libc",
+ "r-efi",
+ "wasi 0.14.7+wasi-0.2.4",
+]
+
+[[package]]
 name = "gimli"
 version = "0.32.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "cfg-if",
  "crunchy",
  "zerocopy",
+]
+
+[[package]]
+name = "hashbrown"
+version = "0.15.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1"
+dependencies = [
+ "foldhash",
+]
+
+[[package]]
+name = "hashlink"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7382cf6263419f2d8df38c55d7da83da5c18aef87fc7a7fc1fb1e344edfe14c1"
+dependencies = [
+ "hashbrown",
 ]
 
 [[package]]
···
 checksum = "58f929b4d672ea937a23a1ab494143d968337a5f47e56d0815df1e0890ddf174"
 
 [[package]]
+name = "libsqlite3-sys"
+version = "0.35.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "133c182a6a2c87864fe97778797e46c7e999672690dc9fa3ee8e241aa4a9c13f"
+dependencies = [
+ "pkg-config",
+ "vcpkg",
+]
+
+[[package]]
+name = "linux-raw-sys"
+version = "0.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df1d3c3b53da64cf5760482273a98e575c651a67eec7f77df96b5b642de8f039"
+
+[[package]]
 name = "lock_api"
 version = "0.4.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "78bed444cc8a2160f01cbcf811ef18cac863ad68ae8ca62092e8db51d51c761c"
 dependencies = [
  "libc",
- "wasi",
+ "wasi 0.11.1+wasi-snapshot-preview1",
  "windows-sys 0.59.0",
 ]
 
···
 checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
 
 [[package]]
+name = "pkg-config"
+version = "0.3.32"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
+
+[[package]]
 name = "plotters"
 version = "0.3.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 dependencies = [
  "proc-macro2",
 ]
+
+[[package]]
+name = "r-efi"
+version = "5.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f"
 
 [[package]]
 name = "rayon"
···
 name = "repo-stream"
 version = "0.1.1"
 dependencies = [
+ "bincode",
  "clap",
  "criterion",
  "env_logger",
···
  "iroh-car",
  "log",
  "multibase",
+ "rusqlite",
  "serde",
  "serde_bytes",
  "serde_ipld_dagcbor",
+ "sha2",
+ "tempfile",
  "thiserror 2.0.17",
  "tokio",
 ]
 
 [[package]]
+name = "rusqlite"
+version = "0.37.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "165ca6e57b20e1351573e3729b958bc62f0e48025386970b6e4d29e7a7e71f3f"
+dependencies = [
+ "bitflags",
+ "fallible-iterator",
+ "fallible-streaming-iterator",
+ "hashlink",
+ "libsqlite3-sys",
+ "smallvec",
+]
+
+[[package]]
 name = "rustc-demangle"
 version = "0.1.26"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "56f7d92ca342cea22a06f2121d944b4fd82af56988c270852495420f961d4ace"
+
+[[package]]
+name = "rustix"
+version = "1.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd15f8a2c5551a84d56efdc1cd049089e409ac19a3072d5037a17fd70719ff3e"
+dependencies = [
+ "bitflags",
+ "errno",
+ "libc",
+ "linux-raw-sys",
+ "windows-sys 0.60.2",
+]
 
 [[package]]
 name = "rustversion"
···
  "ryu",
  "serde",
  "serde_core",
+]
+
+[[package]]
+name = "sha2"
+version = "0.10.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283"
+dependencies = [
+ "cfg-if",
+ "cpufeatures",
+ "digest",
 ]
 
 [[package]]
···
 ]
 
 [[package]]
+name = "tempfile"
+version = "3.23.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2d31c77bdf42a745371d260a26ca7163f1e0924b64afa0b688e61b5a9fa02f16"
+dependencies = [
+ "fastrand",
+ "getrandom",
+ "once_cell",
+ "rustix",
+ "windows-sys 0.60.2",
+]
+
+[[package]]
 name = "thiserror"
 version = "1.0.69"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 ]
 
 [[package]]
+name = "typenum"
+version = "1.19.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb"
+
+[[package]]
 name = "unicode-ident"
 version = "1.0.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 checksum = "eb066959b24b5196ae73cb057f45598450d2c5f71460e98c49b738086eff9c06"
 
 [[package]]
+name = "unty"
+version = "0.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6d49784317cd0d1ee7ec5c716dd598ec5b4483ea832a2dced265471cc0f690ae"
+
+[[package]]
 name = "utf8parse"
 version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
 
 [[package]]
+name = "vcpkg"
+version = "0.2.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "accd4ea62f7bb7a82fe23066fb0957d48ef677f6eeb8215f372f52e48bb32426"
+
+[[package]]
+name = "version_check"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a"
+
+[[package]]
+name = "virtue"
+version = "0.0.18"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "051eb1abcf10076295e815102942cc58f9d5e3b4560e46e53c21e8ff6f3af7b1"
+
+[[package]]
 name = "walkdir"
 version = "2.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 version = "0.11.1+wasi-snapshot-preview1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
+
+[[package]]
+name = "wasi"
+version = "0.14.7+wasi-0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "883478de20367e224c0090af9cf5f9fa85bed63a95c1abf3afc5c083ebc06e8c"
+dependencies = [
+ "wasip2",
+]
+
+[[package]]
+name = "wasip2"
+version = "1.0.1+wasi-0.2.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0562428422c63773dad2c345a1882263bbf4d65cf3f42e90921f787ef5ad58e7"
+dependencies = [
+ "wit-bindgen",
+]
 
 [[package]]
 name = "wasm-bindgen"
···
 version = "0.53.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650"
+
+[[package]]
+name = "wit-bindgen"
+version = "0.46.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f17a85883d4e6d00e8a97c586de764dabcc06133f7f1d55dce5cdc070ad7fe59"
 
 [[package]]
 name = "zerocopy"

Cargo.toml

···
 repository = "https://tangled.org/@microcosm.blue/repo-stream"
 
 [dependencies]
+bincode = { version = "2.0.1", features = ["serde"] }
 futures = "0.3.31"
 futures-core = "0.3.31"
 ipld-core = { version = "0.4.2", features = ["serde"] }
 iroh-car = "0.5.1"
 log = "0.4.28"
 multibase = "0.9.2"
+rusqlite = "0.37.0"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_bytes = "0.11.19"
 serde_ipld_dagcbor = "0.6.4"
+sha2 = "0.10.9"
 thiserror = "2.0.17"
-tokio = "1.47.1"
+tokio = { version = "1.47.1", features = ["rt", "sync"] }
 
 [dev-dependencies]
 clap = { version = "4.5.48", features = ["derive"] }
 criterion = { version = "0.7.0", features = ["async_tokio"] }
 env_logger = "0.11.8"
 multibase = "0.9.2"
+tempfile = "3.23.0"
 tokio = { version = "1.47.1", features = ["full"] }
 
 [profile.profiling]
 inherits = "release"
 debug = true
+
+# [profile.release]
+# debug = true
 
 [[bench]]
 name = "non-huge-cars"

benches/huge-car.rs

···
 extern crate repo_stream;
-use futures::TryStreamExt;
-use iroh_car::CarReader;
-use std::convert::Infallible;
+use repo_stream::Driver;
 use std::path::{Path, PathBuf};
 
 use criterion::{Criterion, criterion_group, criterion_main};
···
     });
 }
 
-async fn drive_car(filename: impl AsRef<Path>) {
+async fn drive_car(filename: impl AsRef<Path>) -> usize {
     let reader = tokio::fs::File::open(filename).await.unwrap();
     let reader = tokio::io::BufReader::new(reader);
-    let reader = CarReader::new(reader).await.unwrap();
 
-    let root = reader
-        .header()
-        .roots()
-        .first()
-        .ok_or("missing root")
+    let mut driver = match Driver::load_car(reader, |block| block.len(), 1024)
+        .await
         .unwrap()
-        .clone();
-
-    let stream = std::pin::pin!(reader.stream());
-
-    let (_commit, v) =
-        repo_stream::drive::Vehicle::init(root, stream, |block| Ok::<_, Infallible>(block.len()))
-            .await
-            .unwrap();
-    let mut record_stream = std::pin::pin!(v.stream());
+    {
+        Driver::Memory(_, mem_driver) => mem_driver,
+        Driver::Disk(_) => panic!("not doing disk for benchmark"),
+    };
 
-    while let Some(_) = record_stream.try_next().await.unwrap() {
-        // just here for the drive
+    let mut n = 0;
+    while let Some(pairs) = driver.next_chunk(256).await.unwrap() {
+        n += pairs.len();
     }
+    n
 }
 
 criterion_group!(benches, criterion_benchmark);

benches/non-huge-cars.rs

···
 extern crate repo_stream;
-use futures::TryStreamExt;
-use iroh_car::CarReader;
-use std::convert::Infallible;
+use repo_stream::Driver;
 
 use criterion::{Criterion, criterion_group, criterion_main};
 
···
     });
 }
 
-async fn drive_car(bytes: &[u8]) {
-    let reader = CarReader::new(bytes).await.unwrap();
-
-    let root = reader
-        .header()
-        .roots()
-        .first()
-        .ok_or("missing root")
+async fn drive_car(bytes: &[u8]) -> usize {
+    let mut driver = match Driver::load_car(bytes, |block| block.len(), 32)
+        .await
         .unwrap()
-        .clone();
-
-    let stream = std::pin::pin!(reader.stream());
+    {
+        Driver::Memory(_, mem_driver) => mem_driver,
+        Driver::Disk(_) => panic!("not benching big cars here"),
+    };
 
-    let (_commit, v) =
-        repo_stream::drive::Vehicle::init(root, stream, |block| Ok::<_, Infallible>(block.len()))
-            .await
-            .unwrap();
-    let mut record_stream = std::pin::pin!(v.stream());
-
-    while let Some(_) = record_stream.try_next().await.unwrap() {
-        // just here for the drive
+    let mut n = 0;
+    while let Some(pairs) = driver.next_chunk(256).await.unwrap() {
+        n += pairs.len();
     }
+    n
 }
 
 criterion_group!(benches, criterion_benchmark);

examples/disk-read-file/main.rs

···
+/*!
+Read a CAR file by spilling to disk
+*/
+
+extern crate repo_stream;
+use clap::Parser;
+use repo_stream::{DiskBuilder, Driver, DriverBuilder};
+use std::path::PathBuf;
+
+#[derive(Debug, Parser)]
+struct Args {
+    #[arg()]
+    car: PathBuf,
+    #[arg()]
+    tmpfile: PathBuf,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    env_logger::init();
+
+    let Args { car, tmpfile } = Args::parse();
+
+    // repo-stream takes an AsyncRead as input. wrapping a filesystem read in
+    // BufReader can provide a really significant performance win.
+    let reader = tokio::fs::File::open(car).await?;
+    let reader = tokio::io::BufReader::new(reader);
+
+    log::info!("hello! reading the car...");
+
+    // in this example we only bother handling CARs that are too big for memory
+    // `noop` helper means: do no block processing, store the raw blocks
+    let driver = match DriverBuilder::new()
+        .with_mem_limit_mb(10) // how much memory can be used before disk spill
+        .load_car(reader)
+        .await?
+    {
+        Driver::Memory(_, _) => panic!("try this on a bigger car"),
+        Driver::Disk(big_stuff) => {
+            // we reach here if the repo was too big and needs to be spilled to
+            // disk to continue
+
+            // set up a disk store we can spill to
+            let disk_store = DiskBuilder::new().open(tmpfile).await?;
+
+            // do the spilling, get back a (similar) driver
+            let (commit, driver) = big_stuff.finish_loading(disk_store).await?;
+
+            // at this point you might want to fetch the account's signing key
+            // via the DID from the commit, and then verify the signature.
+            log::warn!("big's comit: {:?}", commit);
+
+            // pop the driver back out to get some code indentation relief
+            driver
+        }
+    };
+
+    // collect some random stats about the blocks
+    let mut n = 0;
+    let mut zeros = 0;
+
+    log::info!("walking...");
+
+    // this example uses the disk driver's channel mode: the tree walking is
+    // spawned onto a blocking thread, and we get chunks of rkey+blocks back
+    let (mut rx, join) = driver.to_channel(512);
+    while let Some(r) = rx.recv().await {
+        let pairs = r?;
+
+        // keep a count of the total number of blocks seen
+        n += pairs.len();
+
+        for (_, block) in pairs {
+            // for each block, count how many bytes are equal to '0'
+            // (this is just an example, you probably want to do something more
+            // interesting)
+            zeros += block.into_iter().filter(|&b| b == b'0').count()
+        }
+    }
+
+    log::info!("arrived! joining rx...");
+
+    // clean up the database. would be nice to do this in drop so it happens
+    // automatically, but some blocking work happens, so that's not allowed in
+    // async rust. 🤷‍♀️
+    join.await?.reset_store().await?;
+
+    log::info!("done. n={n} zeros={zeros}");
+
+    Ok(())
+}

examples/read-file/main.rs

···
+/*!
+Read a CAR file with in-memory processing
+*/
+
 extern crate repo_stream;
 use clap::Parser;
-use futures::TryStreamExt;
-use iroh_car::CarReader;
-use std::convert::Infallible;
+use repo_stream::{Driver, DriverBuilder};
 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 DriverBuilder::new()
+        .with_block_processor(|block| block.len())
+        .load_car(reader)
+        .await?
+    {
+        Driver::Memory(commit, mem_driver) => (commit, mem_driver),
+        Driver::Disk(_) => panic!("this example doesn't handle big CARs"),
+    };
 
     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! total records={n}");
 
     Ok(())
 }

readme.md

···
 # repo-stream
 
-Fast and (aspirationally) robust atproto CAR file processing in rust
+Efficient and robust atproto CAR file processing in rust
+
+todo
+
+- [ ] get an *emtpy* car for the test suite
+- [ ] implement a max size on disk limit
+
+
+-----
+
+older stuff (to clean up):
 
 
 current car processing times (records processed into their length usize, phil's dev machine):
···
   -> yeah the commit is returned from init
 - [ ] spec compliance todos
   - [x] assert that keys are ordered and fail if not
-  - [ ] verify node mst depth from key (possibly pending [interop test fixes](https://github.com/bluesky-social/atproto-interop-tests/issues/5))
+  - [x] verify node mst depth from key (possibly pending [interop test fixes](https://github.com/bluesky-social/atproto-interop-tests/issues/5))
 - [ ] performance todos
   - [x] consume the serialized nodes into a mutable efficient format
     - [ ] maybe customize the deserialize impl to do that directly?

src/disk.rs

···
+/*!
+Disk storage for blocks on disk
+
+Currently this uses sqlite. In testing sqlite wasn't the fastest, but it seemed
+to be the best behaved in terms of both on-disk space usage and memory usage.
+
+```no_run
+# use repo_stream::{DiskBuilder, DiskError};
+# #[tokio::main]
+# async fn main() -> Result<(), DiskError> {
+let store = DiskBuilder::new()
+    .with_cache_size_mb(32)
+    .with_max_stored_mb(1024) // errors when >1GiB of processed blocks are inserted
+    .open("/some/path.db".into()).await?;
+# Ok(())
+# }
+```
+*/
+
+use crate::drive::DriveError;
+use rusqlite::OptionalExtension;
+use std::path::PathBuf;
+
+#[derive(Debug, thiserror::Error)]
+pub enum DiskError {
+    /// A wrapped database error
+    ///
+    /// (The wrapped err should probably be obscured to remove public-facing
+    /// sqlite bits)
+    #[error(transparent)]
+    DbError(#[from] rusqlite::Error),
+    /// A tokio blocking task failed to join
+    #[error("Failed to join a tokio blocking task: {0}")]
+    JoinError(#[from] tokio::task::JoinError),
+    /// The total size of stored blocks exceeded the allowed size
+    ///
+    /// If you need to process *really* big CARs, you can configure a higher
+    /// limit.
+    #[error("Maximum disk size reached")]
+    MaxSizeExceeded,
+    #[error("this error was replaced, seeing this is a bug.")]
+    #[doc(hidden)]
+    Stolen,
+}
+
+impl DiskError {
+    /// hack for ownership challenges with the disk driver
+    pub(crate) fn steal(&mut self) -> Self {
+        let mut swapped = DiskError::Stolen;
+        std::mem::swap(self, &mut swapped);
+        swapped
+    }
+}
+
+/// Builder-style disk store setup
+pub struct DiskBuilder {
+    /// Database in-memory cache allowance
+    ///
+    /// Default: 32 MiB
+    pub cache_size_mb: usize,
+    /// Database stored block size limit
+    ///
+    /// Default: 10 GiB
+    ///
+    /// Note: actual size on disk may be more, but should approximately scale
+    /// with this limit
+    pub max_stored_mb: usize,
+}
+
+impl Default for DiskBuilder {
+    fn default() -> Self {
+        Self {
+            cache_size_mb: 32,
+            max_stored_mb: 10 * 1024, // 10 GiB
+        }
+    }
+}
+
+impl DiskBuilder {
+    /// Begin configuring the storage with defaults
+    pub fn new() -> Self {
+        Default::default()
+    }
+    /// Set the in-memory cache allowance for the database
+    ///
+    /// Default: 32 MiB
+    pub fn with_cache_size_mb(mut self, size: usize) -> Self {
+        self.cache_size_mb = size;
+        self
+    }
+    /// Set the approximate stored block size limit
+    ///
+    /// Default: 10 GiB
+    pub fn with_max_stored_mb(mut self, max: usize) -> Self {
+        self.max_stored_mb = max;
+        self
+    }
+    /// Open and initialize the actual disk storage
+    pub async fn open(self, path: PathBuf) -> Result<DiskStore, DiskError> {
+        DiskStore::new(path, self.cache_size_mb, self.max_stored_mb).await
+    }
+}
+
+/// On-disk block storage
+pub struct DiskStore {
+    conn: rusqlite::Connection,
+    max_stored: usize,
+    stored: usize,
+}
+
+impl DiskStore {
+    /// Initialize a new disk store
+    pub async fn new(
+        path: PathBuf,
+        cache_mb: usize,
+        max_stored_mb: usize,
+    ) -> Result<Self, DiskError> {
+        let max_stored = max_stored_mb * 2_usize.pow(20);
+        let conn = tokio::task::spawn_blocking(move || {
+            let conn = rusqlite::Connection::open(path)?;
+
+            let sqlite_one_mb = -(2_i64.pow(10)); // negative is kibibytes for sqlite cache_size
+
+            // conn.pragma_update(None, "journal_mode", "OFF")?;
+            // conn.pragma_update(None, "journal_mode", "MEMORY")?;
+            conn.pragma_update(None, "journal_mode", "WAL")?;
+            // conn.pragma_update(None, "wal_autocheckpoint", "0")?; // this lets things get a bit big on disk
+            conn.pragma_update(None, "synchronous", "OFF")?;
+            conn.pragma_update(
+                None,
+                "cache_size",
+                (cache_mb as i64 * sqlite_one_mb).to_string(),
+            )?;
+            Self::reset_tables(&conn)?;
+
+            Ok::<_, DiskError>(conn)
+        })
+        .await??;
+
+        Ok(Self {
+            conn,
+            max_stored,
+            stored: 0,
+        })
+    }
+    pub(crate) fn get_writer(&'_ mut self) -> Result<SqliteWriter<'_>, DiskError> {
+        let tx = self.conn.transaction()?;
+        Ok(SqliteWriter {
+            tx,
+            stored: &mut self.stored,
+            max: self.max_stored,
+        })
+    }
+    pub(crate) fn get_reader<'conn>(&'conn self) -> Result<SqliteReader<'conn>, DiskError> {
+        let select_stmt = self.conn.prepare("SELECT val FROM blocks WHERE key = ?1")?;
+        Ok(SqliteReader { select_stmt })
+    }
+    /// Drop and recreate the kv table
+    pub async fn reset(self) -> Result<Self, DiskError> {
+        tokio::task::spawn_blocking(move || {
+            Self::reset_tables(&self.conn)?;
+            Ok(self)
+        })
+        .await?
+    }
+    fn reset_tables(conn: &rusqlite::Connection) -> Result<(), DiskError> {
+        conn.execute("DROP TABLE IF EXISTS blocks", ())?;
+        conn.execute(
+            "CREATE TABLE blocks (
+                key  BLOB PRIMARY KEY NOT NULL,
+                val  BLOB NOT NULL
+            ) WITHOUT ROWID",
+            (),
+        )?;
+        Ok(())
+    }
+}
+
+pub(crate) struct SqliteWriter<'conn> {
+    tx: rusqlite::Transaction<'conn>,
+    stored: &'conn mut usize,
+    max: usize,
+}
+
+impl SqliteWriter<'_> {
+    pub(crate) fn put_many(
+        &mut self,
+        kv: impl Iterator<Item = Result<(Vec<u8>, Vec<u8>), DriveError>>,
+    ) -> Result<(), DriveError> {
+        let mut insert_stmt = self
+            .tx
+            .prepare_cached("INSERT INTO blocks (key, val) VALUES (?1, ?2)")
+            .map_err(DiskError::DbError)?;
+        for pair in kv {
+            let (k, v) = pair?;
+            *self.stored += v.len();
+            if *self.stored > self.max {
+                return Err(DiskError::MaxSizeExceeded.into());
+            }
+            insert_stmt.execute((k, v)).map_err(DiskError::DbError)?;
+        }
+        Ok(())
+    }
+    pub fn commit(self) -> Result<(), DiskError> {
+        self.tx.commit()?;
+        Ok(())
+    }
+}
+
+pub(crate) struct SqliteReader<'conn> {
+    select_stmt: rusqlite::Statement<'conn>,
+}
+
+impl SqliteReader<'_> {
+    pub(crate) fn get(&mut self, key: Vec<u8>) -> rusqlite::Result<Option<Vec<u8>>> {
+        self.select_stmt
+            .query_one((&key,), |row| row.get(0))
+            .optional()
+    }
+}

src/drive.rs

···
-//! Consume an MST block stream, producing an ordered stream of records
+//! Consume a CAR from an AsyncRead, producing an ordered stream of records
 
-use futures::{Stream, TryStreamExt};
+use crate::disk::{DiskError, DiskStore};
+use crate::process::Processable;
 use ipld_core::cid::Cid;
+use iroh_car::CarReader;
+use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
-use std::error::Error;
+use std::convert::Infallible;
+use tokio::{io::AsyncRead, sync::mpsc};
 
 use crate::mst::{Commit, Node};
-use crate::walk::{Step, Trip, Walker};
+use crate::walk::{Step, WalkError, 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>),
+    WalkError(#[from] WalkError),
+    #[error("CAR file had no roots")]
+    MissingRoot,
+    #[error("Storage error")]
+    StorageError(#[from] DiskError),
+    #[error("Encode error: {0}")]
+    BincodeEncodeError(#[from] bincode::error::EncodeError),
+    #[error("Tried to send on a closed channel")]
+    ChannelSendError, // SendError takes <T> which we don't need
+    #[error("Failed to join a task: {0}")]
+    JoinError(#[from] tokio::task::JoinError),
 }
 
-type CarBlock<E> = Result<(Cid, Vec<u8>), E>;
+#[derive(Debug, thiserror::Error)]
+pub enum DecodeError {
+    #[error(transparent)]
+    BincodeDecodeError(#[from] bincode::error::DecodeError),
+    #[error("extra bytes remained after decoding")]
+    ExtraGarbage,
+}
+
+/// An in-order chunk of Rkey + (processed) Block pairs
+pub type BlockChunk<T> = Vec<(String, T)>;
 
-#[derive(Debug)]
-pub enum MaybeProcessedBlock<T, E> {
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub(crate) 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> {
+    /// TODO this is probably a little broken
+    fn get_size(&self) -> usize {
+        use std::{cmp::max, mem::size_of};
+
+        // enum is always as big as its biggest member?
+        let base_size = max(size_of::<Vec<u8>>(), size_of::<T>());
+
+        let extra = match self {
+            Self::Raw(bytes) => bytes.len(),
+            Self::Processed(t) => t.get_size(),
+        };
+
+        base_size + extra
+    }
 }
 
-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
+impl<T> MaybeProcessedBlock<T> {
+    fn maybe(process: fn(Vec<u8>) -> T, data: Vec<u8>) -> Self {
+        if Node::could_be(&data) {
+            MaybeProcessedBlock::Raw(data)
+        } else {
+            MaybeProcessedBlock::Processed(process(data))
+        }
+    }
+}
+
+/// Read a CAR file, buffering blocks in memory or to disk
+pub enum Driver<R: AsyncRead + Unpin, T: Processable> {
+    /// All blocks fit within the memory limit
     ///
-    /// This will eagerly consume blocks until the `Commit` object is found.
-    /// *Usually* the it's the first block, but there is no guarantee.
+    /// You probably want to check the commit's signature. You can go ahead and
+    /// walk the MST right away.
+    Memory(Commit, MemDriver<T>),
+    /// Blocks exceed the memory limit
     ///
-    /// ### Parameters
+    /// You'll need to provide a disk storage to continue. The commit will be
+    /// returned and can be validated only once all blocks are loaded.
+    Disk(NeedDisk<R, T>),
+}
+
+/// Builder-style driver setup
+pub struct DriverBuilder {
+    pub mem_limit_mb: usize,
+}
+
+impl Default for DriverBuilder {
+    fn default() -> Self {
+        Self { mem_limit_mb: 16 }
+    }
+}
+
+impl DriverBuilder {
+    /// Begin configuring the driver with defaults
+    pub fn new() -> Self {
+        Default::default()
+    }
+    /// Set the in-memory size limit, in MiB
     ///
-    /// `root`: CID of the commit object that is the root of the MST
+    /// Default: 16 MiB
+    pub fn with_mem_limit_mb(self, new_limit: usize) -> Self {
+        Self {
+            mem_limit_mb: new_limit,
+        }
+    }
+    /// Set the block processor
     ///
-    /// `block_stream`: Input stream of raw CAR blocks
+    /// Default: noop, raw blocks will be emitted
+    pub fn with_block_processor<T: Processable>(
+        self,
+        p: fn(Vec<u8>) -> T,
+    ) -> DriverBuilderWithProcessor<T> {
+        DriverBuilderWithProcessor {
+            mem_limit_mb: self.mem_limit_mb,
+            block_processor: p,
+        }
+    }
+    /// Begin processing an atproto MST from a CAR file
+    pub async fn load_car<R: AsyncRead + Unpin>(
+        self,
+        reader: R,
+    ) -> Result<Driver<R, Vec<u8>>, DriveError> {
+        Driver::load_car(reader, crate::process::noop, self.mem_limit_mb).await
+    }
+}
+
+/// Builder-style driver intermediate step
+///
+/// start from `DriverBuilder`
+pub struct DriverBuilderWithProcessor<T: Processable> {
+    pub mem_limit_mb: usize,
+    pub block_processor: fn(Vec<u8>) -> T,
+}
+
+impl<T: Processable> DriverBuilderWithProcessor<T> {
+    /// Set the in-memory size limit, in MiB
     ///
-    /// `process`: record-transforming callback:
+    /// Default: 16 MiB
+    pub fn with_mem_limit_mb(mut self, new_limit: usize) -> Self {
+        self.mem_limit_mb = new_limit;
+        self
+    }
+    /// Begin processing an atproto MST from a CAR file
+    pub async fn load_car<R: AsyncRead + Unpin>(
+        self,
+        reader: R,
+    ) -> Result<Driver<R, T>, DriveError> {
+        Driver::load_car(reader, self.block_processor, self.mem_limit_mb).await
+    }
+}
+
+impl<R: AsyncRead + Unpin, T: Processable> Driver<R, T> {
+    /// Begin processing an atproto MST from a CAR file
     ///
-    /// 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();
+    /// Blocks will be loaded, processed, and buffered in memory. If the entire
+    /// processed size is under the `mem_limit_mb` limit, a `Driver::Memory`
+    /// will be returned along with a `Commit` ready for validation.
+    ///
+    /// If the `mem_limit_mb` limit is reached before loading all blocks, the
+    /// partial state will be returned as `Driver::Disk(needed)`, which can be
+    /// resumed by providing a `SqliteStorage` for on-disk block storage.
+    pub async fn load_car(
+        reader: R,
+        process: fn(Vec<u8>) -> T,
+        mem_limit_mb: usize,
+    ) -> Result<Driver<R, T>, DriveError> {
+        let max_size = mem_limit_mb * 2_usize.pow(20);
+        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;
 
-        while let Some((cid, data)) = block_stream
-            .try_next()
-            .await
-            .map_err(|e| DriveError::CarBlockError(e.into()))?
-        {
+        // try to load all the blocks into memory
+        let mut mem_size = 0;
+        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)
-                    .map_err(|e| DriveError::BadCommit(e.into()))?;
+                let c: Commit = serde_ipld_dagcbor::from_slice(&data)?;
                 commit = Some(c);
-                break;
-            } else {
-                blocks.insert(
-                    cid,
-                    if Node::could_be(&data) {
-                        MaybeProcessedBlock::Raw(data)
-                    } else {
-                        MaybeProcessedBlock::Processed(process(&data))
-                    },
-                );
+                continue;
+            }
+
+            // remaining possible types: node, record, other. optimistically process
+            let maybe_processed = MaybeProcessedBlock::maybe(process, data);
+
+            // stash (maybe processed) blocks in memory as long as we have room
+            mem_size += std::mem::size_of::<Cid>() + maybe_processed.get_size();
+            mem_blocks.insert(cid, maybe_processed);
+            if mem_size >= max_size {
+                return Ok(Driver::Disk(NeedDisk {
+                    car,
+                    root,
+                    process,
+                    max_size,
+                    mem_blocks,
+                    commit,
+                }));
             }
         }
 
-        // we either broke out or read all the blocks without finding the commit...
+        // 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 me = Self {
-            block_stream,
-            blocks,
-            walker,
-            process,
-        };
-        Ok((commit, me))
+        Ok(Driver::Memory(
+            commit,
+            MemDriver {
+                blocks: mem_blocks,
+                walker,
+                process,
+            },
+        ))
+    }
+}
+
+/// 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(Vec<u8>) -> T,
+}
+
+impl<T: Processable> MemDriver<T> {
+    /// Step through the record outputs, in rkey order
+    pub async fn next_chunk(&mut self, n: usize) -> Result<Option<BlockChunk<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
+            match self.walker.step(&mut self.blocks, self.process)? {
+                Step::Missing(cid) => return Err(DriveError::MissingBlock(cid)),
+                Step::Finish => break,
+                Step::Found { rkey, data } => {
+                    out.push((rkey, data));
+                    continue;
+                }
+            };
+        }
+
+        if out.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(out))
+        }
+    }
+}
+
+/// A partially memory-loaded car file that needs disk spillover to continue
+pub struct NeedDisk<R: AsyncRead + Unpin, T: Processable> {
+    car: CarReader<R>,
+    root: Cid,
+    process: fn(Vec<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(crate) fn decode<T: Processable>(bytes: &[u8]) -> Result<T, DecodeError> {
+    let (t, n) = bincode::serde::decode_from_slice(bytes, bincode::config::standard())?;
+    if n != bytes.len() {
+        return Err(DecodeError::ExtraGarbage);
     }
+    Ok(t)
+}
 
-    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) {
-                    MaybeProcessedBlock::Raw(data)
-                } else {
-                    MaybeProcessedBlock::Processed((self.process)(&data))
-                },
-            );
-            if cid == cid_needed {
-                return Ok(());
+impl<R: AsyncRead + Unpin, T: Processable + Send + 'static> NeedDisk<R, T> {
+    pub async fn finish_loading(
+        mut self,
+        mut store: DiskStore,
+    ) -> Result<(Commit, DiskDriver<T>), DriveError> {
+        // move store in and back out so we can manage lifetimes
+        // dump mem blocks into the store
+        store = tokio::task::spawn(async move {
+            let mut writer = store.get_writer()?;
+
+            let kvs = self
+                .mem_blocks
+                .into_iter()
+                .map(|(k, v)| Ok(encode(v).map(|v| (k.to_bytes(), v))?));
+
+            writer.put_many(kvs)?;
+            writer.commit()?;
+            Ok::<_, DriveError>(store)
+        })
+        .await??;
+
+        let (tx, mut rx) = mpsc::channel::<Vec<(Cid, MaybeProcessedBlock<T>)>>(2);
+
+        let store_worker = tokio::task::spawn_blocking(move || {
+            let mut writer = store.get_writer()?;
+
+            while let Some(chunk) = rx.blocking_recv() {
+                let kvs = chunk
+                    .into_iter()
+                    .map(|(k, v)| Ok(encode(v).map(|v| (k.to_bytes(), v))?));
+                writer.put_many(kvs)?;
             }
+
+            writer.commit()?;
+            Ok::<_, DriveError>(store)
+        }); // await later
+
+        // dump the rest to disk (in chunks)
+        log::debug!("dumping the rest of the stream...");
+        loop {
+            let mut mem_size = 0;
+            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 = MaybeProcessedBlock::maybe(self.process, data);
+                mem_size += std::mem::size_of::<Cid>() + maybe_processed.get_size();
+                chunk.push((cid, maybe_processed));
+                if mem_size >= self.max_size {
+                    // soooooo if we're setting the db cache to max_size and then letting
+                    // multiple chunks in the queue that are >= max_size, then at any time
+                    // we might be using some multiple of max_size?
+                    break;
+                }
+            }
+            if chunk.is_empty() {
+                break;
+            }
+            tx.send(chunk)
+                .await
+                .map_err(|_| DriveError::ChannelSendError)?;
         }
+        drop(tx);
+        log::debug!("done. waiting for worker to finish...");
 
-        // if we never found the block
-        Err(DriveError::MissingBlock(cid_needed))
+        store = store_worker.await??;
+
+        log::debug!("worker finished.");
+
+        let commit = self.commit.ok_or(DriveError::MissingCommit)?;
+
+        let walker = Walker::new(commit.data);
+
+        Ok((
+            commit,
+            DiskDriver {
+                process: self.process,
+                state: Some(BigState { store, walker }),
+            },
+        ))
+    }
+}
+
+struct BigState {
+    store: DiskStore,
+    walker: Walker,
+}
+
+/// MST walker that reads from disk instead of an in-memory hashmap
+pub struct DiskDriver<T: Clone> {
+    process: fn(Vec<u8>) -> T,
+    state: Option<BigState>,
+}
+
+// for doctests only
+#[doc(hidden)]
+pub fn _get_fake_disk_driver() -> DiskDriver<Vec<u8>> {
+    use crate::process::noop;
+    DiskDriver {
+        process: noop,
+        state: None,
+    }
+}
+
+impl<T: Processable + Send + 'static> DiskDriver<T> {
+    /// Walk the MST returning up to `n` rkey + record pairs
+    ///
+    /// ```no_run
+    /// # use repo_stream::{drive::{DiskDriver, DriveError, _get_fake_disk_driver}, process::noop};
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<(), DriveError> {
+    /// # let mut disk_driver = _get_fake_disk_driver();
+    /// while let Some(pairs) = disk_driver.next_chunk(256).await? {
+    ///     for (rkey, record) in pairs {
+    ///         println!("{rkey}: size={}", record.len());
+    ///     }
+    /// }
+    /// let store = disk_driver.reset_store().await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub async fn next_chunk(&mut self, n: usize) -> Result<Option<BlockChunk<T>>, DriveError> {
+        let process = self.process;
+
+        // state should only *ever* be None transiently while inside here
+        let mut state = self.state.take().expect("DiskDriver must have Some(state)");
+
+        // the big pain here is that we don't want to leave self.state in an
+        // invalid state (None), so all the error paths have to make sure it
+        // comes out again.
+        let (state, res) = tokio::task::spawn_blocking(
+            move || -> (BigState, Result<BlockChunk<T>, DriveError>) {
+                let mut reader_res = state.store.get_reader();
+                let reader: &mut _ = match reader_res {
+                    Ok(ref mut r) => r,
+                    Err(ref mut e) => {
+                        // unfortunately we can't return the error directly because
+                        // (for some reason) it's attached to the lifetime of the
+                        // reader?
+                        // hack a mem::swap so we can get it out :/
+                        let e_swapped = e.steal();
+                        // the pain: `state` *has to* outlive the reader
+                        drop(reader_res);
+                        return (state, Err(e_swapped.into()));
+                    }
+                };
+
+                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 step = match state.walker.disk_step(reader, process) {
+                        Ok(s) => s,
+                        Err(e) => {
+                            // the pain: `state` *has to* outlive the reader
+                            drop(reader_res);
+                            return (state, Err(e.into()));
+                        }
+                    };
+                    match step {
+                        Step::Missing(cid) => {
+                            // the pain: `state` *has to* outlive the reader
+                            drop(reader_res);
+                            return (state, Err(DriveError::MissingBlock(cid)));
+                        }
+                        Step::Finish => break,
+                        Step::Found { rkey, data } => out.push((rkey, data)),
+                    };
+                }
+
+                // `state` *has to* outlive the reader
+                drop(reader_res);
+
+                (state, Ok::<_, DriveError>(out))
+            },
+        )
+        .await?; // on tokio JoinError, we'll be left with invalid state :(
+
+        // *must* restore state before dealing with the actual result
+        self.state = Some(state);
+
+        let out = res?;
+
+        if out.is_empty() {
+            Ok(None)
+        } else {
+            Ok(Some(out))
+        }
     }
 
-    /// Manually step through the record outputs
-    pub async fn next_record(&mut self) -> Result<Option<(String, T)>, DriveError<PE>> {
+    fn read_tx_blocking(
+        &mut self,
+        n: usize,
+        tx: mpsc::Sender<Result<BlockChunk<T>, DriveError>>,
+    ) -> Result<(), mpsc::error::SendError<Result<BlockChunk<T>, DriveError>>> {
+        let BigState { store, walker } = self.state.as_mut().expect("valid state");
+        let mut reader = match store.get_reader() {
+            Ok(r) => r,
+            Err(e) => return tx.blocking_send(Err(e.into())),
+        };
+
         loop {
-            // 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))),
-            };
+            let mut out: BlockChunk<T> = 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 step = match walker.disk_step(&mut reader, self.process) {
+                    Ok(s) => s,
+                    Err(e) => return tx.blocking_send(Err(e.into())),
+                };
+
+                match step {
+                    Step::Missing(cid) => {
+                        return tx.blocking_send(Err(DriveError::MissingBlock(cid)));
+                    }
+                    Step::Finish => return Ok(()),
+                    Step::Found { rkey, data } => {
+                        out.push((rkey, data));
+                        continue;
+                    }
+                };
+            }
 
-            // load blocks until we reach that cid
-            self.drive_until(cid_needed).await?;
+            if out.is_empty() {
+                break;
+            }
+            tx.blocking_send(Ok(out))?;
         }
+
+        Ok(())
     }
 
-    /// 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)))
-        })
+    /// Spawn the disk reading task into a tokio blocking thread
+    ///
+    /// The idea is to avoid so much sending back and forth to the blocking
+    /// thread, letting a blocking task do all the disk reading work and sending
+    /// records and rkeys back through an `mpsc` channel instead.
+    ///
+    /// This might also allow the disk work to continue while processing the
+    /// records. It's still not yet clear if this method actually has much
+    /// benefit over just using `.next_chunk(n)`.
+    ///
+    /// ```no_run
+    /// # use repo_stream::{drive::{DiskDriver, DriveError, _get_fake_disk_driver}, process::noop};
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<(), DriveError> {
+    /// # let mut disk_driver = _get_fake_disk_driver();
+    /// let (mut rx, join) = disk_driver.to_channel(512);
+    /// while let Some(recvd) = rx.recv().await {
+    ///     let pairs = recvd?;
+    ///     for (rkey, record) in pairs {
+    ///         println!("{rkey}: size={}", record.len());
+    ///     }
+    ///
+    /// }
+    /// let store = join.await?.reset_store().await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn to_channel(
+        mut self,
+        n: usize,
+    ) -> (
+        mpsc::Receiver<Result<BlockChunk<T>, DriveError>>,
+        tokio::task::JoinHandle<Self>,
+    ) {
+        let (tx, rx) = mpsc::channel::<Result<BlockChunk<T>, DriveError>>(1);
+
+        // sketch: this worker is going to be allowed to execute without a join handle
+        let chan_task = tokio::task::spawn_blocking(move || {
+            if let Err(mpsc::error::SendError(_)) = self.read_tx_blocking(n, tx) {
+                log::debug!("big car reader exited early due to dropped receiver channel");
+            }
+            self
+        });
+
+        (rx, chan_task)
+    }
+
+    /// Reset the disk storage so it can be reused. You must call this.
+    ///
+    /// Ideally we'd put this in an `impl Drop`, but since it makes blocking
+    /// calls, that would be risky in an async context. For now you just have to
+    /// carefully make sure you call it.
+    ///
+    /// The sqlite store is returned, so it can be reused for another
+    /// `DiskDriver`.
+    pub async fn reset_store(mut self) -> Result<DiskStore, DriveError> {
+        let BigState { store, .. } = self.state.take().expect("valid state");
+        Ok(store.reset().await?)
     }
 }

src/lib.rs

···
-//! Fast and robust atproto CAR file processing in rust
-//!
-//! For now see the [examples](https://tangled.org/@microcosm.blue/repo-stream/tree/main/examples)
+/*!
+A robust CAR file -> MST walker for atproto
+
+Small CARs have their blocks buffered in memory. If a configurable memory limit
+is reached while reading blocks, CAR reading is suspended, and can be continued
+by providing disk storage to buffer the CAR blocks instead.
+
+A `process` function can be provided for tasks where records are transformed
+into a smaller representation, to save memory (and disk) during block reading.
+
+Once blocks are loaded, the MST is walked and emitted as chunks of pairs of
+`(rkey, processed_block)` pairs, in order (depth first, left-to-right).
+
+Some MST validations are applied
+- Keys must appear in order
+- Keys must be at the correct MST tree depth
+
+`iroh_car` additionally applies a block size limit of `2MiB`.
+
+```
+use repo_stream::{Driver, DriverBuilder, DiskBuilder};
+
+# #[tokio::main]
+# async fn main() -> Result<(), Box<dyn std::error::Error>> {
+# let reader = include_bytes!("../car-samples/tiny.car").as_slice();
+let mut total_size = 0;
 
-pub mod drive;
+match DriverBuilder::new()
+    .with_mem_limit_mb(10)
+    .with_block_processor(|rec| rec.len()) // block processing: just extract the raw record size
+    .load_car(reader)
+    .await?
+{
+
+    // if all blocks fit within memory
+    Driver::Memory(_commit, mut driver) => {
+        while let Some(chunk) = driver.next_chunk(256).await? {
+            for (_rkey, size) in chunk {
+                total_size += size;
+            }
+        }
+    },
+
+    // if the CAR was too big for in-memory processing
+    Driver::Disk(paused) => {
+        // set up a disk store we can spill to
+        let store = DiskBuilder::new().open("some/path.db".into()).await?;
+        // do the spilling, get back a (similar) driver
+        let (_commit, mut driver) = paused.finish_loading(store).await?;
+
+        while let Some(chunk) = driver.next_chunk(256).await? {
+            for (_rkey, size) in chunk {
+                total_size += size;
+            }
+        }
+
+        // clean up the disk store (drop tables etc)
+        driver.reset_store().await?;
+    }
+};
+println!("sum of size of all records: {total_size}");
+# Ok(())
+# }
+```
+
+Disk spilling suspends and returns a `Driver::Disk(paused)` instead of going
+ahead and eagerly using disk I/O. This means you have to write a bit more code
+to handle both cases, but it allows you to have finer control over resource
+usage. For example, you can drive a number of parallel memory CAR workers, and
+separately have a different number of disk workers picking up suspended disk
+tasks from a queue.
+
+Find more [examples in the repo](https://tangled.org/@microcosm.blue/repo-stream/tree/main/examples).
+
+*/
+
 pub mod mst;
-pub mod walk;
+mod walk;
+
+pub mod disk;
+pub mod drive;
+pub mod process;
+
+pub use disk::{DiskBuilder, DiskError, DiskStore};
+pub use drive::{DriveError, Driver, DriverBuilder};
+pub use mst::Commit;
+pub use process::Processable;

src/mst.rs

···
 /// MST node data schema
 #[derive(Debug, Deserialize, PartialEq)]
 #[serde(deny_unknown_fields)]
-pub struct Node {
+pub(crate) struct Node {
     /// link to sub-tree Node on a lower level and with all keys sorting before
     /// keys at this node
     #[serde(rename = "l")]
···
     /// so if a block *could be* a node, any record converter must postpone
     /// processing. if it turns out it happens to be a very node-looking record,
     /// well, sorry, it just has to only be processed later when that's known.
-    pub fn could_be(bytes: impl AsRef<[u8]>) -> bool {
+    pub(crate) fn could_be(bytes: impl AsRef<[u8]>) -> bool {
         const NODE_FINGERPRINT: [u8; 3] = [
             0xA2, // map length 2 (for "l" and "e" keys)
             0x61, // text length 1
···
     /// with an empty array of entries. This is the only situation in which a
     /// tree may contain an empty leaf node which does not either contain keys
     /// ("entries") or point to a sub-tree containing entries.
-    ///
-    /// TODO: to me this is slightly unclear with respect to `l` (ask someone).
-    /// ...is that what "The top of the tree must not be a an empty node which
-    /// only points to a sub-tree." is referring to?
-    pub fn is_empty(&self) -> bool {
+    pub(crate) fn is_empty(&self) -> bool {
         self.left.is_none() && self.entries.is_empty()
     }
 }
···
 /// TreeEntry object
 #[derive(Debug, Deserialize, PartialEq)]
 #[serde(deny_unknown_fields)]
-pub struct Entry {
+pub(crate) struct Entry {
     /// count of bytes shared with previous TreeEntry in this Node (if any)
     #[serde(rename = "p")]
     pub prefix_len: usize,

src/process.rs

···
+/*!
+Record processor function output trait
+
+The return type must satisfy the `Processable` trait, which requires:
+
+- `Clone` because two rkeys can refer to the same record by CID, which may
+  only appear once in the CAR file.
+- `Serialize + DeserializeOwned` so it can be spilled to disk.
+
+One required function must be implemented, `get_size()`: this should return the
+approximate total off-stack size of the type. (the on-stack size will be added
+automatically via `std::mem::get_size`).
+
+Note that it is **not guaranteed** that the `process` function will run on a
+block before storing it in memory or on disk: it's not possible to know if a
+block is a record without actually walking the MST, so the best we can do is
+apply `process` to any block that we know *cannot* be an MST node, and otherwise
+store the raw block bytes.
+
+Here's a silly processing function that just collects 'eyy's found in the raw
+record bytes
+
+```
+# use repo_stream::Processable;
+# use serde::{Serialize, Deserialize};
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct Eyy(usize, String);
+
+impl Processable for Eyy {
+    fn get_size(&self) -> usize {
+        // don't need to compute the usize, it's on the stack
+        self.1.capacity() // in-mem size from the string's capacity, in bytes
+    }
+}
+
+fn process(raw: Vec<u8>) -> Vec<Eyy> {
+    let mut out = Vec::new();
+    let to_find = "eyy".as_bytes();
+    for i in 0..(raw.len() - 3) {
+        if &raw[i..(i+3)] == to_find {
+            out.push(Eyy(i, "eyy".to_string()));
+        }
+    }
+    out
+}
+```
+
+The memory sizing stuff is a little sketch but probably at least approximately
+works.
+*/
+
+use serde::{Serialize, de::DeserializeOwned};
+
+/// Output trait for record processing
+pub trait Processable: Clone + Serialize + DeserializeOwned {
+    /// Any additional in-memory size taken by the processed type
+    ///
+    /// Do not include stack size (`std::mem::size_of`)
+    fn get_size(&self) -> usize;
+}
+
+/// Processor that just returns the raw blocks
+#[inline]
+pub fn noop(block: Vec<u8>) -> Vec<u8> {
+    block
+}
+
+impl Processable for u8 {
+    fn get_size(&self) -> usize {
+        0
+    }
+}
+
+impl Processable for usize {
+    fn get_size(&self) -> usize {
+        0 // no additional space taken, just its stack size (newtype is free)
+    }
+}
+
+impl<Item: Sized + Processable> Processable for Vec<Item> {
+    fn get_size(&self) -> usize {
+        let slot_size = std::mem::size_of::<Item>();
+        let direct_size = slot_size * self.capacity();
+        let items_referenced_size: usize = self.iter().map(|item| item.get_size()).sum();
+        direct_size + items_referenced_size
+    }
+}

src/walk.rs

···
 //! Depth-first MST traversal
 
-use crate::drive::MaybeProcessedBlock;
+use crate::disk::SqliteReader;
+use crate::drive::{DecodeError, MaybeProcessedBlock};
 use crate::mst::Node;
+use crate::process::Processable;
 use ipld_core::cid::Cid;
+use sha2::{Digest, Sha256};
 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> {
-    #[error("empty mst nodes are not allowed")]
-    NodeEmpty,
+pub enum WalkError {
+    #[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,
+    MstError(#[from] MstError),
+    #[error("storage error: {0}")]
+    StorageError(#[from] rusqlite::Error),
+    #[error("Decode error: {0}")]
+    DecodeError(#[from] DecodeError),
 }
 
 /// Errors from invalid Rkeys
-#[derive(Debug, thiserror::Error)]
-pub enum RkeyError {
+#[derive(Debug, PartialEq, thiserror::Error)]
+pub enum MstError {
     #[error("Failed to compute an rkey due to invalid prefix_len")]
     EntryPrefixOutOfbounds,
     #[error("RKey was not utf-8")]
     EntryRkeyNotUtf8(#[from] std::string::FromUtf8Error),
+    #[error("Nodes cannot be empty (except for an entirely empty MST)")]
+    EmptyNode,
+    #[error("Found an entry with rkey at the wrong depth")]
+    WrongDepth,
+    #[error("Lost track of our depth (possible bug?)")]
+    LostDepth,
+    #[error("MST depth underflow: depth-0 node with child trees")]
+    DepthUnderflow,
+    #[error("Encountered an rkey out of order while walking the MST")]
+    RkeyOutOfOrder,
 }
 
 /// 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!
-    Step { rkey: String, data: T },
+    Found { rkey: String, data: T },
 }
 
 #[derive(Debug, Clone, PartialEq)]
 enum Need {
-    Node(Cid),
+    Node { depth: Depth, cid: Cid },
     Record { rkey: String, cid: Cid },
 }
 
-fn push_from_node(stack: &mut Vec<Need>, node: &Node) -> Result<(), RkeyError> {
-    let mut entries = Vec::with_capacity(node.entries.len());
+#[derive(Debug, Clone, Copy, PartialEq)]
+enum Depth {
+    Root,
+    Depth(u32),
+}
+
+impl Depth {
+    fn from_key(key: &[u8]) -> Self {
+        let mut zeros = 0;
+        for byte in Sha256::digest(key) {
+            let leading = byte.leading_zeros();
+            zeros += leading;
+            if leading < 8 {
+                break;
+            }
+        }
+        Self::Depth(zeros / 2) // truncating divide (rounds down)
+    }
+    fn next_expected(&self) -> Result<Option<u32>, MstError> {
+        match self {
+            Self::Root => Ok(None),
+            Self::Depth(d) => d.checked_sub(1).ok_or(MstError::DepthUnderflow).map(Some),
+        }
+    }
+}
 
+fn push_from_node(stack: &mut Vec<Need>, node: &Node, parent_depth: Depth) -> Result<(), MstError> {
+    // empty nodes are not allowed in the MST
+    // ...except for a single one for empty MST, but we wouldn't be pushing that
+    if node.is_empty() {
+        return Err(MstError::EmptyNode);
+    }
+
+    let mut entries = Vec::with_capacity(node.entries.len());
     let mut prefix = vec![];
+    let mut this_depth = parent_depth.next_expected()?;
+
     for entry in &node.entries {
         let mut rkey = vec![];
         let pre_checked = prefix
             .get(..entry.prefix_len)
-            .ok_or(RkeyError::EntryPrefixOutOfbounds)?;
+            .ok_or(MstError::EntryPrefixOutOfbounds)?;
         rkey.extend_from_slice(pre_checked);
         rkey.extend_from_slice(&entry.keysuffix);
+
+        let Depth::Depth(key_depth) = Depth::from_key(&rkey) else {
+            return Err(MstError::WrongDepth);
+        };
+
+        // this_depth is `none` if we are the deepest child (directly below root)
+        // in that case we accept whatever highest depth is claimed
+        let expected_depth = match this_depth {
+            Some(d) => d,
+            None => {
+                this_depth = Some(key_depth);
+                key_depth
+            }
+        };
+
+        // all keys we find should be this depth
+        if key_depth != expected_depth {
+            return Err(MstError::DepthUnderflow);
+        }
+
         prefix = rkey.clone();
 
         entries.push(Need::Record {
···
             cid: entry.value,
         });
         if let Some(ref tree) = entry.tree {
-            entries.push(Need::Node(*tree));
+            entries.push(Need::Node {
+                depth: Depth::Depth(key_depth),
+                cid: *tree,
+            });
         }
     }
 
     entries.reverse();
     stack.append(&mut entries);
 
+    let d = this_depth.ok_or(MstError::LostDepth)?;
+
     if let Some(tree) = node.left {
-        stack.push(Need::Node(tree));
+        stack.push(Need::Node {
+            depth: Depth::Depth(d),
+            cid: tree,
+        });
     }
     Ok(())
 }
···
 impl Walker {
     pub fn new(tree_root_cid: Cid) -> Self {
         Self {
-            stack: vec![Need::Node(tree_root_cid)],
+            stack: vec![Need::Node {
+                depth: Depth::Root,
+                cid: tree_root_cid,
+            }],
             prev: "".to_string(),
         }
     }
 
     /// 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(Vec<u8>) -> T,
+    ) -> Result<Step<T>, WalkError> {
         loop {
-            let Some(mut need) = self.stack.last() else {
+            let Some(need) = self.stack.last_mut() else {
                 log::trace!("tried to walk but we're actually done.");
                 return Ok(Step::Finish);
             };
 
-            match &mut need {
-                Need::Node(cid) => {
+            match need {
+                &mut Need::Node { depth, cid } => {
                     log::trace!("need node {cid:?}");
-                    let Some(block) = blocks.remove(cid) else {
+                    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(WalkError::BadCommitFingerprint);
                     };
                     let node = serde_ipld_dagcbor::from_slice::<Node>(&data)
-                        .map_err(|e| Trip::BadCommit(e.into()))?;
+                        .map_err(WalkError::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)?;
+                    push_from_node(&mut self.stack, &node, depth)?;
                 }
                 Need::Record { rkey, cid } => {
                     log::trace!("need record {cid:?}");
+                    // note that we cannot *remove* a record block, sadly, since
+                    // there can be multiple rkeys pointing to the same cid.
                     let Some(data) = blocks.get_mut(cid) else {
+                        return Ok(Step::Missing(*cid));
+                    };
+                    let rkey = rkey.clone();
+                    let data = match data {
+                        MaybeProcessedBlock::Raw(data) => process(data.to_vec()),
+                        MaybeProcessedBlock::Processed(t) => t.clone(),
+                    };
+
+                    // found node, make sure we remember
+                    self.stack.pop();
+
+                    // rkeys *must* be in order or else the tree is invalid (or
+                    // we have a bug)
+                    if rkey <= self.prev {
+                        return Err(MstError::RkeyOutOfOrder)?;
+                    }
+                    self.prev = rkey.clone();
+
+                    return Ok(Step::Found { rkey, data });
+                }
+            }
+        }
+    }
+
+    /// blocking!!!!!!
+    pub fn disk_step<T: Processable>(
+        &mut self,
+        reader: &mut SqliteReader,
+        process: impl Fn(Vec<u8>) -> T,
+    ) -> Result<Step<T>, WalkError> {
+        loop {
+            let Some(need) = self.stack.last_mut() else {
+                log::trace!("tried to walk but we're actually done.");
+                return Ok(Step::Finish);
+            };
+
+            match need {
+                &mut Need::Node { depth, 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(WalkError::BadCommitFingerprint);
+                    };
+                    let node = serde_ipld_dagcbor::from_slice::<Node>(&data)
+                        .map_err(WalkError::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, depth).map_err(WalkError::MstError)?;
+                }
+                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::Rest(*cid));
+                        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(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);
+                        return Err(MstError::RkeyOutOfOrder)?;
                     }
                     self.prev = rkey.clone();
 
-                    return Ok(Step::Step { rkey, data });
+                    return Ok(Step::Found { rkey, data });
                 }
             }
         }
···
 #[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_depth_spec_0() {
+        let d = Depth::from_key(b"2653ae71");
+        assert_eq!(d, Depth::Depth(0))
+    }
+
+    #[test]
+    fn test_depth_spec_1() {
+        let d = Depth::from_key(b"blue");
+        assert_eq!(d, Depth::Depth(1))
+    }
+
+    #[test]
+    fn test_depth_spec_4() {
+        let d = Depth::from_key(b"app.bsky.feed.post/454397e440ec");
+        assert_eq!(d, Depth::Depth(4))
+    }
+
+    #[test]
+    fn test_depth_spec_8() {
+        let d = Depth::from_key(b"app.bsky.feed.post/9adeb165882c");
+        assert_eq!(d, Depth::Depth(8))
+    }
+
+    #[test]
+    fn test_depth_ietf_draft_0() {
+        let d = Depth::from_key(b"key1");
+        assert_eq!(d, Depth::Depth(0))
+    }
+
+    #[test]
+    fn test_depth_ietf_draft_1() {
+        let d = Depth::from_key(b"key7");
+        assert_eq!(d, Depth::Depth(1))
+    }
+
+    #[test]
+    fn test_depth_ietf_draft_4() {
+        let d = Depth::from_key(b"key515");
+        assert_eq!(d, Depth::Depth(4))
+    }
+
+    #[test]
+    fn test_depth_interop() {
+        // examples from https://github.com/bluesky-social/atproto-interop-tests/blob/main/mst/key_heights.json
+        for (k, expected) in [
+            ("", 0),
+            ("asdf", 0),
+            ("blue", 1),
+            ("2653ae71", 0),
+            ("88bfafc7", 2),
+            ("2a92d355", 4),
+            ("884976f5", 6),
+            ("app.bsky.feed.post/454397e440ec", 4),
+            ("app.bsky.feed.post/9adeb165882c", 8),
+        ] {
+            let d = Depth::from_key(k.as_bytes());
+            assert_eq!(d, Depth::Depth(expected), "key: {}", k);
+        }
+    }
 
     #[test]
-    fn test_next_from_node_empty() {
-        let node = Node {
+    fn test_push_empty_fails() {
+        let empty_node = Node {
             left: None,
             entries: vec![],
         };
         let mut stack = vec![];
-        push_from_node(&mut stack, &node).unwrap();
-        assert_eq!(stack.last(), None);
+        let err = push_from_node(&mut stack, &empty_node, Depth::Depth(4));
+        assert_eq!(err, Err(MstError::EmptyNode));
     }
 
     #[test]
-    fn test_needs_from_node_just_left() {
+    fn test_push_one_node() {
         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());
+        push_from_node(&mut stack, &node, Depth::Depth(4)).unwrap();
+        assert_eq!(
+            stack.last(),
+            Some(Need::Node {
+                depth: Depth::Depth(3),
+                cid: 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()),
-    //             ]
-    //         );
-    //     }
 }

tests/non-huge-cars.rs

···
 extern crate repo_stream;
-use futures::TryStreamExt;
-use iroh_car::CarReader;
-use std::convert::Infallible;
+use repo_stream::Driver;
 
 const TINY_CAR: &'static [u8] = include_bytes!("../car-samples/tiny.car");
 const LITTLE_CAR: &'static [u8] = include_bytes!("../car-samples/little.car");
 const MIDSIZE_CAR: &'static [u8] = include_bytes!("../car-samples/midsize.car");
 
 async fn test_car(bytes: &[u8], expected_records: usize, expected_sum: usize) {
-    let reader = CarReader::new(bytes).await.unwrap();
-
-    let root = reader
-        .header()
-        .roots()
-        .first()
-        .ok_or("missing root")
+    let mut driver = match Driver::load_car(bytes, |block| block.len(), 10 /* MiB */)
+        .await
         .unwrap()
-        .clone();
-
-    let stream = std::pin::pin!(reader.stream());
-
-    let (_commit, v) =
-        repo_stream::drive::Vehicle::init(root, stream, |block| Ok::<_, Infallible>(block.len()))
-            .await
-            .unwrap();
-    let mut record_stream = std::pin::pin!(v.stream());
+    {
+        Driver::Memory(_commit, mem_driver) => mem_driver,
+        Driver::Disk(_) => panic!("too big"),
+    };
 
     let mut records = 0;
     let mut sum = 0;
     let mut found_bsky_profile = false;
     let mut prev_rkey = "".to_string();
-    while let Some((rkey, size)) = record_stream.try_next().await.unwrap() {
-        records += 1;
-        sum += size;
-        if rkey == "app.bsky.actor.profile/self" {
-            found_bsky_profile = true;
+
+    while let Some(pairs) = driver.next_chunk(256).await.unwrap() {
+        for (rkey, size) in pairs {
+            records += 1;
+            sum += size;
+            if rkey == "app.bsky.actor.profile/self" {
+                found_bsky_profile = true;
+            }
+            assert!(rkey > prev_rkey, "rkeys are streamed in order");
+            prev_rkey = rkey;
         }
-        assert!(rkey > prev_rkey, "rkeys are streamed in order");
-        prev_rkey = rkey;
     }
+
     assert_eq!(records, expected_records);
     assert_eq!(sum, expected_sum);
     assert!(found_bsky_profile);