From f6e8bbca6fbca8dc2e39ab169576f524e25bd68e Mon Sep 17 00:00:00 2001 From: Cyril Plisko Date: Mon, 20 Jul 2026 17:04:09 +0300 Subject: [PATCH 1/2] ENG-6268: migrate plateau to axum 0.8 - catalog: drop unused axum dependency - transport: bump utoipa 4->5 (required transitively by utoipa-swagger-ui's axum 0.8 support); adjust PartitionSelector/SchemaChunk schema derivation and IntoParams field overrides for utoipa 5's stricter generics/schema model - server: axum 0.6->0.8, tower-http 0.7, utoipa-swagger-ui 9; route syntax (:name -> {name}), Server::bind -> axum::serve, FromRequest -> FromRequest, boxed(Full::new(...)) -> Body::from(...), swap axum::headers::ContentType for a direct headers dependency (axum-extra isn't needed here - the code never does typed-header extraction), and add OptionalFromRequestParts for the custom Query extractor since axum no longer blanket-supports Option for arbitrary FromRequestParts impls - client: fix iteration_request sending JSON null instead of an empty TopicIterator when no position is given - silently tolerated by axum 0.6's lenient Option> error handling, correctly rejected by axum 0.8's stricter Json extractor No dependency lock with fitzroy (platform consumes plateau via a git tag pin, not a path/workspace dependency). --- Cargo.lock | 311 ++++++++++++++++++---------------- Cargo.toml | 5 + catalog/Cargo.toml | 1 - client/src/lib.rs | 2 +- server/Cargo.toml | 10 +- server/src/axum_util/query.rs | 20 ++- server/src/http.rs | 42 +++-- server/src/http/chunk.rs | 67 ++++---- transport/Cargo.toml | 2 +- transport/src/lib.rs | 7 +- 10 files changed, 252 insertions(+), 215 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4bf3e4c..2715bbf 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,12 @@ version = "1.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe" +[[package]] +name = "adler2" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" + [[package]] name = "ahash" version = "0.8.11" @@ -115,6 +121,15 @@ version = "1.0.99" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b0674a1ddeecb70197781e945de4b3b8ffb61fa939a5597bcf48503737663100" +[[package]] +name = "arbitrary" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d036a3c4ab069c7b410a2ce876bd74808d2d0888a82667669f8e783a898bf1" +dependencies = [ + "derive_arbitrary", +] + [[package]] name = "argminmax" version = "0.6.2" @@ -204,7 +219,7 @@ dependencies = [ "arrow-schema", "arrow-select", "atoi", - "base64 0.22.0", + "base64 0.22.1", "chrono", "half", "lexical-core", @@ -420,52 +435,54 @@ dependencies = [ [[package]] name = "axum" -version = "0.6.20" +version = "0.8.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3b829e4e32b91e643de6eafe82b1d90675f5874230191a4ffbc1b336dec4d6bf" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" dependencies = [ - "async-trait", "axum-core", - "bitflags 1.3.2", "bytes", + "form_urlencoded", "futures-util", - "headers", - "http 0.2.12", - "http-body 0.4.6", - "hyper 0.14.28", + "http 1.3.1", + "http-body 1.0.1", + "http-body-util", + "hyper 1.6.0", + "hyper-util", "itoa", "matchit", "memchr", "mime", "percent-encoding", "pin-project-lite", - "rustversion", - "serde", + "serde_core", "serde_json", "serde_path_to_error", "serde_urlencoded", - "sync_wrapper", + "sync_wrapper 1.0.2", "tokio", "tower", "tower-layer", "tower-service", + "tracing", ] [[package]] name = "axum-core" -version = "0.3.4" +version = "0.5.6" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "759fa577a247914fd3f7f76d62972792636412fbfd634cd452f6a385a74d2d2c" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" dependencies = [ - "async-trait", "bytes", - "futures-util", - "http 0.2.12", - "http-body 0.4.6", + "futures-core", + "http 1.3.1", + "http-body 1.0.1", + "http-body-util", "mime", - "rustversion", + "pin-project-lite", + "sync_wrapper 1.0.2", "tower-layer", "tower-service", + "tracing", ] [[package]] @@ -492,7 +509,7 @@ dependencies = [ "cc", "cfg-if", "libc", - "miniz_oxide", + "miniz_oxide 0.7.2", "object", "rustc-demangle", ] @@ -505,9 +522,9 @@ checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567" [[package]] name = "base64" -version = "0.22.0" +version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9475866fec1451be56a3c2400fd081ff546538961565ccb5b7142cbd22bc7a51" +checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" [[package]] name = "base64ct" @@ -601,9 +618,9 @@ dependencies = [ [[package]] name = "bumpalo" -version = "3.15.4" +version = "3.20.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7ff69b9dd49fd426c69a0db9fc04dd934cdb6645ff000864d98f7e2af8830eaa" +checksum = "72f5acc6cb2ba439de613abc23857ec3d78374d8ed5ac84e9d11336e87da8649" [[package]] name = "bytemuck" @@ -1058,6 +1075,17 @@ dependencies = [ "serde", ] +[[package]] +name = "derive_arbitrary" +version = "1.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e567bd82dcff979e4b03460c307b3cdc9e96fde3d73bed1496d2bc75d9dd62a" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.104", +] + [[package]] name = "digest" version = "0.10.7" @@ -1070,27 +1098,6 @@ dependencies = [ "subtle", ] -[[package]] -name = "dirs" -version = "5.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "44c45a9d03d6676652bcb5e724c7e988de1acad23a711b5217ab9cbecbec2225" -dependencies = [ - "dirs-sys", -] - -[[package]] -name = "dirs-sys" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "520f05a5cbd335fae5a99ff7a6ab8627577660ee5cfd6a94a6a929b52ff0321c" -dependencies = [ - "libc", - "option-ext", - "redox_users", - "windows-sys 0.48.0", -] - [[package]] name = "dlv-list" version = "0.5.2" @@ -1239,12 +1246,13 @@ dependencies = [ [[package]] name = "flate2" -version = "1.0.28" +version = "1.1.9" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "46303f565772937ffe1d394a4fac6f411c6013172fadde9dcdb1e147a086940e" +checksum = "843fba2746e448b37e26a819579957415c8cef339bf08564fe8b7ddbd959573c" dependencies = [ "crc32fast", - "miniz_oxide", + "miniz_oxide 0.8.9", + "zlib-rs", ] [[package]] @@ -1550,13 +1558,28 @@ checksum = "06683b93020a07e3dbcf5f8c0f6d40080d725bea7936fc01ad345c01b97dc270" dependencies = [ "base64 0.21.7", "bytes", - "headers-core", + "headers-core 0.2.0", "http 0.2.12", "httpdate", "mime", "sha1", ] +[[package]] +name = "headers" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3314d5adb5d94bcdf56771f2e50dbbc80bb4bdf88967526706205ac9eff24eb" +dependencies = [ + "base64 0.22.1", + "bytes", + "headers-core 0.3.0", + "http 1.3.1", + "httpdate", + "mime", + "sha1", +] + [[package]] name = "headers-core" version = "0.2.0" @@ -1566,6 +1589,15 @@ dependencies = [ "http 0.2.12", ] +[[package]] +name = "headers-core" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "54b4a22553d4242c49fddb9ba998a99962b5cc6f22cb5a3482bec22522403ce4" +dependencies = [ + "http 1.3.1", +] + [[package]] name = "heck" version = "0.4.1" @@ -1726,12 +1758,6 @@ dependencies = [ "pin-project-lite", ] -[[package]] -name = "http-range-header" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "add0ab9360ddbd88cfeb3bd9574a1d85cfdfa14db10b3e21d3700dbc4328758f" - [[package]] name = "httparse" version = "1.10.1" @@ -2149,17 +2175,6 @@ version = "0.2.8" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4ec2a862134d2a7d32d7983ddcdd1c4923530833c9f2ea1a44fc5fa473989058" -[[package]] -name = "libredox" -version = "0.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85c833ca1e66078851dba29046874e38f08b2c883700aa29a03ddd3b23814ee8" -dependencies = [ - "bitflags 2.9.1", - "libc", - "redox_syscall 0.4.1", -] - [[package]] name = "libsqlite3-sys" version = "0.27.0" @@ -2193,6 +2208,12 @@ dependencies = [ "scopeguard", ] +[[package]] +name = "lockfree-object-pool" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9374ef4228402d4b7e403e5838cb880d9ee663314b0a900d5a6aabf0c213552e" + [[package]] name = "log" version = "0.4.21" @@ -2245,9 +2266,9 @@ dependencies = [ [[package]] name = "matchit" -version = "0.7.3" +version = "0.8.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" [[package]] name = "md-5" @@ -2290,7 +2311,7 @@ version = "0.17.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b166dea96003ee2531cf14833efedced545751d800f03535801d833313f8c15" dependencies = [ - "base64 0.22.0", + "base64 0.22.1", "http-body-util", "hyper 1.6.0", "hyper-rustls 0.27.7", @@ -2352,6 +2373,16 @@ dependencies = [ "adler", ] +[[package]] +name = "miniz_oxide" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" +dependencies = [ + "adler2", + "simd-adler32", +] + [[package]] name = "mio" version = "1.0.3" @@ -2570,12 +2601,6 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ff011a302c396a5197692431fc1948019154afc178baf7d8e37367442a4601cf" -[[package]] -name = "option-ext" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d" - [[package]] name = "ordered-multimap" version = "0.6.0" @@ -2798,7 +2823,6 @@ dependencies = [ "arrow-json", "arrow-schema", "arrow-select", - "axum", "bytes", "bytesize", "chrono", @@ -2928,6 +2952,7 @@ dependencies = [ "chrono", "config", "futures", + "headers 0.4.1", "humantime-serde", "metrics", "metrics-exporter-prometheus", @@ -3330,30 +3355,6 @@ dependencies = [ "syn 2.0.104", ] -[[package]] -name = "proc-macro-error" -version = "1.0.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da25490ff9892aab3fcf7c36f08cfb902dd3e71ca0f9f9517bea02a73a5ce38c" -dependencies = [ - "proc-macro-error-attr", - "proc-macro2", - "quote", - "syn 1.0.109", - "version_check", -] - -[[package]] -name = "proc-macro-error-attr" -version = "1.0.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1be40180e52ecc98ad80b184934baf3d0d29f979574e439af5a55274b35f869" -dependencies = [ - "proc-macro2", - "quote", - "version_check", -] - [[package]] name = "proc-macro2" version = "1.0.95" @@ -3545,17 +3546,6 @@ dependencies = [ "bitflags 1.3.2", ] -[[package]] -name = "redox_users" -version = "0.4.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a18479200779601e498ada4e8c1e1f50e3ee19deb0259c25825a98b5603b2cb4" -dependencies = [ - "getrandom 0.2.12", - "libredox", - "thiserror 1.0.63", -] - [[package]] name = "regex" version = "1.10.4" @@ -3626,7 +3616,7 @@ dependencies = [ "serde", "serde_json", "serde_urlencoded", - "sync_wrapper", + "sync_wrapper 0.1.2", "system-configuration", "tokio", "tokio-rustls 0.24.1", @@ -3718,7 +3708,6 @@ dependencies = [ "proc-macro2", "quote", "rust-embed-utils", - "shellexpand", "syn 2.0.104", "walkdir", ] @@ -4131,7 +4120,7 @@ version = "3.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2c85f8e96d1d6857f13768fcbd895fcb06225510022a2774ed8b5150581847b0" dependencies = [ - "base64 0.22.0", + "base64 0.22.1", "chrono", "hex", "indexmap 1.9.3", @@ -4186,15 +4175,6 @@ dependencies = [ "lazy_static", ] -[[package]] -name = "shellexpand" -version = "3.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "da03fa3b94cc19e3ebfc88c4229c49d8f08cdbd1228870a45f0ffdf84988e14b" -dependencies = [ - "dirs", -] - [[package]] name = "shlex" version = "1.3.0" @@ -4220,6 +4200,12 @@ dependencies = [ "rand_core 0.6.4", ] +[[package]] +name = "simd-adler32" +version = "0.3.10" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" + [[package]] name = "simdutf8" version = "0.1.5" @@ -4633,6 +4619,12 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2047c6ded9c721764247e62cd3b03c09ffc529b2ba5b10ec482ae507a4a70160" +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + [[package]] name = "sysinfo" version = "0.30.7" @@ -4974,14 +4966,14 @@ dependencies = [ [[package]] name = "tower" -version = "0.4.13" +version = "0.5.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8fa9be0de6cf49e536ce1851f987bd21a43b771b09473c3549a6c853db37c1c" +checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4" dependencies = [ "futures-core", "futures-util", - "pin-project", "pin-project-lite", + "sync_wrapper 1.0.2", "tokio", "tower-layer", "tower-service", @@ -4990,17 +4982,15 @@ dependencies = [ [[package]] name = "tower-http" -version = "0.4.4" +version = "0.7.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "61c5bb1d698276a2443e5ecfabc1008bf15a36c12e6a7176e7bf089ea9131140" +checksum = "b11f75e912b0c2be01b63d8cf8057b8c3f97cf34abb3d431a3a4c8675498e233" dependencies = [ "bitflags 2.9.1", "bytes", - "futures-core", - "futures-util", - "http 0.2.12", - "http-body 0.4.6", - "http-range-header", + "http 1.3.1", + "http-body 1.0.1", + "percent-encoding", "pin-project-lite", "tower-layer", "tower-service", @@ -5009,15 +4999,15 @@ dependencies = [ [[package]] name = "tower-layer" -version = "0.3.2" +version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c20c8dbed6283a09604c3e69b4b7eeb54e298b8a600d4d5ecb5ad39de609f1d0" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" [[package]] name = "tower-service" -version = "0.3.2" +version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6bc1c9ce2b5135ac7f93c72918fc37feb872bdc6a5533a8b85eb4b86bfdae52" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" [[package]] name = "tracing" @@ -5204,9 +5194,9 @@ checksum = "711b9620af191e0cdc7468a8d14e709c3dcdb115b36f838e601583af800a370a" [[package]] name = "utoipa" -version = "4.2.0" +version = "5.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "272ebdfbc99111033031d2f10e018836056e4d2c8e2acda76450ec7974269fa7" +checksum = "8bde15df68e80b16c7d16b9616e80770ad158988daa56a27dccd1e55558b0160" dependencies = [ "indexmap 2.10.0", "serde", @@ -5216,11 +5206,10 @@ dependencies = [ [[package]] name = "utoipa-gen" -version = "4.2.0" +version = "5.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d3c9f4d08338c1bfa70dde39412a040a884c6f318b3d09aaaf3437a1e52027fc" +checksum = "6ba0b99ee52df3028635d93840c797102da61f8a7bb3cf751032455895b52ef8" dependencies = [ - "proc-macro-error", "proc-macro2", "quote", "regex", @@ -5229,16 +5218,18 @@ dependencies = [ [[package]] name = "utoipa-swagger-ui" -version = "4.0.0" +version = "9.0.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "154517adf0d0b6e22e8e1f385628f14fcaa3db43531dc74303d3edef89d6dfe5" +checksum = "d047458f1b5b65237c2f6dc6db136945667f40a7668627b3490b9513a3d43a55" dependencies = [ "axum", + "base64 0.22.1", "mime_guess", "regex", "rust-embed", "serde", "serde_json", + "url", "utoipa", "zip", ] @@ -5298,7 +5289,7 @@ dependencies = [ "bytes", "futures-channel", "futures-util", - "headers", + "headers 0.3.9", "http 0.2.12", "hyper 0.14.28", "log", @@ -5732,14 +5723,36 @@ checksum = "525b4ec142c6b68a2d10f01f7bbf6755599ca3f81ea53b8431b7dd348f5fdb2d" [[package]] name = "zip" -version = "0.6.6" +version = "3.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "760394e246e4c28189f19d488c058bf16f564016aefac5d32bb1f3b51d5e9261" +checksum = "12598812502ed0105f607f941c386f43d441e00148fce9dec3ca5ffb0bde9308" dependencies = [ - "byteorder", + "arbitrary", "crc32fast", - "crossbeam-utils", "flate2", + "indexmap 2.10.0", + "memchr", + "zopfli", +] + +[[package]] +name = "zlib-rs" +version = "0.6.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b142a20ec14a91d5bc708c1dc21b080c550113d8aa77afa29635673a65dd02c5" + +[[package]] +name = "zopfli" +version = "0.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5019f391bac5cf252e93bbcc53d039ffd62c7bfb7c150414d61369afe57e946" +dependencies = [ + "bumpalo", + "crc32fast", + "lockfree-object-pool", + "log", + "once_cell", + "simd-adler32", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 3ec9e05..f0efe4d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,9 @@ license = "MIT OR Apache-2.0" [workspace.dependencies] +axum = "0.8" chrono = "0.4" +headers = "0.4" reqwest = { version = "0.11", default-features = false, features = [ "json", "rustls-tls", @@ -32,7 +34,10 @@ reqwest = { version = "0.11", default-features = false, features = [ ] } test-log = { version = "0.2", default-features = false, features = ["trace"] } thiserror = "2.0" +tower-http = { version = "0.7", features = ["trace"] } tracing = "0.1" +utoipa = { version = "5", features = ["axum_extras", "chrono"] } +utoipa-swagger-ui = { version = "9", features = ["axum"] } plateau-catalog = { path = "./catalog" } plateau-cli = { path = "./cli" } diff --git a/catalog/Cargo.toml b/catalog/Cargo.toml index b0ef97a..7a2b420 100644 --- a/catalog/Cargo.toml +++ b/catalog/Cargo.toml @@ -10,7 +10,6 @@ authors.workspace = true [dependencies] anyhow = "1" -axum = { version = "0.6", features = ["headers"] } bytes = "1.6" bytesize = { version = "1.1.0", features = ["serde"] } config = "0.14" diff --git a/client/src/lib.rs b/client/src/lib.rs index 5c3a7fd..c7fdf1e 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -304,7 +304,7 @@ impl Client { Ok(match position.into() { Some(position) => base_request.json(&position), - None => base_request.json(&{}), + None => base_request.json(&TopicIterator::default()), }) } diff --git a/server/Cargo.toml b/server/Cargo.toml index ef22650..1ed4c15 100644 --- a/server/Cargo.toml +++ b/server/Cargo.toml @@ -10,7 +10,6 @@ authors.workspace = true [dependencies] anyhow = "1" -axum = { version = "0.6", features = ["headers"] } bytes = "1.6" bytesize = { version = "1.1.0", features = ["serde"] } config = "0.14" @@ -25,13 +24,14 @@ toml = "0.7" tracing = "0.1" tokio-stream = { version = "0.1", features = ["signal"] } tokio = { version = "1", features = ["full"] } -tower-http = { version = "0.4", features = ["trace"] } -# TODO: 0.7.4 adds a deprecation warning that will need to be fixed down the road -utoipa = { version = "4", features = ["axum_extras"] } -utoipa-swagger-ui = { version = "4", features = ["axum"] } +axum.workspace = true chrono.workspace = true +headers.workspace = true thiserror.workspace = true +tower-http.workspace = true +utoipa.workspace = true +utoipa-swagger-ui.workspace = true plateau-catalog.workspace = true plateau-client = { workspace = true, features = ["replicate"] } diff --git a/server/src/axum_util/query.rs b/server/src/axum_util/query.rs index 3e542f8..caa6879 100644 --- a/server/src/axum_util/query.rs +++ b/server/src/axum_util/query.rs @@ -12,7 +12,6 @@ pub enum QueryRejection { FailedToDeserializeQueryString, } -#[axum::async_trait] impl extract::FromRequestParts for Query where T: de::DeserializeOwned, @@ -41,3 +40,22 @@ impl response::IntoResponse for QueryRejection { http::StatusCode::NOT_ACCEPTABLE.into_response() } } + +impl extract::OptionalFromRequestParts for Query +where + T: de::DeserializeOwned, + S: Send + Sync, +{ + type Rejection = QueryRejection; + + async fn from_request_parts( + parts: &mut http::request::Parts, + state: &S, + ) -> Result, Self::Rejection> { + Ok( + >::from_request_parts(parts, state) + .await + .ok(), + ) + } +} diff --git a/server/src/http.rs b/server/src/http.rs index 713fa50..e1f0c7b 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -5,13 +5,17 @@ use std::sync::Arc; use std::time::{Duration, SystemTime}; use anyhow::Result; -use axum::{ - body::Body, - extract::{DefaultBodyLimit, FromRef, Path, State}, - http::{header::ACCEPT, HeaderMap, Request}, - routing::{get, post}, - Json, Router, Server, -}; +use axum::extract::DefaultBodyLimit; +use axum::extract::FromRef; +use axum::extract::Path; +use axum::extract::Request; +use axum::extract::State; +use axum::http::header::ACCEPT; +use axum::http::HeaderMap; +use axum::routing::get; +use axum::routing::post; +use axum::Json; +use axum::Router; use chrono::{DateTime, Utc}; use futures::{Future, FutureExt}; @@ -136,19 +140,19 @@ pub async fn serve( .route("/ok", get(healthcheck)) .route("/topics", get(get_topics)) .route( - "/topic/:topic_name/partition/:partition_name/records", + "/topic/{topic_name}/partition/{partition_name}/records", get(partition_get_records), ) .route( - "/topic/:topic_name/partition/:partition_name", + "/topic/{topic_name}/partition/{partition_name}", post(topic_append).layer(DefaultBodyLimit::max(config.http.max_append_bytes)), ) - .route("/topic/:topic_name/records", post(topic_iterate_route)) - .route("/topic/:topic_name", get(topic_get_info)) + .route("/topic/{topic_name}/records", post(topic_iterate_route)) + .route("/topic/{topic_name}", get(topic_get_info)) .route("/info", get(get_info)) .layer( TraceLayer::new(log_codes.into_make_classifier()) - .make_span_with(|request: &Request| { + .make_span_with(|request: &Request| { tracing::span!( target: "plateau::http", tracing::Level::INFO, @@ -166,17 +170,20 @@ pub async fn serve( ) .with_state(AppState(catalog, Arc::clone(&config))); - let server = Server::bind(&config.http.bind).serve(filter.into_make_service()); - let addr = server.local_addr(); + let listener = tokio::net::TcpListener::bind(&config.http.bind) + .await + .unwrap(); + let addr = listener.local_addr().unwrap(); - let fut = server.with_graceful_shutdown(FutureExt::map(rx_shutdown, |_| ())); + let server = + axum::serve(listener, filter).with_graceful_shutdown(FutureExt::map(rx_shutdown, |_| ())); let span = tracing::info_span!("Server::run", ?addr); tracing::info!(parent: &span, %addr, "listening"); ( addr, tx_shutdown, - Box::pin(async move { fut.instrument(span).await.unwrap_or(()) }), + Box::pin(async move { server.await.unwrap_or(()) }.instrument(span)), ) } @@ -228,7 +235,7 @@ async fn get_topics( responses( (status = 200, description = "Span of inserted records", body = Inserted), ), - request_body(content = SchemaChunk, content_type = "application/vnd.apache.arrow.file"), + request_body(content = Vec, content_type = "application/vnd.apache.arrow.file"), )] async fn topic_append( State(AppState(catalog, _config)): State, @@ -599,7 +606,6 @@ async fn get_info( Inserted, Partitions, // PartitionFilter, - crate::transport::ArrowSchemaChunk, Span, Topic, Topics, diff --git a/server/src/http/chunk.rs b/server/src/http/chunk.rs index 53ca803..1cd9f60 100644 --- a/server/src/http/chunk.rs +++ b/server/src/http/chunk.rs @@ -1,15 +1,14 @@ -use axum::{ - async_trait, - body::{boxed, Full, HttpBody}, - extract::{ - rejection::{BytesRejection, FailedToBufferBody}, - FromRef, FromRequest, - }, - headers::ContentType, - http::{header::CONTENT_TYPE, Request, StatusCode}, - response::Response, - BoxError, RequestExt as _, -}; +use axum::body::Body; +use axum::extract::rejection::BytesRejection; +use axum::extract::rejection::FailedToBufferBody; +use axum::extract::FromRef; +use axum::extract::FromRequest; +use axum::extract::Request; +use axum::http::header::CONTENT_TYPE; +use axum::http::StatusCode; +use axum::response::Response; +use axum::RequestExt as _; +use headers::ContentType; use bytes::Bytes; use std::io::{Cursor, Write}; @@ -35,18 +34,14 @@ const CONTENT_TYPE_PANDAS_RECORD: &str = "application/json; format=pandas-record pub(crate) struct SchemaChunkRequest(pub(crate) SchemaChunk); -#[async_trait] -impl FromRequest for SchemaChunkRequest +impl FromRequest for SchemaChunkRequest where - B: HttpBody + Send + 'static, - B::Data: Send, - B::Error: Into, Config: FromRef, S: Send + Sync, { type Rejection = ErrorReply; - async fn from_request(req: Request, state: &S) -> Result { + async fn from_request(req: Request, state: &S) -> Result { let config = Config::from_ref(state); let max_append_bytes = config.http.max_append_bytes; @@ -59,20 +54,20 @@ where ))?; if content_type == CONTENT_TYPE_ARROW { - let bytes = match req.with_limited_body() { - Ok(req) => req.extract::(), - Err(req) => req.extract::(), - } - .await - .map_err(|e| { - if let BytesRejection::FailedToBufferBody(FailedToBufferBody::LengthLimitError(_)) = - e - { - return ErrorReply::PayloadTooLarge(max_append_bytes); - } - - ErrorReply::Arrow(ArrowError::from_external_error(Box::new(e))) - })?; + let bytes = req + .with_limited_body() + .extract::() + .await + .map_err(|e| { + if let BytesRejection::FailedToBufferBody( + FailedToBufferBody::LengthLimitError(_), + ) = e + { + return ErrorReply::PayloadTooLarge(max_append_bytes); + } + + ErrorReply::Arrow(ArrowError::from_external_error(Box::new(e))) + })?; deserialize_request(bytes).await } else { @@ -144,14 +139,14 @@ pub(crate) fn to_reply( Response::builder() .header("Content-Type", CONTENT_TYPE_ARROW) .status(StatusCode::OK) - .body(boxed(Full::new(Bytes::from(bytes)))) + .body(Body::from(bytes)) .map_err(|_| ErrorReply::Unknown) } None | Some("*/*") | Some(CONTENT_TYPE_JSON) | Some(CONTENT_TYPE_PANDAS_RECORD) => { Response::builder() .header("Content-Type", CONTENT_TYPE_PANDAS_RECORD) .status(StatusCode::OK) - .body(boxed(Full::new(Bytes::from("[]")))) + .body(Body::from("[]")) .map_err(|_| ErrorReply::Unknown) } Some(other) => Err(ErrorReply::CannotEmit(other.to_string())), @@ -196,7 +191,7 @@ pub(crate) fn to_reply( .get("status") .unwrap_or(&"{}".to_string()), ) - .body(boxed(Full::new(Bytes::from(bytes)))) + .body(Body::from(bytes)) .map_err(|_| ErrorReply::Unknown) } None | Some("*/*") | Some(CONTENT_TYPE_JSON) | Some(CONTENT_TYPE_PANDAS_RECORD) => { @@ -233,7 +228,7 @@ pub(crate) fn to_reply( .unwrap_or(&"{}".to_string()), ) .status(StatusCode::OK) - .body(boxed(Full::new(Bytes::from(bytes)))) + .body(Body::from(bytes)) .map_err(|_| ErrorReply::Unknown) } Some(other) => Err(ErrorReply::CannotEmit(other.to_string())), diff --git a/transport/Cargo.toml b/transport/Cargo.toml index 3370944..ff01d20 100644 --- a/transport/Cargo.toml +++ b/transport/Cargo.toml @@ -29,10 +29,10 @@ arrow-ipc = "55.2.0" strum = { version = "0.26", features = ["derive"] } thiserror = "1" regex = "1.10" -utoipa = { version = "4", features = ["axum_extras"] } chrono = { version = "0.4", features = ["serde"] } tracing.workspace = true +utoipa.workspace = true [features] rweb = ["dep:rweb"] diff --git a/transport/src/lib.rs b/transport/src/lib.rs index 6020cad..4b0d602 100644 --- a/transport/src/lib.rs +++ b/transport/src/lib.rs @@ -264,6 +264,7 @@ pub struct RecordQuery { pub data_focus: DataFocus, #[serde(default)] #[cfg_attr(feature = "clap", arg(skip))] + #[param(value_type = Vec)] pub partition_filter: PartitionFilter, } @@ -327,6 +328,7 @@ pub struct TopicIterationQuery { pub data_focus: DataFocus, #[serde(default)] #[cfg_attr(feature = "clap", clap(skip))] + #[param(value_type = Vec)] pub partition_filter: PartitionFilter, } @@ -336,7 +338,7 @@ pub struct TopicIterationQuery { /// begin with "regex:" to signify that any partition matching the following string can be converted. pub type PartitionFilter = Option>; -#[derive(Clone, Debug, serde::Serialize, serde::Deserialize, utoipa::ToSchema)] +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] #[serde(from = "String", into = "String")] pub enum PartitionSelector { /// The exact name of a partition. @@ -533,8 +535,7 @@ pub const CONTENT_TYPE_JSON: &str = "application/json"; pub type SegmentChunk = RecordBatch; /// A [SegmentChunk] packaged with its associated [ArrowSchema]. -#[derive(Debug, Clone, PartialEq, ToSchema)] -#[aliases(ArrowSchemaChunk = SchemaChunk)] +#[derive(Debug, Clone, PartialEq)] pub struct SchemaChunk + Clone + PartialEq> { pub schema: S, pub chunk: SegmentChunk, From 670f0000262c56a5438b64cf684233c3fccc6423 Mon Sep 17 00:00:00 2001 From: Cyril Plisko Date: Mon, 20 Jul 2026 18:35:36 +0300 Subject: [PATCH 2/2] ENG-6268: decouple topic_iterate's input params from axum types topic_iterate is the one plateau-server function fitzroy calls into directly (not over HTTP) - it took Option>/HeaderMap/ Option> and immediately unwrapped them as its first three lines, so the axum-typed parameters were never actually needed by the function body. Moved that unwrapping into topic_iterate_route (the actual axum handler) instead, so topic_iterate now takes plain domain types (TopicIterationQuery, Option<&str>, TopicIterator). This removes half of the hard axum-version coupling between plateau-server and fitzroy - callers no longer need matching Query/HeaderMap/Json types to call this function. The return type (axum::response::Response) is still coupled, since the function genuinely does Accept-header content negotiation and builds a real HTTP response - decoupling that needs a plain intermediate representation and touches both plateau's and fitzroy's side, left for whoever picks up ENG-6220. No behavior change - pure signature refactor. --- server/src/http.rs | 21 +++++++++++---------- 1 file changed, 11 insertions(+), 10 deletions(-) diff --git a/server/src/http.rs b/server/src/http.rs index e1f0c7b..d5487f4 100644 --- a/server/src/http.rs +++ b/server/src/http.rs @@ -323,24 +323,25 @@ async fn topic_iterate_route( position: Option>, ) -> Result { let max_page = config.http.max_page; - topic_iterate(topic_name, query, headers, position, catalog, max_page).await + let query = query.map(|Query(query)| query).unwrap_or_default(); + let accept = headers.get(ACCEPT).and_then(|header| header.to_str().ok()); + let position = position.map(|Json(value)| value).unwrap_or_default(); + topic_iterate(topic_name, query, accept, position, catalog, max_page).await } +// Takes plain domain types rather than axum extractors so this can be called +// directly (e.g. from fitzroy) without depending on plateau-server's axum +// version. pub async fn topic_iterate( topic_name: String, - query: Option>, - headers: HeaderMap, - position: Option>, + query: TopicIterationQuery, + accept: Option<&str>, + position: TopicIterator, catalog: Arc, max_page: RowLimit, ) -> Result { - let query = query.map(|Query(query)| query).unwrap_or_default(); - let content = headers.get(ACCEPT).and_then(|header| header.to_str().ok()); - let position = position.map(|Json(value)| value); - let topic = catalog.get_topic(&topic_name).await; let page_size = RowLimit::records(query.page_size.unwrap_or(1000)).min(max_page); - let position = position.unwrap_or_default(); let partition_filter = query.partition_filter; let order: Ordering = query.order.unwrap_or(TopicIterationOrder::Asc).into(); @@ -371,7 +372,7 @@ pub async fn topic_iterate( ); } - chunk::to_reply(content, result.batch, query.data_focus) + chunk::to_reply(accept, result.batch, query.data_focus) } #[utoipa::path(