down to one file (still repros)

examples/disk-read-file/main.rs

···
 
 extern crate repo_stream;
 use clap::Parser;
-use repo_stream::{DiskStore, load_car};
 use std::path::PathBuf;
 use std::time::Instant;
+use fjall::{Database, Keyspace, KeyspaceCreateOptions, Error as FjallError};
+use std::collections::HashMap;
+use iroh_car::CarReader;
+
+#[derive(Debug, thiserror::Error)]
+pub enum DriveError {
+    #[error("Error from iroh_car: {0}")]
+    CarReader(#[from] iroh_car::Error),
+    #[error("Storage error")]
+    StorageError(#[from] FjallError),
+}
 
 #[derive(Debug, Parser)]
 struct Args {
···
     let t0 = Instant::now();
 
     // set up a disk store we can spill to
-    let disk_store = DiskStore::new(tmpfile).await?;
+    let mut 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 mut store = load_car(reader, disk_store).await?;
+    let mut mem_blocks = HashMap::new();
+    let mut car = CarReader::new(reader).await?;
+    // load some blocks into memory (idk why doing this helps repro)
+    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}");
+            break;
+        }
+    }
+    // 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)) = car.next_block().await? {
+        store.put(cid.to_bytes(), data)?;
+    }
 
     // at this point you might want to fetch the account's signing key
     // via the DID from the commit, and then verify the signature.
···
 
     Ok(())
 }
+
+
+/// On-disk block storage
+pub struct DiskStore {
+    #[allow(unused)]
+    db: Database,
+    ks: Keyspace,
+    current: usize,
+    all_keys: HashMap<Vec<u8>, usize>,
+}
+
+impl DiskStore {
+    /// Initialize a new disk store
+    pub async fn new(
+        path: PathBuf,
+    ) -> Result<Self, FjallError> {
+        let db = Database::builder(path).open()?;
+        let ks = db.keyspace("z", KeyspaceCreateOptions::default)?;
+
+        Ok(Self {
+            db,
+            ks,
+            current: 0,
+            all_keys: Default::default(),
+        })
+    }
+    pub(crate) fn put(&mut self, k: Vec<u8>, v: Vec<u8>) -> Result<(), DriveError> {
+        self.ks.insert(k.clone(), v)?;
+        self.all_keys.insert(k.clone(), self.current);
+        self.current += 1;
+        Ok(())
+    }
+
+    pub fn check_keys(&mut self) {
+        log::info!("checking keys...");
+        for guard in self.ks.iter() {
+            let key = guard.key().unwrap();
+            self.all_keys.remove(key.as_ref());
+        }
+        if self.all_keys.len() == 0 {
+            log::info!("all keys found.");
+        }
+        for (leftover, i) in &self.all_keys {
+            log::warn!("fjall key missing: {i} {leftover:?}");
+        }
+    }
+}
+
+
+
+

src/disk.rs

···
-use crate::drive::DriveError;
-use fjall::{Database, Keyspace, KeyspaceCreateOptions, Error as FjallError};
-use std::path::PathBuf;
-use std::collections::HashMap;
-
-#[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] FjallError),
-}
-
-/// On-disk block storage
-pub struct DiskStore {
-    #[allow(unused)]
-    db: Database,
-    ks: Keyspace,
-    current: usize,
-    all_keys: HashMap<Vec<u8>, usize>,
-}
-
-impl DiskStore {
-    /// Initialize a new disk store
-    pub async fn new(
-        path: PathBuf,
-    ) -> Result<Self, DiskError> {
-        let db = Database::builder(path).open()?;
-        let ks = db.keyspace("z", KeyspaceCreateOptions::default)?;
-
-        Ok(Self {
-            db,
-            ks,
-            current: 0,
-            all_keys: Default::default(),
-        })
-    }
-    pub(crate) fn put(&mut self, k: Vec<u8>, v: Vec<u8>) -> Result<(), DriveError> {
-        self.ks.insert(k.clone(), v).map_err(DiskError::DbError)?;
-        self.all_keys.insert(k.clone(), self.current);
-        self.current += 1;
-        Ok(())
-    }
-
-    pub fn check_keys(&mut self) {
-        log::info!("checking keys...");
-        for guard in self.ks.iter() {
-            let key = guard.key().unwrap();
-            self.all_keys.remove(key.as_ref());
-        }
-        if self.all_keys.len() == 0 {
-            log::info!("all keys found.");
-        }
-        for (leftover, i) in &self.all_keys {
-            log::warn!("fjall key missing: {i} {leftover:?}");
-        }
-    }
-}

src/drive.rs

···
-//! Consume a CAR from an AsyncRead, producing an ordered stream of records
-
-use crate::disk::{DiskError, DiskStore};
-use iroh_car::CarReader;
-use std::collections::HashMap;
-use tokio::io::AsyncRead;
-
-
-/// Errors that can happen while consuming and emitting blocks and records
-#[derive(Debug, thiserror::Error)]
-pub enum DriveError {
-    #[error("Error from iroh_car: {0}")]
-    CarReader(#[from] iroh_car::Error),
-    #[error("Storage error")]
-    StorageError(#[from] DiskError),
-}
-
-
-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?;
-
-    // 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}");
-            break;
-        }
-    }
-
-    // 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)) = car.next_block().await? {
-        store.put(cid.to_bytes(), data)?;
-    }
-    log::debug!("done.");
-    Ok(store)
-
-}

src/lib.rs

···
 Find more [examples in the repo](https://tangled.org/@microcosm.blue/repo-stream/tree/main/examples).
 
 */
-
-pub mod disk;
-pub mod drive;
-
-pub use disk::{DiskError, DiskStore};
-pub use drive::{DriveError, load_car};