diff --git a/Cargo.lock b/Cargo.lock index 67ffa80a..628a4203 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,17 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "ahash" +version = "0.7.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "891477e0c6a8957309ee5c45a6368af3ae14bb510732d2684ffa19af310920f9" +dependencies = [ + "getrandom 0.2.17", + "once_cell", + "version_check", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -11,6 +22,15 @@ dependencies = [ "memchr", ] +[[package]] +name = "android_system_properties" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "819e7219dbd41043ac279b19830f2efc897156490d7fd6ea916720117ee66311" +dependencies = [ + "libc", +] + [[package]] name = "anstream" version = "0.6.21" @@ -61,6 +81,21 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "anyerror" +version = "0.1.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71add24cc141a1e8326f249b74c41cfd217aeb2a67c9c6cf9134d175469afd49" +dependencies = [ + "serde", +] + +[[package]] +name = "arrayvec" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7c02d123df017efcdfbd739ef81735b36c5ba83ec3c59c80a9d7ecc718f92e50" + [[package]] name = "autocfg" version = "1.5.0" @@ -73,24 +108,122 @@ version = "2.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "812e12b5285cc515a9c72a5c1d3b6d46a19dac5acfef5265968c166106e31dd3" +[[package]] +name = "bitvec" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1bc2832c24239b0141d5674bb9174f9d68a8b5b3f2753311927c172ca46f7e9c" +dependencies = [ + "funty", + "radium", + "tap", + "wyz", +] + +[[package]] +name = "borsh" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1da5ab77c1437701eeff7c88d968729e7766172279eab0676857b3d63af7a6f" +dependencies = [ + "borsh-derive", + "cfg_aliases", +] + +[[package]] +name = "borsh-derive" +version = "1.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0686c856aa6aac0c4498f936d7d6a02df690f614c03e4d906d1018062b5c5e2c" +dependencies = [ + "once_cell", + "proc-macro-crate", + "proc-macro2", + "quote", + "syn 2.0.114", +] + [[package]] name = "bumpalo" version = "3.19.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5dd9dc738b7a8311c7ade152424974d8115f2cdad61e8dab8dac9f2362298510" +[[package]] +name = "byte-unit" +version = "5.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8c6d47a4e2961fb8721bcfc54feae6455f2f64e7054f9bc67e875f0e77f4c58d" +dependencies = [ + "rust_decimal", + "schemars", + "serde", + "utf8-width", +] + +[[package]] +name = "bytecheck" +version = "0.6.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23cdc57ce23ac53c931e88a43d06d070a6fd142f2617be5855eb75efc9beb1c2" +dependencies = [ + "bytecheck_derive", + "ptr_meta", + "simdutf8", +] + +[[package]] +name = "bytecheck_derive" +version = "0.6.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3db406d29fbcd95542e92559bed4d8ad92636d1ca8b3b72ede10b4bcc010e659" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "bytes" version = "1.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e748733b7cbc798e1434b6ac524f0c1ff2ab456fe201501e6497c8417a4fc33" +[[package]] +name = "cc" +version = "1.2.55" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47b26a0954ae34af09b50f0de26458fa95369a0d478d8236d3f93082b219bd29" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "cfg_aliases" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" + +[[package]] +name = "chrono" +version = "0.4.43" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fac4744fb15ae8337dc853fee7fb3f4e48c0fbaa23d0afe49c447b4fab126118" +dependencies = [ + "iana-time-zone", + "js-sys", + "num-traits", + "wasm-bindgen", + "windows-link", +] + [[package]] name = "clap" version = "4.5.57" @@ -122,7 +255,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.114", ] [[package]] @@ -137,6 +270,12 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + [[package]] name = "crc32fast" version = "1.5.0" @@ -146,14 +285,43 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "derive_more" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a9b99b9cbbe49445b21764dc0625032a89b145a2642e67603e1c936f5458d05" +dependencies = [ + "derive_more-impl", +] + +[[package]] +name = "derive_more-impl" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", + "unicode-xid", +] + +[[package]] +name = "dyn-clone" +version = "1.0.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" + [[package]] name = "ember-cluster" version = "0.2.0" dependencies = [ "bytes", - "rand", + "openraft", + "rand 0.9.2", "serde", - "thiserror", + "serde_json", + "thiserror 2.0.18", "tokio", "tracing", "uuid", @@ -166,7 +334,7 @@ dependencies = [ "bytes", "crc32fast", "tempfile", - "thiserror", + "thiserror 2.0.18", "tracing", ] @@ -176,7 +344,7 @@ version = "0.2.0" dependencies = [ "bytes", "itoa", - "thiserror", + "thiserror 2.0.18", ] [[package]] @@ -209,13 +377,19 @@ dependencies = [ "ember-persistence", "ember-protocol", "ordered-float", - "rand", + "rand 0.9.2", "tempfile", - "thiserror", + "thiserror 2.0.18", "tokio", "tracing", ] +[[package]] +name = "equivalent" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f" + [[package]] name = "errno" version = "0.3.14" @@ -232,6 +406,118 @@ version = "2.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "37909eebbb50d72f9059c3b6d82c0463f2ff062c9e95845c43a6c9c0355411be" +[[package]] +name = "find-msvc-tools" +version = "0.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" + +[[package]] +name = "funty" +version = "2.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c" + +[[package]] +name = "futures" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65bc07b1a8bc7c85c5f2e110c476c7389b4554ba72af57d8445ea63a576b0876" +dependencies = [ + "futures-channel", + "futures-core", + "futures-executor", + "futures-io", + "futures-sink", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-channel" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" +dependencies = [ + "futures-core", + "futures-sink", +] + +[[package]] +name = "futures-core" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05f29059c0c2090612e8d742178b0580d2dc940c837851ad723096f87af6663e" + +[[package]] +name = "futures-executor" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e28d1d997f585e54aebc3f97d39e72338912123a67330d723fdbb564d646c9f" +dependencies = [ + "futures-core", + "futures-task", + "futures-util", +] + +[[package]] +name = "futures-io" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9e5c1b78ca4aae1ac06c48a526a655760685149f0d465d21f37abfe57ce075c6" + +[[package]] +name = "futures-macro" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "162ee34ebcb7c64a8abebc059ce0fee27c2262618d7b60ed8faf72fef13c3650" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", +] + +[[package]] +name = "futures-sink" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" + +[[package]] +name = "futures-task" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988" + +[[package]] +name = "futures-util" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" +dependencies = [ + "futures-channel", + "futures-core", + "futures-io", + "futures-macro", + "futures-sink", + "futures-task", + "memchr", + "pin-project-lite", + "pin-utils", + "slab", +] + +[[package]] +name = "getrandom" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff2abc00be7fca6ebc474524697ae276ad847ad0a6b3faa4bcb027e9a4614ad0" +dependencies = [ + "cfg-if", + "libc", + "wasi", +] + [[package]] name = "getrandom" version = "0.3.4" @@ -244,12 +530,61 @@ dependencies = [ "wasip2", ] +[[package]] +name = "hashbrown" +version = "0.12.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888" +dependencies = [ + "ahash", +] + +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" + [[package]] name = "heck" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "iana-time-zone" +version = "0.1.65" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e31bc9ad994ba00e440a8aa5c9ef0ec67d5cb5e5cb0cc7f8b744a35b389cc470" +dependencies = [ + "android_system_properties", + "core-foundation-sys", + "iana-time-zone-haiku", + "js-sys", + "log", + "wasm-bindgen", + "windows-core", +] + +[[package]] +name = "iana-time-zone-haiku" +version = "0.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f31827a206f56af32e590ba56d5d2d085f558508192593743f16b2306495269f" +dependencies = [ + "cc", +] + +[[package]] +name = "indexmap" +version = "2.13.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017" +dependencies = [ + "equivalent", + "hashbrown 0.16.1", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -305,6 +640,12 @@ version = "0.4.29" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "5e5032e24019045c762d3c0f28f5b6b8bbf38563a65908389bf7978758920897" +[[package]] +name = "maplit" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3e2e65a1a2e43cfcb47a895c4c8b10d1f4a61097f9f254f183aee60cad9c651d" + [[package]] name = "matchers" version = "0.2.0" @@ -361,6 +702,42 @@ version = "1.70.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" +[[package]] +name = "openraft" +version = "0.9.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc22bb6823c606299be05f3cc0d2ac30216412e05352eaf192a481c12ea055fc" +dependencies = [ + "anyerror", + "byte-unit", + "chrono", + "clap", + "derive_more", + "futures", + "maplit", + "openraft-macros", + "rand 0.8.5", + "serde", + "thiserror 1.0.69", + "tokio", + "tracing", + "tracing-futures", + "validit", +] + +[[package]] +name = "openraft-macros" +version = "0.9.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8e5c7db6c8f2137b45a63096e09ac5a89177799b4bb0073915a5f41ee156651" +dependencies = [ + "chrono", + "proc-macro2", + "quote", + "semver", + "syn 2.0.114", +] + [[package]] name = "ordered-float" version = "5.1.0" @@ -393,12 +770,38 @@ dependencies = [ "windows-link", ] +[[package]] +name = "pin-project" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "677f1add503faace112b9f1373e43e9e054bfdd22ff1a63c1bc485eaec6a6a8a" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "1.1.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e918e4ff8c4549eb882f14b3a4bc8c8bc93de829416eacf579f1207a8fbf861" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", +] + [[package]] name = "pin-project-lite" version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -408,6 +811,15 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "proc-macro-crate" +version = "3.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "219cb19e96be00ab2e37d6e299658a0cfa83e52429179969b0f0121b4ac46983" +dependencies = [ + "toml_edit", +] + [[package]] name = "proc-macro2" version = "1.0.106" @@ -417,6 +829,26 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "ptr_meta" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0738ccf7ea06b608c10564b31debd4f5bc5e197fc8bfe088f68ae5ce81e7a4f1" +dependencies = [ + "ptr_meta_derive", +] + +[[package]] +name = "ptr_meta_derive" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "16b845dbfca988fa33db069c0e230574d15a3088f147a87b64c7589eb662c9ac" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "quote" version = "1.0.44" @@ -432,14 +864,41 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "radium" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc33ff2d4973d518d823d61aa239014831e521c75da58e3df4840d3f47749d09" + +[[package]] +name = "rand" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" +dependencies = [ + "libc", + "rand_chacha 0.3.1", + "rand_core 0.6.4", +] + [[package]] name = "rand" version = "0.9.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6db2770f06117d490610c7488547d543617b21bfa07796d7a12f6f1bd53850d1" dependencies = [ - "rand_chacha", - "rand_core", + "rand_chacha 0.9.0", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_chacha" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" +dependencies = [ + "ppv-lite86", + "rand_core 0.6.4", ] [[package]] @@ -449,7 +908,16 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.9.5", +] + +[[package]] +name = "rand_core" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" +dependencies = [ + "getrandom 0.2.17", ] [[package]] @@ -458,7 +926,7 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" dependencies = [ - "getrandom", + "getrandom 0.3.4", ] [[package]] @@ -470,6 +938,26 @@ dependencies = [ "bitflags", ] +[[package]] +name = "ref-cast" +version = "1.0.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f354300ae66f76f1c85c5f84693f0ce81d747e2c3f21a45fef496d89c960bf7d" +dependencies = [ + "ref-cast-impl", +] + +[[package]] +name = "ref-cast-impl" +version = "1.0.25" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7186006dcb21920990093f30e3dea63b7d6e977bf1256be20c3563a5db070da" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", +] + [[package]] name = "regex-automata" version = "0.4.14" @@ -487,6 +975,60 @@ version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a96887878f22d7bad8a3b6dc5b7440e0ada9a245242924394987b21cf2210a4c" +[[package]] +name = "rend" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "71fe3824f5629716b1589be05dacd749f6aa084c87e00e016714a8cdfccc997c" +dependencies = [ + "bytecheck", +] + +[[package]] +name = "rkyv" +version = "0.7.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2297bf9c81a3f0dc96bc9521370b88f054168c29826a75e89c55ff196e7ed6a1" +dependencies = [ + "bitvec", + "bytecheck", + "bytes", + "hashbrown 0.12.3", + "ptr_meta", + "rend", + "rkyv_derive", + "seahash", + "tinyvec", + "uuid", +] + +[[package]] +name = "rkyv_derive" +version = "0.7.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84d7b42d4b8d06048d3ac8db0eb31bcb942cbeb709f0b5f2b2ebde398d3038f5" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + +[[package]] +name = "rust_decimal" +version = "1.40.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61f703d19852dbf87cbc513643fa81428361eb6940f1ac14fd58155d295a3eb0" +dependencies = [ + "arrayvec", + "borsh", + "bytes", + "num-traits", + "rand 0.8.5", + "rkyv", + "serde", + "serde_json", +] + [[package]] name = "rustix" version = "1.1.3" @@ -506,12 +1048,36 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +[[package]] +name = "schemars" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a2b42f36aa1cd011945615b92222f6bf73c599a102a300334cd7f8dbeec726cc" +dependencies = [ + "dyn-clone", + "ref-cast", + "serde", + "serde_json", +] + [[package]] name = "scopeguard" version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" +[[package]] +name = "seahash" +version = "4.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c107b6f4780854c8b126e228ea8869f4d7b71260f962fefb57b996b8959ba6b" + +[[package]] +name = "semver" +version = "1.0.27" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d767eb0aabc880b29956c35734170f26ed551a859dbd361d140cdbeca61ab1e2" + [[package]] name = "serde" version = "1.0.228" @@ -539,7 +1105,20 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.114", +] + +[[package]] +name = "serde_json" +version = "1.0.149" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "83fc039473c5595ace860d8c4fafa220ff474b3fc6bfdb4293327f1a37e94d86" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", ] [[package]] @@ -551,6 +1130,12 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shlex" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" + [[package]] name = "signal-hook-registry" version = "1.4.8" @@ -561,6 +1146,18 @@ dependencies = [ "libc", ] +[[package]] +name = "simdutf8" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "smallvec" version = "1.15.1" @@ -583,6 +1180,17 @@ version = "0.11.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" +[[package]] +name = "syn" +version = "1.0.109" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72b64191b275b66ffe2469e8af2c1cfe3bafa67b529ead792a6d0160888b4237" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + [[package]] name = "syn" version = "2.0.114" @@ -594,6 +1202,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "tap" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369" + [[package]] name = "tempfile" version = "3.24.0" @@ -601,19 +1215,39 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "655da9c7eb6305c55742045d5a8d2037996d61d8de95806335c7c86ce0f82e9c" dependencies = [ "fastrand", - "getrandom", + "getrandom 0.3.4", "once_cell", "rustix", "windows-sys 0.61.2", ] +[[package]] +name = "thiserror" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6aaf5339b578ea85b50e080feb250a3e8ae8cfcdff9a461c9ec2904bc923f52" +dependencies = [ + "thiserror-impl 1.0.69", +] + [[package]] name = "thiserror" version = "2.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4288b5bcbc7920c07a1149a35cf9590a2aa808e0bc1eafaade0b80947865fbc4" dependencies = [ - "thiserror-impl", + "thiserror-impl 2.0.18", +] + +[[package]] +name = "thiserror-impl" +version = "1.0.69" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", ] [[package]] @@ -624,7 +1258,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.114", ] [[package]] @@ -636,6 +1270,21 @@ dependencies = [ "cfg-if", ] +[[package]] +name = "tinyvec" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa5fdc3bce6191a1dbc8c02d5c8bffcf557bafa17c124c5264a458f1b0613fa" +dependencies = [ + "tinyvec_macros", +] + +[[package]] +name = "tinyvec_macros" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" + [[package]] name = "tokio" version = "1.49.0" @@ -661,7 +1310,37 @@ checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.114", +] + +[[package]] +name = "toml_datetime" +version = "0.7.5+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92e1cfed4a3038bc5a127e35a2d360f145e1f4b971b551a2ba5fd7aedf7e1347" +dependencies = [ + "serde_core", +] + +[[package]] +name = "toml_edit" +version = "0.23.10+spec-1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84c8b9f757e028cee9fa244aea147aab2a9ec09d5325a9b01e0a49730c2b5269" +dependencies = [ + "indexmap", + "toml_datetime", + "toml_parser", + "winnow", +] + +[[package]] +name = "toml_parser" +version = "1.0.6+spec-1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3198b4b0a8e11f09dd03e133c0280504d0801269e9afa46362ffde1cbeebf44" +dependencies = [ + "winnow", ] [[package]] @@ -683,7 +1362,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.114", ] [[package]] @@ -696,6 +1375,16 @@ dependencies = [ "valuable", ] +[[package]] +name = "tracing-futures" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "97d095ae15e245a057c8e8451bab9b3ee1e1f68e9ba2b4fbc18d0ac5237835f2" +dependencies = [ + "pin-project", + "tracing", +] + [[package]] name = "tracing-log" version = "0.2.0" @@ -731,6 +1420,18 @@ version = "1.0.22" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9312f7c4f6ff9069b165498234ce8be658059c6728633667c526e27dc2cf1df5" +[[package]] +name = "unicode-xid" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" + +[[package]] +name = "utf8-width" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1292c0d970b54115d14f2492fe0170adf21d68a1de108eebc51c1df4f346a091" + [[package]] name = "utf8parse" version = "0.2.2" @@ -743,18 +1444,33 @@ version = "1.20.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ee48d38b119b0cd71fe4141b30f5ba9c7c5d9f4e7a3a8b4a674e4b6ef789976f" dependencies = [ - "getrandom", + "getrandom 0.3.4", "js-sys", "serde_core", "wasm-bindgen", ] +[[package]] +name = "validit" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4efba0434d5a0a62d4f22070b44ce055dc18cb64d4fa98276aa523dadfaba0e7" +dependencies = [ + "anyerror", +] + [[package]] name = "valuable" version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "version_check" +version = "0.9.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -802,7 +1518,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.114", "wasm-bindgen-shared", ] @@ -815,12 +1531,65 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "windows-core" +version = "0.62.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" +dependencies = [ + "windows-implement", + "windows-interface", + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-implement" +version = "0.60.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", +] + +[[package]] +name = "windows-interface" +version = "0.59.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.114", +] + [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + [[package]] name = "windows-sys" version = "0.60.2" @@ -904,12 +1673,30 @@ version = "0.53.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" +[[package]] +name = "winnow" +version = "0.7.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a5364e9d77fcdeeaa6062ced926ee3381faa2ee02d3eb83a5c27a8825540829" +dependencies = [ + "memchr", +] + [[package]] name = "wit-bindgen" version = "0.51.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5" +[[package]] +name = "wyz" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "05f360fc0b24296329c78fda852a1e9ae82de9cf7b27dae4b7f62f118f77b9ed" +dependencies = [ + "tap", +] + [[package]] name = "zerocopy" version = "0.8.38" @@ -927,5 +1714,11 @@ checksum = "8a616990af1a287837c4fe6596ad77ef57948f787e46ce28e166facc0cc1cb75" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.114", ] + +[[package]] +name = "zmij" +version = "1.0.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ff05f8caa9038894637571ae6b9e29466c1f4f829d26c9b28f869a29cbe3445" diff --git a/crates/ember-cluster/Cargo.toml b/crates/ember-cluster/Cargo.toml index d8c6e274..7481b846 100644 --- a/crates/ember-cluster/Cargo.toml +++ b/crates/ember-cluster/Cargo.toml @@ -16,7 +16,9 @@ bytes = { workspace = true } rand = { workspace = true } tracing = { workspace = true } serde = { version = "1", features = ["derive"] } +serde_json = "1" uuid = { version = "1", features = ["v4", "serde"] } +openraft = { version = "0.9", features = ["serde"] } [dev-dependencies] tokio = { workspace = true, features = ["test-util", "macros"] } diff --git a/crates/ember-cluster/src/lib.rs b/crates/ember-cluster/src/lib.rs index f4c13ea9..7605d858 100644 --- a/crates/ember-cluster/src/lib.rs +++ b/crates/ember-cluster/src/lib.rs @@ -32,11 +32,16 @@ mod error; mod gossip; mod message; +mod raft; mod slots; mod topology; pub use error::ClusterError; pub use gossip::{GossipConfig, GossipEngine, GossipEvent, MemberState, MemberStatus}; pub use message::{GossipMessage, MemberInfo, NodeUpdate}; +pub use raft::{ + ClusterCommand, ClusterResponse, ClusterSnapshot, ClusterStateData, Storage as RaftStorage, + TypeConfig, +}; pub use slots::{key_slot, SlotMap, SlotRange, SLOT_COUNT}; pub use topology::{ClusterHealth, ClusterNode, ClusterState, NodeFlags, NodeId, NodeRole}; diff --git a/crates/ember-cluster/src/raft.rs b/crates/ember-cluster/src/raft.rs new file mode 100644 index 00000000..846bb193 --- /dev/null +++ b/crates/ember-cluster/src/raft.rs @@ -0,0 +1,594 @@ +//! Raft consensus for cluster configuration. +//! +//! Uses openraft to achieve consensus on cluster topology changes. +//! Only configuration changes go through Raft - data operations use +//! primary-replica async replication for lower latency. + +use std::collections::BTreeMap; +use std::fmt::Debug; +use std::io::Cursor; +use std::ops::RangeBounds; +use std::sync::Arc; + +use openraft::storage::{LogState, RaftLogReader, RaftSnapshotBuilder, Snapshot}; +use openraft::{ + BasicNode, Entry, EntryPayload, LogId, OptionalSend, RaftStorage, RaftTypeConfig, SnapshotMeta, + StorageError, StorageIOError, StoredMembership, Vote, +}; +use serde::{Deserialize, Serialize}; +use tokio::sync::RwLock; + +use crate::{NodeId, SlotRange}; + +/// Type configuration for openraft. +#[derive(Debug, Clone, Copy, Default, Eq, PartialEq, Ord, PartialOrd, Serialize, Deserialize)] +pub struct TypeConfig; + +impl RaftTypeConfig for TypeConfig { + type D = ClusterCommand; + type R = ClusterResponse; + type Node = BasicNode; + type NodeId = u64; + type Entry = Entry; + type SnapshotData = Cursor>; + type AsyncRuntime = openraft::TokioRuntime; + type Responder = openraft::impls::OneshotResponder; +} + +/// Commands that modify cluster configuration. +/// +/// These are replicated through Raft to ensure all nodes agree +/// on the cluster topology. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub enum ClusterCommand { + /// Add a new node to the cluster. + AddNode { + node_id: NodeId, + raft_id: u64, + addr: String, + is_primary: bool, + }, + /// Remove a node from the cluster. + RemoveNode { node_id: NodeId }, + /// Assign slots to a node. + AssignSlots { + node_id: NodeId, + slots: Vec, + }, + /// Promote a replica to primary (during failover). + PromoteReplica { replica_id: NodeId }, + /// Mark a slot as migrating. + BeginMigration { slot: u16, from: NodeId, to: NodeId }, + /// Complete a slot migration. + CompleteMigration { slot: u16, new_owner: NodeId }, +} + +/// Response from applying a cluster command. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +pub enum ClusterResponse { + Ok, + Error(String), +} + +/// State machine snapshot. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ClusterSnapshot { + pub last_applied: Option>, + pub last_membership: StoredMembership, + /// Serialized cluster state. + pub state_data: Vec, +} + +/// Internal cluster state managed by the state machine. +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ClusterStateData { + /// Node ID to raft ID mapping. + pub nodes: BTreeMap, + /// Slot assignments. + pub slots: BTreeMap, + /// Ongoing migrations. + pub migrations: BTreeMap, +} + +/// Information about a cluster node. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NodeInfo { + pub node_id: String, + pub raft_id: u64, + pub addr: String, + pub is_primary: bool, + pub slots: Vec, +} + +/// State of an ongoing slot migration. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MigrationState { + pub from: String, + pub to: String, +} + +/// Combined log and state machine storage for Raft. +#[derive(Debug)] +pub struct Storage { + vote: RwLock>>, + log: RwLock>>, + last_purged: RwLock>>, + last_applied: RwLock>>, + last_membership: RwLock>, + snapshot: RwLock>, + state: Arc>, +} + +#[derive(Debug, Clone)] +struct StoredSnapshot { + meta: SnapshotMeta, + data: Vec, +} + +impl Default for Storage { + fn default() -> Self { + Self::new() + } +} + +impl Storage { + pub fn new() -> Self { + Self { + vote: RwLock::new(None), + log: RwLock::new(BTreeMap::new()), + last_purged: RwLock::new(None), + last_applied: RwLock::new(None), + last_membership: RwLock::new(StoredMembership::default()), + snapshot: RwLock::new(None), + state: Arc::new(RwLock::new(ClusterStateData::default())), + } + } + + pub fn state(&self) -> Arc> { + Arc::clone(&self.state) + } + + fn apply_command(cmd: &ClusterCommand, state: &mut ClusterStateData) -> ClusterResponse { + match cmd { + ClusterCommand::AddNode { + node_id, + raft_id, + addr, + is_primary, + } => { + let key = node_id.0.to_string(); + state.nodes.insert( + key.clone(), + NodeInfo { + node_id: key, + raft_id: *raft_id, + addr: addr.clone(), + is_primary: *is_primary, + slots: Vec::new(), + }, + ); + ClusterResponse::Ok + } + + ClusterCommand::RemoveNode { node_id } => { + let key = node_id.0.to_string(); + state.nodes.remove(&key); + state.slots.retain(|_, owner| owner != &key); + ClusterResponse::Ok + } + + ClusterCommand::AssignSlots { node_id, slots } => { + let key = node_id.0.to_string(); + if let Some(node) = state.nodes.get_mut(&key) { + node.slots = slots.clone(); + for slot_range in slots { + for slot in slot_range.start..=slot_range.end { + state.slots.insert(slot, key.clone()); + } + } + ClusterResponse::Ok + } else { + ClusterResponse::Error(format!("node {} not found", node_id)) + } + } + + ClusterCommand::PromoteReplica { replica_id } => { + let key = replica_id.0.to_string(); + if let Some(node) = state.nodes.get_mut(&key) { + node.is_primary = true; + ClusterResponse::Ok + } else { + ClusterResponse::Error(format!("replica {} not found", replica_id)) + } + } + + ClusterCommand::BeginMigration { slot, from, to } => { + state.migrations.insert( + *slot, + MigrationState { + from: from.0.to_string(), + to: to.0.to_string(), + }, + ); + ClusterResponse::Ok + } + + ClusterCommand::CompleteMigration { slot, new_owner } => { + state.migrations.remove(slot); + let key = new_owner.0.to_string(); + state.slots.insert(*slot, key); + ClusterResponse::Ok + } + } + } +} + +impl RaftLogReader for Arc { + async fn try_get_log_entries + Clone + Debug + OptionalSend>( + &mut self, + range: RB, + ) -> Result>, StorageError> { + let log = self.log.read().await; + Ok(log.range(range).map(|(_, v)| v.clone()).collect()) + } +} + +impl RaftSnapshotBuilder for Arc { + async fn build_snapshot(&mut self) -> Result, StorageError> { + let last_applied = *self.last_applied.read().await; + let membership = self.last_membership.read().await.clone(); + let state = self.state.read().await; + + let state_data = + serde_json::to_vec(&*state).map_err(|e| StorageIOError::write_snapshot(None, &e))?; + + let snapshot = ClusterSnapshot { + last_applied, + last_membership: membership.clone(), + state_data, + }; + + let data = + serde_json::to_vec(&snapshot).map_err(|e| StorageIOError::write_snapshot(None, &e))?; + + let snapshot_id = last_applied + .map(|id| format!("{}-{}", id.leader_id, id.index)) + .unwrap_or_else(|| "0-0".to_string()); + + let meta = SnapshotMeta { + last_log_id: last_applied, + last_membership: membership, + snapshot_id, + }; + + // Store the snapshot + *self.snapshot.write().await = Some(StoredSnapshot { + meta: meta.clone(), + data: data.clone(), + }); + + Ok(Snapshot { + meta, + snapshot: Box::new(Cursor::new(data)), + }) + } +} + +impl RaftStorage for Arc { + type LogReader = Self; + type SnapshotBuilder = Self; + + async fn get_log_state(&mut self) -> Result, StorageError> { + let log = self.log.read().await; + let last = log.iter().next_back().map(|(_, e)| e.log_id); + let purged = *self.last_purged.read().await; + + Ok(LogState { + last_purged_log_id: purged, + last_log_id: last, + }) + } + + async fn save_vote(&mut self, vote: &Vote) -> Result<(), StorageError> { + *self.vote.write().await = Some(*vote); + Ok(()) + } + + async fn read_vote(&mut self) -> Result>, StorageError> { + Ok(*self.vote.read().await) + } + + async fn get_log_reader(&mut self) -> Self::LogReader { + Arc::clone(self) + } + + async fn append_to_log(&mut self, entries: I) -> Result<(), StorageError> + where + I: IntoIterator> + Send, + { + let mut log = self.log.write().await; + for entry in entries { + log.insert(entry.log_id.index, entry); + } + Ok(()) + } + + async fn delete_conflict_logs_since( + &mut self, + log_id: LogId, + ) -> Result<(), StorageError> { + let mut log = self.log.write().await; + let to_remove: Vec<_> = log.range(log_id.index..).map(|(k, _)| *k).collect(); + for key in to_remove { + log.remove(&key); + } + Ok(()) + } + + async fn purge_logs_upto(&mut self, log_id: LogId) -> Result<(), StorageError> { + let mut log = self.log.write().await; + let to_remove: Vec<_> = log.range(..=log_id.index).map(|(k, _)| *k).collect(); + for key in to_remove { + log.remove(&key); + } + *self.last_purged.write().await = Some(log_id); + Ok(()) + } + + async fn last_applied_state( + &mut self, + ) -> Result<(Option>, StoredMembership), StorageError> { + let last_applied = *self.last_applied.read().await; + let membership = self.last_membership.read().await.clone(); + Ok((last_applied, membership)) + } + + async fn apply_to_state_machine( + &mut self, + entries: &[Entry], + ) -> Result, StorageError> { + let mut results = Vec::new(); + let mut state = self.state.write().await; + + for entry in entries { + *self.last_applied.write().await = Some(entry.log_id); + + match &entry.payload { + EntryPayload::Blank => { + results.push(ClusterResponse::Ok); + } + EntryPayload::Normal(cmd) => { + let result = Storage::apply_command(cmd, &mut state); + results.push(result); + } + EntryPayload::Membership(m) => { + *self.last_membership.write().await = + StoredMembership::new(Some(entry.log_id), m.clone()); + results.push(ClusterResponse::Ok); + } + } + } + + Ok(results) + } + + async fn get_snapshot_builder(&mut self) -> Self::SnapshotBuilder { + Arc::clone(self) + } + + async fn begin_receiving_snapshot( + &mut self, + ) -> Result>>, StorageError> { + Ok(Box::new(Cursor::new(Vec::new()))) + } + + async fn install_snapshot( + &mut self, + meta: &SnapshotMeta, + snapshot: Box>>, + ) -> Result<(), StorageError> { + let data = snapshot.into_inner(); + let snap: ClusterSnapshot = serde_json::from_slice(&data) + .map_err(|e| StorageIOError::read_snapshot(Some(meta.signature()), &e))?; + + *self.last_applied.write().await = snap.last_applied; + *self.last_membership.write().await = snap.last_membership; + + let state_data: ClusterStateData = serde_json::from_slice(&snap.state_data) + .map_err(|e| StorageIOError::read_snapshot(Some(meta.signature()), &e))?; + *self.state.write().await = state_data; + + *self.snapshot.write().await = Some(StoredSnapshot { + meta: meta.clone(), + data, + }); + + Ok(()) + } + + async fn get_current_snapshot( + &mut self, + ) -> Result>, StorageError> { + let snap = self.snapshot.read().await; + Ok(snap.as_ref().map(|s| Snapshot { + meta: s.meta.clone(), + snapshot: Box::new(Cursor::new(s.data.clone())), + })) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use openraft::CommittedLeaderId; + + /// Helper to create a LogId for tests. + fn log_id(term: u64, index: u64) -> LogId { + LogId::new(CommittedLeaderId::new(term, 0), index) + } + + #[tokio::test] + async fn storage_add_node() { + let storage = Arc::new(Storage::new()); + let mut storage_clone = Arc::clone(&storage); + + let node_id = NodeId::new(); + let entry = Entry { + log_id: log_id(1, 1), + payload: EntryPayload::Normal(ClusterCommand::AddNode { + node_id, + raft_id: 1, + addr: "127.0.0.1:6379".to_string(), + is_primary: true, + }), + }; + + let results = storage_clone + .apply_to_state_machine(&[entry]) + .await + .unwrap(); + assert_eq!(results, vec![ClusterResponse::Ok]); + + let state_arc = storage.state(); + let state = state_arc.read().await; + assert!(state.nodes.contains_key(&node_id.0.to_string())); + } + + #[tokio::test] + async fn storage_assign_slots() { + let storage = Arc::new(Storage::new()); + let mut storage_clone = Arc::clone(&storage); + + let node_id = NodeId::new(); + + // Add node first + let add_entry = Entry { + log_id: log_id(1, 1), + payload: EntryPayload::Normal(ClusterCommand::AddNode { + node_id, + raft_id: 1, + addr: "127.0.0.1:6379".to_string(), + is_primary: true, + }), + }; + storage_clone + .apply_to_state_machine(&[add_entry]) + .await + .unwrap(); + + // Assign slots + let assign_entry = Entry { + log_id: log_id(1, 2), + payload: EntryPayload::Normal(ClusterCommand::AssignSlots { + node_id, + slots: vec![SlotRange::new(0, 5460)], + }), + }; + let results = storage_clone + .apply_to_state_machine(&[assign_entry]) + .await + .unwrap(); + assert_eq!(results, vec![ClusterResponse::Ok]); + + let state_arc = storage.state(); + let state = state_arc.read().await; + assert_eq!(state.slots.get(&0), Some(&node_id.0.to_string())); + assert_eq!(state.slots.get(&5460), Some(&node_id.0.to_string())); + } + + #[tokio::test] + async fn storage_migration() { + let storage = Arc::new(Storage::new()); + let mut storage_clone = Arc::clone(&storage); + + let node1 = NodeId::new(); + let node2 = NodeId::new(); + + // Add nodes + let entries: Vec> = [node1, node2] + .iter() + .enumerate() + .map(|(i, node_id)| Entry { + log_id: log_id(1, i as u64 + 1), + payload: EntryPayload::Normal(ClusterCommand::AddNode { + node_id: *node_id, + raft_id: i as u64 + 1, + addr: format!("127.0.0.1:{}", 6379 + i), + is_primary: true, + }), + }) + .collect(); + storage_clone + .apply_to_state_machine(&entries) + .await + .unwrap(); + + // Begin migration + let begin_entry = Entry { + log_id: log_id(1, 3), + payload: EntryPayload::Normal(ClusterCommand::BeginMigration { + slot: 100, + from: node1, + to: node2, + }), + }; + storage_clone + .apply_to_state_machine(&[begin_entry]) + .await + .unwrap(); + + { + let state_arc = storage.state(); + let state = state_arc.read().await; + assert!(state.migrations.contains_key(&100)); + } + + // Complete migration + let complete_entry = Entry { + log_id: log_id(1, 4), + payload: EntryPayload::Normal(ClusterCommand::CompleteMigration { + slot: 100, + new_owner: node2, + }), + }; + storage_clone + .apply_to_state_machine(&[complete_entry]) + .await + .unwrap(); + + { + let state_arc = storage.state(); + let state = state_arc.read().await; + assert!(!state.migrations.contains_key(&100)); + assert_eq!(state.slots.get(&100), Some(&node2.0.to_string())); + } + } + + #[tokio::test] + async fn storage_log_operations() { + let storage = Arc::new(Storage::new()); + let mut storage_clone = Arc::clone(&storage); + + let entry = Entry:: { + log_id: log_id(1, 1), + payload: EntryPayload::Blank, + }; + + storage_clone.append_to_log(vec![entry]).await.unwrap(); + + let state = storage_clone.get_log_state().await.unwrap(); + assert_eq!(state.last_log_id, Some(log_id(1, 1))); + } + + #[tokio::test] + async fn storage_vote() { + let storage = Arc::new(Storage::new()); + let mut storage_clone = Arc::clone(&storage); + + let vote = Vote::new(1, 1); + storage_clone.save_vote(&vote).await.unwrap(); + + let read_vote = storage_clone.read_vote().await.unwrap(); + assert_eq!(read_vote, Some(vote)); + } +}