From 62854ab9c2f2a00da5ed7106d84a58d930dfbe4c Mon Sep 17 00:00:00 2001 From: phil Date: Wed, 12 Nov 2025 15:19:38 -0500 Subject: [PATCH] sketch out repo-stream coordindator --- Cargo.lock | 296 +++++++++++++++++++++------ spacedust/Cargo.toml | 7 + spacedust/src/bin/import_car_file.rs | 24 +++ spacedust/src/bin/import_scraped.rs | 194 ++++++++++++++++++ spacedust/src/bin/scrape_pds.rs | 138 +++++++++++++ spacedust/src/lib.rs | 1 + spacedust/src/storage/car/drive.rs | 0 spacedust/src/storage/car/mod.rs | 0 spacedust/src/storage/car/walk.rs | 0 spacedust/src/storage/fjall/mod.rs | 8 + spacedust/src/storage/mod.rs | 8 + 11 files changed, 616 insertions(+), 60 deletions(-) create mode 100644 spacedust/src/bin/import_car_file.rs create mode 100644 spacedust/src/bin/import_scraped.rs create mode 100644 spacedust/src/bin/scrape_pds.rs create mode 100644 spacedust/src/storage/car/drive.rs create mode 100644 spacedust/src/storage/car/mod.rs create mode 100644 spacedust/src/storage/car/walk.rs create mode 100644 spacedust/src/storage/fjall/mod.rs create mode 100644 spacedust/src/storage/mod.rs diff --git a/Cargo.lock b/Cargo.lock index 41e7296..7b77c50 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -112,9 +112,9 @@ dependencies = [ [[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" @@ -192,7 +192,7 @@ dependencies = [ "nom", "num-traits", "rusticata-macros", - "thiserror 2.0.16", + "thiserror 2.0.17", "time", ] @@ -644,7 +644,7 @@ dependencies = [ "axum", "handlebars", "serde", - "thiserror 2.0.16", + "thiserror 2.0.17", ] [[package]] @@ -674,6 +674,16 @@ 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" version = "0.21.7" @@ -838,6 +848,12 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" 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" @@ -850,6 +866,12 @@ 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" version = "0.1.13+1.0.8" @@ -891,6 +913,15 @@ dependencies = [ "serde", ] +[[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" @@ -976,7 +1007,7 @@ dependencies = [ "multihash", "serde", "serde_bytes", - "unsigned-varint", + "unsigned-varint 0.8.0", ] [[package]] @@ -1086,6 +1117,12 @@ 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" version = "0.1.0" @@ -1579,7 +1616,7 @@ dependencies = [ "slog-bunyan", "slog-json", "slog-term", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tokio-rustls 0.25.0", "toml 0.9.7", @@ -1783,10 +1820,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b25ad44cd4360a0448a9b5a0a6f1c7a621101cca4578706d43c9a821418aebc" dependencies = [ "byteorder", - "byteview", + "byteview 0.6.1", "dashmap", "log", - "lsm-tree", + "lsm-tree 2.10.4", "path-absolutize", "std-semaphore", "tempfile", @@ -1799,16 +1836,32 @@ version = "2.11.2" 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", +] + [[package]] name = "flate2" version = "1.1.2" @@ -1891,7 +1944,7 @@ dependencies = [ "mixtrics", "pin-project", "serde", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tracing", ] @@ -1911,7 +1964,7 @@ dependencies = [ "parking_lot", "pin-project", "serde", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "twox-hash", ] @@ -1944,7 +1997,7 @@ dependencies = [ "parking_lot", "pin-project", "serde", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tracing", ] @@ -1976,7 +2029,7 @@ dependencies = [ "pin-project", "rand 0.9.1", "serde", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tracing", "twox-hash", @@ -2220,7 +2273,7 @@ dependencies = [ "pest_derive", "serde", "serde_json", - "thiserror 2.0.16", + "thiserror 2.0.17", "walkdir", ] @@ -2345,7 +2398,7 @@ dependencies = [ "once_cell", "rand 0.9.1", "ring", - "thiserror 2.0.16", + "thiserror 2.0.17", "tinyvec", "tokio", "tracing", @@ -2368,7 +2421,7 @@ dependencies = [ "rand 0.9.1", "resolv-conf", "smallvec", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tracing", ] @@ -2799,6 +2852,22 @@ dependencies = [ "serde", ] +[[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" @@ -2863,7 +2932,7 @@ dependencies = [ "metrics", "serde", "serde_json", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tokio-tungstenite 0.26.2", "url", @@ -3118,7 +3187,7 @@ dependencies = [ "anyhow", "fluent-uri", "nom", - "thiserror 2.0.16", + "thiserror 2.0.17", "tinyjson", ] @@ -3186,9 +3255,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[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", @@ -3208,6 +3277,28 @@ dependencies = [ "xxhash-rust", ] +[[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" @@ -3229,9 +3320,9 @@ dependencies = [ [[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" @@ -3296,6 +3387,17 @@ dependencies = [ "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]] name = "match_cfg" version = "0.1.0" @@ -3384,7 +3486,7 @@ dependencies = [ "metrics", "metrics-util 0.20.0", "quanta", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tracing", ] @@ -3531,11 +3633,12 @@ dependencies = [ [[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", ] @@ -3548,7 +3651,7 @@ checksum = "6b430e7953c29dd6a09afc29ff0bb69c6e306329ee6794700aee27b76a1aea8d" dependencies = [ "core2", "serde", - "unsigned-varint", + "unsigned-varint 0.8.0", ] [[package]] @@ -3926,7 +4029,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1db05f56d34358a8b1066f67cbb203ee3e7ed2ba674a6263a1d5ec6db2204323" dependencies = [ "memchr", - "thiserror 2.0.16", + "thiserror 2.0.17", "ucd-trie", ] @@ -4036,7 +4139,7 @@ dependencies = [ "rusqlite", "serde", "serde_json", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tracing-subscriber", ] @@ -4079,7 +4182,7 @@ dependencies = [ "smallvec", "sync_wrapper", "tempfile", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tokio-rustls 0.26.2", "tokio-stream", @@ -4123,7 +4226,7 @@ dependencies = [ "serde_json", "serde_urlencoded", "serde_yaml", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", ] @@ -4142,7 +4245,7 @@ dependencies = [ "quote", "regex", "syn 2.0.106", - "thiserror 2.0.16", + "thiserror 2.0.17", ] [[package]] @@ -4269,9 +4372,9 @@ dependencies = [ [[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", @@ -4291,7 +4394,7 @@ dependencies = [ "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", @@ -4312,7 +4415,7 @@ dependencies = [ "rustls 0.23.31", "rustls-pki-types", "slab", - "thiserror 2.0.16", + "thiserror 2.0.17", "tinyvec", "tracing", "web-time", @@ -4537,11 +4640,33 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" 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", @@ -4581,6 +4706,7 @@ dependencies = [ "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", ] @@ -4962,9 +5088,9 @@ dependencies = [ [[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" @@ -4984,11 +5110,12 @@ dependencies = [ [[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]] @@ -5035,6 +5162,18 @@ dependencies = [ "serde", ] +[[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" @@ -5068,7 +5207,7 @@ dependencies = [ "percent-encoding", "ryu", "serde", - "thiserror 2.0.16", + "thiserror 2.0.17", ] [[package]] @@ -5156,6 +5295,17 @@ dependencies = [ "unsafe-libyaml", ] +[[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" @@ -5220,7 +5370,7 @@ checksum = "297f631f50729c8c99b84667867963997ec0b50f32b2a7dbcab828ef0541e8bb" dependencies = [ "num-bigint", "num-traits", - "thiserror 2.0.16", + "thiserror 2.0.17", "time", ] @@ -5262,7 +5412,7 @@ dependencies = [ "rustls 0.23.31", "serde", "serde_json", - "thiserror 2.0.16", + "thiserror 2.0.17", "time", "tokio", "tokio-util", @@ -5355,25 +5505,32 @@ dependencies = [ 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", @@ -5506,9 +5663,9 @@ checksum = "f764005d11ee5f36500a149ace24e00e3da98b0158b3e2d53a7495660d3f4d60" [[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", @@ -5539,11 +5696,11 @@ dependencies = [ [[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]] @@ -5559,9 +5716,9 @@ dependencies = [ [[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", @@ -5993,7 +6150,7 @@ dependencies = [ "native-tls", "rand 0.9.1", "sha1", - "thiserror 2.0.16", + "thiserror 2.0.17", "url", "utf-8", ] @@ -6011,7 +6168,7 @@ dependencies = [ "log", "rand 0.9.1", "sha1", - "thiserror 2.0.16", + "thiserror 2.0.17", "utf-8", ] @@ -6054,7 +6211,7 @@ dependencies = [ "http", "jetstream", "log", - "lsm-tree", + "lsm-tree 2.10.4", "metrics", "metrics-exporter-prometheus 0.17.2", "schemars", @@ -6064,7 +6221,7 @@ dependencies = [ "serde_qs", "sha2", "tempfile", - "thiserror 2.0.16", + "thiserror 2.0.17", "tikv-jemallocator", "tokio", "tokio-util", @@ -6115,6 +6272,12 @@ version = "0.2.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861" +[[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" @@ -6193,7 +6356,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "62fc7c4ce161f049607ecea654dca3f2d727da5371ae85e2e4f14ce2b98ed67c" dependencies = [ "byteorder", - "byteview", + "byteview 0.6.1", "interval-heap", "log", "path-absolutize", @@ -6341,6 +6504,19 @@ dependencies = [ "unicode-ident", ] +[[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" @@ -6400,7 +6576,7 @@ dependencies = [ "reqwest", "serde", "serde_json", - "thiserror 2.0.16", + "thiserror 2.0.17", "tokio", "tokio-util", "url", @@ -6758,7 +6934,7 @@ dependencies = [ "nom", "oid-registry", "rusticata-macros", - "thiserror 2.0.16", + "thiserror 2.0.17", "time", ] diff --git a/spacedust/Cargo.toml b/spacedust/Cargo.toml index d30a08c..be97208 100644 --- a/spacedust/Cargo.toml +++ b/spacedust/Cargo.toml @@ -4,22 +4,29 @@ version = "0.1.0" 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" diff --git a/spacedust/src/bin/import_car_file.rs b/spacedust/src/bin/import_car_file.rs new file mode 100644 index 0000000..8ec55b6 --- /dev/null +++ b/spacedust/src/bin/import_car_file.rs @@ -0,0 +1,24 @@ +use clap::Parser; +use std::path::PathBuf; +use spacedust::storage::car; + +type Result = std::result::Result>; + +#[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(()) +} diff --git a/spacedust/src/bin/import_scraped.rs b/spacedust/src/bin/import_scraped.rs new file mode 100644 index 0000000..a901ded --- /dev/null +++ b/spacedust/src/bin/import_scraped.rs @@ -0,0 +1,194 @@ +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 = anyhow::Result; //std::result::Result>; + + +#[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>) -> 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( + f: R, + disk_tx: &async_channel::Sender>, +) -> Result> { + 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( + car_rx: async_channel::Receiver, + disk_tx: async_channel::Sender>, + n: Arc, +) -> 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( + needed: NeedDisk, + 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( + worker_id: usize, + disk_rx: async_channel::Receiver>, + folder: PathBuf, + n: Arc, + disk_workers_active: Arc, +) -> 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::>::new(); + + + let (cars_tx, cars_rx) = async_channel::bounded(2); + set.spawn(get_cars(cars_folder, cars_tx)); + + let n: Arc = Arc::new(0.into()); + let disk_workers_active: Arc = 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(()) +} diff --git a/spacedust/src/bin/scrape_pds.rs b/spacedust/src/bin/scrape_pds.rs new file mode 100644 index 0000000..af94fef --- /dev/null +++ b/spacedust/src/bin/scrape_pds.rs @@ -0,0 +1,138 @@ +use tokio::io::AsyncWriteExt; +use clap::Parser; +use std::path::PathBuf; +use reqwest::Url; +use tokio::{sync::mpsc, time}; +use serde::Deserialize; + +type Result = std::result::Result>; + +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, + repos: Vec, +} + +fn get_pds_dids(client: reqwest::Client, mut pds: Url) -> mpsc::Receiver { + 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(()) +} diff --git a/spacedust/src/lib.rs b/spacedust/src/lib.rs index 915f061..a25093e 100644 --- a/spacedust/src/lib.rs +++ b/spacedust/src/lib.rs @@ -3,6 +3,7 @@ pub mod delay; pub mod error; pub mod removable_delay_queue; pub mod server; +pub mod storage; pub mod subscriber; use jetstream::events::CommitEvent; diff --git a/spacedust/src/storage/car/drive.rs b/spacedust/src/storage/car/drive.rs new file mode 100644 index 0000000..e69de29 diff --git a/spacedust/src/storage/car/mod.rs b/spacedust/src/storage/car/mod.rs new file mode 100644 index 0000000..e69de29 diff --git a/spacedust/src/storage/car/walk.rs b/spacedust/src/storage/car/walk.rs new file mode 100644 index 0000000..e69de29 diff --git a/spacedust/src/storage/fjall/mod.rs b/spacedust/src/storage/fjall/mod.rs new file mode 100644 index 0000000..58b8e82 --- /dev/null +++ b/spacedust/src/storage/fjall/mod.rs @@ -0,0 +1,8 @@ +use crate::storage::Storage; + +struct FjallStorage { +} + +impl Storage for FjallStorage { + fn import_car() { todo!() } +} diff --git a/spacedust/src/storage/mod.rs b/spacedust/src/storage/mod.rs new file mode 100644 index 0000000..b5dc74b --- /dev/null +++ b/spacedust/src/storage/mod.rs @@ -0,0 +1,8 @@ +pub mod car; +pub mod fjall; + +pub trait Storage { + fn import_car() { + + } +} -- 2.51.2