yep repros

examples/disk-read-file/main.rs

···
 
 extern crate repo_stream;
 use clap::Parser;
-use repo_stream::{DiskBuilder, Driver};
+use repo_stream::{DiskStore, load_car};
 use std::path::PathBuf;
 use std::time::Instant;
 
···
     log::info!("hello! reading the car...");
     let t0 = Instant::now();
 
+    // set up a disk store we can spill to
+    let disk_store = DiskStore::new(tmpfile).await?;
+
     // 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 paused = Driver::load_car(reader).await?;
-
-    // set up a disk store we can spill to
-    let disk_store = DiskBuilder::new().open(tmpfile).await?;
-
-    // do the spilling, get back a driver
-    let mut store = paused.finish_loading(disk_store).await?;
+    let mut store = load_car(reader, 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.

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 fjall::{Database, Keyspace, KeyspaceCreateOptions, Error as FjallError};
 use std::path::PathBuf;
···
     /// sqlite bits)
     #[error(transparent)]
     DbError(#[from] FjallError),
-    /// 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,
-}
-
-/// Builder-style disk store setup
-#[derive(Debug, Clone)]
-pub struct DiskBuilder {}
-
-impl Default for DiskBuilder {
-    fn default() -> Self {
-        Self {}
-    }
-}
-
-impl DiskBuilder {
-    /// Begin configuring the storage with defaults
-    pub fn new() -> Self {
-        Default::default()
-    }
-    /// Open and initialize the actual disk storage
-    pub async fn open(&self, path: PathBuf) -> Result<DiskStore, DiskError> {
-        DiskStore::new(path).await
-    }
 }
 
 /// On-disk block storage
···
     pub async fn new(
         path: PathBuf,
     ) -> Result<Self, DiskError> {
-        let (db, ks) = tokio::task::spawn_blocking(move || {
-            let db = Database::builder(path).open()?;
-            let ks = db.keyspace("z", KeyspaceCreateOptions::default)?;
-
-            Ok::<_, DiskError>((db, ks))
-        })
-        .await??;
+        let db = Database::builder(path).open()?;
+        let ks = db.keyspace("z", KeyspaceCreateOptions::default)?;
 
         Ok(Self {
             db,

src/drive.rs

···
 //! Consume a CAR from an AsyncRead, producing an ordered stream of records
 
 use crate::disk::{DiskError, DiskStore};
-use ipld_core::cid::Cid;
 use iroh_car::CarReader;
 use std::collections::HashMap;
-use std::convert::Infallible;
 use tokio::io::AsyncRead;
 
 
···
 pub enum DriveError {
     #[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("The Commit block reference by the root was not found")]
-    MissingCommit,
-    #[error("The MST block {0} could not be found")]
-    MissingBlock(Cid),
-    #[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),
 }
 
-#[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)>;
-
-/// Read a CAR file, buffering blocks in memory or to disk
-pub enum Driver<R: AsyncRead + Unpin> {
-    /// Blocks exceed the memory limit
-    ///
-    /// 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>),
-}
-
-impl<R: AsyncRead + Unpin> Driver<R> {
-    /// Begin processing an atproto MST from a CAR file
-    ///
-    /// 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,
-    ) -> Result<NeedDisk<R>, DriveError> {
-        let mut mem_blocks = HashMap::new();
+pub async fn load_car<R: AsyncRead + Unpin>(
+    reader: R,
+    mut store: DiskStore,
+) -> Result<DiskStore, DriveError> {
+    let mut mem_blocks = HashMap::new();
 
-        let mut car = CarReader::new(reader).await?;
+    let mut car = CarReader::new(reader).await?;
 
-        // try to load all the blocks into memory
-        let mut n = 0;
+    // try to load all the blocks into memory
+    let mut n = 0;
 
-        while let Some((cid, data)) = car.next_block().await? {
-            mem_blocks.insert(cid, data);
-            n += 1;
-            if n > 20_000 {
-                log::info!("stopping for disk dump at {n}");
-                return Ok(NeedDisk {
-                    car,
-                    mem_blocks,
-                });
-            }
+    while let Some((cid, data)) = car.next_block().await? {
+        mem_blocks.insert(cid, data);
+        n += 1;
+        if n >= 20_000 {
+            log::info!("stopping for disk dump at {n}");
+            break;
         }
-        panic!("car too small");
     }
-}
 
-
-/// A partially memory-loaded car file that needs disk spillover to continue
-pub struct NeedDisk<R: AsyncRead + Unpin> {
-    car: CarReader<R>,
-    mem_blocks: HashMap<Cid, Vec<u8>>,
-}
-
-impl<R: AsyncRead + Unpin> NeedDisk<R> {
-    pub async fn finish_loading(
-        mut self,
-        mut store: DiskStore,
-    ) -> Result<DiskStore, DriveError> {
-        // dump mem blocks into the store
-        for (k, v) in self.mem_blocks {
-            store.put(k.to_bytes(), v)?;
-        }
+    // dump mem blocks into the store
+    for (k, v) in mem_blocks {
+        store.put(k.to_bytes(), v)?;
+    }
 
-        // dump the rest to disk (in chunks)
-        log::debug!("dumping the rest of the stream...");
-        while let Some((cid, data)) = self.car.next_block().await? {
-            store.put(cid.to_bytes(), data)?;
-        }
-        log::debug!("done.");
-        Ok(store)
+    // dump the rest to disk (in chunks)
+    log::debug!("dumping the rest of the stream...");
+    while let Some((cid, data)) = car.next_block().await? {
+        store.put(cid.to_bytes(), data)?;
     }
+    log::debug!("done.");
+    Ok(store)
+
 }

src/lib.rs

···
 pub mod disk;
 pub mod drive;
 
-pub use disk::{DiskBuilder, DiskError, DiskStore};
-pub use drive::{DriveError, Driver, NeedDisk};
+pub use disk::{DiskError, DiskStore};
+pub use drive::{DriveError, load_car};