diff --git a/Cargo.lock b/Cargo.lock index 051e4dd378de..7ccf9f089a60 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -342,6 +342,16 @@ dependencies = [ "syn 2.0.117", ] +[[package]] +name = "asyncband" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94a214ba60d6231afd0e805e3c27c45a1626d9debaa5a5061c45a1ea1b2f1ed0" +dependencies = [ + "hashbrown 0.17.1", + "slab", +] + [[package]] name = "atomic" version = "0.6.1" @@ -689,9 +699,9 @@ dependencies = [ "http 1.4.1", "http-body 1.0.1", "http-body-util", - "md-5", + "md-5 0.10.6", "pin-project-lite", - "sha1", + "sha1 0.10.6", "sha2 0.10.9", "tracing", ] @@ -927,7 +937,7 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", - "sha1", + "sha1 0.10.6", "sync_wrapper 1.0.2", "tokio", "tokio-tungstenite", @@ -1044,6 +1054,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64-simd" version = "0.8.0" @@ -1556,7 +1572,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] @@ -1683,6 +1699,12 @@ version = "0.9.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + [[package]] name = "const-random" version = "0.1.18" @@ -2150,7 +2172,7 @@ version = "0.6.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1a467a65c5e759bce6e65eaf91cc29f466cdc57cb65777bd646872a8a1fd4de" dependencies = [ - "const-oid", + "const-oid 0.9.6", "zeroize", ] @@ -2160,7 +2182,7 @@ version = "0.7.10" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" dependencies = [ - "const-oid", + "const-oid 0.9.6", "pem-rfc7468", "zeroize", ] @@ -2296,6 +2318,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" dependencies = [ "block-buffer 0.12.1", + "const-oid 0.10.2", "crypto-common 0.2.2", "ctutils", ] @@ -2329,7 +2352,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -3045,16 +3068,22 @@ dependencies = [ "dynamo-runtime", "dynamo-sidecar-common", "futures", + "lru 0.16.4", + "opendal", "prost 0.13.5", "rand 0.9.4", "reqwest 0.12.28", + "rmp-serde", + "serde", "serde_json", + "tempfile", "tokio", "tokio-util", "tonic 0.13.1", "tonic-build 0.13.1", "tracing", "url", + "zstd", ] [[package]] @@ -3345,7 +3374,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -4575,7 +4604,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.4", + "socket2 0.5.10", "system-configuration 0.6.1", "tokio", "tower-service", @@ -4908,7 +4937,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -4967,10 +4996,12 @@ checksum = "4603d3033e49e2b0e31229fcab20a5d40089c607d975cd9c80551dc69eed9102" dependencies = [ "jiff-static", "jiff-tzdb-platform", + "js-sys", "log", "portable-atomic", "portable-atomic-util", "serde_core", + "wasm-bindgen", "windows-link 0.2.1", ] @@ -5659,9 +5690,9 @@ dependencies = [ [[package]] name = "log" -version = "0.4.30" +version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "616ec5685824bcc94416c6d4a7a446eea774a31efd7062c8480ba6fd06d7a6e5" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" [[package]] name = "loop9" @@ -5786,12 +5817,32 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest 0.11.3", +] + [[package]] name = "md5" version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7ebb8d8732c6a6df3d8f032a82911cfc747e00efb95cc46e8d0acd5b5b88570c" +[[package]] +name = "mea" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c709842c4ce65cb91e2666ad5319dfc1efc3af0d34f02075eddca9000d9f8afb" +dependencies = [ + "hashbrown 0.17.1", + "slab", +] + [[package]] name = "memchr" version = "2.8.1" @@ -6290,7 +6341,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -6610,10 +6661,10 @@ dependencies = [ "humantime", "hyper 1.10.1", "itertools 0.14.0", - "md-5", + "md-5 0.10.6", "parking_lot", "percent-encoding", - "quick-xml", + "quick-xml 0.38.4", "rand 0.9.4", "reqwest 0.12.28", "ring", @@ -6705,11 +6756,121 @@ dependencies = [ "serde", "serde_json", "serde_with", - "sha1", + "sha1 0.10.6", "sha2 0.10.9", "thiserror 2.0.18", ] +[[package]] +name = "opendal" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f950151f9587a51a7bed70a15fa0cff464eae96e41ae7499f97067bdafdf43eb" +dependencies = [ + "opendal-core", + "opendal-http-transport-reqwest", + "opendal-layer-retry", + "opendal-layer-timeout", + "opendal-service-fs", + "opendal-service-s3", +] + +[[package]] +name = "opendal-core" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a43405d217dfdfb543f58847336d3af672897dd1939bb7dcf314b63cf364f1c9" +dependencies = [ + "anyhow", + "asyncband", + "base64 0.23.1", + "bytes", + "futures", + "http 1.4.1", + "jiff", + "log", + "md-5 0.11.0", + "percent-encoding", + "quick-xml 0.41.0", + "reqsign-core", + "serde", + "serde_json", + "tokio", + "url", + "uuid", + "web-time", +] + +[[package]] +name = "opendal-http-transport-reqwest" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "401999057db611e592f883fcf2cbd6754ff37af587deaadd07b8c1398b2b6b06" +dependencies = [ + "bytes", + "futures", + "http 1.4.1", + "http-body 1.0.1", + "opendal-core", + "reqwest 0.13.4", +] + +[[package]] +name = "opendal-layer-retry" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06306202c97c54fb41bdbbdcb854c8798823f0022ae7aac7a40c8a1023aa83a6" +dependencies = [ + "backon", + "log", + "opendal-core", +] + +[[package]] +name = "opendal-layer-timeout" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6bd334cbd0a0bc934146733e74a80a5faf8db8e014781a25f6d9d80d7b87c981" +dependencies = [ + "opendal-core", + "tokio", +] + +[[package]] +name = "opendal-service-fs" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb9caf04d6d38713299dd4abac984b95ab16ee20e1ff09160f495c5a64644083" +dependencies = [ + "bytes", + "log", + "opendal-core", + "serde", + "tokio", + "xattr", +] + +[[package]] +name = "opendal-service-s3" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "388b1d39b62535c62803754ebef89808859558697366dbedd0299345887ba461" +dependencies = [ + "base64 0.23.1", + "bytes", + "crc-fast", + "http 1.4.1", + "log", + "md-5 0.11.0", + "opendal-core", + "quick-xml 0.41.0", + "reqsign-aws-v4", + "reqsign-core", + "reqsign-file-read-tokio", + "serde", + "url", +] + [[package]] name = "openssl-probe" version = "0.2.1" @@ -7457,7 +7618,7 @@ version = "0.13.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf" dependencies = [ - "heck 0.5.0", + "heck 0.4.1", "itertools 0.14.0", "log", "multimap", @@ -7477,7 +7638,7 @@ version = "0.14.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "343d3bd7056eda839b03204e68deff7d1b13aba7af2b2fd16890697274262ee7" dependencies = [ - "heck 0.5.0", + "heck 0.4.1", "itertools 0.14.0", "log", "multimap", @@ -7626,6 +7787,16 @@ dependencies = [ "serde", ] +[[package]] +name = "quick-xml" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" +dependencies = [ + "memchr", + "serde", +] + [[package]] name = "quick_cache" version = "0.7.0" @@ -7651,7 +7822,7 @@ dependencies = [ "quinn-udp", "rustc-hash 2.1.2", "rustls", - "socket2 0.6.4", + "socket2 0.5.10", "thiserror 2.0.18", "tokio", "tracing", @@ -7664,6 +7835,7 @@ version = "0.11.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83" dependencies = [ + "aws-lc-rs", "bytes", "getrandom 0.4.2", "lru-slab", @@ -7689,9 +7861,9 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.4", + "socket2 0.5.10", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.59.0", ] [[package]] @@ -8022,6 +8194,75 @@ version = "1.9.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba39f3699c378cd8970968dcbff9c43159ea4cfbd88d43c00b22f2ef10a435d2" +[[package]] +name = "reqsign-aws-core" +version = "3.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bac4749b7dfa7bfaccd01eb03e9dc795ed37e3f20d6f0f38e2c67ee85ad6bc86" +dependencies = [ + "bytes", + "form_urlencoded", + "hex", + "http 1.4.1", + "log", + "percent-encoding", + "quick-xml 0.41.0", + "reqsign-core", + "rust-ini", + "serde", + "serde_json", + "serde_urlencoded", + "sha1 0.11.0", +] + +[[package]] +name = "reqsign-aws-v4" +version = "3.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff250f0fd0b913fbd565e405acc553da0f13bde30bfb5403178c9d0313cdc15f" +dependencies = [ + "bytes", + "http 1.4.1", + "log", + "quick-xml 0.41.0", + "reqsign-aws-core", + "reqsign-core", + "serde", +] + +[[package]] +name = "reqsign-core" +version = "3.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff052daffb0599681c50f85c59e7236438976efe991ab864edd9f3b235501a0f" +dependencies = [ + "anyhow", + "base64 0.23.1", + "bytes", + "futures", + "hex", + "hmac 0.13.0", + "http 1.4.1", + "jiff", + "log", + "mea", + "percent-encoding", + "sha1 0.11.0", + "sha2 0.11.0", + "windows-sys 0.61.2", +] + +[[package]] +name = "reqsign-file-read-tokio" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3235df90a6bca681aa47dd86f2393d122a6d77042aa8a7c81e218cd45c5bfc0" +dependencies = [ + "anyhow", + "reqsign-core", + "tokio", +] + [[package]] name = "reqwest" version = "0.11.27" @@ -8126,6 +8367,7 @@ dependencies = [ "mime_guess", "percent-encoding", "pin-project-lite", + "quinn", "rustls", "rustls-pki-types", "rustls-platform-verifier", @@ -8411,7 +8653,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -8479,7 +8721,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -8952,6 +9194,17 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.3", +] + [[package]] name = "sha2" version = "0.10.9" @@ -9590,7 +9843,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -10486,7 +10739,7 @@ dependencies = [ "httparse", "log", "rand 0.9.4", - "sha1", + "sha1 0.10.6", "thiserror 2.0.18", "utf-8", ] @@ -11160,7 +11413,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] @@ -11628,6 +11881,16 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "xattr" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" +dependencies = [ + "libc", + "rustix 1.1.4", +] + [[package]] name = "xmlparser" version = "0.13.6" diff --git a/lib/bindings/python/Cargo.lock b/lib/bindings/python/Cargo.lock index 8b8da101d6ee..a98cb902f8a8 100644 --- a/lib/bindings/python/Cargo.lock +++ b/lib/bindings/python/Cargo.lock @@ -324,6 +324,16 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "asyncband" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94a214ba60d6231afd0e805e3c27c45a1626d9debaa5a5061c45a1ea1b2f1ed0" +dependencies = [ + "hashbrown 0.17.1", + "slab", +] + [[package]] name = "atomic" version = "0.6.1" @@ -441,7 +451,7 @@ dependencies = [ "fastrand", "hex", "http 1.4.2", - "sha1", + "sha1 0.10.6", "time", "tokio", "tracing", @@ -842,7 +852,7 @@ dependencies = [ "serde_json", "serde_path_to_error", "serde_urlencoded", - "sha1", + "sha1 0.10.6", "sync_wrapper", "tokio", "tokio-tungstenite", @@ -927,6 +937,12 @@ version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" +[[package]] +name = "base64" +version = "0.23.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ac07cdecf99051d9a5238b80f35af32cdeba5b336e55d957b318b50137e18da5" + [[package]] name = "base64-simd" version = "0.8.0" @@ -1381,7 +1397,7 @@ version = "3.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "faf9468729b8cbcea668e36183cb69d317348c2e08e994829fb56ebfdfbaac34" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] @@ -1554,6 +1570,16 @@ dependencies = [ "libc", ] +[[package]] +name = "crc-fast" +version = "1.10.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e75b2483e97a5a7da73ac68a05b629f9c53cff58d8ed1c77866079e18b00dba5" +dependencies = [ + "digest 0.10.7", + "spin 0.10.1", +] + [[package]] name = "crc32c" version = "0.6.8" @@ -2016,7 +2042,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -2548,9 +2574,13 @@ dependencies = [ "dynamo-runtime", "dynamo-sidecar-common", "futures", + "lru", + "opendal", "prost 0.13.5", "rand 0.9.4", "reqwest 0.12.28", + "rmp-serde", + "serde", "serde_json", "tokio", "tokio-util", @@ -2558,6 +2588,7 @@ dependencies = [ "tonic-build 0.13.1", "tracing", "url", + "zstd", ] [[package]] @@ -2789,7 +2820,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -3044,7 +3075,7 @@ dependencies = [ "fastrand", "futures-core", "futures-sink", - "spin", + "spin 0.9.8", ] [[package]] @@ -3866,7 +3897,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.4", + "socket2 0.5.10", "system-configuration", "tokio", "tower-service", @@ -4229,10 +4260,12 @@ checksum = "4603d3033e49e2b0e31229fcab20a5d40089c607d975cd9c80551dc69eed9102" dependencies = [ "jiff-static", "jiff-tzdb-platform", + "js-sys", "log", "portable-atomic", "portable-atomic-util", "serde_core", + "wasm-bindgen", "windows-link", ] @@ -4826,12 +4859,32 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "md-5" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69b6441f590336821bb897fb28fc622898ccceb1d6cea3fde5ea86b090c4de98" +dependencies = [ + "cfg-if", + "digest 0.11.3", +] + [[package]] name = "md5" version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7ebb8d8732c6a6df3d8f032a82911cfc747e00efb95cc46e8d0acd5b5b88570c" +[[package]] +name = "mea" +version = "0.6.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c709842c4ce65cb91e2666ad5319dfc1efc3af0d34f02075eddca9000d9f8afb" +dependencies = [ + "hashbrown 0.17.1", + "slab", +] + [[package]] name = "memchr" version = "2.8.2" @@ -5294,7 +5347,7 @@ version = "0.50.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7957b9740744892f114936ab4a57b3f487491bbeafaf8083688b16841a4240e5" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -5598,10 +5651,10 @@ dependencies = [ "humantime", "hyper", "itertools 0.14.0", - "md-5", + "md-5 0.10.6", "parking_lot", "percent-encoding", - "quick-xml", + "quick-xml 0.38.4", "rand 0.9.4", "reqwest 0.12.28", "ring", @@ -5687,11 +5740,121 @@ dependencies = [ "serde", "serde_json", "serde_with", - "sha1", + "sha1 0.10.6", "sha2 0.10.9", "thiserror 2.0.18", ] +[[package]] +name = "opendal" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f950151f9587a51a7bed70a15fa0cff464eae96e41ae7499f97067bdafdf43eb" +dependencies = [ + "opendal-core", + "opendal-http-transport-reqwest", + "opendal-layer-retry", + "opendal-layer-timeout", + "opendal-service-fs", + "opendal-service-s3", +] + +[[package]] +name = "opendal-core" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a43405d217dfdfb543f58847336d3af672897dd1939bb7dcf314b63cf364f1c9" +dependencies = [ + "anyhow", + "asyncband", + "base64 0.23.1", + "bytes", + "futures", + "http 1.4.2", + "jiff", + "log", + "md-5 0.11.0", + "percent-encoding", + "quick-xml 0.41.0", + "reqsign-core", + "serde", + "serde_json", + "tokio", + "url", + "uuid", + "web-time", +] + +[[package]] +name = "opendal-http-transport-reqwest" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "401999057db611e592f883fcf2cbd6754ff37af587deaadd07b8c1398b2b6b06" +dependencies = [ + "bytes", + "futures", + "http 1.4.2", + "http-body 1.0.1", + "opendal-core", + "reqwest 0.13.4", +] + +[[package]] +name = "opendal-layer-retry" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "06306202c97c54fb41bdbbdcb854c8798823f0022ae7aac7a40c8a1023aa83a6" +dependencies = [ + "backon", + "log", + "opendal-core", +] + +[[package]] +name = "opendal-layer-timeout" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6bd334cbd0a0bc934146733e74a80a5faf8db8e014781a25f6d9d80d7b87c981" +dependencies = [ + "opendal-core", + "tokio", +] + +[[package]] +name = "opendal-service-fs" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb9caf04d6d38713299dd4abac984b95ab16ee20e1ff09160f495c5a64644083" +dependencies = [ + "bytes", + "log", + "opendal-core", + "serde", + "tokio", + "xattr", +] + +[[package]] +name = "opendal-service-s3" +version = "0.59.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "388b1d39b62535c62803754ebef89808859558697366dbedd0299345887ba461" +dependencies = [ + "base64 0.23.1", + "bytes", + "crc-fast", + "http 1.4.2", + "log", + "md-5 0.11.0", + "opendal-core", + "quick-xml 0.41.0", + "reqsign-aws-v4", + "reqsign-core", + "reqsign-file-read-tokio", + "serde", + "url", +] + [[package]] name = "openssl-probe" version = "0.2.1" @@ -6607,6 +6770,16 @@ dependencies = [ "serde", ] +[[package]] +name = "quick-xml" +version = "0.41.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e660451e55124f798a69a5af3f49ccfbefbd41910eefd25caf2393e1f3473ec1" +dependencies = [ + "memchr", + "serde", +] + [[package]] name = "quick_cache" version = "0.7.0" @@ -6632,7 +6805,7 @@ dependencies = [ "quinn-udp", "rustc-hash 2.1.2", "rustls", - "socket2 0.6.4", + "socket2 0.5.10", "thiserror 2.0.18", "tokio", "tracing", @@ -6645,6 +6818,7 @@ version = "0.11.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83" dependencies = [ + "aws-lc-rs", "bytes", "getrandom 0.4.2", "lru-slab", @@ -6670,9 +6844,9 @@ dependencies = [ "cfg_aliases", "libc", "once_cell", - "socket2 0.6.4", + "socket2 0.5.10", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.59.0", ] [[package]] @@ -6978,6 +7152,75 @@ version = "0.8.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d6f6ff9a378485b298a5286656da665ba74413d36db0979633275d2e708145d4" +[[package]] +name = "reqsign-aws-core" +version = "3.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bac4749b7dfa7bfaccd01eb03e9dc795ed37e3f20d6f0f38e2c67ee85ad6bc86" +dependencies = [ + "bytes", + "form_urlencoded", + "hex", + "http 1.4.2", + "log", + "percent-encoding", + "quick-xml 0.41.0", + "reqsign-core", + "rust-ini", + "serde", + "serde_json", + "serde_urlencoded", + "sha1 0.11.0", +] + +[[package]] +name = "reqsign-aws-v4" +version = "3.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff250f0fd0b913fbd565e405acc553da0f13bde30bfb5403178c9d0313cdc15f" +dependencies = [ + "bytes", + "http 1.4.2", + "log", + "quick-xml 0.41.0", + "reqsign-aws-core", + "reqsign-core", + "serde", +] + +[[package]] +name = "reqsign-core" +version = "3.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff052daffb0599681c50f85c59e7236438976efe991ab864edd9f3b235501a0f" +dependencies = [ + "anyhow", + "base64 0.23.1", + "bytes", + "futures", + "hex", + "hmac", + "http 1.4.2", + "jiff", + "log", + "mea", + "percent-encoding", + "sha1 0.11.0", + "sha2 0.11.0", + "windows-sys 0.61.2", +] + +[[package]] +name = "reqsign-file-read-tokio" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3235df90a6bca681aa47dd86f2393d122a6d77042aa8a7c81e218cd45c5bfc0" +dependencies = [ + "anyhow", + "reqsign-core", + "tokio", +] + [[package]] name = "reqwest" version = "0.12.28" @@ -7046,6 +7289,7 @@ dependencies = [ "mime_guess", "percent-encoding", "pin-project-lite", + "quinn", "rustls", "rustls-pki-types", "rustls-platform-verifier", @@ -7232,7 +7476,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -7300,7 +7544,7 @@ dependencies = [ "security-framework", "security-framework-sys", "webpki-root-certs", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -7731,6 +7975,17 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "digest 0.11.3", +] + [[package]] name = "sha2" version = "0.10.9" @@ -7916,6 +8171,12 @@ dependencies = [ "lock_api", ] +[[package]] +name = "spin" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3" + [[package]] name = "spki" version = "0.7.3" @@ -8290,7 +8551,7 @@ dependencies = [ "getrandom 0.4.2", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.2", + "windows-sys 0.59.0", ] [[package]] @@ -9080,7 +9341,7 @@ dependencies = [ "httparse", "log", "rand 0.9.4", - "sha1", + "sha1 0.10.6", "thiserror 2.0.18", "utf-8", ] @@ -9679,7 +9940,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.61.2", + "windows-sys 0.48.0", ] [[package]] @@ -10113,6 +10374,16 @@ version = "0.6.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4" +[[package]] +name = "xattr" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" +dependencies = [ + "libc", + "rustix 1.1.4", +] + [[package]] name = "xmlparser" version = "0.13.6" diff --git a/lib/llm/src/protocols/common/extensions.rs b/lib/llm/src/protocols/common/extensions.rs index 527e85095018..e198cd60c180 100644 --- a/lib/llm/src/protocols/common/extensions.rs +++ b/lib/llm/src/protocols/common/extensions.rs @@ -46,6 +46,13 @@ pub struct RouterParams { pub struct MetadataUpload { #[serde(deserialize_with = "deserialize_metadata_upload_url")] pub url: String, + + #[serde( + default, + skip_serializing_if = "Option::is_none", + deserialize_with = "deserialize_metadata_upload_fallback_url" + )] + pub fallback_url: Option, } fn deserialize_metadata_upload_url<'de, D>(deserializer: D) -> Result @@ -62,6 +69,26 @@ where Ok(url.to_string()) } +fn deserialize_metadata_upload_fallback_url<'de, D>( + deserializer: D, +) -> Result, D::Error> +where + D: serde::Deserializer<'de>, +{ + Option::::deserialize(deserializer)? + .map(|url| { + let url = url.trim(); + if url.is_empty() { + Err(serde::de::Error::custom( + "metadata_upload.fallback_url must not be empty", + )) + } else { + Ok(url.to_string()) + } + }) + .transpose() +} + /// Causal trigger that produced an incoming agent request. #[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)] #[serde(rename_all = "snake_case")] @@ -992,16 +1019,21 @@ mod tests { } #[test] - fn metadata_upload_parses_url() { + fn metadata_upload_parses_urls() { let nvext: NvExt = serde_json::from_value(serde_json::json!({ "metadata_upload": { - "url": " s3://bucket/root/rollouts " + "url": " s3://bucket/root/rollouts ", + "fallback_url": " fs:///var/tmp/rollouts " } })) .unwrap(); let upload = nvext.metadata_upload.as_ref().unwrap(); assert_eq!(upload.url, "s3://bucket/root/rollouts"); + assert_eq!( + upload.fallback_url.as_deref(), + Some("fs:///var/tmp/rollouts") + ); assert!(!NvExtResponseFieldSelection::from_nvext(Some(&nvext)).engine_data); assert!( @@ -1018,6 +1050,15 @@ mod tests { })) .is_err() ); + assert!( + serde_json::from_value::(serde_json::json!({ + "metadata_upload": { + "url": "s3://bucket/root/rollouts", + "fallback_url": "" + } + })) + .is_err() + ); assert!( serde_json::from_value::(serde_json::json!({ "metadata_upload": { diff --git a/lib/sidecar/sglang/Cargo.toml b/lib/sidecar/sglang/Cargo.toml index 596192eecfe9..31ac7166cb04 100644 --- a/lib/sidecar/sglang/Cargo.toml +++ b/lib/sidecar/sglang/Cargo.toml @@ -29,13 +29,24 @@ async-stream = { workspace = true } async-trait = { workspace = true } clap = { version = "4", features = ["derive", "env"] } futures = { workspace = true } +lru = { workspace = true } +opendal = { version = "0.59.1", default-features = false, features = [ + "http-transport-reqwest", + "layers-retry", + "layers-timeout", + "services-fs", + "services-s3", +] } rand = { workspace = true } reqwest = { workspace = true } +rmp-serde = { workspace = true } +serde = { workspace = true } serde_json = { workspace = true } tokio = { workspace = true } tokio-util = { workspace = true } tracing = { workspace = true } url = { workspace = true } +zstd = "0.13" prost = { workspace = true } tonic = { workspace = true } @@ -45,3 +56,4 @@ tonic-build = { workspace = true } [dev-dependencies] dynamo-backend-common = { workspace = true, features = ["testing"] } +tempfile = { workspace = true } diff --git a/lib/sidecar/sglang/src/args.rs b/lib/sidecar/sglang/src/args.rs index 153bec1549a7..705d13dd5191 100644 --- a/lib/sidecar/sglang/src/args.rs +++ b/lib/sidecar/sglang/src/args.rs @@ -5,6 +5,8 @@ use dynamo_sidecar_common::SidecarArgs; +use crate::metadata_upload::MetadataUploadArgs; + /// Parsed sidecar arguments. #[derive(clap::Parser, Debug, Clone)] #[command( @@ -22,4 +24,7 @@ pub struct Args { /// wildcard addresses. #[arg(long, env = "SGLANG_DISAGGREGATION_BOOTSTRAP_HOST")] pub bootstrap_host: Option, + + #[command(flatten)] + pub metadata_upload: MetadataUploadArgs, } diff --git a/lib/sidecar/sglang/src/engine.rs b/lib/sidecar/sglang/src/engine.rs index 17c2dae87056..066e73024113 100644 --- a/lib/sidecar/sglang/src/engine.rs +++ b/lib/sidecar/sglang/src/engine.rs @@ -22,11 +22,12 @@ use tokio_util::sync::CancellationToken; use crate::args::Args; use crate::client::{self, Client, Discovery, Pool}; +use crate::metadata_upload::{MetadataDelivery, MetadataUploadService}; use crate::native_http::{self, NativeHttp}; use crate::proto as pb; use crate::protocol::{ - build_generate_request, disaggregated_params_to_json, engine_data_from_meta, extract_logprobs, - meta_u32, output_ids_to_u32, terminal_from_meta, + build_generate_request, disaggregated_params_to_json, meta_u32, output_ids_to_u32, + terminal_from_meta, }; pub struct SglangSidecarEngine { @@ -35,6 +36,7 @@ pub struct SglangSidecarEngine { disaggregation_mode: DisaggregationMode, bootstrap_host: Option, bootstrap_port: Option, + metadata_upload: Option, state: OnceCell, cancel: CancellationToken, } @@ -75,7 +77,6 @@ impl SglangSidecarEngine { "route-to-encoder is not supported by the SGLang sidecar", )); } - let endpoint = args.sidecar.grpc_endpoint; let transport = args.sidecar.grpc.config(); let discovery = bootstrap_discover(&endpoint, &transport)?; @@ -102,6 +103,11 @@ impl SglangSidecarEngine { ); let common = args.sidecar.common; + let metadata_upload = if common.enable_rl && !disaggregation_mode.is_prefill() { + Some(MetadataUploadService::new(args.metadata_upload)?) + } else { + None + }; let config = WorkerConfig { namespace: common.namespace, component: if disaggregation_mode == DisaggregationMode::Aggregated { @@ -135,6 +141,7 @@ impl SglangSidecarEngine { disaggregation_mode, bootstrap_host, bootstrap_port, + metadata_upload, state: OnceCell::new(), cancel: CancellationToken::new(), }, @@ -277,6 +284,10 @@ impl LLMEngine for SglangSidecarEngine { self.bootstrap_host.as_deref(), self.bootstrap_port, )?; + let metadata_delivery = match &self.metadata_upload { + Some(service) => service.delivery_for(&request).await?, + None => MetadataDelivery::Inline, + }; let prefill_handoff = if self.disaggregation_mode.is_prefill() { grpc_request .disaggregated_params @@ -353,15 +364,6 @@ impl LLMEngine for SglangSidecarEngine { break; } }; - let (log_probs, top_logprobs) = - match extract_logprobs(&response.meta_info, return_tokens_as_ids) { - Ok(values) => values, - Err(err) => { - yield Err(err); - break; - } - }; - if is_prefill { if response.finished { let mut terminal = match terminal_from_meta( @@ -395,12 +397,26 @@ impl LLMEngine for SglangSidecarEngine { break; } }; - let engine_data = match engine_data_from_meta(&response.meta_info, true) { - Ok(engine_data) => engine_data, - Err(error) => { + let fields = tokio::select! { + biased; + _ = ctx.stopped() => None, + _ = cancel.cancelled() => None, + result = metadata_delivery.finish( + &response.meta_info, + return_tokens_as_ids, + ) => Some(result), + }; + let ((log_probs, top_logprobs), engine_data) = match fields { + Some(Ok(fields)) => fields, + Some(Err(error)) => { yield Err(error); break; } + None => { + yield Ok(LLMEngineOutput::cancelled() + .with_usage(usage(observed_prompt_tokens, generated))); + break; + } }; terminal.token_ids = token_ids; terminal.log_probs = log_probs; @@ -411,8 +427,10 @@ impl LLMEngine for SglangSidecarEngine { } if !token_ids.is_empty() { - let engine_data = match engine_data_from_meta(&response.meta_info, false) { - Ok(engine_data) => engine_data, + let ((log_probs, top_logprobs), engine_data) = match metadata_delivery + .chunk_fields(&response.meta_info, return_tokens_as_ids) + { + Ok(fields) => fields, Err(error) => { yield Err(error); break; diff --git a/lib/sidecar/sglang/src/lib.rs b/lib/sidecar/sglang/src/lib.rs index 7e8968fc3207..578b0e3929ea 100644 --- a/lib/sidecar/sglang/src/lib.rs +++ b/lib/sidecar/sglang/src/lib.rs @@ -15,6 +15,7 @@ pub mod args; pub mod client; pub mod engine; +mod metadata_upload; mod native_http; /// Generated SGLang gRPC types, temporarily exposed for the Mocker server diff --git a/lib/sidecar/sglang/src/metadata_upload.rs b/lib/sidecar/sglang/src/metadata_upload.rs new file mode 100644 index 000000000000..b02a7bcb0346 --- /dev/null +++ b/lib/sidecar/sglang/src/metadata_upload.rs @@ -0,0 +1,527 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Out-of-band storage for SGLang response metadata. + +use std::collections::HashMap; +use std::io::Write; +use std::sync::Arc; + +use dynamo_backend_common::{DynamoError, PreprocessedRequest}; +use lru::LruCache; +use opendal::layers::{RetryLayer, TimeoutLayer}; +use opendal::{Buffer, Operator}; +use serde::{Deserialize, Serialize}; +use serde_json::Value; +use tokio::sync::{Mutex, OnceCell}; + +use crate::client; +use crate::protocol::{ExtractedLogprobs, engine_data_from_meta, extract_logprobs}; + +mod config; +pub(crate) use config::MetadataUploadArgs; +use config::OperatorPolicy; + +const OUTPUT_PATH: &str = "choice_0.msgpack.zst"; +type CachedOperator = Arc>>; +type ResponseFields = (ExtractedLogprobs, Option); + +pub(crate) struct MetadataUploadService { + operators: OperatorCache, +} + +pub(crate) enum MetadataDelivery { + Inline, + Upload(Box), +} + +pub(crate) struct MetadataUploader { + primary: Operator, + fallback: Option, +} + +pub(crate) struct OperatorCache { + operators: Mutex>, + policy: OperatorPolicy, + #[cfg(test)] + build_count: std::sync::atomic::AtomicUsize, +} + +#[derive(Serialize)] +struct MetadataPayload<'a> { + schema_version: u8, + metadata: &'a Value, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct MetadataUploadConfig { + url: String, + fallback_url: Option, +} + +impl OperatorCache { + fn new(policy: OperatorPolicy) -> Self { + // Ensure built-in services are registered when this crate is linked statically. + opendal::install_default(); + Self { + operators: Mutex::new(LruCache::new(policy.capacity)), + policy, + #[cfg(test)] + build_count: std::sync::atomic::AtomicUsize::new(0), + } + } + + async fn get(&self, raw: String, field: &'static str) -> Result { + let url = normalize_url(raw, field)?; + let cell = { + let mut operators = self.operators.lock().await; + if let Some(cell) = operators.get(&url) { + cell.clone() + } else { + let cell = Arc::new(OnceCell::new()); + operators.put(url.clone(), cell.clone()); + cell + } + }; + let policy = self.policy; + let operator = cell + .get_or_init(|| async { + #[cfg(test)] + self.build_count + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let build_url = url.clone(); + tokio::task::spawn_blocking(move || build_operator(&build_url, policy)) + .await + .map_err(|error| { + client::protocol_error(format!( + "metadata upload setup task failed: {error}" + )) + })? + }) + .await; + operator.clone() + } + + #[cfg(test)] + async fn insert(&self, url: &str, operator: Operator) { + let cell = OnceCell::new(); + cell.set(Ok(operator)).expect("new OnceCell must be empty"); + self.operators + .lock() + .await + .put(url.to_string(), Arc::new(cell)); + } +} + +impl MetadataUploadService { + pub(crate) fn new(args: MetadataUploadArgs) -> Result { + Ok(Self { + operators: OperatorCache::new(args.try_into()?), + }) + } + + pub(crate) async fn delivery_for( + &self, + request: &PreprocessedRequest, + ) -> Result { + let Some(config) = request + .extra_args + .as_ref() + .and_then(Value::as_object) + .and_then(|extra| extra.get("nvext")) + .and_then(Value::as_object) + .and_then(|nvext| nvext.get("metadata_upload")) + else { + return Ok(MetadataDelivery::Inline); + }; + if config.is_null() { + return Ok(MetadataDelivery::Inline); + } + let config = MetadataUploadConfig::deserialize(config).map_err(|error| { + client::invalid_arg(format!("invalid extra_args.nvext.metadata_upload: {error}")) + })?; + let primary = self.operators.get(config.url, "url"); + let fallback = async { + match config.fallback_url { + Some(url) => self.operators.get(url, "fallback_url").await.map(Some), + None => Ok(None), + } + }; + let (primary, fallback) = tokio::join!(primary, fallback); + let primary = primary?; + let fallback = fallback?; + Ok(MetadataDelivery::Upload(Box::new(MetadataUploader { + primary, + fallback, + }))) + } +} + +impl MetadataDelivery { + pub(crate) fn chunk_fields( + &self, + meta: &HashMap, + return_tokens_as_ids: bool, + ) -> Result { + match self { + Self::Inline => Ok(( + extract_logprobs(meta, return_tokens_as_ids)?, + engine_data_from_meta(meta, false)?, + )), + Self::Upload(_) => Ok(((None, None), None)), + } + } + + pub(crate) async fn finish( + &self, + meta: &HashMap, + return_tokens_as_ids: bool, + ) -> Result { + match self { + Self::Inline => Ok(( + extract_logprobs(meta, return_tokens_as_ids)?, + engine_data_from_meta(meta, true)?, + )), + Self::Upload(uploader) => { + uploader.upload(grpc_metadata(meta)).await?; + Ok(((None, None), None)) + } + } + } +} + +impl MetadataUploader { + pub(crate) async fn upload(&self, metadata: Value) -> Result<(), DynamoError> { + if metadata.as_object().is_some_and(|value| value.is_empty()) { + return Ok(()); + } + let compressed: Buffer = tokio::task::spawn_blocking(move || encode(metadata)) + .await + .map_err(|error| { + client::protocol_error(format!("SGLang metadata encoding task failed: {error}")) + })?? + .into(); + match write(&self.primary, compressed.clone()).await { + Ok(()) => Ok(()), + Err(primary_error) => { + let Some(fallback) = self.fallback.as_ref() else { + return Err(primary_error); + }; + tracing::warn!(error = %primary_error, "primary metadata upload failed; attempting fallback"); + write(fallback, compressed).await + } + } + } +} + +fn normalize_url(raw: String, field: &str) -> Result { + let raw = raw.trim(); + if raw.is_empty() { + return Err(client::invalid_arg(format!( + "extra_args.nvext.metadata_upload.{field} must not be empty" + ))); + } + Ok(raw.to_string()) +} + +fn build_operator(raw: &str, policy: OperatorPolicy) -> Result { + Operator::from_uri(raw) + .map(|operator| configure_operator(operator, policy)) + .map_err(|error| { + client::invalid_arg(format!( + "could not configure `{}` metadata upload destination: {error}", + raw.split_once(':') + .map(|(scheme, _)| scheme) + .unwrap_or("unknown") + )) + }) +} + +fn configure_operator(operator: Operator, policy: OperatorPolicy) -> Operator { + let retry = RetryLayer::new() + .with_factor(policy.retry_factor) + .with_min_delay(policy.retry_min_delay) + .with_max_delay(policy.retry_max_delay) + .with_max_times(policy.retry_max_times); + let retry = if policy.retry_jitter { + retry.with_jitter() + } else { + retry + }; + let timeout = TimeoutLayer::new() + .with_timeout(policy.timeout) + .with_io_timeout(policy.io_timeout); + // Timeout must be inside RetryLayer so a timeout cannot drop a stateful + // retry future before OpenDAL restores its body state. + operator.layer(timeout).layer(retry) +} + +async fn write(operator: &Operator, data: Buffer) -> Result<(), DynamoError> { + operator + .write(OUTPUT_PATH, data) + .await + .map(|_| ()) + .map_err(|error| upload_error(operator, error)) +} + +fn upload_error(operator: &Operator, error: opendal::Error) -> DynamoError { + client::protocol_error(format!( + "could not upload SGLang metadata to {}: {error}", + operator.info().root() + )) +} + +fn encode(metadata: Value) -> Result, DynamoError> { + let payload = MetadataPayload { + schema_version: 1, + metadata: &metadata, + }; + let msgpack = rmp_serde::to_vec_named(&payload).map_err(|error| { + client::protocol_error(format!("could not serialize SGLang metadata: {error}")) + })?; + let mut encoder = zstd::stream::write::Encoder::new(Vec::new(), 0).map_err(|error| { + client::protocol_error(format!( + "could not initialize SGLang metadata compression: {error}" + )) + })?; + // Python's ZstdDecompressor.decompress(), used by the existing fsspec + // reader, requires the uncompressed size in the frame header. + encoder + .set_pledged_src_size(Some(msgpack.len() as u64)) + .map_err(|error| { + client::protocol_error(format!( + "could not configure SGLang metadata compression: {error}" + )) + })?; + encoder.include_contentsize(true).map_err(|error| { + client::protocol_error(format!( + "could not configure SGLang metadata compression: {error}" + )) + })?; + encoder.write_all(&msgpack).map_err(|error| { + client::protocol_error(format!("could not compress SGLang metadata: {error}")) + })?; + encoder.finish().map_err(|error| { + client::protocol_error(format!( + "could not finish SGLang metadata compression: {error}" + )) + }) +} + +pub(crate) fn grpc_metadata(meta: &HashMap) -> Value { + Value::Object( + meta.iter() + .map(|(key, raw)| { + let value = + serde_json::from_str(raw).unwrap_or_else(|_| Value::String(raw.clone())); + (key.clone(), value) + }) + .collect(), + ) +} + +#[cfg(test)] +mod tests { + use std::io::Cursor; + use std::num::NonZeroUsize; + use std::path::Path; + use std::time::Duration; + + use dynamo_backend_common::{OutputOptions, SamplingOptions, StopConditions}; + use serde_json::{Value, json}; + + use super::config::OperatorPolicy; + use super::{MetadataDelivery, MetadataUploadService, OUTPUT_PATH, OperatorCache}; + + fn policy(capacity: usize) -> OperatorPolicy { + OperatorPolicy { + capacity: NonZeroUsize::new(capacity).unwrap(), + timeout: Duration::from_secs(10), + io_timeout: Duration::from_secs(10), + retry_max_times: 3, + retry_min_delay: Duration::from_millis(1), + retry_max_delay: Duration::from_millis(10), + retry_factor: 2.0, + retry_jitter: false, + } + } + + fn operator_cache(capacity: usize) -> OperatorCache { + OperatorCache::new(policy(capacity)) + } + + fn service(capacity: usize) -> MetadataUploadService { + MetadataUploadService { + operators: operator_cache(capacity), + } + } + + fn fs_uri(path: &Path) -> String { + format!("fs://{}", path.display()) + } + + fn request(extra_args: Value) -> dynamo_backend_common::PreprocessedRequest { + dynamo_backend_common::PreprocessedRequest::builder() + .model("model".to_string()) + .token_ids(vec![1]) + .sampling_options(SamplingOptions::default()) + .output_options(OutputOptions::default()) + .stop_conditions(StopConditions::default()) + .extra_args(Some(extra_args)) + .build() + .unwrap() + } + + #[tokio::test] + async fn request_configuration_is_optional_and_strict() { + let configured = request(json!({ + "nvext": {"metadata_upload": {"url": "fs:///tmp/metadata"}} + })); + assert!(matches!( + service(2).delivery_for(&configured).await.unwrap(), + MetadataDelivery::Upload(_) + )); + + let unconfigured = request(json!({"nvext": {}})); + assert!(matches!( + service(2).delivery_for(&unconfigured).await.unwrap(), + MetadataDelivery::Inline + )); + + let invalid = request(json!({ + "nvext": {"metadata_upload": {"url": "fs:///tmp", "format": "json"}} + })); + assert!(service(2).delivery_for(&invalid).await.is_err()); + + let invalid_fallback = request(json!({ + "nvext": {"metadata_upload": { + "url": "fs:///tmp", + "fallback_url": "unknown://destination" + }} + })); + assert!(service(2).delivery_for(&invalid_fallback).await.is_err()); + } + + #[tokio::test] + async fn delivery_mode_owns_inline_response_fields() { + let meta = std::collections::HashMap::from([ + ("output_token_logprobs".into(), r#"[[-0.1,101,"a"]]"#.into()), + ("routed_experts".into(), "[1,2]".into()), + ]); + let inline = service(2) + .delivery_for(&request(json!({"nvext": {}}))) + .await + .unwrap(); + let ((logprobs, _), engine_data) = inline.chunk_fields(&meta, false).unwrap(); + assert_eq!(logprobs.unwrap(), vec![-0.1]); + assert_eq!(engine_data.unwrap()["routed_experts"], json!([1, 2])); + + opendal::OperatorRegistry::get().register::("metadata-fields"); + let remote = service(2) + .delivery_for(&request(json!({ + "nvext": {"metadata_upload": {"url": "metadata-fields://rollout"}} + }))) + .await + .unwrap(); + assert_eq!( + remote.chunk_fields(&meta, false).unwrap(), + ((None, None), None) + ); + } + + #[tokio::test] + async fn uploads_python_compatible_msgpack_zstd_payload() { + let directory = tempfile::tempdir().unwrap(); + let url = fs_uri(directory.path()); + let configured = request(json!({ + "nvext": {"metadata_upload": {"url": url}} + })); + let delivery = service(2).delivery_for(&configured).await.unwrap(); + delivery + .finish( + &std::collections::HashMap::from([ + ("id".into(), "sglang-1".into()), + ("finish_reason".into(), r#"{"type":"stop"}"#.into()), + ("output_token_logprobs".into(), r#"[[-0.1,101,"a"]]"#.into()), + ]), + false, + ) + .await + .unwrap(); + + let compressed = std::fs::read(directory.path().join("choice_0.msgpack.zst")).unwrap(); + assert!( + zstd::zstd_safe::get_frame_content_size(&compressed) + .unwrap() + .is_some(), + "the frame must advertise its content size for Python compatibility" + ); + let msgpack = zstd::stream::decode_all(Cursor::new(compressed)).unwrap(); + let payload: Value = rmp_serde::from_slice(&msgpack).unwrap(); + assert_eq!(payload["schema_version"], 1); + assert_eq!(payload["metadata"]["id"], "sglang-1"); + assert_eq!(payload["metadata"]["output_token_logprobs"][0][1], 101); + } + + #[tokio::test] + async fn uses_fallback_only_after_primary_failure() { + let directory = tempfile::tempdir().unwrap(); + let primary = directory.path().join("primary-is-a-file"); + std::fs::write(&primary, b"primary sentinel").unwrap(); + let fallback = directory.path().join("fallback"); + let configured = request(json!({ + "nvext": {"metadata_upload": { + "url": fs_uri(&primary), + "fallback_url": fs_uri(&fallback) + }} + })); + let delivery = service(2).delivery_for(&configured).await.unwrap(); + delivery + .finish( + &std::collections::HashMap::from([("id".into(), "sglang-2".into())]), + false, + ) + .await + .unwrap(); + + assert_eq!(std::fs::read(primary).unwrap(), b"primary sentinel"); + assert!(fallback.join(OUTPUT_PATH).is_file()); + } + + #[tokio::test] + async fn accepts_a_registered_custom_scheme() { + opendal::OperatorRegistry::get().register::("custom-metadata"); + let configured = request(json!({ + "nvext": {"metadata_upload": {"url": "custom-metadata://rollout"}} + })); + + assert!(matches!( + service(2).delivery_for(&configured).await.unwrap(), + MetadataDelivery::Upload(_) + )); + } + + #[tokio::test] + async fn operator_cache_evicts_least_recently_used_url() { + let directory = tempfile::tempdir().unwrap(); + let first = fs_uri(&directory.path().join("first")); + let second = fs_uri(&directory.path().join("second")); + let third = fs_uri(&directory.path().join("third")); + let cache = operator_cache(2); + + cache.get(first.clone(), "url").await.unwrap(); + cache.get(second.clone(), "url").await.unwrap(); + cache.get(first.clone(), "url").await.unwrap(); + cache.get(third.clone(), "url").await.unwrap(); + + let operators = cache.operators.lock().await; + assert!(operators.peek(&first).is_some()); + assert!(operators.peek(&second).is_none()); + assert!(operators.peek(&third).is_some()); + } +} + +#[cfg(test)] +#[path = "metadata_upload/resilience_tests.rs"] +mod resilience_tests; diff --git a/lib/sidecar/sglang/src/metadata_upload/config.rs b/lib/sidecar/sglang/src/metadata_upload/config.rs new file mode 100644 index 000000000000..c38945b9dd0a --- /dev/null +++ b/lib/sidecar/sglang/src/metadata_upload/config.rs @@ -0,0 +1,199 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::num::{NonZeroU64, NonZeroUsize}; +use std::time::Duration; + +use dynamo_backend_common::DynamoError; + +use crate::client; + +#[derive(clap::Args, Debug, Clone)] +pub struct MetadataUploadArgs { + /// Maximum number of OpenDAL operators retained by the metadata upload LRU. + #[arg( + long = "metadata-upload-operator-cache-capacity", + env = "DYN_SGLANG_METADATA_UPLOAD_OPERATOR_CACHE_CAPACITY", + default_value = "64" + )] + pub operator_cache_capacity: NonZeroUsize, + + /// OpenDAL control-operation timeout in seconds. + #[arg( + long = "metadata-upload-timeout-secs", + env = "DYN_SGLANG_METADATA_UPLOAD_TIMEOUT_SECS", + default_value = "3" + )] + pub timeout_secs: NonZeroU64, + + /// Timeout in seconds for each OpenDAL I/O attempt. + #[arg( + long = "metadata-upload-io-timeout-secs", + env = "DYN_SGLANG_METADATA_UPLOAD_IO_TIMEOUT_SECS", + default_value = "3" + )] + pub io_timeout_secs: NonZeroU64, + + /// Maximum retries after the initial OpenDAL operation attempt. + #[arg( + long = "metadata-upload-retry-max-times", + env = "DYN_SGLANG_METADATA_UPLOAD_RETRY_MAX_TIMES", + default_value = "3" + )] + pub retry_max_times: usize, + + /// Initial exponential retry delay in milliseconds. + #[arg( + long = "metadata-upload-retry-min-delay-ms", + env = "DYN_SGLANG_METADATA_UPLOAD_RETRY_MIN_DELAY_MS", + default_value_t = 100 + )] + pub retry_min_delay_ms: u64, + + /// Maximum exponential retry delay in milliseconds. + #[arg( + long = "metadata-upload-retry-max-delay-ms", + env = "DYN_SGLANG_METADATA_UPLOAD_RETRY_MAX_DELAY_MS", + default_value_t = 1_000 + )] + pub retry_max_delay_ms: u64, + + /// Exponential retry multiplier; must be at least 1.0. + #[arg( + long = "metadata-upload-retry-factor", + env = "DYN_SGLANG_METADATA_UPLOAD_RETRY_FACTOR", + default_value_t = 2.0 + )] + pub retry_factor: f32, + + /// Add random jitter to exponential retry delays. + #[arg(long = "metadata-upload-retry-jitter", env = "DYN_SGLANG_METADATA_UPLOAD_RETRY_JITTER", default_value_t = true, action = clap::ArgAction::Set)] + pub retry_jitter: bool, +} + +#[derive(Clone, Copy)] +pub(super) struct OperatorPolicy { + pub capacity: NonZeroUsize, + pub timeout: Duration, + pub io_timeout: Duration, + pub retry_max_times: usize, + pub retry_min_delay: Duration, + pub retry_max_delay: Duration, + pub retry_factor: f32, + pub retry_jitter: bool, +} + +impl TryFrom for OperatorPolicy { + type Error = DynamoError; + + fn try_from(args: MetadataUploadArgs) -> Result { + if !args.retry_factor.is_finite() || args.retry_factor < 1.0 { + return Err(client::invalid_arg( + "metadata-upload-retry-factor must be finite and at least 1.0", + )); + } + if args.retry_max_delay_ms < args.retry_min_delay_ms { + return Err(client::invalid_arg( + "metadata-upload-retry-max-delay-ms must be greater than or equal to metadata-upload-retry-min-delay-ms", + )); + } + Ok(Self { + capacity: args.operator_cache_capacity, + timeout: Duration::from_secs(args.timeout_secs.get()), + io_timeout: Duration::from_secs(args.io_timeout_secs.get()), + retry_max_times: args.retry_max_times, + retry_min_delay: Duration::from_millis(args.retry_min_delay_ms), + retry_max_delay: Duration::from_millis(args.retry_max_delay_ms), + retry_factor: args.retry_factor, + retry_jitter: args.retry_jitter, + }) + } +} + +#[cfg(test)] +mod tests { + use clap::{CommandFactory, Parser}; + + use crate::args::Args; + + #[test] + fn policy_is_configurable_by_cli_and_env() { + let args = Args::try_parse_from([ + "dynamo-sglang-sidecar", + "--grpc-endpoint", + "127.0.0.1:30000", + "--metadata-upload-operator-cache-capacity", + "8", + "--metadata-upload-timeout-secs", + "11", + "--metadata-upload-io-timeout-secs", + "12", + "--metadata-upload-retry-max-times", + "4", + "--metadata-upload-retry-min-delay-ms", + "20", + "--metadata-upload-retry-max-delay-ms", + "200", + "--metadata-upload-retry-factor", + "1.5", + "--metadata-upload-retry-jitter=false", + ]) + .unwrap(); + let policy = args.metadata_upload; + assert_eq!(policy.operator_cache_capacity.get(), 8); + assert_eq!(policy.timeout_secs.get(), 11); + assert_eq!(policy.io_timeout_secs.get(), 12); + assert_eq!(policy.retry_max_times, 4); + assert_eq!(policy.retry_min_delay_ms, 20); + assert_eq!(policy.retry_max_delay_ms, 200); + assert_eq!(policy.retry_factor, 1.5); + assert!(!policy.retry_jitter); + + let command = Args::command(); + for (id, env) in [ + ( + "operator_cache_capacity", + "DYN_SGLANG_METADATA_UPLOAD_OPERATOR_CACHE_CAPACITY", + ), + ("timeout_secs", "DYN_SGLANG_METADATA_UPLOAD_TIMEOUT_SECS"), + ( + "io_timeout_secs", + "DYN_SGLANG_METADATA_UPLOAD_IO_TIMEOUT_SECS", + ), + ( + "retry_max_times", + "DYN_SGLANG_METADATA_UPLOAD_RETRY_MAX_TIMES", + ), + ( + "retry_min_delay_ms", + "DYN_SGLANG_METADATA_UPLOAD_RETRY_MIN_DELAY_MS", + ), + ( + "retry_max_delay_ms", + "DYN_SGLANG_METADATA_UPLOAD_RETRY_MAX_DELAY_MS", + ), + ("retry_factor", "DYN_SGLANG_METADATA_UPLOAD_RETRY_FACTOR"), + ("retry_jitter", "DYN_SGLANG_METADATA_UPLOAD_RETRY_JITTER"), + ] { + let arg = command + .get_arguments() + .find(|arg| arg.get_id() == id) + .unwrap(); + assert_eq!(arg.get_env().unwrap(), env); + } + } + + #[test] + fn zero_retries_is_valid() { + let args = Args::try_parse_from([ + "dynamo-sglang-sidecar", + "--grpc-endpoint", + "127.0.0.1:30000", + "--metadata-upload-retry-max-times", + "0", + ]) + .unwrap(); + + assert_eq!(args.metadata_upload.retry_max_times, 0); + } +} diff --git a/lib/sidecar/sglang/src/metadata_upload/resilience_tests.rs b/lib/sidecar/sglang/src/metadata_upload/resilience_tests.rs new file mode 100644 index 000000000000..c38516a0c7b3 --- /dev/null +++ b/lib/sidecar/sglang/src/metadata_upload/resilience_tests.rs @@ -0,0 +1,306 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use std::num::NonZeroUsize; +use std::sync::Arc; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::Duration; + +use dynamo_backend_common::{OutputOptions, PreprocessedRequest, SamplingOptions, StopConditions}; +use opendal::raw::*; +use opendal::{ + Buffer, Builder, Capability, Error, ErrorKind, Metadata, MetadataBuilder, OperationContext, + Operator, Result, +}; +use serde_json::json; + +use super::config::OperatorPolicy; +use super::{MetadataUploadService, OperatorCache, configure_operator}; + +fn policy(capacity: usize) -> OperatorPolicy { + OperatorPolicy { + capacity: NonZeroUsize::new(capacity).unwrap(), + timeout: Duration::from_secs(10), + io_timeout: Duration::from_secs(10), + retry_max_times: 3, + retry_min_delay: Duration::from_millis(1), + retry_max_delay: Duration::from_millis(10), + retry_factor: 2.0, + retry_jitter: false, + } +} + +fn operator_cache(capacity: usize) -> OperatorCache { + OperatorCache::new(policy(capacity)) +} + +fn request() -> PreprocessedRequest { + PreprocessedRequest::builder() + .model("model".to_string()) + .token_ids(vec![1]) + .sampling_options(SamplingOptions::default()) + .output_options(OutputOptions::default()) + .stop_conditions(StopConditions::default()) + .extra_args(Some(json!({ + "nvext": {"metadata_upload": { + "url": "test://primary", + "fallback_url": "test://fallback" + }} + }))) + .build() + .unwrap() +} + +#[tokio::test] +async fn concurrent_cache_misses_build_one_operator() { + let directory = tempfile::tempdir().unwrap(); + let url = format!("fs://{}", directory.path().display()); + let cache = Arc::new(operator_cache(2)); + let barrier = Arc::new(tokio::sync::Barrier::new(33)); + let mut tasks = Vec::new(); + for _ in 0..32 { + let cache = cache.clone(); + let url = url.clone(); + let barrier = barrier.clone(); + tasks.push(tokio::spawn(async move { + barrier.wait().await; + cache.get(url, "url").await + })); + } + barrier.wait().await; + for task in tasks { + task.await.unwrap().unwrap(); + } + assert_eq!(cache.build_count.load(Ordering::Relaxed), 1); +} + +#[tokio::test] +async fn concurrent_cache_misses_share_construction_failure() { + let cache = Arc::new(operator_cache(2)); + let barrier = Arc::new(tokio::sync::Barrier::new(33)); + let mut tasks = Vec::new(); + for _ in 0..32 { + let cache = cache.clone(); + let barrier = barrier.clone(); + tasks.push(tokio::spawn(async move { + barrier.wait().await; + cache.get("unknown://destination".to_string(), "url").await + })); + } + barrier.wait().await; + for task in tasks { + assert!(task.await.unwrap().is_err()); + } + assert_eq!(cache.build_count.load(Ordering::Relaxed), 1); +} + +#[tokio::test] +async fn retries_temporary_errors_until_success() { + let (operator, attempts) = test_operator(WriteBehavior::FailUntil(2), policy(2)); + + operator + .write("metadata", Buffer::from("data")) + .await + .unwrap(); + + assert_eq!(attempts.load(Ordering::Relaxed), 3); +} + +#[tokio::test] +async fn stops_after_configured_retry_limit() { + let mut retry_policy = policy(2); + retry_policy.retry_max_times = 2; + let (operator, attempts) = test_operator(WriteBehavior::FailUntil(usize::MAX), retry_policy); + + assert!( + operator + .write("metadata", Buffer::from("data")) + .await + .is_err() + ); + assert_eq!(attempts.load(Ordering::Relaxed), 3); +} + +#[tokio::test] +async fn timeout_exhausts_retries_before_fallback() { + let cache = operator_cache(2); + let mut timeout_policy = policy(2); + timeout_policy.io_timeout = Duration::from_millis(5); + timeout_policy.retry_max_times = 1; + let (primary, primary_attempts) = test_operator(WriteBehavior::Pending, timeout_policy); + let (fallback, fallback_attempts) = test_operator(WriteBehavior::Success, policy(2)); + cache.insert("test://primary", primary).await; + cache.insert("test://fallback", fallback).await; + let service = MetadataUploadService { operators: cache }; + let delivery = service.delivery_for(&request()).await.unwrap(); + + delivery + .finish( + &std::collections::HashMap::from([("id".into(), "timeout".into())]), + false, + ) + .await + .unwrap(); + + assert_eq!(primary_attempts.load(Ordering::Relaxed), 2); + assert_eq!(fallback_attempts.load(Ordering::Relaxed), 1); +} + +#[derive(Clone, Copy, Debug)] +enum WriteBehavior { + Success, + FailUntil(usize), + Pending, +} + +#[derive(Clone, Debug)] +struct TestBuilder { + behavior: WriteBehavior, + attempts: Arc, +} + +impl Default for TestBuilder { + fn default() -> Self { + Self { + behavior: WriteBehavior::Success, + attempts: Arc::new(AtomicUsize::new(0)), + } + } +} + +impl Builder for TestBuilder { + type Config = (); + + fn build(self) -> Result { + Ok(TestService { + behavior: self.behavior, + attempts: self.attempts, + }) + } +} + +#[derive(Clone, Debug)] +struct TestService { + behavior: WriteBehavior, + attempts: Arc, +} + +impl Service for TestService { + type Reader = (); + type Writer = TestWriter; + type Lister = (); + type Deleter = (); + type Copier = (); + type Composer = (); + + fn info(&self) -> ServiceInfo { + ServiceInfo::with_scheme("metadata-test") + } + + fn capability(&self) -> Capability { + Capability { + write: true, + write_can_multi: true, + ..Default::default() + } + } + + async fn create_dir( + &self, + _: &OperationContext, + _: &str, + _: OpCreateDir, + ) -> Result { + unsupported() + } + + async fn stat(&self, _: &OperationContext, _: &str, _: OpStat) -> Result { + unsupported() + } + + fn read(&self, _: &OperationContext, _: &str, _: OpRead) -> Result { + unsupported() + } + + fn write(&self, _: &OperationContext, _: &str, _: OpWrite) -> Result { + Ok(TestWriter { + behavior: self.behavior, + attempts: self.attempts.clone(), + size: 0, + }) + } + + fn delete(&self, _: &OperationContext) -> Result { + unsupported() + } + + fn list(&self, _: &OperationContext, _: &str, _: OpList) -> Result { + unsupported() + } + + fn copy(&self, _: &OperationContext, _: &str, _: &str, _: OpCopy) -> Result { + unsupported() + } + + async fn rename( + &self, + _: &OperationContext, + _: &str, + _: &str, + _: OpRename, + ) -> Result { + unsupported() + } + + async fn presign(&self, _: &OperationContext, _: &str, _: OpPresign) -> Result { + unsupported() + } +} + +#[derive(Debug)] +struct TestWriter { + behavior: WriteBehavior, + attempts: Arc, + size: usize, +} + +impl oio::Write for TestWriter { + async fn write(&mut self, data: Buffer) -> Result<()> { + self.size += data.len(); + Ok(()) + } + + async fn close(&mut self) -> Result { + let attempt = self.attempts.fetch_add(1, Ordering::Relaxed) + 1; + match self.behavior { + WriteBehavior::Success => Ok(file_metadata(self.size)), + WriteBehavior::FailUntil(limit) if attempt <= limit => { + Err(Error::new(ErrorKind::Unexpected, "injected temporary failure").set_temporary()) + } + WriteBehavior::FailUntil(_) => Ok(file_metadata(self.size)), + WriteBehavior::Pending => std::future::pending().await, + } + } + + async fn abort(&mut self) -> Result<()> { + Ok(()) + } +} + +fn test_operator(behavior: WriteBehavior, policy: OperatorPolicy) -> (Operator, Arc) { + let attempts = Arc::new(AtomicUsize::new(0)); + let operator = Operator::new(TestBuilder { + behavior, + attempts: attempts.clone(), + }) + .unwrap(); + (configure_operator(operator, policy), attempts) +} + +fn unsupported() -> Result { + Err(Error::new(ErrorKind::Unsupported, "not used by this test")) +} + +fn file_metadata(size: usize) -> Metadata { + MetadataBuilder::file(size as u64).build() +}