From ff3fe59fcccef2e7a13d4da175282847b45f5497 Mon Sep 17 00:00:00 2001 From: Nelson Dominguez Date: Sat, 28 Feb 2026 15:28:47 +0100 Subject: [PATCH 1/3] Add HTTP layer --- Cargo.lock | 570 ++++++++++++++++++++++++++++++++++++ Cargo.toml | 7 +- minikv/Cargo.toml | 15 + minikv/src/cli.rs | 210 +++++++++++++ minikv/src/error.rs | 15 + minikv/src/http/handlers.rs | 489 +++++++++++++++++++++++++++++++ minikv/src/http/mod.rs | 22 ++ minikv/src/http/query.rs | 209 +++++++++++++ minikv/src/http/s3.rs | 128 ++++++++ minikv/src/main.rs | 178 ++++++++++- 10 files changed, 1840 insertions(+), 3 deletions(-) create mode 100644 minikv/src/cli.rs create mode 100644 minikv/src/error.rs create mode 100644 minikv/src/http/handlers.rs create mode 100644 minikv/src/http/mod.rs create mode 100644 minikv/src/http/query.rs create mode 100644 minikv/src/http/s3.rs diff --git a/Cargo.lock b/Cargo.lock index 468a305..5b4c9b3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -11,6 +11,62 @@ dependencies = [ "memchr", ] +[[package]] +name = "anstream" +version = "0.6.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "43d5b281e737544384e969a5ccad3f1cdd24b48086a0fc1b2a5262a26b8f4f4a" +dependencies = [ + "anstyle", + "anstyle-parse", + "anstyle-query", + "anstyle-wincon", + "colorchoice", + "is_terminal_polyfill", + "utf8parse", +] + +[[package]] +name = "anstyle" +version = "1.0.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5192cca8006f1fd4f7237516f40fa183bb07f8fbdfedaa0036de5ea9b0b45e78" + +[[package]] +name = "anstyle-parse" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7644824f0aa2c7b9384579234ef10eb7efb6a0deb83f9630a49594dd9c15c2" +dependencies = [ + "utf8parse", +] + +[[package]] +name = "anstyle-query" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "40c48f72fd53cd289104fc64099abca73db4166ad86ea0b4341abe65af83dadc" +dependencies = [ + "windows-sys 0.61.2", +] + +[[package]] +name = "anstyle-wincon" +version = "3.0.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "291e6a250ff86cd4a820112fb8898808a366d8f9f58ce16d1f538353ad55747d" +dependencies = [ + "anstyle", + "once_cell_polyfill", + "windows-sys 0.61.2", +] + +[[package]] +name = "anyhow" +version = "1.0.102" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" + [[package]] name = "arrayref" version = "0.3.9" @@ -33,6 +89,17 @@ dependencies = [ "serde_json", ] +[[package]] +name = "async-trait" +version = "0.1.89" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "atomic-waker" version = "1.1.2" @@ -61,6 +128,61 @@ dependencies = [ "fs_extra", ] +[[package]] +name = "axum" +version = "0.7.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" +dependencies = [ + "async-trait", + "axum-core", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "rustversion", + "serde", + "serde_json", + "serde_path_to_error", + "serde_urlencoded", + "sync_wrapper", + "tokio", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "axum-core" +version = "0.4.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09f2bd6146b97ae3359fa0cc6d6b376d9539582c7b4220f041a33ec24c226199" +dependencies = [ + "async-trait", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "rustversion", + "sync_wrapper", + "tower-layer", + "tower-service", + "tracing", +] + [[package]] name = "base64" version = "0.22.1" @@ -135,6 +257,46 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "clap" +version = "4.5.60" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2797f34da339ce31042b27d23607e051786132987f595b02ba4f6a6dffb7030a" +dependencies = [ + "clap_builder", + "clap_derive", +] + +[[package]] +name = "clap_builder" +version = "4.5.60" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24a241312cea5059b13574bb9b3861cabf758b879c15190b37b6d6fd63ab6876" +dependencies = [ + "anstream", + "anstyle", + "clap_lex", + "strsim", +] + +[[package]] +name = "clap_derive" +version = "4.5.55" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a92793da1a46a5f2a02a6f4c46c6496b28c43638adea8306fcb0caa1634f24e5" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "clap_lex" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a822ea5bc7590f9d40f1ba12c0dc3c2760f3482c6984db1573ad11031420831" + [[package]] name = "cmake" version = "0.1.57" @@ -144,6 +306,12 @@ dependencies = [ "cc", ] +[[package]] +name = "colorchoice" +version = "1.0.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b05b61dc5112cbb17e4b6cd61790d9845d13888356391624cbe7e41efeac1e75" + [[package]] name = "combine" version = "4.6.7" @@ -298,6 +466,12 @@ version = "1.0.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" +[[package]] +name = "foldhash" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -449,6 +623,19 @@ dependencies = [ "wasm-bindgen", ] +[[package]] +name = "getrandom" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "139ef39800118c7683f2fd3c98c1b23c09ae076556b435f8e9064ae108aaeeec" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasip2", + "wasip3", +] + [[package]] name = "h2" version = "0.4.13" @@ -474,12 +661,27 @@ version = "0.14.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +[[package]] +name = "hashbrown" +version = "0.15.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" +dependencies = [ + "foldhash", +] + [[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 = "hermit-abi" version = "0.5.2" @@ -680,6 +882,12 @@ dependencies = [ "zerovec", ] +[[package]] +name = "id-arena" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d3067d79b975e8844ca9eb072e16b31c3c1c36928edf9c6789548c524d0d954" + [[package]] name = "idna" version = "1.1.0" @@ -709,6 +917,8 @@ checksum = "7714e70437a7dc3ac8eb7e6f8df75fd8eb422675fc7678aff7364301092b1017" dependencies = [ "equivalent", "hashbrown 0.16.1", + "serde", + "serde_core", ] [[package]] @@ -733,6 +943,12 @@ dependencies = [ "serde", ] +[[package]] +name = "is_terminal_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6cb138bb79a146c1bd460005623e142ef0181e3d0219cb493e02f7d08a35695" + [[package]] name = "itoa" version = "1.0.17" @@ -787,6 +1003,12 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" +[[package]] +name = "leb128fmt" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09edd9e8b54e49e587e4f6295a7d29c3ea94d469cb40ab8ca70b288248a81db2" + [[package]] name = "libc" version = "0.2.182" @@ -820,15 +1042,53 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" +[[package]] +name = "matchers" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9" +dependencies = [ + "regex-automata", +] + +[[package]] +name = "matchit" +version = "0.7.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" + [[package]] name = "memchr" version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "minikv" version = "0.1.0" +dependencies = [ + "axum", + "bytes", + "clap", + "dashmap", + "futures", + "minikv-core", + "quick-xml", + "rand 0.8.5", + "serde", + "serde_json", + "thiserror 2.0.18", + "tokio", + "tracing", + "tracing-subscriber", + "uuid", +] [[package]] name = "minikv-core" @@ -860,6 +1120,15 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "nu-ansi-term" +version = "0.50.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "num_cpus" version = "1.17.0" @@ -876,6 +1145,12 @@ version = "1.21.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42f5e15c9953c5e4ccceeb2e7382a716482c34515315f7b03532b8b4e8393d2d" +[[package]] +name = "once_cell_polyfill" +version = "1.70.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "384b8ab6d37215f3c5301a95a4accb5d64aa607f1fcb26a11b5303878451b4fe" + [[package]] name = "openssl-probe" version = "0.2.1" @@ -941,6 +1216,16 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "prettyplease" +version = "0.2.37" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" +dependencies = [ + "proc-macro2", + "syn", +] + [[package]] name = "proc-macro2" version = "1.0.106" @@ -950,6 +1235,16 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "quick-xml" +version = "0.31.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1004a344b30a54e2ee58d66a71b32d2db2feb0a31f9a2d302bf0536f15de2a33" +dependencies = [ + "memchr", + "serde", +] + [[package]] name = "quinn" version = "0.11.9" @@ -1034,6 +1329,17 @@ dependencies = [ "rand_hc", ] +[[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" @@ -1054,6 +1360,16 @@ dependencies = [ "rand_core 0.5.1", ] +[[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]] name = "rand_chacha" version = "0.9.0" @@ -1073,6 +1389,15 @@ dependencies = [ "getrandom 0.1.16", ] +[[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]] name = "rand_core" version = "0.9.5" @@ -1284,6 +1609,12 @@ dependencies = [ "snap", ] +[[package]] +name = "ryu" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f" + [[package]] name = "same-file" version = "1.0.6" @@ -1331,6 +1662,12 @@ dependencies = [ "libc", ] +[[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" @@ -1374,6 +1711,38 @@ dependencies = [ "zmij", ] +[[package]] +name = "serde_path_to_error" +version = "0.1.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "10a9ff822e371bb5403e391ecd83e182e0e77ba7f6fe0160b795797109d1b457" +dependencies = [ + "itoa", + "serde", + "serde_core", +] + +[[package]] +name = "serde_urlencoded" +version = "0.7.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" +dependencies = [ + "form_urlencoded", + "itoa", + "ryu", + "serde", +] + +[[package]] +name = "sharded-slab" +version = "0.1.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6" +dependencies = [ + "lazy_static", +] + [[package]] name = "shlex" version = "1.3.0" @@ -1424,6 +1793,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" +[[package]] +name = "strsim" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7da8b5736845d9f2fcb837ea5d9e2628564b3b043a70948a3f0b778838c5fb4f" + [[package]] name = "subtle" version = "2.6.1" @@ -1501,6 +1876,15 @@ dependencies = [ "syn", ] +[[package]] +name = "thread_local" +version = "1.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185" +dependencies = [ + "cfg-if", +] + [[package]] name = "tinystr" version = "0.8.2" @@ -1590,6 +1974,7 @@ dependencies = [ "tokio", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -1628,6 +2013,7 @@ version = "0.1.44" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63e71662fa4b2a2c3a26f570f037eb95bb1f85397f3cd8076caed2f026a6d100" dependencies = [ + "log", "pin-project-lite", "tracing-attributes", "tracing-core", @@ -1651,6 +2037,36 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "db97caf9d906fbde555dd62fa95ddba9eecfd14cb388e4f491a66d74cd5fb79a" dependencies = [ "once_cell", + "valuable", +] + +[[package]] +name = "tracing-log" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3" +dependencies = [ + "log", + "once_cell", + "tracing-core", +] + +[[package]] +name = "tracing-subscriber" +version = "0.3.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f30143827ddab0d256fd843b7a66d164e9f271cfa0dde49142c5ca0ca291f1e" +dependencies = [ + "matchers", + "nu-ansi-term", + "once_cell", + "regex-automata", + "sharded-slab", + "smallvec", + "thread_local", + "tracing", + "tracing-core", + "tracing-log", ] [[package]] @@ -1665,6 +2081,12 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unicode-xid" +version = "0.2.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ebc1c04c71510c7f702b52b7c350734c9ff1295c464a03335b00bb84fc54f853" + [[package]] name = "untrusted" version = "0.9.0" @@ -1689,6 +2111,29 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6c140620e7ffbb22c2dee59cafe6084a59b5ffc27a8859a5f0d494b5d52b6be" +[[package]] +name = "utf8parse" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" + +[[package]] +name = "uuid" +version = "1.21.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b672338555252d43fd2240c714dc444b8c6fb0a5c5335e65a07bba7742735ddb" +dependencies = [ + "getrandom 0.4.1", + "js-sys", + "wasm-bindgen", +] + +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "walkdir" version = "2.5.0" @@ -1729,6 +2174,15 @@ dependencies = [ "wit-bindgen", ] +[[package]] +name = "wasip3" +version = "0.4.0+wasi-0.3.0-rc-2026-01-06" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5428f8bf88ea5ddc08faddef2ac4a67e390b88186c703ce6dbd955e1c145aca5" +dependencies = [ + "wit-bindgen", +] + [[package]] name = "wasm-bindgen" version = "0.2.114" @@ -1788,6 +2242,28 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "wasm-encoder" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "990065f2fe63003fe337b932cfb5e3b80e0b4d0f5ff650e6985b1048f62c8319" +dependencies = [ + "leb128fmt", + "wasmparser", +] + +[[package]] +name = "wasm-metadata" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909" +dependencies = [ + "anyhow", + "indexmap", + "wasm-encoder", + "wasmparser", +] + [[package]] name = "wasm-streams" version = "0.5.0" @@ -1801,6 +2277,18 @@ dependencies = [ "web-sys", ] +[[package]] +name = "wasmparser" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe" +dependencies = [ + "bitflags", + "hashbrown 0.15.5", + "indexmap", + "semver", +] + [[package]] name = "web-sys" version = "0.3.91" @@ -2117,6 +2605,88 @@ name = "wit-bindgen" version = "0.51.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d7249219f66ced02969388cf2bb044a09756a083d0fab1e566056b04d9fbcaa5" +dependencies = [ + "wit-bindgen-rust-macro", +] + +[[package]] +name = "wit-bindgen-core" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea61de684c3ea68cb082b7a88508a8b27fcc8b797d738bfc99a82facf1d752dc" +dependencies = [ + "anyhow", + "heck", + "wit-parser", +] + +[[package]] +name = "wit-bindgen-rust" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21" +dependencies = [ + "anyhow", + "heck", + "indexmap", + "prettyplease", + "syn", + "wasm-metadata", + "wit-bindgen-core", + "wit-component", +] + +[[package]] +name = "wit-bindgen-rust-macro" +version = "0.51.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c0f9bfd77e6a48eccf51359e3ae77140a7f50b1e2ebfe62422d8afdaffab17a" +dependencies = [ + "anyhow", + "prettyplease", + "proc-macro2", + "quote", + "syn", + "wit-bindgen-core", + "wit-bindgen-rust", +] + +[[package]] +name = "wit-component" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2" +dependencies = [ + "anyhow", + "bitflags", + "indexmap", + "log", + "serde", + "serde_derive", + "serde_json", + "wasm-encoder", + "wasm-metadata", + "wasmparser", + "wit-parser", +] + +[[package]] +name = "wit-parser" +version = "0.244.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736" +dependencies = [ + "anyhow", + "id-arena", + "indexmap", + "log", + "semver", + "serde", + "serde_derive", + "serde_json", + "unicode-xid", + "wasmparser", +] [[package]] name = "writeable" diff --git a/Cargo.toml b/Cargo.toml index a77b041..da80326 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,6 +14,8 @@ keywords = ["storage", "distributed", "leveldb", "object-store"] categories = ["database", "network-programming"] [workspace.dependencies] +minikv-core = { path = "minikv-core" } + thiserror = "2" tracing = "0.1" @@ -26,8 +28,9 @@ dashmap = "6.1" tokio = { version = "1.49", default-features = false } reqwest = { version = "0.13", default-features = false } bytes = { version = "1.11" } - -# Chosen over `leveldb` crate which requires C++ LevelDB via FFI. +uuid = { version = "1", features = ["v4"] } +rand = "0.8" +futures = "0.3" rusty-leveldb = "1" # Testing diff --git a/minikv/Cargo.toml b/minikv/Cargo.toml index 3290225..6216031 100644 --- a/minikv/Cargo.toml +++ b/minikv/Cargo.toml @@ -11,3 +11,18 @@ keywords.workspace = true categories.workspace = true [dependencies] +minikv-core = { workspace = true } +tracing = { workspace = true } +dashmap = { workspace = true } +thiserror = { workspace = true } +bytes = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +futures = { workspace = true } +rand = { workspace = true } +uuid = { workspace = true } +tokio = { workspace = true, features = ["full"] } +clap = { version = "4", features = ["derive", "env"] } +tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } +axum = { version = "0.7", features = [] } +quick-xml = { version = "0.31", features = ["serialize"] } diff --git a/minikv/src/cli.rs b/minikv/src/cli.rs new file mode 100644 index 0000000..f8c3b19 --- /dev/null +++ b/minikv/src/cli.rs @@ -0,0 +1,210 @@ +/// CLI definition for `minikv`. +/// +/// # Subcommands +/// - `server` run the HTTP metadata coordinator +/// - `rebuild` reconstruct LevelDB from volume server autoindex +/// - `rebalance` move all keys to their ideal volume set +/// - `print-nginx-config` emit the nginx volume server config to stdout +use std::path::PathBuf; +use std::time::Duration; + +use clap::{Args, Parser, Subcommand}; + +#[derive(Parser, Debug)] +#[command( + name = "minikv", + version, + about = "Distributed object storage coordinator", + long_about = None, +)] +pub struct Cli { + #[command(subcommand)] + pub command: Command, +} + +#[derive(Subcommand, Debug)] +pub enum Command { + /// Run the HTTP metadata coordinator server. + Server(ServerArgs), + + /// Rebuild LevelDB metadata from volume server autoindex listings. + /// + /// WARNING: This clears the entire DB before scanning. + Rebuild(CommonArgs), + + /// Rebalance all keys to their ideal volume set. + Rebalance(CommonArgs), + + /// Print the nginx volume server configuration to stdout. + PrintNginxConfig, +} + +/// Arguments shared by server, rebuild, and rebalance. +#[derive(Args, Debug, Clone)] +pub struct CommonArgs { + /// Path to the LevelDB database directory. + #[arg(long, env = "MINIKV_DB", required = true)] + pub db: PathBuf, + + /// Comma-separated list of volume server addresses (host:port). + #[arg(long, env = "MINIKV_VOLUMES", required = true, value_delimiter = ',')] + pub volumes: Vec, + + /// Number of replicas to maintain per object. + #[arg(long, env = "MINIKV_REPLICAS", default_value = "3")] + pub replicas: usize, + + /// Number of sub-volume shards per volume server. + /// Use 1 to disable sub-volume path components. + #[arg(long, env = "MINIKV_SUBVOLUMES", default_value = "10")] + pub subvolumes: usize, + + /// Timeout for HEAD probes to volume servers (e.g. "1s", "500ms"). + #[arg(long, env = "MINIKV_VOLTIMEOUT", default_value = "1s", value_parser = parse_duration)] + pub voltimeout: Duration, +} + +/// Additional arguments for the `server` subcommand. +#[derive(Args, Debug)] +pub struct ServerArgs { + #[command(flatten)] + pub common: CommonArgs, + + /// Port to listen on. + #[arg(long, env = "MINIKV_PORT", default_value = "3000")] + pub port: u16, + + /// Optional fallback server for keys missing from all volumes. + #[arg(long, env = "MINIKV_FALLBACK")] + pub fallback: Option, + + /// Public-facing addresses for each volume server, used in Location + /// redirect headers returned to clients. + /// + /// Must be the same length as --volumes and in the same order. + /// Each entry is the externally-reachable host:port for the corresponding + /// --volumes entry. + /// + /// Example: + /// --volumes=volume1:8080,volume2:8080,volume3:8080 + /// --public-volumes=localhost:8001,localhost:8002,localhost:8003 + /// + /// When omitted, --volumes addresses are used as-is in Location headers + /// (correct for bare-metal deployments where internal == external). + #[arg( + long, + env = "MINIKV_PUBLIC_VOLUMES", + value_delimiter = ',', + requires = "volumes" + )] + pub public_volumes: Option>, + + /// Require UNLINK (soft-delete) before a hard DELETE is allowed. + #[arg(long, env = "MINIKV_PROTECT", default_value = "false")] + pub protect: bool, + + /// Compute and store a BLAKE3 checksum for every uploaded object. + #[arg(long, env = "MINIKV_CHECKSUM", default_value = "true")] + pub checksum: bool, + + /// Enable X-Accel-Redirect mode for GET/HEAD responses. + /// + /// When enabled, the coordinator returns `X-Accel-Redirect: /accel//` + /// instead of `302 Location`. A frontend nginx must be configured with: + /// + /// proxy_pass http://coordinator:3000; + /// location ~ ^/accel/([^/]+)/(.+)$ { + /// internal; + /// proxy_pass http://$1/$2; + /// } + /// + /// This allows nginx to stream the object body while the coordinator controls + /// all response headers, including Content-Type from stored object metadata. + /// + /// When disabled (default), GET/HEAD returns a standard 302 redirect. + #[arg(long, env = "MINIKV_ACCEL_REDIRECT", default_value = "false")] + pub accel_redirect: bool, + + /// Enable verbose structured logging. + #[arg(short, long, env = "MINIKV_VERBOSE", default_value = "false")] + pub verbose: bool, +} + +/// Parse a human-friendly duration string such as "1s" or "500ms". +fn parse_duration(s: &str) -> Result { + if let Some(ms) = s.strip_suffix("ms") { + ms.parse::() + .map(Duration::from_millis) + .map_err(|e| e.to_string()) + } else if let Some(secs) = s.strip_suffix('s') { + secs.parse::() + .map(Duration::from_secs) + .map_err(|e| e.to_string()) + } else { + // Fall back: treat as integer milliseconds. + s.parse::() + .map(Duration::from_millis) + .map_err(|_| format!("invalid duration '{s}'. Use '1s' or '500ms'")) + } +} + +/// Validate common arguments and panic with a clear message on bad input. +pub fn validate_common(args: &CommonArgs) { + if args.volumes.is_empty() { + eprintln!("error: --volumes must contain at least one volume server"); + std::process::exit(1); + } + if args.replicas == 0 { + eprintln!("error: --replicas must be ≥ 1"); + std::process::exit(1); + } + if args.volumes.len() < args.replicas { + eprintln!( + "error: need at least as many volumes ({}) as replicas ({})", + args.volumes.len(), + args.replicas + ); + std::process::exit(1); + } + if args.subvolumes == 0 { + eprintln!("error: --subvolumes must be ≥ 1"); + std::process::exit(1); + } +} + +/// Validate server-only arguments. +#[allow(unused)] +pub fn validate_server(args: &ServerArgs) { + validate_common(&args.common); + if let Some(ref pv) = args.public_volumes + && pv.len() != args.common.volumes.len() + { + eprintln!( + "error: --public-volumes has {} entries but --volumes has {}; they must match 1:1", + pv.len(), + args.common.volumes.len() + ); + std::process::exit(1); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_duration_seconds() { + assert_eq!(parse_duration("1s").unwrap(), Duration::from_secs(1)); + assert_eq!(parse_duration("30s").unwrap(), Duration::from_secs(30)); + } + + #[test] + fn parse_duration_millis() { + assert_eq!(parse_duration("500ms").unwrap(), Duration::from_millis(500)); + } + + #[test] + fn parse_duration_invalid() { + assert!(parse_duration("1h").is_err()); + } +} diff --git a/minikv/src/error.rs b/minikv/src/error.rs new file mode 100644 index 0000000..61f59f7 --- /dev/null +++ b/minikv/src/error.rs @@ -0,0 +1,15 @@ +/// Unified error type for minikv. +#[derive(Debug, thiserror::Error)] +pub enum Error { + /// Standard I/O errors (temp file handling for multipart uploads). + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + + /// XML deserialization errors for S3-compatible API bodies. + #[error("XML parse error: {0}")] + XmlParse(String), + + /// Minikv core library error + #[error(transparent)] + Core(#[from] minikv_core::Error), +} diff --git a/minikv/src/http/handlers.rs b/minikv/src/http/handlers.rs new file mode 100644 index 0000000..17f8dde --- /dev/null +++ b/minikv/src/http/handlers.rs @@ -0,0 +1,489 @@ +/// HTTP request handlers +/// +/// All standard and custom methods (UNLINK, REBALANCE) are handled here. +/// +/// # Method dispatch +/// ```text +/// GET / HEAD → 302 redirect to volume server (after probe) +/// PUT → write to replicas +/// POST ?uploads → initiate multipart +/// POST ?uploadId=X → complete multipart +/// POST ?delete → batch delete +/// DELETE → hard delete (requires prior UNLINK if protect=true) +/// UNLINK → soft delete +/// REBALANCE → move key to ideal volumes +/// GET ?list → list active keys +/// GET ?unlinked→ list soft-deleted keys +/// ``` +use std::sync::Arc; + +use axum::body::Body; +use axum::extract::{Path, RawQuery, State}; +use axum::http::{HeaderMap, Method, StatusCode}; +use axum::response::{IntoResponse, Response}; +use bytes::Bytes; +use rand::seq::SliceRandom; +use tracing::{debug, info, instrument, warn}; +use uuid::Uuid; + +use crate::http::query::handle_query; +use crate::http::s3::{CompleteMultipartUpload, Delete}; +use minikv_core::hashing::key_to_path; +use minikv_core::rebalance::rebalance_key; +use minikv_core::record::Deleted; +use minikv_core::replication::remote_head; +use minikv_core::state::AppState; +use minikv_core::volumes::key_to_volume; + +/// Top-level axum handler for all routes (`/*key`). +/// +/// Non-standard methods (UNLINK, REBALANCE) are routed via `axum::routing::any` +/// and dispatched here by inspecting `req.method()`. +#[instrument(skip(state, headers, body), fields(method = %method, key = %key))] +pub async fn handle( + method: Method, + State(state): State>, + Path(key): Path, + RawQuery(raw_query): RawQuery, + headers: HeaderMap, + body: Bytes, +) -> Response { + let key_bytes = key.as_bytes(); + let raw_query = raw_query.unwrap_or_default(); + + info!(method = %method, key, "request"); + + // GET with a non-empty query string → list/query handler. + if method == Method::GET && !raw_query.is_empty() { + return handle_query(state, key_bytes, &raw_query).await; + } + + // Build the per-operation lock key. For multipart part uploads + // (PUT ?partNumber=N) we lock key+partNumber so concurrent part + // uploads don't collide. + let lock_key = if method == Method::PUT || method == Method::POST { + let part_number = extract_query_param(&raw_query, "partNumber"); + if let Some(pn) = part_number { + format!("{key}?partNumber={pn}") + } else { + key.clone() + } + } else { + key.clone() + }; + + // Acquire per-key write lock for mutating methods. + let needs_lock = method == Method::PUT + || method == Method::POST + || method == Method::DELETE + || method.as_str() == "UNLINK" + || method.as_str() == "REBALANCE"; + + let _guard = if needs_lock { + match state.key_lock.try_lock(&lock_key) { + Some(g) => Some(g), + None => { + debug!(key, "key locked by concurrent operation"); + return StatusCode::CONFLICT.into_response(); + } + } + } else { + None + }; + + match method.as_str() { + "GET" | "HEAD" => handle_get_head(&state, key_bytes, &method).await, + "PUT" => handle_put(&state, key_bytes, &raw_query, headers, body).await, + "POST" => handle_post(&state, key_bytes, &raw_query, headers, body).await, + "DELETE" => { + let status = state.delete(key_bytes, false).await; + StatusCode::from_u16(status) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) + .into_response() + } + "UNLINK" => { + let status = state.delete(key_bytes, true).await; + StatusCode::from_u16(status) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) + .into_response() + } + "REBALANCE" => handle_rebalance(&state, key_bytes).await, + _ => StatusCode::METHOD_NOT_ALLOWED.into_response(), + } +} + +/// Handle GET and HEAD. Return 302 redirect to the object on a volume server. +async fn handle_get_head(state: &AppState, key: &[u8], _method: &Method) -> Response { + let rec = state.get_record(key).await; + let kp = key_to_path(key); + + // Build response with optional hash header. + let mut resp_builder = Response::builder(); + if let Some(ref hash) = rec.hash { + resp_builder = resp_builder.header("Content-Blake3", hash); + } + + let remote: String; + + if rec.deleted == Deleted::Soft || rec.deleted == Deleted::Hard { + // Key doesn't exist; check fallback. + if let Some(ref fb) = state.fallback { + remote = format!("http://{fb}{}", String::from_utf8_lossy(key)); + } else { + return ( + StatusCode::NOT_FOUND, + [(axum::http::header::CONTENT_LENGTH, "0")], + ) + .into_response(); + } + } else { + // Key exists. Check which volumes report it as present. + let kvolumes = key_to_volume(key, &state.volumes, state.replicas, state.subvolumes); + let balance = if minikv_core::volumes::needs_rebalance(&rec.volumes, &kvolumes) { + "unbalanced" + } else { + "balanced" + }; + + resp_builder = resp_builder + .header("Key-Balance", balance) + .header("Key-Volumes", rec.volumes.join(",")); + + // Probe volumes in random order; use the first responsive one. + let mut shuffled = rec.volumes.clone(); + shuffled.shuffle(&mut rand::thread_rng()); + + let mut found_remote: Option = None; + for vol in &shuffled { + let candidate = format!("http://{vol}{kp}"); + match remote_head(&state.http_client, &candidate, state.vol_timeout).await { + Ok(true) => { + found_remote = Some(candidate); + break; + } + Ok(false) => {} + Err(e) => warn!(?e, vol, "HEAD probe error"), + } + } + + match found_remote { + Some(r) => remote = r, + None => { + return ( + StatusCode::NOT_FOUND, + [(axum::http::header::CONTENT_LENGTH, "0")], + ) + .into_response(); + } + } + } + + // ------------------------------------------------------------------ + // Content-Type header strategy for X-Accel-Redirect mode: + // + // When nginx processes X-Accel-Redirect it makes an internal subrequest + // to /accel/... and the client response comes from THAT subrequest (the + // volume server), NOT from this coordinator response. Headers set here + // are discarded by nginx unless we use the variable persistence trick: + // + // 1. We set X-Content-Type on the coordinator response. + // 2. nginx captures it as $upstream_http_x_content_type — this variable + // persists across the internal redirect (same ngx_http_request_t). + // 3. The internal /accel/ location uses: + // proxy_hide_header Content-Type; + // add_header Content-Type $upstream_http_x_content_type always; + // overriding the volume server's application/octet-stream with the + // coordinator's stored MIME type. + // + // In plain 302 mode, Content-Type on the coordinator response is + // irrelevant (client follows the redirect to nginx directly), but + // we emit it anyway for HEAD requests that inspect coordinator headers. + // ------------------------------------------------------------------ + if let Some(ref ct) = rec.content_type { + resp_builder = resp_builder + .header(axum::http::header::CONTENT_TYPE, ct.as_str()) + // X-Content-Type persists as $upstream_http_x_content_type + // across nginx's internal redirect — see nginx-frontend.conf. + .header("X-Content-Type", ct.as_str()); + } + + let public_remote = rewrite_location(&remote, &state.vol_rewrite); + + if state.accel_redirect { + let accel_path = build_accel_path(&remote); + resp_builder + .status(StatusCode::OK) + .header("X-Accel-Redirect", accel_path) + .header(axum::http::header::CONTENT_LENGTH, "0") + .body(Body::empty()) + .unwrap() + .into_response() + } else { + resp_builder + .status(StatusCode::FOUND) + .header(axum::http::header::LOCATION, &public_remote) + .header(axum::http::header::CONTENT_LENGTH, "0") + .body(Body::empty()) + .unwrap() + .into_response() + } +} + +/// Handle PUT. Write a new object or a multipart part. +async fn handle_put( + state: &AppState, + key: &[u8], + raw_query: &str, + headers: HeaderMap, + body: Bytes, +) -> Response { + if body.is_empty() { + return StatusCode::LENGTH_REQUIRED.into_response(); // 411 + } + + let rec = state.get_record(key).await; + if rec.deleted == Deleted::No { + return StatusCode::FORBIDDEN.into_response(); // 403, no overwrites with PUT + } + + // Multipart part upload? + if let (Some(pn_str), Some(upload_id)) = ( + extract_query_param(raw_query, "partNumber"), + extract_query_param(raw_query, "uploadId"), + ) { + if !state.upload_ids.contains_key(&upload_id) { + return StatusCode::FORBIDDEN.into_response(); + } + + let part_num: u32 = match pn_str.parse() { + Ok(n) => n, + Err(_) => return StatusCode::BAD_REQUEST.into_response(), + }; + + return handle_put_part(upload_id, part_num, body).await; + } + + // Extract and store Content-Type from the PUT request headers. + // Stripped of parameters: "image/jpeg; charset=utf-8" -> "image/jpeg". + let content_type = headers + .get(axum::http::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + .map(|ct| ct.split(';').next().unwrap_or(ct).trim().to_string()); + + // Normal PUT. + let status = state.write_to_replicas(key, body, content_type).await; + StatusCode::from_u16(status) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) + .into_response() +} + +/// Write a multipart part to a temp file. +async fn handle_put_part(upload_id: String, part_num: u32, body: Bytes) -> Response { + let path = std::env::temp_dir().join(format!("{upload_id}-{part_num}")); + match tokio::fs::write(&path, &body).await { + Ok(()) => StatusCode::OK.into_response(), + Err(e) => { + warn!(?e, ?path, "failed to write part file"); + StatusCode::FORBIDDEN.into_response() + } + } +} + +/// Handle POST. Multipart initiation/completion or batch delete. +async fn handle_post( + state: &AppState, + key: &[u8], + raw_query: &str, + headers: HeaderMap, + body: Bytes, +) -> Response { + let rec = state.get_record(key).await; + if rec.deleted == Deleted::No { + return StatusCode::FORBIDDEN.into_response(); + } + + // ?uploads => initiate multipart upload. + if raw_query == "uploads" { + let upload_id = Uuid::new_v4().to_string(); + state.upload_ids.insert(upload_id.clone(), ()); + let xml = format!( + "{upload_id}" + ); + return (StatusCode::OK, xml).into_response(); + } + + // ?delete => batch delete. + if raw_query == "delete" { + return handle_batch_delete(state, key, body).await; + } + + // ?uploadId=X => complete multipart upload. + if let Some(upload_id) = extract_query_param(raw_query, "uploadId") { + return handle_complete_multipart(state, key, upload_id, headers, body).await; + } + + StatusCode::BAD_REQUEST.into_response() +} + +/// POST ?delete => batch-delete multiple keys. +async fn handle_batch_delete(state: &AppState, prefix: &[u8], body: Bytes) -> Response { + let xml = match std::str::from_utf8(&body) { + Ok(s) => s, + Err(_) => return StatusCode::BAD_REQUEST.into_response(), + }; + + let del = match Delete::parse(xml) { + Ok(d) => d, + Err(e) => { + warn!(?e, "batch delete XML parse error"); + return StatusCode::INTERNAL_SERVER_ERROR.into_response(); + } + }; + + let prefix_str = String::from_utf8_lossy(prefix); + for subkey in del.keys() { + let full_key = format!("{prefix_str}/{subkey}"); + let status = state.delete(full_key.as_bytes(), false).await; + if status != 204 { + return StatusCode::from_u16(status) + .unwrap_or(StatusCode::INTERNAL_SERVER_ERROR) + .into_response(); + } + } + + StatusCode::NO_CONTENT.into_response() +} + +/// POST ?uploadId=X => complete a multipart upload. +async fn handle_complete_multipart( + state: &AppState, + key: &[u8], + upload_id: String, + headers: HeaderMap, + body: Bytes, +) -> Response { + if !state.upload_ids.contains_key(&upload_id) { + return StatusCode::FORBIDDEN.into_response(); + } + state.upload_ids.remove(&upload_id); + + let xml = match std::str::from_utf8(&body) { + Ok(s) => s, + Err(_) => return StatusCode::BAD_REQUEST.into_response(), + }; + + let cmu = match CompleteMultipartUpload::parse(xml) { + Ok(c) => c, + Err(e) => { + warn!(?e, "complete multipart XML parse error"); + return StatusCode::INTERNAL_SERVER_ERROR.into_response(); + } + }; + + // Read and concatenate part files. + let mut combined: Vec = Vec::new(); + let tmp = std::env::temp_dir(); + + for part_num in cmu.part_numbers() { + let path = tmp.join(format!("{upload_id}-{part_num}")); + match tokio::fs::read(&path).await { + Ok(data) => { + combined.extend_from_slice(&data); + // Clean up temp file immediately after reading. + let _ = tokio::fs::remove_file(&path).await; + } + Err(e) => { + warn!(?e, ?path, "multipart: missing part file"); + return StatusCode::FORBIDDEN.into_response(); + } + } + } + + let content_type = headers + .get(axum::http::header::CONTENT_TYPE) + .and_then(|v| v.to_str().ok()) + // Strip parameters (e.g. "text/html; charset=utf-8" -> "text/html") + // so we store a clean canonical MIME type. + .map(|ct| ct.split(';').next().unwrap_or(ct).trim().to_string()); + + let status = state + .write_to_replicas(key, Bytes::from(combined), content_type) + .await; + let xml_resp = ""; + ( + StatusCode::from_u16(status).unwrap_or(StatusCode::INTERNAL_SERVER_ERROR), + xml_resp, + ) + .into_response() +} + +/// Handle REBALANCE => move a key to its ideal volumes. +async fn handle_rebalance(state: &AppState, key: &[u8]) -> Response { + let rec = state.get_record(key).await; + if rec.deleted != Deleted::No { + return StatusCode::NOT_FOUND.into_response(); + } + + let kvolumes = key_to_volume(key, &state.volumes, state.replicas, state.subvolumes); + + match rebalance_key(state, key, &rec.volumes, &kvolumes).await { + Ok(true) => StatusCode::NO_CONTENT.into_response(), // 204 + Ok(false) => StatusCode::BAD_REQUEST.into_response(), // 400 + Err(e) => { + warn!(?e, "REBALANCE error"); + StatusCode::INTERNAL_SERVER_ERROR.into_response() + } + } +} + +/// Build the X-Accel-Redirect internal path from a full volume URL. +/// +/// `http://volume1:8080/sv09/a2/38/abc=` → `/accel/volume1:8080/sv09/a2/38/abc=` +fn build_accel_path(url: &str) -> String { + match url.strip_prefix("http://") { + Some(rest) => format!("/accel/{rest}"), + None => format!("/accel/{url}"), + } +} + +/// Rewrite an internal volume URL to its public-facing equivalent. +/// +/// Used in 302 redirect mode. In X-Accel-Redirect mode this is a no-op +/// since the client never sees the volume URL. +fn rewrite_location(url: &str, vol_rewrite: &std::collections::HashMap) -> String { + if vol_rewrite.is_empty() { + return url.to_string(); + } + let without_scheme = match url.strip_prefix("http://") { + Some(s) => s, + None => return url.to_string(), + }; + match without_scheme.split_once('/') { + Some((host_port, path)) => { + let public = vol_rewrite + .get(host_port) + .map(|s| s.as_str()) + .unwrap_or(host_port); + format!("http://{public}/{path}") + } + None => { + let public = vol_rewrite + .get(without_scheme) + .map(|s| s.as_str()) + .unwrap_or(without_scheme); + format!("http://{public}") + } + } +} + +/// Extract a named parameter from a raw query string. +fn extract_query_param(raw: &str, name: &str) -> Option { + for part in raw.split('&') { + if let Some((k, v)) = part.split_once('=') + && k == name + { + return Some(v.to_string()); + } + } + None +} diff --git a/minikv/src/http/mod.rs b/minikv/src/http/mod.rs new file mode 100644 index 0000000..cd736e1 --- /dev/null +++ b/minikv/src/http/mod.rs @@ -0,0 +1,22 @@ +pub mod handlers; +pub mod query; +pub mod s3; + +use std::sync::Arc; + +use axum::Router; +use axum::routing::any; + +use minikv_core::state::AppState; + +/// Build the axum `Router` wired to `AppState`. +/// +/// All paths (including the root `/`) are routed to the single dispatch +/// handler, which inspects the HTTP method and query string internally. +pub fn build_router(state: Arc) -> Router { + Router::new() + // Catch-all: every path, every method. + // Non-standard methods (UNLINK, REBALANCE) require `any`. + .route("/*key", any(handlers::handle)) + .with_state(state) +} diff --git a/minikv/src/http/query.rs b/minikv/src/http/query.rs new file mode 100644 index 0000000..d889aae --- /dev/null +++ b/minikv/src/http/query.rs @@ -0,0 +1,209 @@ +/// Query parameter parsing and list/unlinked operations. +/// +/// Handles GET requests with a query string, which are routing differently +/// from plain GET (redirect) requests. +/// +/// Supported operations: +/// - `?list[&start=X][&limit=N]` list active keys under prefix +/// - `?unlinked[&start=X][&limit=N]` list soft-deleted keys +/// - `?list-type=2&prefix=X` S3-style listing +use std::sync::Arc; + +use axum::http::StatusCode; +use axum::response::{IntoResponse, Response}; +use serde::{Deserialize, Serialize}; +use tracing::debug; + +use minikv_core::record::Deleted; +use minikv_core::state::AppState; + +/// JSON response for list operations. +#[derive(Debug, Serialize)] +pub struct ListResponse { + pub next: String, + pub keys: Vec, +} + +/// Query parameters common to list and unlinked operations. +#[allow(unused)] +#[derive(Debug, Deserialize)] +pub struct ListParams { + pub start: Option, + pub limit: Option, +} + +/// Maximum number of keys returned in a single list response before a +/// 413 (Payload Too Large) is returned. Matches Go's hard limit of 1,000,000. +const MAX_KEYS: usize = 1_000_000; + +/// Handle a GET request that has a non-empty query string. +/// +/// Dispatches to: +/// - S3-style listing (`?list-type=2`) +/// - Our own `?list` / `?unlinked` operations +pub async fn handle_query(state: Arc, key_prefix: &[u8], raw_query: &str) -> Response { + // S3-style listing: ?list-type=2&prefix=... + if raw_query.contains("list-type=2") { + return handle_s3_list(state, key_prefix, raw_query).await; + } + + // Extract operation from the first query component (before any &). + let operation = raw_query.split('&').next().unwrap_or(""); + + match operation { + "list" | "unlinked" => handle_list(state, key_prefix, raw_query, operation).await, + _ => StatusCode::FORBIDDEN.into_response(), + } +} + +/// Handle `?list` and `?unlinked` queries. +async fn handle_list( + state: Arc, + key_prefix: &[u8], + raw_query: &str, + operation: &str, +) -> Response { + // Parse optional start and limit from query string. + let start = extract_param(raw_query, "start"); + let limit = extract_param(raw_query, "limit").and_then(|s| s.parse::().ok()); + + debug!(operation, ?start, ?limit, "list query"); + + // Collect matching keys from the DB. + let entries = match state.db.scan_prefix(key_prefix) { + Ok(e) => e, + Err(_) => return StatusCode::INTERNAL_SERVER_ERROR.into_response(), + }; + + let mut keys: Vec = Vec::new(); + let mut next = String::new(); + + for (raw_key, raw_val) in &entries { + // Apply start cursor. + if let Some(ref s) = start + && raw_key.as_slice() < s.as_bytes() + { + continue; + } + + // Decode and filter by deletion state. + let rec = match minikv_core::record::Record::decode(raw_val) { + Ok(r) => r, + Err(_) => continue, + }; + + let include = match operation { + "list" => rec.deleted == Deleted::No, + "unlinked" => rec.deleted == Deleted::Soft, + _ => false, + }; + + if !include { + continue; + } + + // Hard limit guard. + if keys.len() >= MAX_KEYS { + return StatusCode::PAYLOAD_TOO_LARGE.into_response(); + } + + // Soft limit: if reached, set `next` cursor and stop. + if let Some(lim) = limit + && keys.len() == lim + { + next = String::from_utf8_lossy(raw_key).into_owned(); + break; + } + + keys.push(String::from_utf8_lossy(raw_key).into_owned()); + } + + let body = match serde_json::to_vec(&ListResponse { next, keys }) { + Ok(b) => b, + Err(_) => return StatusCode::INTERNAL_SERVER_ERROR.into_response(), + }; + + ( + StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, "application/json")], + body, + ) + .into_response() +} + +/// Handle `?list-type=2` S3-style listing. +async fn handle_s3_list(state: Arc, key_prefix: &[u8], raw_query: &str) -> Response { + // Append the S3 `prefix` parameter to our key prefix. + let s3_prefix = extract_param(raw_query, "prefix").unwrap_or_default(); + let full_prefix = format!("{}{}", String::from_utf8_lossy(key_prefix), s3_prefix); + + let entries = match state.db.scan_prefix(full_prefix.as_bytes()) { + Ok(e) => e, + Err(_) => return StatusCode::INTERNAL_SERVER_ERROR.into_response(), + }; + + let mut xml = String::from(""); + let prefix_len = full_prefix.len(); + + for (raw_key, raw_val) in &entries { + let rec = match minikv_core::record::Record::decode(raw_val) { + Ok(r) => r, + Err(_) => continue, + }; + if rec.deleted != Deleted::No { + continue; + } + let key_str = String::from_utf8_lossy(raw_key); + let suffix = &key_str[prefix_len..]; + xml.push_str(""); + xml.push_str(suffix); + xml.push_str(""); + } + + xml.push_str(""); + + ( + StatusCode::OK, + [(axum::http::header::CONTENT_TYPE, "application/xml")], + xml, + ) + .into_response() +} + +/// Extract a named query parameter from a raw query string. +fn extract_param(raw_query: &str, name: &str) -> Option { + for part in raw_query.split('&') { + if let Some((k, v)) = part.split_once('=') + && k == name + { + return Some(v.to_string()); + } + } + None +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn extract_param_simple() { + assert_eq!( + extract_param("list&start=foo&limit=10", "start"), + Some("foo".to_string()) + ); + } + + #[test] + fn extract_param_missing() { + assert_eq!(extract_param("list", "start"), None); + } + + #[test] + fn extract_param_limit() { + assert_eq!( + extract_param("list&limit=50", "limit"), + Some("50".to_string()) + ); + } +} diff --git a/minikv/src/http/s3.rs b/minikv/src/http/s3.rs new file mode 100644 index 0000000..de59966 --- /dev/null +++ b/minikv/src/http/s3.rs @@ -0,0 +1,128 @@ +/// S3-compatible XML request body parsing. +/// +/// Handles two S3 API shapes: +/// - `CompleteMultipartUpload` finalize a multipart upload. +/// - `Delete` batch-delete multiple objects. +/// +/// Uses `quick-xml` with serde for zero-copy XML deserialization. +use serde::Deserialize; + +use crate::error::Error; + +// ── CompleteMultipartUpload ──────────────────────────────────────────────── + +/// A single part in a CompleteMultipartUpload body. +#[derive(Debug, Deserialize)] +pub struct UploadPart { + #[serde(rename = "PartNumber")] + pub part_number: u32, +} + +/// Body of `POST /?uploadId=`. +/// +/// ```xml +/// +/// 1 +/// 2 +/// +/// ``` +#[derive(Debug, Deserialize)] +#[serde(rename = "CompleteMultipartUpload")] +pub struct CompleteMultipartUpload { + #[serde(rename = "Part", default)] + pub parts: Vec, +} + +impl CompleteMultipartUpload { + pub fn parse(xml: &str) -> Result { + quick_xml::de::from_str(xml) + .map_err(|e| Error::XmlParse(format!("CompleteMultipartUpload: {e}"))) + } + + /// Return part numbers in the order they appear in the XML. + pub fn part_numbers(&self) -> Vec { + self.parts.iter().map(|p| p.part_number).collect() + } +} + +// ── Delete ───────────────────────────────────────────────────────────────── + +/// A single object key in a `Delete` body. +#[derive(Debug, Deserialize)] +pub struct DeleteObject { + #[serde(rename = "Key")] + pub key: String, +} + +/// Body of `POST /?delete`. +/// +/// ```xml +/// +/// key1 +/// key2 +/// +/// ``` +#[derive(Debug, Deserialize)] +#[serde(rename = "Delete")] +pub struct Delete { + #[serde(rename = "Object", default)] + pub objects: Vec, +} + +impl Delete { + pub fn parse(xml: &str) -> Result { + quick_xml::de::from_str(xml).map_err(|e| Error::XmlParse(format!("Delete: {e}"))) + } + + pub fn keys(&self) -> Vec<&str> { + self.objects.iter().map(|o| o.key.as_str()).collect() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parse_complete_multipart_upload() { + let xml = r#" + 1 + 2 + 3 + "#; + + let cmu = CompleteMultipartUpload::parse(xml).unwrap(); + assert_eq!(cmu.part_numbers(), vec![1, 2, 3]); + } + + #[test] + fn parse_complete_multipart_upload_empty() { + let xml = ""; + let cmu = CompleteMultipartUpload::parse(xml).unwrap(); + assert!(cmu.part_numbers().is_empty()); + } + + #[test] + fn parse_delete() { + let xml = r#" + key1 + key2 + "#; + + let del = Delete::parse(xml).unwrap(); + assert_eq!(del.keys(), vec!["key1", "key2"]); + } + + #[test] + fn parse_delete_empty() { + let xml = ""; + let del = Delete::parse(xml).unwrap(); + assert!(del.keys().is_empty()); + } + + #[test] + fn parse_invalid_xml_is_error() { + assert!(CompleteMultipartUpload::parse(" { + todo!("This is not implemented yet") + } + Command::Server(args) => { + init_logging(args.verbose); + cli::validate_common(&args.common); + + info!( + port = args.port, + volumes = ?args.common.volumes, + replicas = args.common.replicas, + subvolumes = args.common.subvolumes, + protect = args.protect, + checksum = args.checksum, + accel_redirect = args.accel_redirect, + "starting minikv server" + ); + + let db = open_db(&args.common); + + let vol_rewrite = args + .public_volumes + .unwrap_or_default() + .into_iter() + .zip(args.common.volumes.iter().cloned()) + .map(|(public, internal)| (internal, public)) + .collect(); + + let state = Arc::new(AppState { + db: Arc::new(db), + key_lock: KeyLock::new(), + upload_ids: DashMap::new(), + volumes: args.common.volumes, + vol_rewrite, + fallback: args.fallback, + replicas: args.common.replicas, + subvolumes: args.common.subvolumes, + protect: args.protect, + checksum: args.checksum, + accel_redirect: args.accel_redirect, + vol_timeout: args.common.voltimeout, + http_client: build_volume_client(), + }); + + let router = http::build_router(Arc::clone(&state)); + let addr = SocketAddr::from(([0, 0, 0, 0], args.port)); + let listener = tokio::net::TcpListener::bind(addr) + .await + .unwrap_or_else(|e| { + error!(?e, addr = %addr, "failed to bind"); + std::process::exit(1); + }); + + info!(addr = %addr, "listening"); + axum::serve(listener, router).await.unwrap_or_else(|e| { + error!(?e, "server error"); + std::process::exit(1); + }); + } + + Command::Rebuild(args) => { + init_logging(false); + cli::validate_common(&args); + + info!(volumes = ?args.volumes, "starting rebuild"); + + let db = open_db(&args); + let state = Arc::new(AppState { + db: Arc::new(db), + key_lock: KeyLock::new(), + upload_ids: DashMap::new(), + volumes: args.volumes, + vol_rewrite: HashMap::new(), + fallback: None, + replicas: args.replicas, + subvolumes: args.subvolumes, + protect: false, + checksum: false, + accel_redirect: false, + vol_timeout: args.voltimeout, + http_client: build_volume_client(), + }); + + rebuild_all(state).await; + } + + Command::Rebalance(args) => { + init_logging(false); + cli::validate_common(&args); + + info!(volumes = ?args.volumes, "starting rebalance"); + + let db = open_db(&args); + let state = Arc::new(AppState { + db: Arc::new(db), + key_lock: KeyLock::new(), + upload_ids: DashMap::new(), + volumes: args.volumes, + vol_rewrite: HashMap::new(), + fallback: None, + replicas: args.replicas, + subvolumes: args.subvolumes, + protect: false, + checksum: false, + accel_redirect: false, + vol_timeout: args.voltimeout, + http_client: build_volume_client(), + }); + + rebalance_all(state).await; + } + } +} + +/// Open LevelDB or exit with a clear error message. +fn open_db(args: &cli::CommonArgs) -> LevelDbStore { + LevelDbStore::open(&args.db).unwrap_or_else(|e| { + eprintln!("error: failed to open LevelDB at {:?}: {e}", args.db); + std::process::exit(1); + }) +} + +/// Initialise `tracing_subscriber` with structured output. +fn init_logging(verbose: bool) { + let filter = if verbose { + EnvFilter::new("debug") + } else { + EnvFilter::from_default_env().add_directive("info".parse().unwrap()) + }; + + tracing_subscriber::fmt() + .with_env_filter(filter) + .with_target(true) + .compact() + .init(); +} + +// /// Emit the nginx volume server configuration to stdout. +// fn print_nginx_config() { +// println!("{}", include_str!("../config/nginx-volume.conf")); +// } From 42d15e0f12ad3ff2576974fbbcbfb85a2800a1ef Mon Sep 17 00:00:00 2001 From: Nelson Dominguez Date: Tue, 3 Mar 2026 10:38:21 +0100 Subject: [PATCH 2/3] Move `MemStore` to tests directory The in-memory metadastore is intended for testing purposes only. --- minikv-core/src/storage/mod.rs | 53 ---------------------------------- minikv-core/tests/rebalance.rs | 50 +++++++++++++++++++++++++++++++- 2 files changed, 49 insertions(+), 54 deletions(-) diff --git a/minikv-core/src/storage/mod.rs b/minikv-core/src/storage/mod.rs index c324b00..b5450d3 100644 --- a/minikv-core/src/storage/mod.rs +++ b/minikv-core/src/storage/mod.rs @@ -42,56 +42,3 @@ pub trait MetadataStore: Send + Sync { /// Primarily used by `rebuild_all` to regenerate the database from scratch. fn delete_all(&self) -> Result<(), Error>; } - -/// In-memory `MetadataStore` backed by a `BTreeMap`. -/// -/// Useful for unit and integration tests that do not require persistent storage. -#[cfg(debug_assertions)] -pub mod mem { - use super::*; - use std::collections::BTreeMap; - use std::sync::Mutex; - - #[derive(Default)] - pub struct MemStore { - inner: Mutex, Vec>>, - } - - impl MetadataStore for MemStore { - fn get(&self, key: &[u8]) -> Result>, Error> { - Ok(self.inner.lock().unwrap().get(key).cloned()) - } - - fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> { - self.inner - .lock() - .unwrap() - .insert(key.to_vec(), value.to_vec()); - Ok(()) - } - - fn delete(&self, key: &[u8]) -> Result<(), Error> { - self.inner.lock().unwrap().remove(key); - Ok(()) - } - - fn scan_prefix(&self, prefix: &[u8]) -> Result, Error> { - let store = self.inner.lock().unwrap(); - Ok(store - .iter() - .filter(|(k, _)| k.starts_with(prefix)) - .map(|(k, v)| (k.clone(), v.clone())) - .collect()) - } - - fn scan_all(&self) -> Result, Error> { - let store = self.inner.lock().unwrap(); - Ok(store.iter().map(|(k, v)| (k.clone(), v.clone())).collect()) - } - - fn delete_all(&self) -> Result<(), Error> { - self.inner.lock().unwrap().clear(); - Ok(()) - } - } -} diff --git a/minikv-core/tests/rebalance.rs b/minikv-core/tests/rebalance.rs index 8fe05e1..9a0be07 100644 --- a/minikv-core/tests/rebalance.rs +++ b/minikv-core/tests/rebalance.rs @@ -8,18 +8,66 @@ //! An in-memory `MemStore` is used as the metadata backend. use dashmap::DashMap; +use minikv_core::Error; +use minikv_core::KeyValuePair; +use minikv_core::MetadataStore; use minikv_core::locking::KeyLock; use minikv_core::rebalance::rebalance_key; use minikv_core::replication::build_volume_client; use minikv_core::state::AppState; -use minikv_core::storage::mem::MemStore; use minikv_core::volumes::needs_rebalance; +use std::collections::BTreeMap; use std::collections::HashMap; use std::sync::Arc; +use std::sync::Mutex; use std::time::Duration; use wiremock::matchers::method; use wiremock::{Mock, MockServer, ResponseTemplate}; +// In-memory `MetadataStore` backed by a `BTreeMap`. +#[derive(Default)] +pub struct MemStore { + inner: Mutex, Vec>>, +} + +impl MetadataStore for MemStore { + fn get(&self, key: &[u8]) -> Result>, Error> { + Ok(self.inner.lock().unwrap().get(key).cloned()) + } + + fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> { + self.inner + .lock() + .unwrap() + .insert(key.to_vec(), value.to_vec()); + Ok(()) + } + + fn delete(&self, key: &[u8]) -> Result<(), Error> { + self.inner.lock().unwrap().remove(key); + Ok(()) + } + + fn scan_prefix(&self, prefix: &[u8]) -> Result, Error> { + let store = self.inner.lock().unwrap(); + Ok(store + .iter() + .filter(|(k, _)| k.starts_with(prefix)) + .map(|(k, v)| (k.clone(), v.clone())) + .collect()) + } + + fn scan_all(&self) -> Result, Error> { + let store = self.inner.lock().unwrap(); + Ok(store.iter().map(|(k, v)| (k.clone(), v.clone())).collect()) + } + + fn delete_all(&self) -> Result<(), Error> { + self.inner.lock().unwrap().clear(); + Ok(()) + } +} + /// Build a test `AppState` backed by `MemStore` and pointing at `volumes`. fn make_state(volumes: Vec) -> Arc { let replicas = volumes.len().min(2); From 360d6d98b19308fb408b041752902d5047b70560 Mon Sep 17 00:00:00 2001 From: Nelson Dominguez Date: Tue, 3 Mar 2026 11:08:04 +0100 Subject: [PATCH 3/3] Add HTTP integration tests --- Cargo.lock | 81 +++++++-- minikv/Cargo.toml | 9 + minikv/src/lib.rs | 5 + minikv/src/main.rs | 16 +- minikv/tests/integration.rs | 338 ++++++++++++++++++++++++++++++++++++ 5 files changed, 421 insertions(+), 28 deletions(-) create mode 100644 minikv/src/lib.rs create mode 100644 minikv/tests/integration.rs diff --git a/Cargo.lock b/Cargo.lock index 5b4c9b3..170fb89 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -97,7 +97,7 @@ checksum = "9035ad2d096bed7955a320ee7e2230574d28fd3c3a0f186cbea1ff3c7eed5dbb" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -288,7 +288,7 @@ dependencies = [ "heck", "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -408,7 +408,7 @@ checksum = "97369cbbc041bc366949bc74d34658d6cda5621039731c6310521892a3a20ae0" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -553,7 +553,7 @@ checksum = "e835b70203e41293343137df5c0664546da5745f82ec9b84d40be8336958447b" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1085,9 +1085,11 @@ dependencies = [ "serde_json", "thiserror 2.0.18", "tokio", + "tower-util", "tracing", "tracing-subscriber", "uuid", + "wiremock", ] [[package]] @@ -1186,6 +1188,26 @@ version = "2.3.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" +[[package]] +name = "pin-project" +version = "0.4.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ef0f924a5ee7ea9cbcea77529dba45f8a9ba9f622419fe3386ca581a3ae9d5a" +dependencies = [ + "pin-project-internal", +] + +[[package]] +name = "pin-project-internal" +version = "0.4.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "851c8d0ce9bebe43790dedfc86614c23494ac9f423dd618d3a61fc693eafe61e" +dependencies = [ + "proc-macro2", + "quote", + "syn 1.0.109", +] + [[package]] name = "pin-project-lite" version = "0.2.16" @@ -1223,7 +1245,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "479ca8adacdd7ce8f1fb39ce9ecccbfe93a3f1344b3d0d97f20bc0196208f62b" dependencies = [ "proc-macro2", - "syn", + "syn 2.0.117", ] [[package]] @@ -1695,7 +1717,7 @@ checksum = "d540f220d3187173da220f885ab66608367b6574e925011a9353e4badda91d79" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1805,6 +1827,17 @@ version = "2.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" +[[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.117" @@ -1833,7 +1866,7 @@ checksum = "728a70f3dbaf5bab7f0c4b1ac8d7ae5ea60a4b5549c8a5914361c99147a709d2" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1862,7 +1895,7 @@ checksum = "4fee6c4efc90059e10f81e6d42c60a18f76588c3d74cb83a0b242a2b6c7504c1" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1873,7 +1906,7 @@ checksum = "ebc4ee7f67670e9b64d05fa4253e753e016c6c95ff35b89b7941d6b856dec1d5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -1935,7 +1968,7 @@ checksum = "af407857209536a95c8e56f8231ef2c2e2aff839b22e07a1ffcbc617e9db9fa5" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2007,6 +2040,18 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" +[[package]] +name = "tower-util" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d1093c19826d33807c72511e68f73b4a0469a3f22c2bd5f7d5212178b4b89674" +dependencies = [ + "futures-core", + "futures-util", + "pin-project", + "tower-service", +] + [[package]] name = "tracing" version = "0.1.44" @@ -2027,7 +2072,7 @@ checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2229,7 +2274,7 @@ dependencies = [ "bumpalo", "proc-macro2", "quote", - "syn", + "syn 2.0.117", "wasm-bindgen-shared", ] @@ -2630,7 +2675,7 @@ dependencies = [ "heck", "indexmap", "prettyplease", - "syn", + "syn 2.0.117", "wasm-metadata", "wit-bindgen-core", "wit-component", @@ -2646,7 +2691,7 @@ dependencies = [ "prettyplease", "proc-macro2", "quote", - "syn", + "syn 2.0.117", "wit-bindgen-core", "wit-bindgen-rust", ] @@ -2713,7 +2758,7 @@ checksum = "b659052874eb698efe5b9e8cf382204678a0086ebf46982b79d6ca3182927e5d" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", "synstructure", ] @@ -2734,7 +2779,7 @@ checksum = "f65c489a7071a749c849713807783f70672b28094011623e200cb86dcb835953" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] @@ -2754,7 +2799,7 @@ checksum = "d71e5d6e06ab090c67b5e44993ec16b72dcbaabc526db883a360057678b48502" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", "synstructure", ] @@ -2794,7 +2839,7 @@ checksum = "eadce39539ca5cb3985590102671f2567e659fca9666581ad3411d59207951f3" dependencies = [ "proc-macro2", "quote", - "syn", + "syn 2.0.117", ] [[package]] diff --git a/minikv/Cargo.toml b/minikv/Cargo.toml index 6216031..8e93d98 100644 --- a/minikv/Cargo.toml +++ b/minikv/Cargo.toml @@ -10,6 +10,10 @@ authors.workspace = true keywords.workspace = true categories.workspace = true +[lib] +name = "minikv_server" +path = "src/lib.rs" + [dependencies] minikv-core = { workspace = true } tracing = { workspace = true } @@ -26,3 +30,8 @@ clap = { version = "4", features = ["derive", "env"] } tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] } axum = { version = "0.7", features = [] } quick-xml = { version = "0.31", features = ["serialize"] } + +[dev-dependencies] +tower-util = { version = "0.3" } +minikv-core = { workspace = true } +wiremock = "0.6" diff --git a/minikv/src/lib.rs b/minikv/src/lib.rs new file mode 100644 index 0000000..033cec3 --- /dev/null +++ b/minikv/src/lib.rs @@ -0,0 +1,5 @@ +pub mod cli; +mod error; +pub mod http; + +pub use error::Error; diff --git a/minikv/src/main.rs b/minikv/src/main.rs index 2a66d4f..b092405 100644 --- a/minikv/src/main.rs +++ b/minikv/src/main.rs @@ -6,10 +6,6 @@ //! - `rebuild` → DB reconstruction from volume servers //! - `rebalance` → key migration to ideal volumes -mod cli; -mod error; -mod http; - use minikv_core::storage; use std::collections::HashMap; @@ -21,12 +17,12 @@ use dashmap::DashMap; use tracing::{error, info}; use tracing_subscriber::EnvFilter; -use cli::{Cli, Command}; use minikv_core::locking::KeyLock; use minikv_core::rebalance::rebalance_all; use minikv_core::rebuild::rebuild_all; use minikv_core::replication::build_volume_client; use minikv_core::state::AppState; +use minikv_server::cli::{Cli, Command, CommonArgs, validate_common}; use storage::leveldb::LevelDbStore; #[tokio::main] @@ -39,7 +35,7 @@ async fn main() { } Command::Server(args) => { init_logging(args.verbose); - cli::validate_common(&args.common); + validate_common(&args.common); info!( port = args.port, @@ -78,7 +74,7 @@ async fn main() { http_client: build_volume_client(), }); - let router = http::build_router(Arc::clone(&state)); + let router = minikv_server::http::build_router(Arc::clone(&state)); let addr = SocketAddr::from(([0, 0, 0, 0], args.port)); let listener = tokio::net::TcpListener::bind(addr) .await @@ -96,7 +92,7 @@ async fn main() { Command::Rebuild(args) => { init_logging(false); - cli::validate_common(&args); + validate_common(&args); info!(volumes = ?args.volumes, "starting rebuild"); @@ -122,7 +118,7 @@ async fn main() { Command::Rebalance(args) => { init_logging(false); - cli::validate_common(&args); + validate_common(&args); info!(volumes = ?args.volumes, "starting rebalance"); @@ -149,7 +145,7 @@ async fn main() { } /// Open LevelDB or exit with a clear error message. -fn open_db(args: &cli::CommonArgs) -> LevelDbStore { +fn open_db(args: &CommonArgs) -> LevelDbStore { LevelDbStore::open(&args.db).unwrap_or_else(|e| { eprintln!("error: failed to open LevelDB at {:?}: {e}", args.db); std::process::exit(1); diff --git a/minikv/tests/integration.rs b/minikv/tests/integration.rs new file mode 100644 index 0000000..bbe6b55 --- /dev/null +++ b/minikv/tests/integration.rs @@ -0,0 +1,338 @@ +use std::collections::{BTreeMap, HashMap}; +/// Integration tests +/// +/// Each test spins up a real axum server with an in-memory metadata store +/// and wiremock volume servers, then exercises the full HTTP flow. +/// +/// # Running +/// ```sh +/// cargo test --test integration +/// ``` +use std::sync::Arc; +use std::time::Duration; + +use axum::body::Body; +use axum::http::{Method, Request, StatusCode}; +use dashmap::DashMap; +use minikv_core::locking::KeyLock; +use minikv_core::replication::build_volume_client; +use minikv_core::state::AppState; +use minikv_core::{Error, KeyValuePair, MetadataStore}; +use minikv_server::http::build_router; +use std::sync::Mutex; +use tower_util::ServiceExt; +use wiremock::matchers::method; +use wiremock::{Mock, MockServer, ResponseTemplate}; + +// In-memory `MetadataStore` backed by a `BTreeMap`. +#[derive(Default)] +pub struct MemStore { + inner: Mutex, Vec>>, +} + +impl MetadataStore for MemStore { + fn get(&self, key: &[u8]) -> Result>, Error> { + Ok(self.inner.lock().unwrap().get(key).cloned()) + } + + fn put(&self, key: &[u8], value: &[u8]) -> Result<(), Error> { + self.inner + .lock() + .unwrap() + .insert(key.to_vec(), value.to_vec()); + Ok(()) + } + + fn delete(&self, key: &[u8]) -> Result<(), Error> { + self.inner.lock().unwrap().remove(key); + Ok(()) + } + + fn scan_prefix(&self, prefix: &[u8]) -> Result, Error> { + let store = self.inner.lock().unwrap(); + Ok(store + .iter() + .filter(|(k, _)| k.starts_with(prefix)) + .map(|(k, v)| (k.clone(), v.clone())) + .collect()) + } + + fn scan_all(&self) -> Result, Error> { + let store = self.inner.lock().unwrap(); + Ok(store.iter().map(|(k, v)| (k.clone(), v.clone())).collect()) + } + + fn delete_all(&self) -> Result<(), Error> { + self.inner.lock().unwrap().clear(); + Ok(()) + } +} + +/// Build a full test app with mock volume servers. +async fn test_app(volumes: Vec) -> axum::Router { + let replicas = volumes.len().min(1); + let state = Arc::new(AppState { + db: Arc::new(MemStore::default()), + key_lock: KeyLock::new(), + upload_ids: DashMap::new(), + volumes, + vol_rewrite: HashMap::new(), + fallback: None, + replicas, + subvolumes: 1, + protect: false, + checksum: true, + accel_redirect: false, + vol_timeout: Duration::from_secs(5), + http_client: build_volume_client(), + }); + build_router(state) +} + +// ── GET on missing key => 404 ─────────────────────────────────────────────── + +#[tokio::test] +async fn get_missing_key_returns_404() { + let app = test_app(vec!["localhost:9999".into()]).await; + + let req = Request::builder() + .method(Method::GET) + .uri("/missing-key") + .body(Body::empty()) + .unwrap(); + + let resp = app.oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); +} + +// ── PUT then GET => 302 redirect ─────────────────────────────────────────── + +#[tokio::test] +async fn put_then_get_redirects_to_volume() { + let volume_server = MockServer::start().await; + + // Volume server accepts PUT. + Mock::given(method("PUT")) + .respond_with(ResponseTemplate::new(201)) + .mount(&volume_server) + .await; + + // Volume server responds to HEAD probe. + Mock::given(method("HEAD")) + .respond_with(ResponseTemplate::new(200)) + .mount(&volume_server) + .await; + + let host = volume_server + .uri() + .strip_prefix("http://") + .unwrap() + .to_string(); + + let app = test_app(vec![host]).await; + + // PUT the object. + let put_req = Request::builder() + .method(Method::PUT) + .uri("/mykey") + .header("Content-Length", "5") + .body(Body::from("hello")) + .unwrap(); + + let put_resp = app.clone().oneshot(put_req).await.unwrap(); + assert_eq!(put_resp.status(), StatusCode::CREATED); + + // GET should return 302 with a Location header. + let get_req = Request::builder() + .method(Method::GET) + .uri("/mykey") + .body(Body::empty()) + .unwrap(); + + let get_resp = app.oneshot(get_req).await.unwrap(); + assert_eq!(get_resp.status(), StatusCode::FOUND); + assert!( + get_resp.headers().contains_key("location"), + "GET response must have Location header" + ); +} + +// ── PUT with empty body => 411 ───────────────────────────────────────────── + +#[tokio::test] +async fn put_empty_body_returns_411() { + let app = test_app(vec!["localhost:9999".into()]).await; + + let req = Request::builder() + .method(Method::PUT) + .uri("/key") + .header("Content-Length", "0") + .body(Body::empty()) + .unwrap(); + + let resp = app.oneshot(req).await.unwrap(); + assert_eq!(resp.status(), StatusCode::LENGTH_REQUIRED); +} + +// ── Double PUT => 403 (no silent overwrite) ──────────────────────────────── + +#[tokio::test] +async fn double_put_returns_403() { + let volume_server = MockServer::start().await; + + Mock::given(method("PUT")) + .respond_with(ResponseTemplate::new(201)) + .mount(&volume_server) + .await; + Mock::given(method("HEAD")) + .respond_with(ResponseTemplate::new(200)) + .mount(&volume_server) + .await; + + let host = volume_server + .uri() + .strip_prefix("http://") + .unwrap() + .to_string(); + + let app = test_app(vec![host]).await; + + let put = |app: axum::Router| async move { + app.oneshot( + Request::builder() + .method(Method::PUT) + .uri("/key") + .header("Content-Length", "5") + .body(Body::from("hello")) + .unwrap(), + ) + .await + .unwrap() + }; + + let r1 = put(app.clone()).await; + assert_eq!(r1.status(), StatusCode::CREATED); + + let r2 = put(app.clone()).await; + assert_eq!(r2.status(), StatusCode::FORBIDDEN); +} + +// ── Concurrent PUT on same key => exactly one 201, others 409 ───────────── + +#[tokio::test] +async fn concurrent_put_same_key_serialised() { + use std::sync::atomic::{AtomicUsize, Ordering}; + + let volume_server = MockServer::start().await; + Mock::given(method("PUT")) + .respond_with(ResponseTemplate::new(201)) + .mount(&volume_server) + .await; + Mock::given(method("HEAD")) + .respond_with(ResponseTemplate::new(200)) + .mount(&volume_server) + .await; + + let host = volume_server + .uri() + .strip_prefix("http://") + .unwrap() + .to_string(); + + let app = test_app(vec![host]).await; + let app = Arc::new(app); + + let created = Arc::new(AtomicUsize::new(0)); + let conflicts = Arc::new(AtomicUsize::new(0)); + let mut handles = vec![]; + + for _ in 0..8 { + let app = Arc::clone(&app); + let created = Arc::clone(&created); + let conflicts = Arc::clone(&conflicts); + handles.push(tokio::spawn(async move { + let req = Request::builder() + .method(Method::PUT) + .uri("/same-key") + .header("Content-Length", "5") + .body(Body::from("hello")) + .unwrap(); + let app = Arc::clone(&app); + let resp = ::clone(&app) + .oneshot(req) + .await + .unwrap(); + match resp.status() { + StatusCode::CREATED => created.fetch_add(1, Ordering::Relaxed), + StatusCode::CONFLICT => conflicts.fetch_add(1, Ordering::Relaxed), + StatusCode::FORBIDDEN => created.load(Ordering::Relaxed), // second PUT + s => panic!("unexpected status {s}"), + }; + })); + } + + for h in handles { + h.await.unwrap(); + } + + // At most one write should have succeeded without conflict. + assert!( + created.load(Ordering::Relaxed) <= 1, + "more than one concurrent PUT succeeded" + ); +} + +// ── DELETE after PUT => 204 ──────────────────────────────────────────────── + +#[tokio::test] +async fn put_then_delete_returns_204() { + let volume_server = MockServer::start().await; + + Mock::given(method("PUT")) + .respond_with(ResponseTemplate::new(201)) + .mount(&volume_server) + .await; + Mock::given(method("DELETE")) + .respond_with(ResponseTemplate::new(204)) + .mount(&volume_server) + .await; + Mock::given(method("HEAD")) + .respond_with(ResponseTemplate::new(200)) + .mount(&volume_server) + .await; + + let host = volume_server + .uri() + .strip_prefix("http://") + .unwrap() + .to_string(); + + let app = test_app(vec![host]).await; + + // PUT + app.clone() + .oneshot( + Request::builder() + .method(Method::PUT) + .uri("/del-key") + .header("Content-Length", "5") + .body(Body::from("hello")) + .unwrap(), + ) + .await + .unwrap(); + + // DELETE + let del_resp = app + .oneshot( + Request::builder() + .method(Method::DELETE) + .uri("/del-key") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + + assert_eq!(del_resp.status(), StatusCode::NO_CONTENT); +}