sketch out repo-stream coordindator

Cargo.lock

···
 
 [[package]]
 name = "anyhow"
-version = "1.0.97"
+version = "1.0.100"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dcfed56ad506cb2c684a14971b8861fdc3baaaae314b9e5f9bb532cbe3ba7a4f"
+checksum = "a23eb6b1614318a8071c9b2521f36b424b2c83db5eb3a0fead4a6c0809af6e61"
 
 [[package]]
 name = "arbitrary"
···
  "nom",
  "num-traits",
  "rusticata-macros",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "time",
 ]
 
···
  "axum",
  "handlebars",
  "serde",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
 ]
 
 [[package]]
···
 version = "0.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4c7f02d4ea65f2c1853089ffd8d2787bdbc63de2f0d29dedbcf8ccdfa0ccd4cf"
+
+[[package]]
+name = "base256emoji"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b5e9430d9a245a77c92176e649af6e275f20839a48389859d1661e9a128d077c"
+dependencies = [
+ "const-str",
+ "match-lookup",
+]
 
 [[package]]
 name = "base64"
···
 checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b"
 
 [[package]]
+name = "byteorder-lite"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495"
+
+[[package]]
 name = "bytes"
 version = "1.10.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 version = "0.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6236364b88b9b6d0bc181ba374cf1ab55ba3ef97a1cb6f8cddad48a273767fb5"
+
+[[package]]
+name = "byteview"
+version = "0.8.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1e6b0e42e210b794e14b152c6fe1a55831e30ef4a0f5dc39d73d714fb5f1906c"
 
 [[package]]
 name = "bzip2-sys"
···
 ]
 
 [[package]]
+name = "cbor4ii"
+version = "0.2.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b544cf8c89359205f4f990d0e6f3828db42df85b5dac95d09157a250eb0749c4"
+dependencies = [
+ "serde",
+]
+
+[[package]]
 name = "cc"
 version = "1.2.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "multihash",
  "serde",
  "serde_bytes",
- "unsigned-varint",
+ "unsigned-varint 0.8.0",
 ]
 
 [[package]]
···
 version = "0.9.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8"
+
+[[package]]
+name = "const-str"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2f421161cb492475f1661ddc9815a745a1c894592070661180fdec3d4872e9c3"
 
 [[package]]
 name = "constellation"
···
  "slog-bunyan",
  "slog-json",
  "slog-term",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tokio-rustls 0.25.0",
  "toml 0.9.7",
···
 checksum = "0b25ad44cd4360a0448a9b5a0a6f1c7a621101cca4578706d43c9a821418aebc"
 dependencies = [
  "byteorder",
- "byteview",
+ "byteview 0.6.1",
  "dashmap",
  "log",
- "lsm-tree",
+ "lsm-tree 2.10.4",
  "path-absolutize",
  "std-semaphore",
  "tempfile",
···
 source = "git+https://github.com/fjall-rs/fjall.git#42d811f7c8cc9004407d520d37d2a1d8d246c03d"
 dependencies = [
  "byteorder",
- "byteview",
+ "byteview 0.6.1",
  "dashmap",
  "log",
- "lsm-tree",
+ "lsm-tree 2.10.4",
  "path-absolutize",
+ "std-semaphore",
+ "tempfile",
+ "xxhash-rust",
+]
+
+[[package]]
+name = "fjall"
+version = "3.0.0-pre.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "467588c1f15d1cfa9e43f02a45cf55d82fa1f12a6ae961b848c520458525600c"
+dependencies = [
+ "byteorder-lite",
+ "byteview 0.8.0",
+ "dashmap",
+ "log",
+ "lsm-tree 3.0.0-pre.0",
  "std-semaphore",
  "tempfile",
  "xxhash-rust",
···
  "mixtrics",
  "pin-project",
  "serde",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tracing",
 ]
···
  "parking_lot",
  "pin-project",
  "serde",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "twox-hash",
 ]
···
  "parking_lot",
  "pin-project",
  "serde",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tracing",
 ]
···
  "pin-project",
  "rand 0.9.1",
  "serde",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tracing",
  "twox-hash",
···
  "pest_derive",
  "serde",
  "serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "walkdir",
 ]
 
···
  "once_cell",
  "rand 0.9.1",
  "ring",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tinyvec",
  "tokio",
  "tracing",
···
  "rand 0.9.1",
  "resolv-conf",
  "smallvec",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tracing",
 ]
···
 ]
 
 [[package]]
+name = "iroh-car"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb7f8cd4cb9aa083fba8b52e921764252d0b4dcb1cd6d120b809dbfe1106e81a"
+dependencies = [
+ "anyhow",
+ "cid",
+ "futures",
+ "serde",
+ "serde_ipld_dagcbor",
+ "thiserror 1.0.69",
+ "tokio",
+ "unsigned-varint 0.7.2",
+]
+
+[[package]]
 name = "is-terminal"
 version = "0.4.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "metrics",
  "serde",
  "serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tokio-tungstenite 0.26.2",
  "url",
···
  "anyhow",
  "fluent-uri",
  "nom",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tinyjson",
 ]
 
···
 
 [[package]]
 name = "lsm-tree"
-version = "2.10.2"
+version = "2.10.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "55b6d7475a8dd22e749186968daacf8e2a77932b061b1bd263157987bbfc0c6c"
+checksum = "799399117a2bfb37660e08be33f470958babb98386b04185288d829df362ea15"
 dependencies = [
  "byteorder",
  "crossbeam-skiplist",
···
 ]
 
 [[package]]
+name = "lsm-tree"
+version = "3.0.0-pre.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "be375d45e348328e78582dffbda4f1709dd52fca27c1a81c7bf6ca134e6335f7"
+dependencies = [
+ "byteorder-lite",
+ "byteview 0.8.0",
+ "crossbeam-skiplist",
+ "enum_dispatch",
+ "interval-heap",
+ "log",
+ "lz4_flex",
+ "quick_cache",
+ "rustc-hash 2.1.1",
+ "self_cell",
+ "sfa",
+ "tempfile",
+ "varint-rs",
+ "xxhash-rust",
+]
+
+[[package]]
 name = "lz4"
 version = "1.28.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 
 [[package]]
 name = "lz4_flex"
-version = "0.11.3"
+version = "0.11.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "75761162ae2b0e580d7e7c390558127e5f01b4194debd6221fd8c207fc80e3f5"
+checksum = "08ab2867e3eeeca90e844d1940eab391c9dc5228783db2ed999acbc0a9ed375a"
 
 [[package]]
 name = "mach2"
···
  "madsim",
  "spin",
  "tokio",
+]
+
+[[package]]
+name = "match-lookup"
+version = "0.1.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1265724d8cb29dbbc2b0f06fffb8bf1a8c0cf73a78eede9ba73a4a66c52a981e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 1.0.109",
 ]
 
 [[package]]
···
  "metrics",
  "metrics-util 0.20.0",
  "quanta",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tracing",
 ]
···
 
 [[package]]
 name = "multibase"
-version = "0.9.1"
+version = "0.9.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9b3539ec3c1f04ac9748a260728e855f261b4977f5c3406612c884564f329404"
+checksum = "8694bb4835f452b0e3bb06dbebb1d6fc5385b6ca1caf2e55fd165c042390ec77"
 dependencies = [
  "base-x",
+ "base256emoji",
  "data-encoding",
  "data-encoding-macro",
 ]
···
 dependencies = [
  "core2",
  "serde",
- "unsigned-varint",
+ "unsigned-varint 0.8.0",
 ]
 
 [[package]]
···
 checksum = "1db05f56d34358a8b1066f67cbb203ee3e7ed2ba674a6263a1d5ec6db2204323"
 dependencies = [
  "memchr",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "ucd-trie",
 ]
 
···
  "rusqlite",
  "serde",
  "serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tracing-subscriber",
 ]
···
  "smallvec",
  "sync_wrapper",
  "tempfile",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tokio-rustls 0.26.2",
  "tokio-stream",
···
  "serde_json",
  "serde_urlencoded",
  "serde_yaml",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
 ]
 
···
  "quote",
  "regex",
  "syn 2.0.106",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
 ]
 
 [[package]]
···
 
 [[package]]
 name = "quick_cache"
-version = "0.6.12"
+version = "0.6.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8f8ed0655cbaf18a26966142ad23b95d8ab47221c50c4f73a1db7d0d2d6e3da8"
+checksum = "9ad6644cb07b7f3488b9f3d2fde3b4c0a7fa367cafefb39dff93a659f76eb786"
 dependencies = [
  "equivalent",
  "hashbrown 0.15.2",
···
  "rustc-hash 2.1.1",
  "rustls 0.23.31",
  "socket2 0.5.9",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tracing",
  "web-time",
···
  "rustls 0.23.31",
  "rustls-pki-types",
  "slab",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tinyvec",
  "tracing",
  "web-time",
···
 checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c"
 
 [[package]]
+name = "repo-stream"
+version = "0.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "727a78c392bd51b1af938e4383f2f6f46ae727cb38394136d1aebab0633faf8e"
+dependencies = [
+ "bincode 2.0.1",
+ "futures",
+ "futures-core",
+ "ipld-core",
+ "iroh-car",
+ "log",
+ "multibase",
+ "rusqlite",
+ "serde",
+ "serde_bytes",
+ "serde_ipld_dagcbor",
+ "sha2",
+ "thiserror 2.0.17",
+ "tokio",
+]
+
+[[package]]
 name = "reqwest"
-version = "0.12.23"
+version = "0.12.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d429f34c8092b2d42c7c93cec323bb4adeb7c67698f70839adec842ec10c7ceb"
+checksum = "9d0946410b9f7b082a427e4ef5c8ff541a88b357bc6c637c40db3a68ac70a36f"
 dependencies = [
  "async-compression",
  "base64 0.22.1",
···
  "url",
  "wasm-bindgen",
  "wasm-bindgen-futures",
+ "wasm-streams",
  "web-sys",
 ]
 
···
 
 [[package]]
 name = "self_cell"
-version = "1.1.0"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c2fdfc24bc566f839a2da4c4295b82db7d25a24253867d5c64355abb5799bdbe"
+checksum = "0f7d95a54511e0c7be3f51e8867aa8cf35148d7b9445d44de2f943e2b206e749"
 
 [[package]]
 name = "semver"
···
 
 [[package]]
 name = "serde_bytes"
-version = "0.11.17"
+version = "0.11.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8437fd221bde2d4ca316d61b90e337e9e702b3820b87d63caa9ba6c02bd06d96"
+checksum = "a5d440709e79d88e51ac01c4b72fc6cb7314017bb7da9eeff678aa94c10e3ea8"
 dependencies = [
  "serde",
+ "serde_core",
 ]
 
 [[package]]
···
 ]
 
 [[package]]
+name = "serde_ipld_dagcbor"
+version = "0.6.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "46182f4f08349a02b45c998ba3215d3f9de826246ba02bb9dddfe9a2a2100778"
+dependencies = [
+ "cbor4ii",
+ "ipld-core",
+ "scopeguard",
+ "serde",
+]
+
+[[package]]
 name = "serde_json"
 version = "1.0.145"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "percent-encoding",
  "ryu",
  "serde",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
 ]
 
 [[package]]
···
 ]
 
 [[package]]
+name = "sfa"
+version = "0.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e5f5f9dc21f55409f15103d5a7e7601b804935923c7fe4746dc806c3a422a038"
+dependencies = [
+ "byteorder-lite",
+ "log",
+ "xxhash-rust",
+]
+
+[[package]]
 name = "sha1"
 version = "0.10.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
 dependencies = [
  "num-bigint",
  "num-traits",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "time",
 ]
 
···
  "rustls 0.23.31",
  "serde",
  "serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "time",
  "tokio",
  "tokio-util",
···
 name = "spacedust"
 version = "0.1.0"
 dependencies = [
+ "anyhow",
+ "async-channel",
  "async-trait",
  "clap",
  "ctrlc",
  "dropshot",
  "env_logger",
+ "fjall 3.0.0-pre.0",
  "futures",
  "http",
+ "ipld-core",
  "jetstream",
  "links",
  "log",
  "metrics",
  "metrics-exporter-prometheus 0.17.2",
  "rand 0.9.1",
+ "repo-stream",
+ "reqwest",
  "schemars",
  "semver",
  "serde",
+ "serde_ipld_dagcbor",
  "serde_json",
  "serde_qs",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tinyjson",
  "tokio",
  "tokio-tungstenite 0.27.0",
···
 
 [[package]]
 name = "tempfile"
-version = "3.19.1"
+version = "3.23.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7437ac7763b9b123ccf33c338a5cc1bac6f69b45a136c19bdd8a65e3916435bf"
+checksum = "2d31c77bdf42a745371d260a26ca7163f1e0924b64afa0b688e61b5a9fa02f16"
 dependencies = [
  "fastrand",
  "getrandom 0.3.3",
···
 
 [[package]]
 name = "thiserror"
-version = "2.0.16"
+version = "2.0.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3467d614147380f2e4e374161426ff399c91084acd2363eaf549172b3d5e60c0"
+checksum = "f63587ca0f12b72a0600bcba1d40081f830876000bb46dd2337a3051618f4fc8"
 dependencies = [
- "thiserror-impl 2.0.16",
+ "thiserror-impl 2.0.17",
 ]
 
 [[package]]
···
 
 [[package]]
 name = "thiserror-impl"
-version = "2.0.16"
+version = "2.0.17"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6c5e1be1c48b9172ee610da68fd9cd2770e7a4056cb3fc98710ee6906f0c7960"
+checksum = "3ff15c8ecd7de3849db632e14d18d2571fa09dfc5ed93479bc4485c7a517c913"
 dependencies = [
  "proc-macro2",
  "quote",
···
  "native-tls",
  "rand 0.9.1",
  "sha1",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "url",
  "utf-8",
 ]
···
  "log",
  "rand 0.9.1",
  "sha1",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "utf-8",
 ]
 
···
  "http",
  "jetstream",
  "log",
- "lsm-tree",
+ "lsm-tree 2.10.4",
  "metrics",
  "metrics-exporter-prometheus 0.17.2",
  "schemars",
···
  "serde_qs",
  "sha2",
  "tempfile",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tikv-jemallocator",
  "tokio",
  "tokio-util",
···
 
 [[package]]
 name = "unsigned-varint"
+version = "0.7.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6889a77d49f1f013504cec6bf97a2c730394adedaeb1deb5ea08949a50541105"
+
+[[package]]
+name = "unsigned-varint"
 version = "0.8.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "eb066959b24b5196ae73cb057f45598450d2c5f71460e98c49b738086eff9c06"
···
 checksum = "62fc7c4ce161f049607ecea654dca3f2d727da5371ae85e2e4f14ce2b98ed67c"
 dependencies = [
  "byteorder",
- "byteview",
+ "byteview 0.6.1",
  "interval-heap",
  "log",
  "path-absolutize",
···
 ]
 
 [[package]]
+name = "wasm-streams"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15053d8d85c7eccdbefef60f06769760a563c7f0a9d6902a13d35c7800b0ad65"
+dependencies = [
+ "futures-util",
+ "js-sys",
+ "wasm-bindgen",
+ "wasm-bindgen-futures",
+ "web-sys",
+]
+
+[[package]]
 name = "web-sys"
 version = "0.3.77"
 source = "registry+https://github.com/rust-lang/crates.io-index"
···
  "reqwest",
  "serde",
  "serde_json",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "tokio",
  "tokio-util",
  "url",
···
  "nom",
  "oid-registry",
  "rusticata-macros",
- "thiserror 2.0.16",
+ "thiserror 2.0.17",
  "time",
 ]
 

spacedust/Cargo.toml

···
 edition = "2024"
 
 [dependencies]
+anyhow = "1.0.100"
+async-channel = "2.5.0"
 async-trait = "0.1.88"
 clap = { version = "4.5.40", features = ["derive"] }
 ctrlc = "3.4.7"
 dropshot = "0.16.2"
 env_logger = "0.11.8"
+fjall = "3.0.0-pre.0"
 futures = "0.3.31"
 http = "1.3.1"
+ipld-core = { version = "0.4.2", features = ["serde"] }
 jetstream = { path = "../jetstream", features = ["metrics"] }
 links = { path = "../links" }
 log = "0.4.27"
 metrics = "0.24.2"
 metrics-exporter-prometheus = { version = "0.17.1", features = ["http-listener"] }
 rand = "0.9.1"
+repo-stream = "0.2.1"
+reqwest = { version = "0.12.24", features = ["json", "stream"] }
 schemars = "0.8.22"
 semver = "1.0.26"
 serde = { version = "1.0.219", features = ["derive"] }
+serde_ipld_dagcbor = "0.6.4"
 serde_json = "1.0.140"
 serde_qs = "1.0.0-rc.3"
 thiserror = "2.0.12"

spacedust/src/bin/import_car_file.rs

···
+use clap::Parser;
+use std::path::PathBuf;
+use spacedust::storage::car;
+
+type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
+
+#[derive(Debug, Parser)]
+struct Args {
+    #[arg()]
+    file: PathBuf,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+
+    let Args { file } = Args::parse();
+
+    let reader = tokio::fs::File::open(file).await?;
+
+    car::import(reader).await?;
+
+    Ok(())
+}

spacedust/src/bin/import_scraped.rs

···
+use clap::Parser;
+use std::sync::{Arc, atomic::{AtomicUsize, Ordering}};
+use std::path::PathBuf;
+use tokio::{task::JoinSet, io::AsyncRead};
+use repo_stream::{DriverBuilder, Driver, DiskBuilder, DiskStore, drive::NeedDisk};
+
+
+type Result<T> = anyhow::Result<T>; //std::result::Result<T, Box<dyn std::error::Error>>;
+
+
+#[derive(Debug, Parser)]
+struct Args {
+    #[arg(long)]
+    cars_folder: PathBuf,
+    #[arg(long)]
+    mem_workers: usize,
+    #[arg(long)]
+    disk_workers: usize,
+    #[arg(long)]
+    disk_folder: PathBuf,
+}
+
+async fn get_cars(cars_folder: PathBuf, tx: async_channel::Sender<tokio::io::BufReader<tokio::fs::File>>) -> Result<()> {
+    let mut dir = tokio::fs::read_dir(cars_folder).await?;
+    while let Some(entry) = dir.next_entry().await? {
+        if !entry.file_type().await?.is_file() {
+            continue;
+        }
+        let reader = tokio::fs::File::open(&entry.path()).await?;
+        let reader = tokio::io::BufReader::new(reader);
+        tx.send(reader).await?;
+    }
+    Ok(())
+}
+
+async fn drive_mem<R: AsyncRead + Unpin + Send + Sync + 'static>(
+    f: R,
+    disk_tx: &async_channel::Sender<NeedDisk<R, usize>>,
+) -> Result<Option<usize>> {
+    let mut n = 0;
+    match DriverBuilder::new()
+        .with_block_processor(|_| 0_usize) // don't care just counting records
+        .with_mem_limit_mb(32)
+        .load_car(f)
+        .await?
+    {
+        Driver::Memory(_commit, mut driver) => {
+            while let Some(chunk) = driver.next_chunk(512).await? {
+                n += chunk.len();
+            }
+            Ok(Some(n))
+        }
+        Driver::Disk(need_disk) => {
+            disk_tx.send(need_disk).await?;
+            Ok(None)
+        }
+    }
+}
+
+async fn mem_worker<R: AsyncRead + Unpin + Send + Sync + 'static>(
+    car_rx: async_channel::Receiver<R>,
+    disk_tx: async_channel::Sender<NeedDisk<R, usize>>,
+    n: Arc<AtomicUsize>,
+) -> Result<()> {
+    while let Ok(f) = car_rx.recv().await {
+        let driven = match drive_mem(f, &disk_tx).await {
+            Ok(d) => d,
+            Err(e) => {
+                eprintln!("failed to drive mem: {e:?}. skipping...");
+                continue;
+            }
+        };
+        if let Some(drove) = driven {
+            n.fetch_add(drove, Ordering::Relaxed);
+        }
+    }
+    Ok(())
+}
+
+async fn drive_disk<R: AsyncRead + Unpin>(
+    needed: NeedDisk<R, usize>,
+    store: DiskStore,
+) -> Result<(usize, DiskStore)> {
+    let (_commit, mut driver) = needed.finish_loading(store).await?;
+    let mut n = 0;
+    while let Some(chunk) = driver.next_chunk(512).await? {
+        n += chunk.len();
+    }
+    let store = driver.reset_store().await?;
+    Ok((n, store))
+}
+
+async fn disk_worker<R: AsyncRead + Unpin>(
+    worker_id: usize,
+    disk_rx: async_channel::Receiver<NeedDisk<R, usize>>,
+    folder: PathBuf,
+    n: Arc<AtomicUsize>,
+    disk_workers_active: Arc<AtomicUsize>,
+) -> Result<()> {
+    let mut file = folder;
+    file.push(format!("disk-worker-{worker_id}.sqlite"));
+    let mut store = DiskBuilder::new()
+        .with_cache_size_mb(128)
+        .open(file.clone())
+        .await?;
+    while let Ok(needed) = disk_rx.recv().await {
+        let active = disk_workers_active.fetch_add(1, Ordering::AcqRel);
+        println!("-> disk workers active: {}", active + 1);
+        let drove = match drive_disk(needed, store).await {
+            Ok((d, s)) => {
+                store = s;
+                d
+            }
+            Err(e) => {
+                eprintln!("failed to drive disk: {e:?}. skipping...");
+                store = DiskBuilder::new()
+                    .with_cache_size_mb(128)
+                    .open(file.clone())
+                    .await?;
+                continue;
+            }
+        };
+        n.fetch_add(drove, Ordering::Relaxed);
+        let were_active = disk_workers_active.fetch_sub(1, Ordering::AcqRel);
+        println!("<- disk workers active: {}", were_active - 1);
+    }
+    Ok(())
+}
+
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+
+    let Args { cars_folder, disk_folder, disk_workers, mem_workers } = Args::parse();
+
+    let mut set = JoinSet::<Result<()>>::new();
+
+
+    let (cars_tx, cars_rx) = async_channel::bounded(2);
+    set.spawn(get_cars(cars_folder, cars_tx));
+
+    let n: Arc<AtomicUsize> = Arc::new(0.into());
+    let disk_workers_active: Arc<AtomicUsize> = Arc::new(0.into());
+
+    set.spawn({
+        let n = n.clone();
+        let mut interval = tokio::time::interval(std::time::Duration::from_secs(10));
+        async move {
+            let mut last_n = n.load(Ordering::Relaxed);
+            loop {
+                interval.tick().await;
+                let n = n.load(Ordering::Relaxed);
+                let diff = n - last_n;
+                println!("rate: {} rec/sec", diff / 10);
+                if diff == 0 {
+                    println!("zero encountered, stopping rate calculation polling.");
+                    break Ok(());
+                }
+                last_n = n;
+            }
+        }
+    });
+
+
+    let (needs_disk_tx, needs_disk_rx) = async_channel::bounded(disk_workers);
+
+
+    for _ in 0..mem_workers {
+        set.spawn(mem_worker(cars_rx.clone(), needs_disk_tx.clone(), n.clone()));
+    }
+    drop(cars_rx);
+    drop(needs_disk_tx);
+
+    tokio::fs::create_dir_all(disk_folder.clone()).await?;
+    for id in 0..disk_workers {
+        set.spawn(disk_worker(
+            id,
+            needs_disk_rx.clone(),
+            disk_folder.clone(),
+            n.clone(),
+            disk_workers_active.clone(),
+        ));
+    }
+    drop(needs_disk_rx);
+
+    while let Some(res) = set.join_next().await {
+        println!("task from set joined: {res:?}");
+    }
+
+    eprintln!("total records processed: {n:?}");
+
+    Ok(())
+}

spacedust/src/bin/scrape_pds.rs

···
+use tokio::io::AsyncWriteExt;
+use clap::Parser;
+use std::path::PathBuf;
+use reqwest::Url;
+use tokio::{sync::mpsc, time};
+use serde::Deserialize;
+
+type Result<T> = std::result::Result<T, Box<dyn std::error::Error>>;
+
+use futures::StreamExt;
+
+#[derive(Debug, Parser)]
+struct Args {
+    #[arg(long)]
+    pds: Url,
+    #[arg(long)]
+    throttle_ms: u64, // 100ms per pds?
+    #[arg(long)]
+    folder: PathBuf,
+}
+
+async fn download_repo(
+    client: &reqwest::Client,
+    mut pds: Url,
+    did: String,
+    mut path: PathBuf,
+) -> Result<()> {
+    path.push(format!("{did}.car"));
+    let f = tokio::fs::File::create(path).await?;
+    let mut w = tokio::io::BufWriter::new(f);
+
+    pds.set_path("/xrpc/com.atproto.sync.getRepo");
+    pds.set_query(Some(&format!("did={did}")));
+    let mut byte_stream = client
+        .get(pds)
+        .send()
+        .await?
+        .bytes_stream();
+
+    while let Some(stuff) = byte_stream.next().await {
+        tokio::io::copy(&mut stuff?.as_ref(), &mut w).await?;
+    }
+    w.flush().await?;
+
+    Ok(())
+}
+
+
+#[derive(Debug, Deserialize)]
+struct RepoInfo {
+    did: String,
+    active: bool,
+}
+
+#[derive(Debug, Deserialize)]
+struct ListReposResponse {
+    cursor: Option<String>,
+    repos: Vec<RepoInfo>,
+}
+
+fn get_pds_dids(client: reqwest::Client, mut pds: Url) -> mpsc::Receiver<String> {
+    let (tx, rx) = mpsc::channel(2);
+    tokio::task::spawn(async move {
+        pds.set_path("/xrpc/com.atproto.sync.listRepos");
+        let mut cursor = None;
+
+        loop {
+            if let Some(c) = cursor {
+                pds.set_query(Some(&format!("cursor={c}")));
+            }
+            let res: ListReposResponse = client
+                .get(pds.clone())
+                .send()
+                .await
+                .expect("to send request")
+                .error_for_status()
+                .expect("to be ok")
+                .json()
+                .await
+                .expect("json response");
+            for repo in res.repos {
+                if repo.active {
+                    tx.send(repo.did).await.expect("to be able to send on the channel");
+                }
+            }
+            cursor = res.cursor;
+            if cursor.is_none() {
+                break;
+            }
+        }
+
+    });
+    rx
+}
+
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    env_logger::init();
+
+    let Args { pds, throttle_ms, folder } = Args::parse();
+
+    tokio::fs::create_dir_all(folder.clone()).await?;
+
+    let client = reqwest::Client::builder()
+        .user_agent("microcosm/spacedust-testing")
+        .build()?;
+
+    let mut dids = get_pds_dids(client.clone(), pds.clone());
+
+    let mut interval = time::interval(time::Duration::from_millis(throttle_ms));
+    let mut oks = 0;
+    let mut single_fails = 0;
+    let mut double_fails = 0;
+
+    while let Some(did) = dids.recv().await {
+        interval.tick().await;
+        println!("did: {did:?}");
+        if let Err(e) = download_repo(&client, pds.clone(), did.clone(), folder.clone()).await {
+            single_fails += 1;
+            eprintln!("failed to download repo for did: {did:?}: {e:?}. retrying in a moment...");
+            tokio::time::sleep(time::Duration::from_secs(3)).await;
+            interval.reset();
+            if let Err(e) = download_repo(&client, pds.clone(), did.clone(), folder.clone()).await {
+                double_fails += 1;
+                eprintln!("failed again: {e:?}. moving on in a moment...");
+                tokio::time::sleep(time::Duration::from_secs(1)).await;
+                continue;
+            }
+        }
+        oks += 1;
+        println!("  -> done. did: {did:?}");
+    }
+
+    eprintln!("got {oks} repos. single fails: {single_fails}; doubles: {double_fails}.");
+
+    Ok(())
+}

spacedust/src/lib.rs

···
 pub mod error;
 pub mod removable_delay_queue;
 pub mod server;
+pub mod storage;
 pub mod subscriber;
 
 use jetstream::events::CommitEvent;

spacedust/src/storage/fjall/mod.rs

···
+use crate::storage::Storage;
+
+struct FjallStorage {
+}
+
+impl Storage for FjallStorage {
+    fn import_car() { todo!() }
+}

spacedust/src/storage/mod.rs

···
+pub mod car;
+pub mod fjall;
+
+pub trait Storage {
+    fn import_car() {
+
+    }
+}