diff --git a/.github/workflows/qc.yml b/.github/workflows/qc.yml index 6d1c899..da7cd2e 100644 --- a/.github/workflows/qc.yml +++ b/.github/workflows/qc.yml @@ -89,8 +89,8 @@ jobs: flags: "" - name: all-features flags: "--all-features" - - name: no-default-features - flags: "--no-default-features" + - name: hf-bucket + flags: "--features hf-bucket" steps: - uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4.3.1 - name: Setup Rust diff --git a/.gitignore b/.gitignore index e693806..3d2d0bc 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,4 @@ /target spool-test +spool-test-bucket .DS_Store diff --git a/CHANGELOG.md b/CHANGELOG.md index c727b7c..f79c806 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- *(hf-bucket)* `HfBucketSink`: terminal sink writing parquet/json/csv + windows to a Hugging Face storage bucket. Uploads go through Xet + (`hf-xet` crate) with chunk deduplication, then register via the + sans-IO `BatchAction` (`POST /api/buckets/{bucket}/batch`). Also adds + `CreateBucketAction`. Opt-in feature `hf-bucket`; requires Rust 1.89+ + (transitive `redb` MSRV). +- *(hf-bucket)* `nea_weather_bucket` example: the reference NEA pipelines + writing to a storage bucket; collectors, records, and config plumbing + are now shared with the dataset example via `examples/common/mod.rs`. + ## [0.4.0](https://github.com/InfiniteUnion/meathook-rs/compare/v0.3.0...v0.4.0) - 2026-07-13 ### Added diff --git a/Cargo.lock b/Cargo.lock index 2b0407a..6730690 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -14,7 +14,7 @@ version = "0.8.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", "const-random", "getrandom 0.3.4", "once_cell", @@ -61,6 +61,21 @@ version = "1.0.102" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +[[package]] +name = "approx" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cab112f0a86d568ea0e627cc1d6be74a1e9cd55214684db5561995f6dad897c6" +dependencies = [ + "num-traits", +] + +[[package]] +name = "arrayvec" +version = "0.7.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3fb67a6e08acf24fdeccbac2cb6ac4305825bd1f117462e0e6f2f193345ad56" + [[package]] name = "arrow" version = "58.3.0" @@ -277,6 +292,17 @@ dependencies = [ "regex-syntax", ] +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "atoi" version = "2.0.0" @@ -332,6 +358,28 @@ version = "2.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b4388bee8683e3d04af747c73422af53102d2bd24d9eadb6cbc100baef4b43f8" +[[package]] +name = "blake3" +version = "1.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6d9e454fc11f76977dc803893aff6304ed33d6a26efae8696573bea74baa27ae" +dependencies = [ + "arrayvec", + "cc", + "cfg-if 1.0.4", + "constant_time_eq", + "cpufeatures", +] + +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + [[package]] name = "brotli" version = "8.0.3" @@ -353,6 +401,17 @@ dependencies = [ "alloc-stdlib", ] +[[package]] +name = "bstr" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6bb31b46c14244e20ee9984b11bf5c992b91fb6939fea616e3512c8baecdbe5f" +dependencies = [ + "memchr", + "regex-automata", + "serde_core", +] + [[package]] name = "bumpalo" version = "3.20.3" @@ -376,7 +435,7 @@ checksum = "f9abbd1bc6865053c427f7198e6af43bfdedc55ab791faed4fbd361d789575ff" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -403,6 +462,12 @@ dependencies = [ "shlex", ] +[[package]] +name = "cfg-if" +version = "0.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4785bdd1c96b2a846b2bd7cc02e86b6b3dbf14e7e53446c4f54c92a361040822" + [[package]] name = "cfg-if" version = "1.0.4" @@ -415,6 +480,17 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if 1.0.4", + "cpufeatures", + "rand_core 0.10.1", +] + [[package]] name = "chrono" version = "0.4.45" @@ -422,7 +498,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1aa79e62e7697b8e29b513a68abacf485adcd1fe8284a4316c5ae868e6633327" dependencies = [ "iana-time-zone", + "js-sys", "num-traits", + "wasm-bindgen", "windows-link", ] @@ -435,6 +513,15 @@ dependencies = [ "cc", ] +[[package]] +name = "colored" +version = "3.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "combine" version = "4.6.7" @@ -445,6 +532,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + [[package]] name = "const-random" version = "0.1.18" @@ -465,6 +558,27 @@ dependencies = [ "tiny-keccak", ] +[[package]] +name = "const-str" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18f12cc9948ed9604230cdddc7c86e270f9401ccbe3c2e98a4378c5e7632212f" + +[[package]] +name = "const_panic" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9603f79528ece8163c496f8932121cb36cfe46259e9c907bb3d8205139d7caa3" +dependencies = [ + "typewit", +] + +[[package]] +name = "constant_time_eq" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d52eff69cd5e647efe296129160853a42795992097e8af39800e1060caeea9b" + [[package]] name = "convert_case" version = "0.11.0" @@ -474,6 +588,16 @@ dependencies = [ "unicode-segmentation", ] +[[package]] +name = "core-foundation" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "core-foundation" version = "0.10.1" @@ -490,12 +614,63 @@ version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" +[[package]] +name = "countio" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9702aee5d1d744c01d82f6915644f950f898e014903385464c773b96fefdecb" +dependencies = [ + "futures-io", +] + +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if 1.0.4", +] + +[[package]] +name = "crossbeam-channel" +version = "0.5.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d85363c37faeca707aef026efa9f3b34d077bce547e48f770770625c6013679e" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61803da095bee82a81bb1a452ecc25d3b2f1416d1897eb86430c6159ef717c17" + [[package]] name = "crunchy" version = "0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "460fbee9c2c2f33933d720630a6a0bac33ba7053db5344fac858d4b8952d77d5" +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + [[package]] name = "csv" version = "1.4.0" @@ -517,15 +692,58 @@ dependencies = [ "memchr", ] +[[package]] +name = "ctor" +version = "1.0.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "914a755b7c2d4af2bdcff7ce1739e2db9a1b81a9b07123d8015786ae03c0980d" +dependencies = [ + "link-section", + "linktime-proc-macro", +] + [[package]] name = "deranged" version = "0.5.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7cd812cc2bc1d69d4764bd80df88b4317eaef9e773c75226407d9bc0876b211c" dependencies = [ + "powerfmt", "serde_core", ] +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer", + "const-oid", + "crypto-common", +] + +[[package]] +name = "dirs" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3e8aa94d75141228480295a7d0e7feb620b1a5ad9f12bc40be62411e38cce4e" +dependencies = [ + "dirs-sys", +] + +[[package]] +name = "dirs-sys" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e01a3366d27ee9890022452ee61b2b63a67e6f13f58900b651ff5665f0bb1fab" +dependencies = [ + "libc", + "option-ext", + "redox_users", + "windows-sys 0.61.2", +] + [[package]] name = "displaydoc" version = "0.2.6" @@ -534,7 +752,7 @@ checksum = "1ac70aa55017e108007fbaf5aa0f54b021c98f92ff8af59d42eda9da96e3dd4f" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -543,6 +761,12 @@ version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" +[[package]] +name = "either" +version = "1.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "252afb9ae5eaa683babdc6a068b3f5726eb19e05070c731f9b2a23a7c3e8ed34" + [[package]] name = "equivalent" version = "1.0.2" @@ -624,6 +848,21 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "futures" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b147ee9d1f6d097cef9ce628cd2ee62288d963e16fb287bd9286455b241382d" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + [[package]] name = "futures-channel" version = "0.3.32" @@ -631,6 +870,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "07bbe89c50d7a535e539b8c17bc0b49bdb77747034daa8087407d655f3f7cc1d" dependencies = [ "futures-core", + "futures-sink", ] [[package]] @@ -639,6 +879,34 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7e3450815272ef58cec6d564423f6e755e25379b217b0bc688e295ba24df6b1d" +[[package]] +name = "futures-executor" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "baf29c38818342a3b26b5b923639e7b1f4a61fc5e76102d4b1981c6dc7a7579d" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.32" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "futures-sink" version = "0.3.32" @@ -657,22 +925,36 @@ version = "0.3.32" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "389ca41296e6190b48053de0321d02a77f32f8a5d2461dd38762c0593805c6d6" dependencies = [ + "futures-channel", "futures-core", + "futures-io", + "futures-macro", + "futures-sink", "futures-task", + "memchr", "pin-project-lite", "slab", ] +[[package]] +name = "gearhash" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c8cf82cf76cd16485e56295a1377c775ce708c9f1a0be6b029076d60a245d213" +dependencies = [ + "cfg-if 0.1.10", +] + [[package]] name = "getrandom" version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", "js-sys", "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "wasm-bindgen", ] @@ -682,7 +964,7 @@ version = "0.3.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", "js-sys", "libc", "r-efi 5.3.0", @@ -696,11 +978,34 @@ version = "0.4.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0de51e6874e94e7bf76d726fc5d13ba782deca734ff60d5bb2fb2607c7406555" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", + "js-sys", "libc", "r-efi 6.0.0", + "rand_core 0.10.1", "wasip2", "wasip3", + "wasm-bindgen", +] + +[[package]] +name = "git-version" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1ad568aa3db0fcbc81f2f116137f263d7304f512a1209b35b85150d3ef88ad19" +dependencies = [ + "git-version-macro", +] + +[[package]] +name = "git-version-macro" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53010ccb100b96a67bc32c0175f0ed1426b31b655d562898e57325f81c023ac0" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", ] [[package]] @@ -729,7 +1034,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ea2d84b969582b4b1864a92dc5d27cd2b77b622a8d79306834f1be5ba20d84b" dependencies = [ "bytemuck", - "cfg-if", + "cfg-if 1.0.4", "crunchy", "num-traits", "zerocopy", @@ -750,12 +1055,43 @@ version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" +[[package]] +name = "heapify" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0049b265b7f201ca9ab25475b22b47fe444060126a51abe00f77d986fc5cc52e" + [[package]] name = "heck" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hf-xet" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c237ef4fb0ce1962a5117f8bd8c74454b41629826a9df17d14a1840ca18f0754" +dependencies = [ + "anyhow", + "async-trait", + "bytes", + "http", + "more-asserts", + "serde", + "serde_json", + "thiserror", + "tokio", + "tokio-util", + "tokio_with_wasm", + "tracing", + "uuid", + "xet-client", + "xet-core-structures", + "xet-data", + "xet-runtime", +] + [[package]] name = "http" version = "1.4.2" @@ -811,6 +1147,15 @@ dependencies = [ "serde", ] +[[package]] +name = "hybrid-array" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "707114b52a152fa7bdb290cd7cd5912d9467273b6d74e21b8d81aca1f8533f6b" +dependencies = [ + "typenum", +] + [[package]] name = "hyper" version = "1.10.1" @@ -865,9 +1210,11 @@ dependencies = [ "percent-encoding", "pin-project-lite", "socket2", + "system-configuration", "tokio", "tower-service", "tracing", + "windows-registry", ] [[package]] @@ -1027,6 +1374,15 @@ version = "2.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d98f6fed1fde3f8c21bc40a1abb88dd75e67924f9cffc3ef95607bad8017f8e2" +[[package]] +name = "itertools" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285" +dependencies = [ + "either", +] + [[package]] name = "itoa" version = "1.0.18" @@ -1039,7 +1395,7 @@ version = "0.22.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5efd9a482cf3a427f00d6b35f14332adc7902ce91efb778580e180ff90fa3498" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", "combine", "jni-macros", "jni-sys", @@ -1060,7 +1416,7 @@ dependencies = [ "quote", "rustc_version", "simd_cesu8", - "syn", + "syn 2.0.117", ] [[package]] @@ -1079,7 +1435,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" dependencies = [ "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1098,7 +1454,7 @@ version = "0.3.100" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f2025f20d7a4fa7785846e7b63d10a76d3f1cee98ee5cb79ea59703f95e42162" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", "futures-util", "wasm-bindgen", ] @@ -1121,9 +1477,26 @@ dependencies = [ "convert_case", "proc-macro2", "quote", - "syn", + "syn 2.0.117", +] + +[[package]] +name = "konst" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f660d5f887e3562f9ab6f4a14988795b694099d66b4f5dedc02d197ba9becb1d" +dependencies = [ + "const_panic", + "konst_proc_macros", + "typewit", ] +[[package]] +name = "konst_proc_macros" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e037a2e1d8d5fdbd49b16a4ea09d5d6401c1f29eca5ff29d03d3824dba16256a" + [[package]] name = "lazy_static" version = "1.5.0" @@ -1205,6 +1578,27 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "libredox" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "28d0a00925a9f930d679b6789b721e3a7f9ed110f41b86d2497caa780c3a070a" +dependencies = [ + "libc", +] + +[[package]] +name = "link-section" +version = "0.19.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "39c29a617ce3df32c08497bdc1ab6e2376e0b17948ac166a2fbe5977c5954cd9" + +[[package]] +name = "linktime-proc-macro" +version = "0.2.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e57c38c1e860fd37c604281cdfb1dd2216977fd76a50f85ba2f388ef3219616" + [[package]] name = "linux-raw-sys" version = "0.12.1" @@ -1280,9 +1674,11 @@ dependencies = [ "base64", "bytes", "csv", + "hf-xet", "http", "humantime-serde", "nea-rs", + "parking_lot", "parquet", "reqwest", "satay-reqwest", @@ -1323,17 +1719,23 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" dependencies = [ "libc", - "wasi", + "wasi 0.11.1+wasi-snapshot-preview1", "windows-sys 0.61.2", ] [[package]] -name = "nea-rs" -version = "0.2.0" +name = "more-asserts" +version = "0.3.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94dfc83d3487e6a3d48c67469d008cb48167008442eacb3492e766ae4e9d9ea0" -dependencies = [ - "http", +checksum = "1fafa6961cabd9c63bcd77a45d7e3b7f3b552b70417831fb0f56db717e72407e" + +[[package]] +name = "nea-rs" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e685262d0633cfc69a8284dd292435d47714951f58a03465b3e0dc3b3b053ca1" +dependencies = [ + "http", "nutype", "regex", "satay-runtime", @@ -1341,6 +1743,15 @@ dependencies = [ "serde_json", ] +[[package]] +name = "ntapi" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3b335231dfd352ffb0f8017f3b6027a4917f7df785ea2143d8af2adc66980ae" +dependencies = [ + "winapi", +] + [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -1409,28 +1820,68 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48a6a6ddc63e6d0cd6b77af9aabae24601b39ad543d29bff88c4761c1fc34700" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", "kinded", "proc-macro2", "quote", "regex", "rustc_version", - "syn", + "syn 2.0.117", "urlencoding", ] +[[package]] +name = "objc2-core-foundation" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" +dependencies = [ + "bitflags", +] + +[[package]] +name = "objc2-io-kit" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "33fafba39597d6dc1fb709123dfa8289d39406734be322956a69f0931c73bb15" +dependencies = [ + "libc", + "objc2-core-foundation", +] + +[[package]] +name = "objc2-system-configuration" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7216bd11cbda54ccabcab84d523dc93b858ec75ecfb3a7d89513fa22464da396" +dependencies = [ + "objc2-core-foundation", +] + [[package]] name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +[[package]] +name = "oneshot" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "269bca4c2591a28585d6bf10d9ed0332b7d76900a1b02bec41bdc3a2cdcda107" + [[package]] name = "openssl-probe" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe" +[[package]] +name = "option-ext" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" + [[package]] name = "ordered-float" version = "2.10.1" @@ -1440,6 +1891,15 @@ dependencies = [ "num-traits", ] +[[package]] +name = "os_str_bytes" +version = "6.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e2355d85b9a3786f481747ced0e0ff2ba35213a1f9bd406ed906554d7af805a1" +dependencies = [ + "memchr", +] + [[package]] name = "parking_lot" version = "0.12.5" @@ -1456,7 +1916,7 @@ version = "0.9.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", "libc", "redox_syscall", "smallvec", @@ -1508,6 +1968,26 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "pin-project" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2466b2336ed02bcdca6b294417127b90ec92038d1d5c4fbeac971a922e0e0924" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c96395f0a926bc13b1c17622aaddda1ecb55d49c8f1bf9777e4d877800a43f8b" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "pin-project-lite" version = "0.2.17" @@ -1551,7 +2031,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" dependencies = [ "proc-macro2", - "syn", + "syn 2.0.117", ] [[package]] @@ -1593,7 +2073,7 @@ dependencies = [ "bytes", "getrandom 0.3.4", "lru-slab", - "rand", + "rand 0.9.4", "ring", "rustc-hash", "rustls", @@ -1647,7 +2127,18 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" dependencies = [ "rand_chacha", - "rand_core", + "rand_core 0.9.5", +] + +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.2", + "rand_core 0.10.1", ] [[package]] @@ -1657,7 +2148,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.9.5", ] [[package]] @@ -1669,6 +2160,21 @@ dependencies = [ "getrandom 0.3.4", ] +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "redb" +version = "3.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ba239c1c1693315d3cc0e601db3b3965543afbf48c41730fdca2f069f510f4a" +dependencies = [ + "libc", +] + [[package]] name = "redox_syscall" version = "0.5.18" @@ -1678,6 +2184,17 @@ dependencies = [ "bitflags", ] +[[package]] +name = "redox_users" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4e608c6638b9c18977b00b475ac1f28d14e84b27d8d42f70e0bf1e3dec127ac" +dependencies = [ + "getrandom 0.2.17", + "libredox", + "thiserror", +] + [[package]] name = "regex" version = "1.12.4" @@ -1716,6 +2233,7 @@ dependencies = [ "base64", "bytes", "futures-core", + "futures-util", "h2", "http", "http-body", @@ -1731,18 +2249,36 @@ dependencies = [ "rustls", "rustls-pki-types", "rustls-platform-verifier", + "serde", + "serde_json", "sync_wrapper", "tokio", "tokio-rustls", + "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", + "wasm-streams", "web-sys", ] +[[package]] +name = "reqwest-middleware" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "07bc3f1384cffa4f274dad2d4ddd73aed32fed8f786d96c6be8aa4e5fd3c3b58" +dependencies = [ + "anyhow", + "async-trait", + "http", + "reqwest", + "thiserror", + "tower-service", +] + [[package]] name = "ring" version = "0.17.14" @@ -1750,7 +2286,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" dependencies = [ "cc", - "cfg-if", + "cfg-if 1.0.4", "getrandom 0.2.17", "libc", "untrusted", @@ -1827,7 +2363,7 @@ version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "26d1e2536ce4f35f4846aa13bff16bd0ff40157cdb14cc056c7b14ba41233ba0" dependencies = [ - "core-foundation", + "core-foundation 0.10.1", "core-foundation-sys", "jni", "log", @@ -1872,6 +2408,12 @@ version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" +[[package]] +name = "safe-transmute" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3944826ff8fa8093089aba3acb4ef44b9446a99a16f3bf4e74af3f77d340ab7d" + [[package]] name = "same-file" version = "1.0.6" @@ -1883,9 +2425,9 @@ dependencies = [ [[package]] name = "satay-reqwest" -version = "0.4.0" +version = "0.16.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "de885e4fe3e7ffb8f1d1c8e9e10f706d9357ab99dde50d633f2140a08f052531" +checksum = "70f0655e46b54da98c263f69b5d94ab39515f4dea7cc4da452d8be72af337fd5" dependencies = [ "reqwest", "satay-runtime", @@ -1894,9 +2436,9 @@ dependencies = [ [[package]] name = "satay-runtime" -version = "0.4.0" +version = "0.16.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "948550184ec0ff3090acc0867c7b843d63ec4761a03f1c12bdecd43b895588d1" +checksum = "e1b1b284a4cad511a645521d9c12673fb02fcf505d23977aa63104fbe3a3378d" dependencies = [ "fast-float2", "http", @@ -1929,7 +2471,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" dependencies = [ "bitflags", - "core-foundation", + "core-foundation 0.10.1", "core-foundation-sys", "libc", "security-framework-sys", @@ -1999,7 +2541,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2015,6 +2557,17 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_repr" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8d3b1629de253c70a0508c3899572da79ca359fdab27c7920ff00406df418906" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.3", +] + [[package]] name = "serde_spanned" version = "1.1.1" @@ -2024,6 +2577,17 @@ dependencies = [ "serde_core", ] +[[package]] +name = "sha2" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" +dependencies = [ + "cfg-if 1.0.4", + "cpufeatures", + "digest", +] + [[package]] name = "sharded-slab" version = "0.1.7" @@ -2033,6 +2597,17 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shellexpand" +version = "3.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32824fab5e16e6c4d86dc1ba84489390419a39f97699852b66480bb87d297ed8" +dependencies = [ + "bstr", + "dirs", + "os_str_bytes", +] + [[package]] name = "shlex" version = "2.0.1" @@ -2105,12 +2680,34 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "static_assertions" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2eb9349b6444b326872e140eb1cf5e7c522154d69e7a0ffb0fb81c06b37543f" + +[[package]] +name = "statrs" +version = "0.18.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2a3fe7c28c6512e766b0874335db33c94ad7b8f9054228ae1c2abd47ce7d335e" +dependencies = [ + "approx", + "num-traits", +] + [[package]] name = "subtle" version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[package]] +name = "symlink" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a" + [[package]] name = "syn" version = "2.0.117" @@ -2122,6 +2719,17 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "syn" +version = "3.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53e9bae58849f64dfa4f5d5ae372c8341f7305f82a3868709269343628b659a3" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "sync_wrapper" version = "1.0.2" @@ -2139,7 +2747,42 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", +] + +[[package]] +name = "sysinfo" +version = "0.38.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92ab6a2f8bfe508deb3c6406578252e491d299cbbf3bc0529ecc3313aee4a52f" +dependencies = [ + "libc", + "memchr", + "ntapi", + "objc2-core-foundation", + "objc2-io-kit", + "windows", +] + +[[package]] +name = "system-configuration" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b" +dependencies = [ + "bitflags", + "core-foundation 0.9.4", + "system-configuration-sys", +] + +[[package]] +name = "system-configuration-sys" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e1d1b10ced5ca923a1fcb8d03e96b8d3268065d724548c0211415ff6ac6bac4" +dependencies = [ + "core-foundation-sys", + "libc", ] [[package]] @@ -2172,7 +2815,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2181,7 +2824,7 @@ version = "1.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", ] [[package]] @@ -2197,11 +2840,12 @@ dependencies = [ [[package]] name = "time" -version = "0.3.48" +version = "0.3.47" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fc1aa89044e7786ffb2ec017acb22cb7de5b0be46d0f21aea2b224b8561e5db2" +checksum = "743bd48c283afc0388f9b8827b976905fb217ad9e647fae3a379a9283c4def2c" dependencies = [ "deranged", + "itoa", "num-conv", "powerfmt", "serde_core", @@ -2211,15 +2855,15 @@ dependencies = [ [[package]] name = "time-core" -version = "0.1.9" +version = "0.1.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9e1c906769ad99c88eaa54e728060edef082f8e358ff32030cb7c7d315e81109" +checksum = "7694e1cfe791f8d31026952abf09c69ca6f6fa4e1a1229e18988f06a04a12dca" [[package]] name = "time-macros" -version = "0.2.28" +version = "0.2.27" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d3bfe86347f0cc659f586f01e26303ccd32418f26f30c7b0309b3ca3a07d695" +checksum = "2e70e4c5a0e0a8a4823ad65dfe1a6930e4f4d756dcd9dd7939022b5e8c501215" dependencies = [ "num-conv", "time-core", @@ -2284,7 +2928,18 @@ checksum = "385a6cb71ab9ab790c5fe8d67f1645e6c450a7ce006a33de03daa956cf70a496" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", +] + +[[package]] +name = "tokio-retry" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a129d95275ebf4c493ec53bf0f8cd95f5ac161bc4f381700809a54f595d4470" +dependencies = [ + "pin-project-lite", + "rand 0.10.2", + "tokio", ] [[package]] @@ -2310,6 +2965,30 @@ dependencies = [ "tokio", ] +[[package]] +name = "tokio_with_wasm" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34e40fbbbd95441133fe9483f522db15dbfd26dc636164ebd8f2dd28759a6aa6" +dependencies = [ + "js-sys", + "tokio", + "tokio_with_wasm_proc", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + +[[package]] +name = "tokio_with_wasm_proc" +version = "0.8.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d01145a2c788d6aae4cd653afec1e8332534d7d783d01897cefcafe4428de992" +dependencies = [ + "quote", + "syn 2.0.117", +] + [[package]] name = "toml" version = "0.9.12+spec-1.1.0" @@ -2336,18 +3015,18 @@ dependencies = [ [[package]] name = "toml_parser" -version = "1.1.2+spec-1.1.0" +version = "1.1.3+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526" +checksum = "1d38ac1cf9b95face32296c0a3ede1fdc270627c9d9c02a7274dd6d960dc4d56" dependencies = [ - "winnow 1.0.3", + "winnow 1.0.4", ] [[package]] name = "toml_writer" -version = "1.1.1+spec-1.1.0" +version = "1.1.2+spec-1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "756daf9b1013ebe47a8776667b466417e2d4c5679d441c26230efd9ef78692db" +checksum = "7d56353a2a665ad0f41a421187180aab746c8c325620617ad883a99a1cbe66d2" [[package]] name = "tower" @@ -2405,6 +3084,19 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-appender" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c" +dependencies = [ + "crossbeam-channel", + "symlink", + "thiserror", + "time", + "tracing-subscriber", +] + [[package]] name = "tracing-attributes" version = "0.1.31" @@ -2413,7 +3105,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2437,6 +3129,16 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "tracing-serde" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "704b1aeb7be0d0a84fc9828cae51dab5970fee5088f83d1dd7ee6f6246fc6ff1" +dependencies = [ + "serde", + "tracing-core", +] + [[package]] name = "tracing-subscriber" version = "0.3.23" @@ -2447,12 +3149,15 @@ dependencies = [ "nu-ansi-term", "once_cell", "regex-automata", + "serde", + "serde_json", "sharded-slab", "smallvec", "thread_local", "tracing", "tracing-core", "tracing-log", + "tracing-serde", ] [[package]] @@ -2467,6 +3172,18 @@ version = "2.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9ea3136b675547379c4bd395ca6b938e5ad3c3d20fad76e7fe85f9e0d011419c" +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + +[[package]] +name = "typewit" +version = "1.15.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "214ca0b2191785cbc06209b9ca1861e048e39b5ba33574b3cedd58363d5bb5f6" + [[package]] name = "unicode-ident" version = "1.0.24" @@ -2515,6 +3232,17 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" +[[package]] +name = "uuid" +version = "1.25.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f053576934f05a761a402421fbbe3d425d9366f75f978806a037b3ca481abecc" +dependencies = [ + "getrandom 0.4.2", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "valuable" version = "0.1.1" @@ -2552,6 +3280,15 @@ 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.3+wasi-0.2.9" @@ -2570,13 +3307,22 @@ dependencies = [ "wit-bindgen 0.51.0", ] +[[package]] +name = "wasite" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66fe902b4a6b8028a753d5424909b764ccf79b7a209eac9bf97e59cda9f71a42" +dependencies = [ + "wasi 0.14.7+wasi-0.2.4", +] + [[package]] name = "wasm-bindgen" version = "0.2.123" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a254a4b10c19a76f09a27640e7ffbf9bc30bf67e16a3bf28aaefa4920fe81563" dependencies = [ - "cfg-if", + "cfg-if 1.0.4", "once_cell", "rustversion", "wasm-bindgen-macro", @@ -2612,7 +3358,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.117", "wasm-bindgen-shared", ] @@ -2647,6 +3393,19 @@ dependencies = [ "wasmparser", ] +[[package]] +name = "wasm-streams" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" +dependencies = [ + "futures-util", + "js-sys", + "wasm-bindgen", + "wasm-bindgen-futures", + "web-sys", +] + [[package]] name = "wasmparser" version = "0.244.0" @@ -2688,6 +3447,35 @@ dependencies = [ "rustls-pki-types", ] +[[package]] +name = "whoami" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "626c4bac6755d76ffc12cb01b2eac751db1996b9e0041de9aa02c8c211ddc82c" +dependencies = [ + "libc", + "libredox", + "objc2-system-configuration", + "wasite", + "web-sys", +] + +[[package]] +name = "winapi" +version = "0.3.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" +dependencies = [ + "winapi-i686-pc-windows-gnu", + "winapi-x86_64-pc-windows-gnu", +] + +[[package]] +name = "winapi-i686-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" + [[package]] name = "winapi-util" version = "0.1.11" @@ -2697,6 +3485,33 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "winapi-x86_64-pc-windows-gnu" +version = "0.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" + +[[package]] +name = "windows" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580" +dependencies = [ + "windows-collections", + "windows-core", + "windows-future", + "windows-numerics", +] + +[[package]] +name = "windows-collections" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610" +dependencies = [ + "windows-core", +] + [[package]] name = "windows-core" version = "0.62.2" @@ -2710,6 +3525,17 @@ dependencies = [ "windows-strings", ] +[[package]] +name = "windows-future" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb" +dependencies = [ + "windows-core", + "windows-link", + "windows-threading", +] + [[package]] name = "windows-implement" version = "0.60.2" @@ -2718,7 +3544,7 @@ checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2729,7 +3555,7 @@ checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2738,6 +3564,27 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-numerics" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26" +dependencies = [ + "windows-core", + "windows-link", +] + +[[package]] +name = "windows-registry" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720" +dependencies = [ + "windows-link", + "windows-result", + "windows-strings", +] + [[package]] name = "windows-result" version = "0.4.1" @@ -2816,6 +3663,15 @@ dependencies = [ "windows_x86_64_msvc 0.53.1", ] +[[package]] +name = "windows-threading" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37" +dependencies = [ + "windows-link", +] + [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" @@ -2920,9 +3776,9 @@ checksum = "df79d97927682d2fd8adb29682d1140b343be4ac0f08fd68b7765d9c059d3945" [[package]] name = "winnow" -version = "1.0.3" +version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0592e1c9d151f854e6fd382574c3a0855250e1d9b2f99d9281c6e6391af352f1" +checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81" [[package]] name = "wit-bindgen" @@ -2960,7 +3816,7 @@ dependencies = [ "heck", "indexmap", "prettyplease", - "syn", + "syn 2.0.117", "wasm-metadata", "wit-bindgen-core", "wit-component", @@ -2976,7 +3832,7 @@ dependencies = [ "prettyplease", "proc-macro2", "quote", - "syn", + "syn 2.0.117", "wit-bindgen-core", "wit-bindgen-rust", ] @@ -3024,6 +3880,147 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "xet-client" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3b8da8cc70aa2e3c500c0400e012df82c656ab9fca47f9f939fffc5afd89aca" +dependencies = [ + "anyhow", + "async-trait", + "base64", + "bytes", + "crc32fast", + "futures", + "http", + "hyper", + "more-asserts", + "rand 0.10.2", + "redb", + "reqwest", + "reqwest-middleware", + "serde", + "serde_json", + "serde_repr", + "statrs", + "tempfile", + "thiserror", + "tokio", + "tokio-retry", + "tokio_with_wasm", + "tracing", + "url", + "urlencoding", + "web-time", + "xet-core-structures", + "xet-runtime", +] + +[[package]] +name = "xet-core-structures" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73503c223783dccc864abde22115e09d12f190448a0baf58ab2c54bc709e2f99" +dependencies = [ + "async-trait", + "base64", + "blake3", + "bytemuck", + "bytes", + "countio", + "futures", + "futures-util", + "getrandom 0.4.2", + "heapify", + "itertools", + "lz4_flex", + "more-asserts", + "rand 0.10.2", + "regex", + "safe-transmute", + "serde", + "static_assertions", + "thiserror", + "tokio", + "tokio-util", + "tracing", + "uuid", + "web-time", + "xet-runtime", +] + +[[package]] +name = "xet-data" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c89052ec5dec2187cad30b86af92cc24fd61c4a57a795f1ff7ff5f38d49184eb" +dependencies = [ + "anyhow", + "async-trait", + "bytes", + "chrono", + "gearhash", + "http", + "itertools", + "more-asserts", + "rand 0.10.2", + "serde", + "serde_json", + "sha2", + "tempfile", + "thiserror", + "tokio", + "tokio-util", + "tokio_with_wasm", + "tracing", + "url", + "uuid", + "web-time", + "xet-client", + "xet-core-structures", + "xet-runtime", +] + +[[package]] +name = "xet-runtime" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af5c60d5eed38ab4c576f4421bae835e7bd07631fb381705605529d2015c106b" +dependencies = [ + "anyhow", + "async-trait", + "bytes", + "chrono", + "colored", + "const-str", + "ctor", + "dirs", + "futures", + "git-version", + "humantime", + "konst", + "libc", + "more-asserts", + "oneshot", + "pin-project", + "rand 0.10.2", + "reqwest", + "serde", + "serde_json", + "shellexpand", + "sysinfo", + "thiserror", + "tokio", + "tokio-util", + "tokio_with_wasm", + "tracing", + "tracing-appender", + "tracing-subscriber", + "web-time", + "whoami", + "winapi", +] + [[package]] name = "yoke" version = "0.8.3" @@ -3043,7 +4040,7 @@ checksum = "de844c262c8848816172cef550288e7dc6c7b7814b4ee56b3e1553f275f1858e" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", "synstructure", ] @@ -3064,7 +4061,7 @@ checksum = "1ae7f38b72ec2a254e2b87ef277cf2cd4fb97cbebf944faa6f33354da0867930" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -3084,7 +4081,7 @@ checksum = "11532158c46691caf0f2593ea8358fed6bbf68a0315e80aae9bd41fbade684a1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", "synstructure", ] @@ -3124,7 +4121,7 @@ checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 3b7f5e1..8fc059a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -2,7 +2,7 @@ name = "meathook-rs" version = "0.4.0" edition = "2024" -rust-version = "1.88" +rust-version = "1.95" description = "A polling runtime with composable, durable sinks" license = "MIT OR Apache-2.0" readme = "README.md" @@ -24,6 +24,7 @@ parquet = ["dep:arrow", "dep:parquet", "dep:serde_arrow"] csv = ["dep:csv"] satay = ["dep:satay-runtime", "dep:satay-reqwest"] huggingface = ["parquet", "satay", "dep:base64"] +hf-bucket = ["huggingface", "dep:hf-xet"] [dependencies] http = "1" @@ -41,19 +42,26 @@ parquet = { version = "58", optional = true } serde_arrow = { version = "0.14.1", features = ["arrow-58"], optional = true } csv = { version = "1", optional = true } base64 = { version = "0.22", optional = true } -satay-runtime = { version = "0.4.0", features = ["json"], optional = true } -satay-reqwest = { version = "0.4.0", optional = true } +satay-runtime = { version = "0.16.2", features = ["json"], optional = true } +satay-reqwest = { version = "0.16.2", optional = true } +hf-xet = { version = "1.6", optional = true } [dev-dependencies] anyhow = "1" bytes = "1" humantime-serde = "1" -nea-rs = "0.2.0" +nea-rs = "0.5.1" +parking_lot = "0.12" tempfile = "3" tokio = { version = "1", features = ["full", "test-util"] } toml = "0.9" tracing-subscriber = { version = "0.3", features = ["env-filter"] } +[[example]] +name = "nea_weather_bucket" +required-features = ["hf-bucket"] + + [[example]] name = "nea_weather" required-features = ["huggingface"] diff --git a/README.md b/README.md index 90bf9a3..3a844af 100644 --- a/README.md +++ b/README.md @@ -219,6 +219,40 @@ happens during shutdown, when every pipeline flushes its last window at once. Set a request timeout on gated clients so a stalled upload cannot hold the gate forever. +## Hugging Face bucket sink + +`HfBucketSink` (feature `hf-bucket`) writes windows to a Hugging Face +storage bucket — `hf://buckets/{namespace}/{bucket}` — instead of a +dataset repo. Buckets have no git layer: no commit queue, no file-count +limits, and Xet chunk deduplication across the whole bucket. + +```rust,ignore +let terminal = HfBucketSink::new(client, "you/your-bucket", token) + .expect("xet session")?; +let sink = SinkStack::new() + .tier(JsonlStore::new("/var/lib/meathook/spool/air_temperature"), FlushPolicy::hourly()) + .terminal(terminal); +``` + +Each window still lands at the same deterministic path +(`data/{pipeline}/{date}/{time}-{fingerprint}.parquet`), uploaded through +Xet and then registered with a single sans-IO `BatchAction` +(`POST /api/buckets/{bucket}/batch`). The upload leg uses the official +`hf-xet` crate — the same client `huggingface_hub` uses — so replays of +identical bytes transfer almost nothing. There is no `CommitGate` +equivalent: buckets accept writes without a per-repository queue. + +The bucket must exist before the first write; create it with +`CreateBucketAction` (409 decodes to `AlreadyExists`) or `hf buckets +create you/your-bucket --private`. + +[`examples/nea_weather_bucket.rs`](examples/nea_weather_bucket.rs) runs +the same three NEA pipelines as the dataset example, sharing collectors, +records, and config plumbing through `examples/common/mod.rs` — only the +terminal sink and its config differ. Build it with +`cargo run --features hf-bucket --example nea_weather_bucket -- +examples/meathook_bucket.toml`. + ## Feature flags | Feature | Default | Implies | Adds | @@ -227,6 +261,7 @@ forever. | `csv` | No | Nothing | `CsvEncoder` | | `satay` | No | Nothing | `SatayCollector` for satay-generated API clients | | `huggingface` | Yes | `parquet`, `satay` | `HfSink` and the sans-IO `CommitAction` | +| `hf-bucket` | No | `huggingface` | `HfBucketSink` for [storage buckets](https://huggingface.co/storage), plus `CreateBucketAction`. Requires Rust 1.89+: the transitive `redb` dependency of `hf-xet` declares MSRV 1.89 | With `--no-default-features`, the crate still provides the core traits, sink combinators, JSONL spool, supervisor, `Encoder`, and `JsonEncoder`. The Satay @@ -293,6 +328,11 @@ cargo test # Timing, spool recovery, Parquet round trips, HF_TOKEN=hf_... MEATHOOK_TEST_REPO=you/meathook-test \ cargo test --test hf_integration -- --ignored # Commit to a scratch repository. + +HF_TOKEN=hf_... MEATHOOK_TEST_BUCKET=you/meathook-test \ + cargo test --features hf-bucket --test hf_bucket_integration -- --ignored + # Write to a scratch storage bucket (created + # if missing; delete with `hf buckets delete`). ``` ## Security diff --git a/examples/common/mod.rs b/examples/common/mod.rs new file mode 100644 index 0000000..ab23159 --- /dev/null +++ b/examples/common/mod.rs @@ -0,0 +1,280 @@ +//! Wiring shared by the `nea_weather*` examples: NEA collectors, row-shaped +//! records, TOML config plumbing, tracing setup, and the two-tier buffering +//! stack (`Tier(MemStore) → Tier(JsonlStore)`) each example finishes with +//! its own terminal sink (`HfSink` for dataset repos, `HfBucketSink` for +//! storage buckets). +//! +//! This file is a module of each example, not an example itself: cargo only +//! discovers `examples/.rs` and `examples//main.rs`. + +use std::collections::HashMap; +use std::path::PathBuf; +use std::time::Duration; + +use anyhow::Context as _; +use meathook::{ + Collector, FlushPolicy, JsonlStore, MemStore, SatayCollector, Sink, SinkStack, Tier, +}; +use nea_rs::{ + AirTemperatureOperationResponse, NeaReadingSnapshot, NeaWeatherStation, Pm25OperationResponse, + RainfallOperationResponse, +}; +use satay_reqwest::ReqwestActionExt as _; +use serde::de::DeserializeOwned; +use serde::{Deserialize, Serialize}; +use time::format_description::well_known::Rfc3339; +use tracing::warn; + +/// Outer memory tier: batch for five minutes or 10,000 records before the +/// records reach the fsynced JSONL tier. +pub const MEMORY_FLUSH_POLICY: FlushPolicy = FlushPolicy::new(Duration::from_secs(300), 10_000); + +/// The stack every `nea_weather*` example builds: +/// memory batching, durable spool, then the example's terminal sink `S`. +pub type TieredStack = Tier, Tier, S>>; + +/// Example configuration. `S` is the example-specific sink section +/// (`[sink.huggingface]` or `[sink.bucket]`). +#[derive(Debug, Deserialize)] +pub struct Config { + pub spool_dir: PathBuf, + pub flush: FlushConfig, + pub sink: S, + #[serde(default)] + pub collectors: HashMap, +} + +#[derive(Debug, Deserialize)] +pub struct FlushConfig { + #[serde(with = "humantime_serde")] + pub every: Duration, + pub max_records: usize, +} + +#[derive(Debug, Deserialize)] +pub struct CollectorConfig { + #[serde(with = "humantime_serde")] + pub interval: Duration, +} + +impl Config { + /// Configured poll interval for a collector, 1m when unlisted. + pub fn interval(&self, collector: &str) -> Duration { + self.collectors + .get(collector) + .map_or(Duration::from_secs(60), |c| c.interval) + } +} + +/// One station reading, flattened row-shape for parquet. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StationReading { + pub station_id: String, + pub station_name: String, + pub timestamp: String, + pub value: f64, +} + +/// One regional PM2.5 reading. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RegionReading { + pub region: String, + pub timestamp: String, + pub value: f64, +} + +fn fmt_ts(ts: time::OffsetDateTime) -> String { + ts.format(&Rfc3339).unwrap_or_else(|_| ts.to_string()) +} + +fn flatten_station_data( + stations: &[NeaWeatherStation], + readings: &[NeaReadingSnapshot], +) -> Vec { + readings + .iter() + .flat_map(|snapshot| { + let timestamp = fmt_ts(snapshot.timestamp); + snapshot.data.iter().filter_map(move |reading| { + let station = stations + .iter() + .find(|station| station.id == reading.station_id)?; + Some(StationReading { + station_id: station.id.clone().into(), + station_name: station.name.clone(), + timestamp: timestamp.clone(), + value: reading.value, + }) + }) + }) + .collect() +} + +fn flatten_air_temperature(response: AirTemperatureOperationResponse) -> Vec { + match response { + AirTemperatureOperationResponse::Ok(ok) => { + flatten_station_data(&ok.data.stations, &ok.data.readings) + } + other => { + warn!(?other, "air_temperature returned non-ok response"); + Vec::new() + } + } +} + +fn flatten_rainfall(response: RainfallOperationResponse) -> Vec { + match response { + RainfallOperationResponse::Ok(ok) => { + flatten_station_data(&ok.data.stations, &ok.data.readings) + } + other => { + warn!(?other, "rainfall returned non-ok response"); + Vec::new() + } + } +} + +fn flatten_pm25(response: Pm25OperationResponse) -> Vec { + match response { + Pm25OperationResponse::Ok(ok) => ok + .data + .items + .iter() + .flat_map(|item| { + let timestamp = fmt_ts(item.timestamp); + let regional = &item.readings.pm25_one_hourly; + [ + ("east", regional.east), + ("west", regional.west), + ("north", regional.north), + ("south", regional.south), + ("central", regional.central), + ] + .map(|(region, value)| RegionReading { + region: region.to_owned(), + timestamp: timestamp.clone(), + value: f64::from(u16::from(value)), + }) + }) + .collect(), + other => { + warn!(?other, "pm25 returned non-ok response"); + Vec::new() + } + } +} + +/// Wiring shared by every pipeline, independent of the terminal sink. +#[derive(Clone)] +pub struct Ctx { + pub client: reqwest::Client, + pub api_key: Option, + pub token: String, + pub spool_dir: PathBuf, + pub policy: FlushPolicy, +} + +impl Ctx { + pub fn from_config(config: &Config) -> anyhow::Result { + Ok(Self { + // The timeout matters beyond hygiene: a stalled upload holds + // sinks (and the dataset example's commit gate) until the OS + // abandons the connection. + client: reqwest::Client::builder() + .timeout(Duration::from_secs(60)) + .build() + .context("building http client")?, + api_key: std::env::var("X_API_KEY").ok(), + token: std::env::var("HF_TOKEN").context("HF_TOKEN must be set")?, + spool_dir: config.spool_dir.clone(), + policy: FlushPolicy::new(config.flush.every, config.flush.max_records), + }) + } + + pub fn api(&self) -> nea_rs::Api { + let api = nea_rs::Api::new(); + match &self.api_key { + Some(key) => api.x_api_key(key.clone()), + None => api, + } + } + + /// In-memory batching, durable spool, then `terminal`. + pub fn tiered(&self, pipeline: &str, terminal: S) -> TieredStack + where + R: Clone + Serialize + DeserializeOwned + Send + 'static, + S: Sink, + { + SinkStack::new() + .tier(MemStore::new(), MEMORY_FLUSH_POLICY) + .tier(JsonlStore::new(self.spool_dir.join(pipeline)), self.policy) + .terminal(terminal) + } +} + +pub fn init_tracing() { + tracing_subscriber::fmt() + .with_env_filter( + tracing_subscriber::EnvFilter::try_from_default_env() + .unwrap_or_else(|_| "info,meathook=debug".into()), + ) + .init(); +} + +pub fn load_config(config_path: &str) -> anyhow::Result> { + toml::from_str( + &std::fs::read_to_string(config_path) + .with_context(|| format!("reading config {config_path}"))?, + ) + .with_context(|| format!("parsing config {config_path}")) +} + +pub fn air_temperature_collector( + ctx: &Ctx, +) -> impl Collector + 'static + use<> { + let api = ctx.api(); + SatayCollector::new( + "air_temperature", + ctx.client.clone(), + move |client| { + let api = api.clone(); + async move { + api.weather_readings() + .air_temperature() + .send_with(&client) + .await + } + }, + flatten_air_temperature, + ) +} + +pub fn rainfall_collector( + ctx: &Ctx, +) -> impl Collector + 'static + use<> { + let api = ctx.api(); + SatayCollector::new( + "rainfall", + ctx.client.clone(), + move |client| { + let api = api.clone(); + async move { api.weather_readings().rainfall().send_with(&client).await } + }, + flatten_rainfall, + ) +} + +pub fn pm25_collector( + ctx: &Ctx, +) -> impl Collector + 'static + use<> { + let api = ctx.api(); + SatayCollector::new( + "pm25", + ctx.client.clone(), + move |client| { + let api = api.clone(); + async move { api.air_quality().pm25().send_with(&client).await } + }, + flatten_pm25, + ) +} diff --git a/examples/meathook_bucket.toml b/examples/meathook_bucket.toml new file mode 100644 index 0000000..b1a0534 --- /dev/null +++ b/examples/meathook_bucket.toml @@ -0,0 +1,28 @@ +# Bucket-sink variant of the example config. Intervals/windows use +# humantime strings. Secrets come from the environment, never from this +# file: +# HF_TOKEN - HuggingFace access token with write access to the bucket +# X_API_KEY - optional data.gov.sg API key (these endpoints work keyless) +# +# The bucket must exist before the first flush: +# hf buckets create zeon256/meathook-test --private + +spool_dir = "./spool-test-bucket" # PVC mount on k8s; separate from the + # dataset example's spool so the two can + # run side by side + +[flush] # FlushPolicy for each pipeline's durable JSONL tier +every = "10m" +max_records = 50_000 + +[sink.bucket] +id = "zeon256/meathook-test" + +[collectors.air_temperature] +interval = "1m" + +[collectors.rainfall] +interval = "5m" + +[collectors.pm25] +interval = "1h" diff --git a/examples/nea_weather.rs b/examples/nea_weather.rs index 16700d9..eccb92e 100644 --- a/examples/nea_weather.rs +++ b/examples/nea_weather.rs @@ -1,5 +1,6 @@ //! Reference meathook consumer: collects NEA (data.gov.sg) realtime weather -//! readings and ships configured parquet windows to a `HuggingFace` dataset. +//! readings and ships configured parquet windows to a `HuggingFace` dataset +//! repo. //! //! Each pipeline's stack is `Tier(MemStore) → Tier(JsonlStore) → HfSink`. //! The outer memory tier batches for five minutes or 10,000 records; its @@ -7,48 +8,22 @@ //! durable tier flushes windows to HF. Records still held in memory remain //! volatile; leftover JSONL segments replay on the next start. //! +//! Collectors, records, and config plumbing are shared with the bucket +//! variant in `examples/nea_weather_bucket.rs` via `examples/common/mod.rs`. +//! //! ```bash //! HF_TOKEN=hf_... cargo run --example nea_weather -- examples/meathook.toml //! ``` -use std::collections::HashMap; -use std::path::PathBuf; -use std::time::Duration; +#[path = "common/mod.rs"] +mod common; -use anyhow::Context as _; -use meathook::{ - CommitGate, FlushPolicy, HfSink, JsonlStore, Meathook, MemStore, Pipeline, SatayCollector, - SinkStack, Tier, -}; -use nea_rs::{ - AirTemperatureOperationResponse, NeaReadingSnapshot, NeaWeatherStation, Pm25OperationResponse, - RainfallOperationResponse, +use common::{ + Config, Ctx, air_temperature_collector, init_tracing, load_config, pm25_collector, + rainfall_collector, }; -use satay_reqwest::ReqwestActionExt as _; -use serde::de::DeserializeOwned; -use serde::{Deserialize, Serialize}; -use time::format_description::well_known::Rfc3339; -use tracing::warn; - -const MEMORY_FLUSH_POLICY: FlushPolicy = FlushPolicy::new(Duration::from_secs(300), 10_000); - -type WeatherSink = Tier, Tier, HfSink>>; - -#[derive(Debug, Deserialize)] -struct Config { - spool_dir: PathBuf, - flush: FlushConfig, - sink: SinkConfig, - #[serde(default)] - collectors: HashMap, -} - -#[derive(Debug, Deserialize)] -struct FlushConfig { - #[serde(with = "humantime_serde")] - every: Duration, - max_records: usize, -} +use meathook::{CommitGate, HfSink, Meathook, Pipeline}; +use serde::Deserialize; #[derive(Debug, Deserialize)] struct SinkConfig { @@ -66,194 +41,33 @@ fn default_branch() -> String { "main".to_owned() } -#[derive(Debug, Deserialize)] -struct CollectorConfig { - #[serde(with = "humantime_serde")] - interval: Duration, -} - -impl Config { - fn interval(&self, collector: &str) -> Duration { - self.collectors - .get(collector) - .map_or(Duration::from_secs(60), |c| c.interval) - } -} - -/// One station reading, flattened row-shape for parquet. -#[derive(Debug, Clone, Serialize, Deserialize)] -struct StationReading { - station_id: String, - station_name: String, - timestamp: String, - value: f64, -} - -/// One regional PM2.5 reading. -#[derive(Debug, Clone, Serialize, Deserialize)] -struct RegionReading { - region: String, - timestamp: String, - value: f64, -} - -fn fmt_ts(ts: time::OffsetDateTime) -> String { - ts.format(&Rfc3339).unwrap_or_else(|_| ts.to_string()) -} - -fn flatten_station_data( - stations: &[NeaWeatherStation], - readings: &[NeaReadingSnapshot], -) -> Vec { - let names: HashMap = stations - .iter() - .map(|s| (s.id.to_string(), s.name.as_str())) - .collect(); - let names = &names; - readings - .iter() - .flat_map(|snapshot| { - let timestamp = fmt_ts(snapshot.timestamp); - snapshot.data.iter().map(move |reading| { - let station_id = reading.station_id.to_string(); - StationReading { - station_name: names.get(&station_id).copied().unwrap_or("").to_owned(), - station_id, - timestamp: timestamp.clone(), - value: reading.value, - } - }) - }) - .collect() -} - -/// Shared wiring context cloned into every pipeline factory. +/// Terminal-sink wiring specific to the dataset example. #[derive(Clone)] -struct Ctx { - client: reqwest::Client, - api_key: Option, +struct SinkCtx { repo: String, branch: String, - token: String, - spool_dir: PathBuf, - policy: FlushPolicy, /// Shared by every pipeline's sink: commits to the one HF repo go out /// one at a time instead of racing into HF's per-repo commit queue. gate: CommitGate, } -impl Ctx { - fn api(&self) -> nea_rs::Api { - let api = nea_rs::Api::new(); - match &self.api_key { - Some(key) => api.x_api_key(key.clone()), - None => api, +impl SinkCtx { + fn from_config(config: &Config) -> Self { + Self { + repo: config.sink.huggingface.repo.clone(), + branch: config.sink.huggingface.branch.clone(), + gate: CommitGate::new(), } } - /// In-memory batching, durable spool, then the terminal HF sink. - fn tiered_hf(&self, pipeline: &str) -> WeatherSink + fn tiered(&self, ctx: &Ctx, pipeline: &str) -> common::TieredStack> where - R: Clone + Serialize + DeserializeOwned + Send + 'static, + R: Clone + serde::Serialize + serde::de::DeserializeOwned + Send + 'static, { - let terminal = HfSink::new(self.client.clone(), self.repo.clone(), self.token.clone()) + let terminal = HfSink::new(ctx.client.clone(), self.repo.clone(), ctx.token.clone()) .branch(self.branch.clone()) .gate(self.gate.clone()); - - SinkStack::new() - .tier(MemStore::new(), MEMORY_FLUSH_POLICY) - .tier(JsonlStore::new(self.spool_dir.join(pipeline)), self.policy) - .terminal(terminal) - } -} - -fn init_tracing() { - tracing_subscriber::fmt() - .with_env_filter( - tracing_subscriber::EnvFilter::try_from_default_env() - .unwrap_or_else(|_| "info,meathook=debug".into()), - ) - .init(); -} - -fn load_config(config_path: &str) -> anyhow::Result { - toml::from_str( - &std::fs::read_to_string(config_path) - .with_context(|| format!("reading config {config_path}"))?, - ) - .with_context(|| format!("parsing config {config_path}")) -} - -fn ctx_from_config(config: &Config) -> anyhow::Result { - Ok(Ctx { - // The timeout matters beyond hygiene: a stalled HF upload holds the - // commit gate's permit, blocking every other pipeline's commit - // until this one is abandoned. - client: reqwest::Client::builder() - .timeout(Duration::from_secs(60)) - .build() - .context("building http client")?, - api_key: std::env::var("X_API_KEY").ok(), - repo: config.sink.huggingface.repo.clone(), - branch: config.sink.huggingface.branch.clone(), - token: std::env::var("HF_TOKEN").context("HF_TOKEN must be set")?, - spool_dir: config.spool_dir.clone(), - policy: FlushPolicy::new(config.flush.every, config.flush.max_records), - gate: CommitGate::new(), - }) -} - -fn flatten_air_temperature(response: AirTemperatureOperationResponse) -> Vec { - match response { - AirTemperatureOperationResponse::Ok(ok) => { - flatten_station_data(&ok.data.stations, &ok.data.readings) - } - other => { - warn!(?other, "air_temperature returned non-ok response"); - Vec::new() - } - } -} - -fn flatten_rainfall(response: RainfallOperationResponse) -> Vec { - match response { - RainfallOperationResponse::Ok(ok) => { - flatten_station_data(&ok.data.stations, &ok.data.readings) - } - other => { - warn!(?other, "rainfall returned non-ok response"); - Vec::new() - } - } -} - -fn flatten_pm25(response: Pm25OperationResponse) -> Vec { - match response { - Pm25OperationResponse::Ok(ok) => ok - .data - .items - .iter() - .flat_map(|item| { - let timestamp = fmt_ts(item.timestamp); - let regional = &item.readings.pm25_one_hourly; - [ - ("east", regional.east), - ("west", regional.west), - ("north", regional.north), - ("south", regional.south), - ("central", regional.central), - ] - .map(|(region, value)| RegionReading { - region: region.to_owned(), - timestamp: timestamp.clone(), - value: f64::from(u16::from(value)), - }) - }) - .collect(), - other => { - warn!(?other, "pm25 returned non-ok response"); - Vec::new() - } + ctx.tiered(pipeline, terminal) } } @@ -264,63 +78,52 @@ async fn main() -> anyhow::Result<()> { let config_path = std::env::args() .nth(1) .unwrap_or_else(|| "examples/meathook.toml".to_owned()); - let config = load_config(&config_path)?; - let ctx = ctx_from_config(&config)?; + let config = load_config::(&config_path)?; + let ctx = Ctx::from_config(&config)?; + let sink = SinkCtx::from_config(&config); let air_temperature = { let ctx = ctx.clone(); + let sink = sink.clone(); let interval = config.interval("air_temperature"); move || { - let api = ctx.api(); - let collector = SatayCollector::new( - "air_temperature", - ctx.client.clone(), - move |client| { - let api = api.clone(); - async move { api.air_temperature().send_with(&client).await } - }, - flatten_air_temperature, - ); - Pipeline::new(collector, ctx.tiered_hf("air_temperature"), interval) - .with_key_fn(|r: &StationReading| (r.station_id.clone(), r.timestamp.clone())) + let collector = air_temperature_collector(&ctx); + Pipeline::new( + collector, + sink.tiered::(&ctx, "air_temperature"), + interval, + ) + .with_key_fn(|r: &common::StationReading| (r.station_id.clone(), r.timestamp.clone())) } }; let rainfall = { let ctx = ctx.clone(); + let sink = sink.clone(); let interval = config.interval("rainfall"); move || { - let api = ctx.api(); - let collector = SatayCollector::new( - "rainfall", - ctx.client.clone(), - move |client| { - let api = api.clone(); - async move { api.rainfall().send_with(&client).await } - }, - flatten_rainfall, - ); - Pipeline::new(collector, ctx.tiered_hf("rainfall"), interval) - .with_key_fn(|r: &StationReading| (r.station_id.clone(), r.timestamp.clone())) + let collector = rainfall_collector(&ctx); + Pipeline::new( + collector, + sink.tiered::(&ctx, "rainfall"), + interval, + ) + .with_key_fn(|r: &common::StationReading| (r.station_id.clone(), r.timestamp.clone())) } }; let pm25 = { let ctx = ctx.clone(); + let sink = sink.clone(); let interval = config.interval("pm25"); move || { - let api = ctx.api(); - let collector = SatayCollector::new( - "pm25", - ctx.client.clone(), - move |client| { - let api = api.clone(); - async move { api.pm25().send_with(&client).await } - }, - flatten_pm25, - ); - Pipeline::new(collector, ctx.tiered_hf("pm25"), interval) - .with_key_fn(|r: &RegionReading| (r.region.clone(), r.timestamp.clone())) + let collector = pm25_collector(&ctx); + Pipeline::new( + collector, + sink.tiered::(&ctx, "pm25"), + interval, + ) + .with_key_fn(|r: &common::RegionReading| (r.region.clone(), r.timestamp.clone())) } }; diff --git a/examples/nea_weather_bucket.rs b/examples/nea_weather_bucket.rs new file mode 100644 index 0000000..6e762e9 --- /dev/null +++ b/examples/nea_weather_bucket.rs @@ -0,0 +1,129 @@ +//! Bucket variant of the reference consumer: the same NEA collectors and +//! buffering tiers as `nea_weather`, but each window is written to a +//! Hugging Face storage bucket (`hf://buckets/{namespace}/{bucket}`) via +//! `HfBucketSink` instead of committed to a dataset repo. +//! +//! Each pipeline's stack is `Tier(MemStore) → Tier(JsonlStore) → +//! HfBucketSink`. Windows land at the same deterministic Hive-style paths, +//! uploaded through Xet with chunk deduplication, then registered with one +//! sans-IO `BatchAction`. Buckets have no commit queue, so there is no +//! `CommitGate` to share; the JSONL spool still owns custody between +//! collection and delivery. +//! +//! ```bash +//! HF_TOKEN=hf_... cargo run --example nea_weather_bucket -- examples/meathook_bucket.toml +//! ``` +//! +//! Requires the `hf-bucket` feature. + +#[path = "common/mod.rs"] +mod common; + +use common::{ + Config, Ctx, air_temperature_collector, init_tracing, load_config, pm25_collector, + rainfall_collector, +}; +use meathook::{HfBucketSink, Meathook, Pipeline}; +use serde::Deserialize; + +#[derive(Debug, Deserialize)] +struct SinkConfig { + bucket: BucketConfig, +} + +#[derive(Debug, Deserialize)] +struct BucketConfig { + /// `{namespace}/{bucket}`; must already exist — create it with + /// `CreateBucketAction` or `hf buckets create`. + id: String, +} + +#[derive(Clone)] +struct SinkCtx { + bucket: String, +} + +impl SinkCtx { + fn from_config(config: &Config) -> Self { + Self { + bucket: config.sink.bucket.id.clone(), + } + } + + fn tiered( + &self, + ctx: &Ctx, + pipeline: &str, + ) -> Result>, meathook::HfBucketSinkError> + where + R: Clone + serde::Serialize + serde::de::DeserializeOwned + Send + 'static, + { + let terminal = + HfBucketSink::new(ctx.client.clone(), self.bucket.clone(), ctx.token.clone())?; + Ok(ctx.tiered(pipeline, terminal)) + } +} + +#[tokio::main] +async fn main() -> anyhow::Result<()> { + init_tracing(); + + let config_path = std::env::args() + .nth(1) + .unwrap_or_else(|| "examples/meathook_bucket.toml".to_owned()); + let config = load_config::(&config_path)?; + let ctx = Ctx::from_config(&config)?; + let sink = SinkCtx::from_config(&config); + + let air_temperature = { + let ctx = ctx.clone(); + let sink = sink.clone(); + let interval = config.interval("air_temperature"); + move || { + let collector = air_temperature_collector(&ctx); + let stack = sink + .tiered::(&ctx, "air_temperature") + .expect("xet session"); + Pipeline::new(collector, stack, interval).with_key_fn(|r: &common::StationReading| { + (r.station_id.clone(), r.timestamp.clone()) + }) + } + }; + + let rainfall = { + let ctx = ctx.clone(); + let sink = sink.clone(); + let interval = config.interval("rainfall"); + move || { + let collector = rainfall_collector(&ctx); + let stack = sink + .tiered::(&ctx, "rainfall") + .expect("xet session"); + Pipeline::new(collector, stack, interval).with_key_fn(|r: &common::StationReading| { + (r.station_id.clone(), r.timestamp.clone()) + }) + } + }; + + let pm25 = { + let ctx = ctx.clone(); + let sink = sink.clone(); + let interval = config.interval("pm25"); + move || { + let collector = pm25_collector(&ctx); + let stack = sink + .tiered::(&ctx, "pm25") + .expect("xet session"); + Pipeline::new(collector, stack, interval) + .with_key_fn(|r: &common::RegionReading| (r.region.clone(), r.timestamp.clone())) + } + }; + + Meathook::builder() + .pipeline(air_temperature) + .pipeline(rainfall) + .pipeline(pm25) + .run() + .await?; + Ok(()) +} diff --git a/src/lib.rs b/src/lib.rs index e474c15..70e6776 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -19,7 +19,9 @@ //! buffering tiers ([`Tier`]), each backed by a pluggable [`Store`]: //! in-memory [`MemStore`], durable write-ahead [`JsonlStore`], or your own. //! Finish the stack with a terminal sink such as `HfSink` (feature -//! `huggingface`); fan out completed sinks with [`SinkExt::tee`]. +//! `huggingface`) or `HfBucketSink` for Hugging Face storage buckets +//! (feature `hf-bucket`); fan out completed sinks with +//! [`SinkExt::tee`]. //! //! Pipelines (collector + sink stack) are supervised by the [`Meathook`] //! runtime: one tokio task each, respawn-on-panic with backoff, and a final @@ -52,5 +54,10 @@ pub use encode::{CsvEncoder, CsvError}; pub use encode::{Encoder, JsonEncoder}; #[cfg(feature = "parquet")] pub use encode::{ParquetCompression, ParquetEncodeError, ParquetEncoder, Uncompressed, Zstd}; +#[cfg(feature = "hf-bucket")] +pub use sink::hf_bucket::{ + BatchAction, BatchOutcome, CreateBucketAction, CreateBucketOutcome, HfBucketSink, + HfBucketSinkError, +}; #[cfg(feature = "huggingface")] pub use sink::huggingface::{CommitGate, HfSink, HfSinkError}; diff --git a/src/pipeline.rs b/src/pipeline.rs index 237c0c4..3e3b5d4 100644 --- a/src/pipeline.rs +++ b/src/pipeline.rs @@ -182,9 +182,11 @@ mod tests { "fake" } - async fn collect(&mut self) -> Result, Infallible> { + fn collect( + &mut self, + ) -> impl Future, Infallible>> + Send { let tick = self.ticks.fetch_add(1, Ordering::SeqCst); - Ok(vec![(tick, 0), (tick + 1, 0)]) + std::future::ready(Ok(vec![(tick, 0), (tick + 1, 0)])) } } @@ -232,13 +234,13 @@ mod tests { "flaky" } - async fn collect(&mut self) -> Result, FlakyError> { + fn collect(&mut self) -> impl Future, FlakyError>> + Send { let call = self.calls.fetch_add(1, Ordering::SeqCst); - if call.is_multiple_of(2) { + std::future::ready(if call.is_multiple_of(2) { Err(FlakyError) } else { Ok(vec![call]) - } + }) } } diff --git a/src/runtime.rs b/src/runtime.rs index e2898dc..154e16e 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -260,10 +260,10 @@ mod tests { "panics-once" } - async fn collect(&mut self) -> Result, Infallible> { + fn collect(&mut self) -> impl Future, Infallible>> + Send { let call = self.global_calls.fetch_add(1, Ordering::SeqCst); assert!(call != 0, "boom"); - Ok(vec![call]) + std::future::ready(Ok(vec![call])) } } diff --git a/src/sink.rs b/src/sink.rs index d0505f3..c4328ca 100644 --- a/src/sink.rs +++ b/src/sink.rs @@ -5,6 +5,9 @@ use std::future::Future; use time::OffsetDateTime; +#[cfg(feature = "hf-bucket")] +pub mod hf_bucket; + #[cfg(feature = "huggingface")] pub mod huggingface; @@ -40,3 +43,37 @@ pub trait Sink: Send { /// flush, startup recovery). fn flush(&mut self) -> impl Future> + Send; } + +/// FNV-1a 64-bit — stable across releases (the fingerprint is load-bearing +/// for replay idempotency: same bytes must map to the same path forever). +pub(crate) fn fingerprint(bytes: &[u8]) -> u64 { + const OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325; + const PRIME: u64 = 0x0100_0000_01b3; + bytes.iter().fold(OFFSET_BASIS, |hash, &byte| { + (hash ^ u64::from(byte)).wrapping_mul(PRIME) + }) +} + +/// Object path keyed by the full window start plus a fingerprint of the +/// encoded bytes. The start (to the second) identifies the window — distinct +/// windows get distinct paths no matter their content; the fingerprint +/// separates repeated drains of one window — paths only ever collide when +/// window *and* content match, and then overwriting is a no-op (see +/// [`HfSink`] docs; bucket sinks share the same contract). +/// +/// [`HfSink`]: crate::sink::huggingface::HfSink +pub(crate) fn object_path(meta: &WindowMeta, content: &[u8], ext: &str) -> String { + let date = meta.start.date(); + format!( + "data/{}/{:04}-{:02}-{:02}/{:02}-{:02}-{:02}-{:016x}.{}", + meta.pipeline, + date.year(), + u8::from(date.month()), + date.day(), + meta.start.hour(), + meta.start.minute(), + meta.start.second(), + fingerprint(content), + ext, + ) +} diff --git a/src/sink/hf_bucket.rs b/src/sink/hf_bucket.rs new file mode 100644 index 0000000..a0e3d1c --- /dev/null +++ b/src/sink/hf_bucket.rs @@ -0,0 +1,644 @@ +//! [`HfBucketSink`]: terminal sink writing encoded window files (parquet +//! by default) to a [Hugging Face storage bucket] +//! (), `hf://buckets/{namespace}/{bucket}`. +//! +//! Two legs, unlike the git-commit [`HfSink`]: +//! +//! 1. Content upload through Xet, the Hub's content-addressed chunk +//! store, via the official `hf-xet` crate. Xet chunks the bytes and +//! deduplicates them against everything already stored, so replaying +//! identical bytes (spool recovery, retried drain) transfers almost +//! nothing. `hf-xet` fetches and refreshes its CAS write tokens from +//! `/api/buckets/{bucket}/xet-write-token` itself. +//! 2. One sans-IO [`BatchAction`] registering the uploaded content under +//! its bucket path: `POST /api/buckets/{bucket}/batch` with a single +//! NDJSON `addFile` line carrying the xet hash — sent through the same +//! satay transport as every other meathook HTTP call. +//! +//! Buckets have no server-side commit queue, so there is no +//! [`CommitGate`] equivalent; transient failures retry in-sink like +//! `HfSink`'s commits. The `/batch` endpoint is non-transactional: if the +//! process dies after the Xet commit but before registration, the chunks +//! sit unreferenced in CAS and the replayed upload deduplicates against +//! them, so a JSONL-tier replay costs almost nothing. +//! +//! [`HfSink`]: crate::sink::huggingface::HfSink +//! [`CommitGate`]: crate::sink::huggingface::CommitGate + +use std::error; +use std::marker::PhantomData; +use std::time::Duration; + +use http::StatusCode; +use http::header; +use satay_reqwest::ReqwestActionExt; +use satay_runtime::{Action, RequestParts, ResponseParts, insert_header, into_request}; +use serde::Serialize; +use serde::de; +use tokio::time::sleep; +use tracing::{info, warn}; +use xet::XetError; +use xet::xet_session::{ + DeduplicationMetrics, HeaderMap, HeaderValue, Sha256Policy, XetSession, XetSessionBuilder, +}; + +use crate::encode::{Encoder, ParquetEncodeError, ParquetEncoder}; +use crate::sink::{Sink, WindowMeta, object_path}; + +/// Error from the `HuggingFace` bucket sink. +#[derive(Debug, thiserror::Error)] +pub enum HfBucketSinkError { + #[error(transparent)] + Encode(E), + #[error("xet upload failed: {0}")] + Xet(#[from] XetError), + #[error("invalid hugging face token: {0}")] + Token(#[from] header::InvalidHeaderValue), + #[error("transport error: {0}")] + Transport(#[from] satay_reqwest::Error), + #[error("hugging face rejected batch ({status}): {body}")] + Rejected { status: StatusCode, body: String }, +} + +/// One `addFile` registration in a `HuggingFace` storage bucket, as a +/// sans-IO [`Action`]: `POST /api/buckets/{bucket}/batch` with an NDJSON +/// payload referencing previously uploaded Xet content by hash. +/// +/// The xet hash is filled in after the content upload; send the action +/// only once the bytes are committed to CAS. +#[derive(Debug, Clone)] +pub struct BatchAction { + pub bucket: String, + pub token: String, + /// Path of the file inside the bucket, e.g. + /// `data/pm25/2026-06-12/08-00-00-abe4fb8a17f5800b.parquet`. + pub path_in_bucket: String, + /// Xet hash of the uploaded content. + pub xet_hash: String, + /// File modification time in Unix milliseconds. + pub mtime_ms: i64, +} + +/// Decoded result of a [`BatchAction`]. +/// +/// Non-2xx responses decode into [`Rejected`](BatchOutcome::Rejected) +/// rather than an error so the typed status/body survive the fixed +/// `satay_runtime::Error` decode signature. +#[derive(Debug, Clone)] +pub enum BatchOutcome { + Added, + Rejected { status: StatusCode, body: String }, +} + +#[derive(Serialize)] +#[serde(rename_all = "camelCase")] +struct AddFileLine<'a> { + #[serde(rename = "type")] + kind: &'a str, + path: &'a str, + xet_hash: &'a str, + mtime: i64, +} + +impl Action for BatchAction { + type Response = BatchOutcome; + + fn request(self) -> Result>, satay_runtime::Error> { + let uri = format!("https://huggingface.co/api/buckets/{}/batch", self.bucket); + + let mut body = serde_json::to_vec(&AddFileLine { + kind: "addFile", + path: &self.path_in_bucket, + xet_hash: &self.xet_hash, + mtime: self.mtime_ms, + })?; + body.push(b'\n'); + + let mut headers = http::HeaderMap::new(); + insert_header( + &mut headers, + "authorization", + &format!("Bearer {}", self.token), + )?; + insert_header(&mut headers, "content-type", "application/x-ndjson")?; + if let Some(auth) = headers.get_mut(header::AUTHORIZATION) { + auth.set_sensitive(true); + } + + into_request(RequestParts { + method: http::Method::POST, + uri, + headers, + body, + }) + } + + fn decode>( + response: ResponseParts, + ) -> Result { + if response.status.is_success() { + Ok(BatchOutcome::Added) + } else { + Ok(BatchOutcome::Rejected { + status: response.status, + body: String::from_utf8_lossy(response.body.as_ref()).into_owned(), + }) + } + } +} + +/// Sans-IO creation of a `HuggingFace` storage bucket: +/// `POST /api/buckets/{namespace}/{name}`. +/// +/// A 409 decodes to [`AlreadyExists`](CreateBucketOutcome::AlreadyExists). +/// A 401/403 can also mean "exists, but this token may not create in the +/// namespace" — verify with `hf buckets info` when that distinction +/// matters. +#[derive(Debug, Clone)] +pub struct CreateBucketAction { + pub namespace: String, + pub name: String, + pub token: String, + pub private: bool, +} + +/// Decoded result of a [`CreateBucketAction`]. +#[derive(Debug, Clone)] +pub enum CreateBucketOutcome { + Created, + AlreadyExists, + Rejected { status: StatusCode, body: String }, +} + +#[derive(Serialize)] +struct CreateBucketBody { + private: bool, +} + +impl Action for CreateBucketAction { + type Response = CreateBucketOutcome; + + fn request(self) -> Result>, satay_runtime::Error> { + let uri = format!( + "https://huggingface.co/api/buckets/{}/{}", + self.namespace, self.name + ); + let body = serde_json::to_vec(&CreateBucketBody { + private: self.private, + })?; + + let mut headers = http::HeaderMap::new(); + insert_header( + &mut headers, + "authorization", + &format!("Bearer {}", self.token), + )?; + insert_header(&mut headers, "content-type", "application/json")?; + if let Some(auth) = headers.get_mut(header::AUTHORIZATION) { + auth.set_sensitive(true); + } + + into_request(RequestParts { + method: http::Method::POST, + uri, + headers, + body, + }) + } + + fn decode>( + response: ResponseParts, + ) -> Result { + if response.status.is_success() { + Ok(CreateBucketOutcome::Created) + } else if response.status == StatusCode::CONFLICT { + Ok(CreateBucketOutcome::AlreadyExists) + } else { + Ok(CreateBucketOutcome::Rejected { + status: response.status, + body: String::from_utf8_lossy(response.body.as_ref()).into_owned(), + }) + } + } +} + +/// Refresh route `hf-xet` uses to obtain CAS write tokens for a bucket. +fn xet_write_token_url(bucket: &str) -> String { + format!("https://huggingface.co/api/buckets/{bucket}/xet-write-token") +} + +/// One Xet session bound to a bucket's write-token refresh route. +struct XetUploader { + session: XetSession, + refresh_url: String, + refresh_headers: HeaderMap, +} + +impl XetUploader { + fn new(bucket: &str, token: &str) -> Result { + let session = XetSessionBuilder::new() + .build() + .map_err(HfBucketSinkError::Xet)?; + let mut refresh_headers = HeaderMap::new(); + let auth = + HeaderValue::from_str(&format!("Bearer {token}")).map_err(HfBucketSinkError::Token)?; + refresh_headers.insert(header::AUTHORIZATION, auth); + Ok(Self { + session, + refresh_url: xet_write_token_url(bucket), + refresh_headers, + }) + } + + /// Upload the bytes, finalize the Xet commit, and return the hash + /// referencing them plus dedup metrics for observability. + async fn upload( + &self, + path: &str, + content: &[u8], + ) -> Result<(String, DeduplicationMetrics), XetError> { + let commit = self + .session + .new_upload_commit()? + .with_token_refresh_url(self.refresh_url.clone(), self.refresh_headers.clone()) + .build() + .await?; + let upload = commit + .upload_bytes(content.to_vec(), Sha256Policy::Skip, Some(path.to_owned())) + .await?; + let metadata = upload.finalize_ingestion().await?; + let uploaded = (metadata.xet_info.hash, metadata.dedup_metrics); + commit.commit().await?; + Ok(uploaded) + } +} + +/// Total upload+register attempts per window (1 initial + 3 retries; +/// backoff between attempts is 2s, 4s, 8s). +const WRITE_ATTEMPTS: u32 = 4; + +/// Whether a failed write is worth retrying in-sink: transport flakes, +/// Xet network failures, contention/server-side statuses. Authentication +/// problems (both legs), bad requests, and encode failures propagate +/// immediately. +fn transient(error: &HfBucketSinkError) -> bool { + match error { + HfBucketSinkError::Xet(XetError::Authentication(_)) + | HfBucketSinkError::Encode(_) + | HfBucketSinkError::Token(_) => false, + HfBucketSinkError::Xet(_) | HfBucketSinkError::Transport(_) => true, + HfBucketSinkError::Rejected { status, .. } => { + *status == StatusCode::TOO_MANY_REQUESTS || status.is_server_error() + } + } +} + +/// Deterministic `mtime` for the batch `addFile` line: the window end, +/// not wall-clock, so replayed windows register byte-identical payloads. +fn mtime_ms(meta: &WindowMeta) -> i64 { + meta.end.unix_timestamp().saturating_mul(1000) + i64::from(meta.end.nanosecond() / 1_000_000) +} + +/// Terminal sink: encodes each ingested window with its [`Encoder`] +/// (parquet by default) and writes it to a `HuggingFace` storage bucket +/// at the same deterministic, Hive-style path [`HfSink`] uses: +/// +/// ```text +/// data/{pipeline}/{YYYY-MM-DD}/{HH}-{MM}-{SS}-{fnv1a64(content)}.{E::EXT} +/// } +/// ``` +/// +/// Replays stay idempotent like `HfSink`'s — and cheaper: Xet +/// deduplication means re-uploading identical bytes transfers almost +/// nothing before the registration overwrites the same path. +/// +/// Transient failures (transport errors, Xet network errors, 429, 5xx) +/// retry a few times in-sink with backoff before the error propagates; +/// an upstream [`Tier`] retains records when this sink errors and +/// retries at its next firing (or replays them on the next start). +/// +/// [`HfSink`]: crate::sink::huggingface::HfSink +/// [`Tier`]: crate::Tier +pub struct HfBucketSink { + client: reqwest::Client, + bucket: String, + token: String, + xet: XetUploader, + encoder: E, + _record: PhantomData, +} + +impl HfBucketSink { + /// Sink writing to `bucket` (e.g. `"zeon256/weather-windows"`). The + /// bucket must already exist — create it with [`CreateBucketAction`] + /// or `hf buckets create`. The token is a `HuggingFace` access token + /// with write access, typically from the `HF_TOKEN` env var. + /// + /// # Errors + /// + /// Fails only if the token is not valid header material or the + /// underlying Xet session cannot start. + pub fn new( + client: reqwest::Client, + bucket: impl Into, + token: impl Into, + ) -> Result { + let bucket = bucket.into(); + let token = token.into(); + Ok(Self { + xet: XetUploader::new(&bucket, &token)?, + client, + bucket, + token, + encoder: ParquetEncoder::default(), + _record: PhantomData, + }) + } +} + +impl HfBucketSink { + /// Swap the wire format this sink ships (parquet by default): + /// + /// ```no_run + /// use meathook::{HfBucketSink, JsonEncoder}; + /// + /// let sink = HfBucketSink::<()>::new(reqwest::Client::new(), "you/bucket", "hf_token") + /// .expect("xet session") + /// .encoder(JsonEncoder); + /// # let _ = sink; + /// ``` + #[must_use] + pub fn encoder(self, encoder: E2) -> HfBucketSink { + HfBucketSink { + client: self.client, + bucket: self.bucket, + token: self.token, + xet: self.xet, + encoder, + _record: PhantomData, + } + } + + /// One upload-and-register round trip: Xet commit, then the batch + /// `addFile` referencing it. + async fn ship( + &self, + mut action: BatchAction, + content: &[u8], + ) -> Result> { + let (xet_hash, metrics) = self.xet.upload(&action.path_in_bucket, content).await?; + action.xet_hash = xet_hash; + match action.send_with(&self.client).await? { + BatchOutcome::Added => Ok(metrics), + BatchOutcome::Rejected { status, body } => { + Err(HfBucketSinkError::Rejected { status, body }) + } + } + } +} + +impl Sink for HfBucketSink +where + R: Serialize + de::DeserializeOwned + Send + 'static, + E: Encoder, +{ + type Error = HfBucketSinkError; + + async fn ingest(&mut self, meta: &WindowMeta, records: Vec) -> Result<(), Self::Error> { + if records.is_empty() { + return Ok(()); + } + let content = self + .encoder + .encode(&records) + .map_err(HfBucketSinkError::Encode)?; + let path_in_bucket = object_path(meta, &content, E::EXT); + let action = BatchAction { + bucket: self.bucket.clone(), + token: self.token.clone(), + path_in_bucket: path_in_bucket.clone(), + xet_hash: String::new(), + mtime_ms: mtime_ms(meta), + }; + + let mut attempt = 1; + let metrics = loop { + match self.ship(action.clone(), &content).await { + Ok(metrics) => break metrics, + Err(error) if attempt < WRITE_ATTEMPTS && transient(&error) => { + let backoff = Duration::from_secs(1 << attempt); + warn!( + pipeline = %meta.pipeline, + %error, + attempt, + ?backoff, + "hugging face bucket write failed; retrying" + ); + sleep(backoff).await; + attempt += 1; + } + Err(error) => return Err(error), + } + }; + info!( + pipeline = %meta.pipeline, + path = %path_in_bucket, + records = records.len(), + bytes = metrics.total_bytes, + uploaded = metrics.new_bytes, + "wrote window to hugging face bucket" + ); + Ok(()) + } + + /// No-op: this terminal sink ships every batch as it is ingested. + fn flush(&mut self) -> impl Future> + Send { + std::future::ready(Ok(())) + } +} + +#[cfg(test)] +mod tests { + use time::macros::datetime; + + use super::*; + use crate::encode::JsonEncoder; + + fn batch_action() -> BatchAction { + BatchAction { + bucket: "zeon256/sg-weather".into(), + token: "hf_secret".into(), + path_in_bucket: "data/pm25/2026-06-12/08-00-00-abe4fb8a17f5800b.parquet".into(), + xet_hash: "96e637d9665bd35477b1908a23f2e254edfba0618dbd2d62f90a6baee7d139cf".into(), + mtime_ms: 1_781_254_800_000, + } + } + + #[test] + fn batch_request_shape() { + let request = batch_action().request().unwrap(); + + assert_eq!(request.method(), http::Method::POST); + assert_eq!( + request.uri(), + "https://huggingface.co/api/buckets/zeon256/sg-weather/batch" + ); + assert_eq!( + request.headers().get("content-type").unwrap(), + "application/x-ndjson" + ); + assert_eq!( + request.headers().get("authorization").unwrap(), + "Bearer hf_secret" + ); + + let body = String::from_utf8(request.body().clone()).unwrap(); + let line: serde_json::Value = serde_json::from_str(body.trim_end()).unwrap(); + assert_eq!(line["type"], "addFile"); + assert_eq!( + line["path"], + "data/pm25/2026-06-12/08-00-00-abe4fb8a17f5800b.parquet" + ); + assert_eq!( + line["xetHash"], + "96e637d9665bd35477b1908a23f2e254edfba0618dbd2d62f90a6baee7d139cf" + ); + assert_eq!(line["mtime"], 1_781_254_800_000_i64); + } + + #[test] + fn batch_decode_success_and_rejection() { + let ok = BatchAction::decode(ResponseParts { + status: StatusCode::OK, + headers: http::HeaderMap::new(), + body: b"".as_slice(), + }) + .unwrap(); + assert!(matches!(ok, BatchOutcome::Added)); + + let rejected = BatchAction::decode(ResponseParts { + status: StatusCode::UNAUTHORIZED, + headers: http::HeaderMap::new(), + body: b"Invalid credentials".as_slice(), + }) + .unwrap(); + match rejected { + BatchOutcome::Rejected { status, body } => { + assert_eq!(status, StatusCode::UNAUTHORIZED); + assert_eq!(body, "Invalid credentials"); + } + other @ BatchOutcome::Added => panic!("expected Rejected, got {other:?}"), + } + } + + #[test] + fn create_bucket_request_shape() { + let request = CreateBucketAction { + namespace: "zeon256".into(), + name: "sg-weather".into(), + token: "hf_secret".into(), + private: true, + } + .request() + .unwrap(); + + assert_eq!(request.method(), http::Method::POST); + assert_eq!( + request.uri(), + "https://huggingface.co/api/buckets/zeon256/sg-weather" + ); + assert_eq!( + request.headers().get("content-type").unwrap(), + "application/json" + ); + + let body: serde_json::Value = serde_json::from_slice(request.body()).unwrap(); + assert_eq!(body, serde_json::json!({ "private": true })); + } + + #[test] + fn create_bucket_decode_created_conflict_rejected() { + let decode = |status, body: &[u8]| { + CreateBucketAction::decode(ResponseParts { + status, + headers: http::HeaderMap::new(), + body, + }) + .unwrap() + }; + + assert!(matches!( + decode(StatusCode::CREATED, b""), + CreateBucketOutcome::Created + )); + assert!(matches!( + decode(StatusCode::CONFLICT, b"already exists"), + CreateBucketOutcome::AlreadyExists + )); + match decode(StatusCode::FORBIDDEN, b"no write permission") { + CreateBucketOutcome::Rejected { status, body } => { + assert_eq!(status, StatusCode::FORBIDDEN); + assert_eq!(body, "no write permission"); + } + other => panic!("expected Rejected, got {other:?}"), + } + } + + #[test] + fn xet_write_token_url_matches_hub_route() { + assert_eq!( + xet_write_token_url("zeon256/sg-weather"), + "https://huggingface.co/api/buckets/zeon256/sg-weather/xet-write-token" + ); + } + + /// The `mtime` is load-bearing for deterministic replay payloads: it + /// must come from the window, never from wall-clock. + #[test] + fn mtime_is_window_end_in_ms() { + let meta = WindowMeta { + pipeline: "pm25".into(), + start: datetime!(2026-06-12 08:00 UTC), + end: datetime!(2026-06-12 09:00 UTC), + }; + assert_eq!(mtime_ms(&meta), 1_781_254_800_000); + } + + #[test] + fn transient_classifies_retryable_failures() { + let rejected = + |status: StatusCode| HfBucketSinkError::::Rejected { + status, + body: String::new(), + }; + assert!(transient(&rejected(StatusCode::TOO_MANY_REQUESTS))); + assert!(transient(&rejected(StatusCode::INTERNAL_SERVER_ERROR))); + assert!(transient(&rejected(StatusCode::SERVICE_UNAVAILABLE))); + assert!(!transient(&rejected(StatusCode::UNAUTHORIZED))); + assert!(transient(&HfBucketSinkError::::Xet( + XetError::Network("connection reset".into()) + ))); + assert!(!transient(&HfBucketSinkError::::Xet( + XetError::Authentication("bad token".into()) + ))); + } + /// The bucket sink shares the dataset sink's path derivation, so the + /// pinned fingerprint must not drift between the two modules. + #[test] + fn object_path_matches_dataset_sink_scheme() { + let meta = WindowMeta { + pipeline: "pm25".into(), + start: datetime!(2026-06-12 08:00 UTC), + end: datetime!(2026-06-12 09:00 UTC), + }; + assert_eq!( + object_path(&meta, b"PARQUET", ::EXT), + "data/pm25/2026-06-12/08-00-00-abe4fb8a17f5800b.parquet" + ); + let json_path = object_path(&meta, b"PARQUET", JsonEncoder::EXT); + assert_eq!( + json_path, + "data/pm25/2026-06-12/08-00-00-abe4fb8a17f5800b.json" + ); + } +} diff --git a/src/sink/huggingface.rs b/src/sink/huggingface.rs index 696643f..8d3c57b 100644 --- a/src/sink/huggingface.rs +++ b/src/sink/huggingface.rs @@ -26,7 +26,7 @@ use tokio::time::sleep; use tracing::{info, warn}; use crate::encode::{Encoder, ParquetEncodeError, ParquetEncoder}; -use crate::sink::{Sink, WindowMeta}; +use crate::sink::{Sink, WindowMeta, object_path}; /// Error from the `HuggingFace` sink. #[derive(Debug, thiserror::Error)] @@ -301,38 +301,6 @@ impl HfSink { } } -/// FNV-1a 64-bit — stable across releases (the fingerprint is load-bearing -/// for replay idempotency: same bytes must map to the same path forever). -fn fingerprint(bytes: &[u8]) -> u64 { - const OFFSET_BASIS: u64 = 0xcbf2_9ce4_8422_2325; - const PRIME: u64 = 0x0100_0000_01b3; - bytes.iter().fold(OFFSET_BASIS, |hash, &byte| { - (hash ^ u64::from(byte)).wrapping_mul(PRIME) - }) -} - -/// Keyed by the full window start plus a fingerprint of the encoded -/// bytes. The start (to the second) identifies the window — distinct -/// windows get distinct paths no matter their content; the fingerprint -/// separates repeated drains of one window — paths only ever collide when -/// window *and* content match, and then overwriting is a no-op (see -/// [`HfSink`] docs). -fn object_path(meta: &WindowMeta, content: &[u8], ext: &str) -> String { - let date = meta.start.date(); - format!( - "data/{}/{:04}-{:02}-{:02}/{:02}-{:02}-{:02}-{:016x}.{}", - meta.pipeline, - date.year(), - u8::from(date.month()), - date.day(), - meta.start.hour(), - meta.start.minute(), - meta.start.second(), - fingerprint(content), - ext, - ) -} - impl Sink for HfSink where R: Serialize + de::DeserializeOwned + Send + 'static, @@ -388,8 +356,8 @@ where } /// No-op: this terminal sink ships every batch as it is ingested. - async fn flush(&mut self) -> Result<(), Self::Error> { - Ok(()) + fn flush(&mut self) -> impl Future> + Send { + std::future::ready(Ok(())) } } diff --git a/src/store/jsonl.rs b/src/store/jsonl.rs index 4ff18c7..ac1ebbf 100644 --- a/src/store/jsonl.rs +++ b/src/store/jsonl.rs @@ -157,46 +157,56 @@ where /// Append records to the window's segment file, fsyncing the file (and /// the directory when the segment is new) before returning. - async fn append(&mut self, window: i64, records: Vec) -> Result<(), JsonlStoreError> { - self.ensure_dir()?; - let path = self.segment_path(window); - - let mut lines = vec![]; - for record in &records { - serde_json::to_writer(&mut lines, record).map_err(JsonlStoreError::Serialize)?; - lines.push(b'\n'); - } + fn append( + &mut self, + window: i64, + records: Vec, + ) -> impl Future> + Send { + let result = (|| { + self.ensure_dir()?; + let path = self.segment_path(window); + + let mut lines = vec![]; + for record in &records { + serde_json::to_writer(&mut lines, record).map_err(JsonlStoreError::Serialize)?; + lines.push(b'\n'); + } - let is_new = !path.exists(); - let mut file = fs::OpenOptions::new() - .create(true) - .append(true) - .open(&path) - .map_err(|e| io_err(&path, e))?; - file.write_all(&lines).map_err(|e| io_err(&path, e))?; - file.sync_all().map_err(|e| io_err(&path, e))?; - if is_new { - fs::File::open(&self.dir) - .and_then(|d| d.sync_all()) - .map_err(|e| io_err(&self.dir, e))?; - } - Ok(()) + let is_new = !path.exists(); + let mut file = fs::OpenOptions::new() + .create(true) + .append(true) + .open(&path) + .map_err(|e| io_err(&path, e))?; + file.write_all(&lines).map_err(|e| io_err(&path, e))?; + file.sync_all().map_err(|e| io_err(&path, e))?; + if is_new { + fs::File::open(&self.dir) + .and_then(|d| d.sync_all()) + .map_err(|e| io_err(&self.dir, e))?; + } + Ok(()) + })(); + std::future::ready(result) } - async fn oldest( + fn oldest( &mut self, after: Option, - ) -> Result>, JsonlStoreError> { - self.ensure_dir()?; - Ok(self - .list_segments()? - .into_iter() - .find(|(window, _)| after.is_none_or(|a| *window > a)) - .map(|(window, path)| JsonlSegment { - window, - path, - _record: PhantomData, - })) + ) -> impl Future>, JsonlStoreError>> + Send { + let result = (|| { + self.ensure_dir()?; + Ok(self + .list_segments()? + .into_iter() + .find(|(window, _)| after.is_none_or(|a| *window > a)) + .map(|(window, path)| JsonlSegment { + window, + path, + _record: PhantomData, + })) + })(); + std::future::ready(result) } fn pipeline_hint(&self) -> Option<&str> { @@ -226,39 +236,42 @@ where self.window } - async fn records(&mut self) -> Result, JsonlStoreError> { - let contents = fs::read_to_string(&self.path).map_err(|e| io_err(&self.path, e))?; - let lines = contents - .lines() - .filter(|l| !l.is_empty()) - .collect::>(); - let mut records = Vec::with_capacity(lines.len()); - let last = lines.len().saturating_sub(1); - for (i, line) in lines.iter().enumerate() { - match serde_json::from_str::(line) { - Ok(record) => records.push(record), - Err(error) if i == last => { - warn!( - path = %self.path.display(), - %error, - "skipping torn final line in store segment (crash mid-append)" - ); - } - Err(error) => { - warn!( - path = %self.path.display(), - line = i, - %error, - "skipping corrupt line in store segment" - ); + fn records(&mut self) -> impl Future, JsonlStoreError>> + Send { + let result = (|| { + let contents = fs::read_to_string(&self.path).map_err(|e| io_err(&self.path, e))?; + let lines = contents + .lines() + .filter(|l| !l.is_empty()) + .collect::>(); + let mut records = Vec::with_capacity(lines.len()); + let last = lines.len().saturating_sub(1); + for (i, line) in lines.iter().enumerate() { + match serde_json::from_str::(line) { + Ok(record) => records.push(record), + Err(error) if i == last => { + warn!( + path = %self.path.display(), + %error, + "skipping torn final line in store segment (crash mid-append)" + ); + } + Err(error) => { + warn!( + path = %self.path.display(), + line = i, + %error, + "skipping corrupt line in store segment" + ); + } } } - } - Ok(records) + Ok(records) + })(); + std::future::ready(result) } - async fn commit(self) -> Result<(), JsonlStoreError> { - fs::remove_file(&self.path).map_err(|e| io_err(&self.path, e)) + fn commit(self) -> impl Future> + Send { + std::future::ready(fs::remove_file(&self.path).map_err(|e| io_err(&self.path, e))) } } diff --git a/src/store/mem.rs b/src/store/mem.rs index 8445832..f049034 100644 --- a/src/store/mem.rs +++ b/src/store/mem.rs @@ -52,28 +52,30 @@ where where Self: 'a; - async fn append(&mut self, window: i64, records: Vec) -> Result<(), Infallible> { + fn append( + &mut self, + window: i64, + records: Vec, + ) -> impl Future> + Send { self.windows.entry(window).or_default().extend(records); - Ok(()) + std::future::ready(Ok(())) } - async fn oldest( + fn oldest( &mut self, after: Option, - ) -> Result>, Infallible> { + ) -> impl Future>, Infallible>> + Send { let lower = after.map_or(Bound::Unbounded, Bound::Excluded); - let Some(window) = self + let window = self .windows .range((lower, Bound::Unbounded)) .next() - .map(|(window, _)| *window) - else { - return Ok(None); - }; - Ok(Some(MemSegment { + .map(|(window, _)| *window); + let segment = window.map(|window| MemSegment { store: self, window, - })) + }); + std::future::ready(Ok(segment)) } } @@ -96,18 +98,18 @@ where self.window } - async fn records(&mut self) -> Result, Infallible> { - Ok(self + fn records(&mut self) -> impl Future, Infallible>> + Send { + std::future::ready(Ok(self .store .windows .get(&self.window) .cloned() - .unwrap_or_default()) + .unwrap_or_default())) } - async fn commit(self) -> Result<(), Infallible> { + fn commit(self) -> impl Future> + Send { self.store.windows.remove(&self.window); - Ok(()) + std::future::ready(Ok(())) } } diff --git a/src/test_util.rs b/src/test_util.rs index 52f30b6..e15b0f0 100644 --- a/src/test_util.rs +++ b/src/test_util.rs @@ -1,7 +1,8 @@ //! Shared fakes for unit tests: a `Vec`-backed sink with a failure toggle. +use parking_lot::Mutex; +use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; -use std::sync::{Arc, Mutex}; use time::OffsetDateTime; @@ -53,7 +54,7 @@ impl SharedSink { impl SharedSink { #[must_use] pub fn batches(&self) -> Vec<(WindowMeta, Vec)> { - self.batches.lock().unwrap().clone() + self.batches.lock().clone() } #[must_use] @@ -68,20 +69,28 @@ impl SharedSink { impl Sink for SharedSink { type Error = TestSinkFailure; - async fn ingest(&mut self, meta: &WindowMeta, records: Vec) -> Result<(), Self::Error> { - if self.fail.load(Ordering::SeqCst) { - return Err(TestSinkFailure); - } - self.batches.lock().unwrap().push((meta.clone(), records)); - Ok(()) + fn ingest( + &mut self, + meta: &WindowMeta, + records: Vec, + ) -> impl Future> + Send { + let result = if self.fail.load(Ordering::SeqCst) { + Err(TestSinkFailure) + } else { + self.batches.lock().push((meta.clone(), records)); + Ok(()) + }; + std::future::ready(result) } - async fn flush(&mut self) -> Result<(), Self::Error> { - if self.fail.load(Ordering::SeqCst) { - return Err(TestSinkFailure); - } - self.flushed.store(true, Ordering::SeqCst); - Ok(()) + fn flush(&mut self) -> impl Future> + Send { + let result = if self.fail.load(Ordering::SeqCst) { + Err(TestSinkFailure) + } else { + self.flushed.store(true, Ordering::SeqCst); + Ok(()) + }; + std::future::ready(result) } } diff --git a/tests/hf_bucket_integration.rs b/tests/hf_bucket_integration.rs new file mode 100644 index 0000000..c1f3f86 --- /dev/null +++ b/tests/hf_bucket_integration.rs @@ -0,0 +1,81 @@ +//! Network integration test against a real (scratch) `HuggingFace` storage +//! bucket. Ignored by default; run with: +//! +//! ```bash +//! HF_TOKEN=hf_... MEATHOOK_TEST_BUCKET=you/meathook-test \ +//! cargo test --features hf-bucket --test hf_bucket_integration -- --ignored +//! ``` +//! +//! The bucket is created (or must already exist) before the window is +//! written; verify afterwards with +//! `hf buckets list you/meathook-test --recursive`. + +#![cfg(feature = "hf-bucket")] + +use std::env; + +use meathook::{CreateBucketAction, CreateBucketOutcome, HfBucketSink, Sink as _, WindowMeta}; +use satay_reqwest::ReqwestActionExt as _; +use serde::{Deserialize, Serialize}; +use time::OffsetDateTime; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct Sample { + station_id: String, + timestamp: String, + value: f64, +} + +#[tokio::test] +#[ignore = "requires HF_TOKEN and MEATHOOK_TEST_BUCKET with write access"] +async fn writes_parquet_window_to_scratch_bucket() { + let token = env::var("HF_TOKEN").expect("HF_TOKEN must be set"); + let bucket = env::var("MEATHOOK_TEST_BUCKET").expect("MEATHOOK_TEST_BUCKET must be set"); + let (namespace, name) = bucket + .split_once('/') + .expect("MEATHOOK_TEST_BUCKET must be {namespace}/{bucket}"); + + let client = reqwest::Client::new(); + let created = CreateBucketAction { + namespace: namespace.to_owned(), + name: name.to_owned(), + token: token.clone(), + private: true, + } + .send_with(&client) + .await + .expect("bucket creation should succeed"); + assert!( + matches!( + created, + CreateBucketOutcome::Created | CreateBucketOutcome::AlreadyExists + ), + "unexpected create outcome: {created:?}" + ); + + let mut sink = HfBucketSink::new(client, bucket, token).expect("xet session should start"); + let now = OffsetDateTime::now_utc(); + let meta = WindowMeta { + pipeline: "integration_test".to_owned(), + start: now, + end: now, + }; + + sink.ingest( + &meta, + vec![ + Sample { + station_id: "S100".into(), + timestamp: now.to_string(), + value: 1.0, + }, + Sample { + station_id: "S117".into(), + timestamp: now.to_string(), + value: 2.0, + }, + ], + ) + .await + .expect("bucket write should succeed"); +}