diff --git a/Cargo.lock b/Cargo.lock index 91070a54..ad3c847c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -17,6 +17,23 @@ version = "1.0.95" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "34ac096ce696dc2fcabef30516bb13c0a68a11d30131d3df6f04711467681b04" +[[package]] +name = "async-trait" +version = "0.1.92" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82f6aeea286b8eb4dd3431a1be1b59d290ace00f5bfd8e2a159bc2a05e2c1667" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "atomic-waker" +version = "1.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1505bd5d3d116872e7271a6d4e16d81d0c8570876c8de68093a09ac269d8aac0" + [[package]] name = "autocfg" version = "1.4.0" @@ -41,6 +58,12 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bitflags" +version = "2.13.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ded4057c258ba199e2d26386d3af3780957ecaee6c4ef4041c6b4b8b97c0b06" + [[package]] name = "block-buffer" version = "0.10.4" @@ -90,12 +113,43 @@ dependencies = [ "either", ] +[[package]] +name = "cc" +version = "1.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "375989decec1a0fa424442bed520fa12ede43a48475ce8c7417d2c69b32b5f6b" +dependencies = [ + "find-msvc-tools", + "shlex", +] + [[package]] name = "cfg-if" version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core 0.10.1", +] + +[[package]] +name = "combine" +version = "4.6.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4356eba6976435498d05a311ee9fc486da3bc81c200d3f0d7dd9703b41913f92" +dependencies = [ + "bytes", + "memchr", +] + [[package]] name = "const-oid" version = "0.9.6" @@ -108,6 +162,32 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" +[[package]] +name = "core-foundation" +version = "0.9.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e195e091a93c46f7102ec7818a2aa394e1e1771c3ab4825963fa03e45afb8f" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2a6cd9ae233e7f62ba4e9353e81a88df7fc8a5987b8d445b4d90c879bd156f6" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "core-foundation-sys" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b" + [[package]] name = "cpufeatures" version = "0.2.17" @@ -126,6 +206,36 @@ dependencies = [ "libc", ] +[[package]] +name = "critical-section" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "790eea4361631c5e7d22598ecd5723ff611904e3344ce8720784c93e3d83d40b" + +[[package]] +name = "crossbeam-channel" +version = "0.5.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "98b0cc327b5bc766e7fda9c9260cc0fa81b43a8e240440422dff70788e3f9ef1" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-epoch" +version = "0.9.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dc74980687109a3b14c72fd458107bf0baa1da1a1a805e178d15501ba9b86d9d" +dependencies = [ + "crossbeam-utils", +] + +[[package]] +name = "crossbeam-utils" +version = "0.8.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6" + [[package]] name = "crypto-bigint" version = "0.5.5" @@ -133,7 +243,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76" dependencies = [ "generic-array", - "rand_core", + "rand_core 0.6.4", "subtle", "zeroize", ] @@ -184,6 +294,12 @@ dependencies = [ "syn 2.0.96", ] +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + [[package]] name = "der" version = "0.7.10" @@ -218,6 +334,17 @@ dependencies = [ "crypto-common 0.2.1", ] +[[package]] +name = "displaydoc" +version = "0.2.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6232dd377dcc64799954cbd3a9bb882e9cdc1308ccd87b1c098f1fb2eaf82a8" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + [[package]] name = "ecdsa" version = "0.16.9" @@ -277,19 +404,25 @@ dependencies = [ "hkdf", "pem-rfc7468", "pkcs8", - "rand_core", + "rand_core 0.6.4", "sec1", "subtle", "zeroize", ] +[[package]] +name = "equivalent" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826" + [[package]] name = "ff" version = "0.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c0b50bfb653653f9ca9095b427bed08ab8d75a137839d9ad64eb11810d5b6393" dependencies = [ - "rand_core", + "rand_core 0.6.4", "subtle", ] @@ -299,6 +432,84 @@ version = "0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" +[[package]] +name = "find-msvc-tools" +version = "0.1.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1c66915cc5391097de7a2b0b972e8505aa33c0df401fd3a6811dbdb32881ba48" + +[[package]] +name = "fnv" +version = "1.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f9eec918d3f24069decb9af1554cad7c880e2da24a9afd88aca000531ab82c1" + +[[package]] +name = "form_urlencoded" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb4cb245038516f5f85277875cdaa4f7d2c9a0fa0468de06ed190163b1581fcf" +dependencies = [ + "percent-encoding", +] + +[[package]] +name = "futures-channel" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b1f9e3d69d39e4862ffed03ed071a76f9a13ba1d9109d355b0f0aa6b15e393c4" +dependencies = [ + "futures-core", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-io" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "53c0fa8157de1303bfffdaa1cc2a673bfffb60102f76b0ef4441659124373fed" + +[[package]] +name = "futures-macro" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fb9654ba8355388abeb8dcb4fc62f511300867002afc858860463bdd9fe0c44" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-macro", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "generic-array" version = "0.14.9" @@ -321,6 +532,18 @@ dependencies = [ "wasi", ] +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "rand_core 0.10.1", +] + [[package]] name = "group" version = "0.13.0" @@ -328,16 +551,111 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0f9ef7462f7c099f518d754361858f86d8a07af53ba9af0fe635bbccb151a63" dependencies = [ "ff", - "rand_core", + "rand_core 0.6.4", "subtle", ] +[[package]] +name = "h2" +version = "0.4.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7d29020232d6aa3fb1daca64c1127cf662cf97f254ae16c18c05b8ab635fc118" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + +[[package]] +name = "hashbrown" +version = "0.17.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" + [[package]] name = "heck" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" +[[package]] +name = "hickory-net" +version = "0.26.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c480823ed7c2c5d0f09c41020cb6b7c28029ce60ec42dc942158dcf22f8e0a4d" +dependencies = [ + "async-trait", + "cfg-if", + "data-encoding", + "futures-channel", + "futures-io", + "futures-util", + "hickory-proto", + "idna", + "ipnet", + "jni", + "rand 0.10.3", + "thiserror", + "tinyvec", + "tokio", + "tracing", + "url", +] + +[[package]] +name = "hickory-proto" +version = "0.26.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "12b92608f679a6fa515dd1d15c1ff89443026e391200a2c840c7afcba482893d" +dependencies = [ + "data-encoding", + "idna", + "ipnet", + "jni", + "once_cell", + "prefix-trie", + "rand 0.10.3", + "ring", + "thiserror", + "tinyvec", + "tracing", + "url", +] + +[[package]] +name = "hickory-resolver" +version = "0.26.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f3da5255c95d5a716857d54b5b8f4e8d67c3484d3beaaaae2ce25063b3ba981" +dependencies = [ + "cfg-if", + "futures-util", + "hickory-net", + "hickory-proto", + "ipconfig", + "ipnet", + "jni", + "moka", + "ndk-context", + "once_cell", + "parking_lot", + "rand 0.10.3", + "resolv-conf", + "smallvec", + "system-configuration", + "thiserror", + "tokio", + "tracing", +] + [[package]] name = "hkdf" version = "0.12.4" @@ -356,6 +674,56 @@ dependencies = [ "digest 0.10.7", ] +[[package]] +name = "hostname" +version = "0.4.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "617aaa3557aef3810a6369d0a99fac8a080891b68bd9f9812a1eeda0c0730cbd" +dependencies = [ + "cfg-if", + "libc", + "windows-link", +] + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ca2a8f2913ee65f60facd6a5905613afaa448497a0230cc41ce022d93290bc2c" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "23169fe34a5fbcdd3f3862e78fb9b6fccd5f02a6dc6f732547005d45631ce71c" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + [[package]] name = "hybrid-array" version = "0.4.8" @@ -365,6 +733,175 @@ dependencies = [ "typenum", ] +[[package]] +name = "hyper" +version = "1.12.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2c3e324da4c95177d6291d4c8730197c0d1822f8a9766814a4a44fa5ab797c9c" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "h2", + "http", + "http-body", + "httpdate", + "pin-project-lite", + "smallvec", + "tokio", +] + +[[package]] +name = "hyper-util" +version = "0.1.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ddc03d96684f9226b8a787cdb71488417b53ab5ea8fdb1dac946cb9431cc8bff" +dependencies = [ + "bytes", + "http", + "http-body", + "hyper", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "icu_collections" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe090a42a0d3140fed8c63263ed1fecc02422cb2f4bfe55cd8de2de1b8e105dd" +dependencies = [ + "displaydoc", + "potential_utf", + "utf8_iter", + "yoke", + "zerofrom", + "zerovec", +] + +[[package]] +name = "icu_locale_core" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "25d29b57c17c6528d0dc4c2141cfa2d08200465acb537a3909c8b9b8a4a984c0" +dependencies = [ + "displaydoc", + "litemap", + "tinystr", + "writeable", + "zerovec", +] + +[[package]] +name = "icu_normalizer" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "404a3b8c486f205c87fe7f23ab6894d84a4ad76d6e4671ae1981ba7def87c599" +dependencies = [ + "icu_collections", + "icu_normalizer_data", + "icu_properties", + "icu_provider", + "smallvec", + "zerovec", +] + +[[package]] +name = "icu_normalizer_data" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "20f0088bfb7441f7a37fa0c9e20cb1e9cba1cac6b9ce8cd2a6dc44294f8e8cbc" + +[[package]] +name = "icu_properties" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ea83d991c56158f7cf898bf86436c1dfe7c07ee82fe88951f2695c9f37a379f" +dependencies = [ + "displaydoc", + "icu_collections", + "icu_locale_core", + "icu_properties_data", + "icu_provider", + "zerotrie", + "zerovec", +] + +[[package]] +name = "icu_properties_data" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b98f29af82d02b163fe50161388350e743bb3107cc8e8c2f7314cb505faddfcf" + +[[package]] +name = "icu_provider" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "413099315636c588ee8f011190b329b971a09a8fae9d157e01dcae8379468514" +dependencies = [ + "displaydoc", + "icu_locale_core", + "writeable", + "yoke", + "zerofrom", + "zerotrie", + "zerovec", +] + +[[package]] +name = "idna" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3b0875f23caa03898994f6ddc501886a45c7d3d62d04d2d90788d47be1b1e4de" +dependencies = [ + "idna_adapter", + "smallvec", + "utf8_iter", +] + +[[package]] +name = "idna_adapter" +version = "1.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714" +dependencies = [ + "icu_normalizer", + "icu_properties", +] + +[[package]] +name = "indexmap" +version = "2.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855" +dependencies = [ + "equivalent", + "hashbrown", +] + +[[package]] +name = "ipconfig" +version = "0.3.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4d40460c0ce33d6ce4b0630ad68ff63d6661961c48b6dba35e5a4d81cfb48222" +dependencies = [ + "socket2", + "widestring", + "windows-registry", + "windows-result", + "windows-sys 0.61.2", +] + +[[package]] +name = "ipnet" +version = "2.12.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "791930b43c0d5973160d90a8f3894509f2b273430f5c5c73b668636d0287c5c0" +dependencies = [ + "serde", +] + [[package]] name = "itertools" version = "0.13.0" @@ -380,6 +917,55 @@ version = "1.0.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d75a2a4b1b190afb6f5425f10f6a8f959d2ea0b9c2b1d79553551850539e4674" +[[package]] +name = "jni" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5efd9a482cf3a427f00d6b35f14332adc7902ce91efb778580e180ff90fa3498" +dependencies = [ + "cfg-if", + "combine", + "jni-macros", + "jni-sys", + "log", + "simd_cesu8", + "thiserror", + "walkdir", + "windows-link", +] + +[[package]] +name = "jni-macros" +version = "0.22.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a00109accc170f0bdb141fed3e393c565b6f5e072365c3bd58f5b062591560a3" +dependencies = [ + "proc-macro2", + "quote", + "rustc_version", + "simd_cesu8", + "syn 2.0.96", +] + +[[package]] +name = "jni-sys" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c6377a88cb3910bee9b0fa88d4f42e1d2da8e79915598f65fb0c7ee14c878af2" +dependencies = [ + "jni-sys-macros", +] + +[[package]] +name = "jni-sys-macros" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "38c0b942f458fe50cdac086d2f946512305e5631e720728f2a61aabcd47a6264" +dependencies = [ + "quote", + "syn 2.0.96", +] + [[package]] name = "js-sys" version = "0.3.77" @@ -398,12 +984,12 @@ checksum = "0529410abe238729a60b108898784df8984c87f6054c9c4fcacc47e4803c1ce1" dependencies = [ "base64", "ed25519-dalek", - "getrandom", + "getrandom 0.2.15", "hmac", "js-sys", "p256", "p384", - "rand", + "rand 0.8.6", "rsa", "serde", "serde_json", @@ -422,9 +1008,9 @@ dependencies = [ [[package]] name = "libc" -version = "0.2.169" +version = "0.2.191" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5aba8db14291edd000dfcc4d620c7ebfb122c613afb886ca8803fa4e128a20a" +checksum = "65cd87c9a6d60536580b316ad985726104730a5151c99845f2490428ef91240a" [[package]] name = "libm" @@ -432,6 +1018,21 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" +[[package]] +name = "litemap" +version = "0.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae" + +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.25" @@ -453,6 +1054,40 @@ version = "2.7.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "78ca9ab1a0babb1e7d5695e3530886289c18cf2f87ec19a575a0abdce112e3a3" +[[package]] +name = "mio" +version = "1.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" +dependencies = [ + "libc", + "wasi", + "windows-sys 0.61.2", +] + +[[package]] +name = "moka" +version = "0.12.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4293f18e7567a1caf3c584855554377025c65e0aa445344d04171f5ad63d19b9" +dependencies = [ + "crossbeam-channel", + "crossbeam-epoch", + "crossbeam-utils", + "equivalent", + "parking_lot", + "portable-atomic", + "smallvec", + "tagptr", + "uuid", +] + +[[package]] +name = "ndk-context" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27b02d87554356db9e9a873add8782d4ea6e3e58ea071a9adb9a2e8ddb884a8b" + [[package]] name = "nu-ansi-term" version = "0.46.0" @@ -474,7 +1109,7 @@ dependencies = [ "num-integer", "num-iter", "num-traits", - "rand", + "rand 0.8.6", "smallvec", "zeroize", ] @@ -514,6 +1149,16 @@ name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +dependencies = [ + "critical-section", + "portable-atomic", +] + +[[package]] +name = "openssl-probe" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e036b411754132fd6b74e94b74e9a13c5cf606fa93d1885c35860d3060820b1" [[package]] name = "overload" @@ -545,6 +1190,29 @@ dependencies = [ "sha2 0.10.9", ] +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "pastey" version = "0.2.1" @@ -560,6 +1228,12 @@ dependencies = [ "base64ct", ] +[[package]] +name = "percent-encoding" +version = "2.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220" + [[package]] name = "pin-project-lite" version = "0.2.16" @@ -593,6 +1267,15 @@ version = "1.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "280dc24453071f1b63954171985a0b0d30058d287960968b9b2aca264c8d4ee6" +[[package]] +name = "potential_utf" +version = "0.1.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661" +dependencies = [ + "zerovec", +] + [[package]] name = "ppv-lite86" version = "0.2.21" @@ -602,6 +1285,17 @@ dependencies = [ "zerocopy", ] +[[package]] +name = "prefix-trie" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4cf6e3177f0684016a5c209b00882e15f8bdd3f3bb48f0491df10cd102d0c6e7" +dependencies = [ + "either", + "ipnet", + "num-traits", +] + [[package]] name = "primeorder" version = "0.13.6" @@ -657,6 +1351,20 @@ dependencies = [ "pyo3-macros", ] +[[package]] +name = "pyo3-async-runtimes" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b877331c7817ccb6379e9d19a06993160141f1d0a386e55276128717df85e3fc" +dependencies = [ + "futures-channel", + "futures-util", + "once_cell", + "pin-project-lite", + "pyo3", + "tokio", +] + [[package]] name = "pyo3-build-config" version = "0.30.1" @@ -702,13 +1410,19 @@ dependencies = [ [[package]] name = "quote" -version = "1.0.38" +version = "1.0.47" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0e4dccaaaf89514f546c693ddc140f729f958c247918a13380cccc6078391acc" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + [[package]] name = "rand" version = "0.8.6" @@ -717,7 +1431,18 @@ checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" dependencies = [ "libc", "rand_chacha", - "rand_core", + "rand_core 0.6.4", +] + +[[package]] +name = "rand" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", ] [[package]] @@ -727,7 +1452,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" dependencies = [ "ppv-lite86", - "rand_core", + "rand_core 0.6.4", ] [[package]] @@ -736,7 +1461,22 @@ version = "0.6.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ec0be4795e2f6a28069bec0b5ff3e2ac9bafc99e6a9a7dc3547996c5c816922c" dependencies = [ - "getrandom", + "getrandom 0.2.15", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", ] [[package]] @@ -783,33 +1523,65 @@ version = "0.8.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c" +[[package]] +name = "resolv-conf" +version = "0.7.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7" + [[package]] name = "restate-sdk-python-core" version = "1.2.0" dependencies = [ + "bytes", + "http", + "http-body", + "http-body-util", + "hyper", + "percent-encoding", "pyo3", + "pyo3-async-runtimes", "restate-sdk-shared-core", + "tokio", + "tokio-stream", + "tokio-util", + "tracing", "tracing-subscriber", ] [[package]] name = "restate-sdk-shared-core" -version = "8.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bfd8d132e369af9310530b57c3e352a81cfad54e79d8d38fc5d7de9026ad2dfa" +version = "8.3.0" +source = "git+https://github.com/restatedev/sdk-shared-core?branch=cloud-exp-revamp#7fbd30d6b219a3540de7f9cfa68f9c643904dca0" dependencies = [ "base64", "bs58", "bytes", "bytes-utils", + "futures-util", + "hickory-resolver", + "hostname", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-util", "jsonwebtoken", "pastey", "prost", + "rand 0.8.6", + "rustls", + "rustls-native-certs", "serde", "sha2 0.11.0", + "socket2", "strum", "thiserror", + "tokio", + "tokio-rustls", + "tokio-util", "tracing", + "uuid", ] [[package]] @@ -822,6 +1594,20 @@ dependencies = [ "subtle", ] +[[package]] +name = "ring" +version = "0.17.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a4689e6c2294d81e88dc6261c768b63bc4fcdb852be6d1352498b114f61383b7" +dependencies = [ + "cc", + "cfg-if", + "getrandom 0.2.15", + "libc", + "untrusted", + "windows-sys 0.52.0", +] + [[package]] name = "rsa" version = "0.9.10" @@ -835,7 +1621,7 @@ dependencies = [ "num-traits", "pkcs1", "pkcs8", - "rand_core", + "rand_core 0.6.4", "signature", "spki", "subtle", @@ -851,17 +1637,87 @@ dependencies = [ "semver", ] +[[package]] +name = "rustls" +version = "0.23.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "48e13bd8c0e9365c43cfa5c9e8f9ad49d3c8444926c9aac819e0e4dc503c8fdf" +dependencies = [ + "once_cell", + "ring", + "rustls-pki-types", + "rustls-webpki", + "subtle", + "zeroize", +] + +[[package]] +name = "rustls-native-certs" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a073f5dc7469f984c52ad2752b63b0807745133b6de880b7b64c1ac4c48aec4" +dependencies = [ + "openssl-probe", + "rustls-pki-types", + "schannel", + "security-framework", +] + +[[package]] +name = "rustls-pki-types" +version = "1.15.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2f4925028c7eb5d1fcdaf196971378ed9d2c1c4efc7dc5d011256f76c99c0a96" +dependencies = [ + "zeroize", +] + +[[package]] +name = "rustls-webpki" +version = "0.103.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4443b1b6d39d7c663139942cc4558262562d2c84123bb6282ef177009ca8b281" +dependencies = [ + "ring", + "rustls-pki-types", + "untrusted", +] + [[package]] name = "rustversion" version = "1.0.19" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f7c45b9784283f1b2e7fb61b42047c2fd678ef0960d4f6f1eba131594cc369d4" +checksum = "f7c45b9784283f1b2e7fb61b42047c2fd678ef0960d4f6f1eba131594cc369d4" + +[[package]] +name = "ryu" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3cb5ba0dc43242ce17de99c180e96db90b235b8a9fdc9543c96d2209116bd9f" + +[[package]] +name = "same-file" +version = "1.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93fc1dc3aaa9bfed95e02e6eadabb4baf7e3078b0bd1b4d7b6b0b68378900502" +dependencies = [ + "winapi-util", +] + +[[package]] +name = "schannel" +version = "0.1.30" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9cb23c424d8ee841bb4d813677ca237e2a610067d594fec130f87d6093495ab5" +dependencies = [ + "windows-sys 0.61.2", +] [[package]] -name = "ryu" -version = "1.0.18" +name = "scopeguard" +version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f3cb5ba0dc43242ce17de99c180e96db90b235b8a9fdc9543c96d2209116bd9f" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" [[package]] name = "sec1" @@ -877,6 +1733,29 @@ dependencies = [ "zeroize", ] +[[package]] +name = "security-framework" +version = "3.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d" +dependencies = [ + "bitflags", + "core-foundation 0.10.1", + "core-foundation-sys", + "libc", + "security-framework-sys", +] + +[[package]] +name = "security-framework-sys" +version = "2.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2691df843ecc5d231c0b14ece2acc3efb62c0a398c7e1d875f3983ce020e3" +dependencies = [ + "core-foundation-sys", + "libc", +] + [[package]] name = "semver" version = "1.0.28" @@ -956,6 +1835,12 @@ dependencies = [ "lazy_static", ] +[[package]] +name = "shlex" +version = "2.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" + [[package]] name = "signature" version = "2.2.0" @@ -963,15 +1848,47 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ "digest 0.10.7", - "rand_core", + "rand_core 0.6.4", +] + +[[package]] +name = "simd_cesu8" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11031e251abf8611c80f460e19dbdeb54a66db918e49c65a7065b46ac7aec520" +dependencies = [ + "rustc_version", + "simdutf8", ] +[[package]] +name = "simdutf8" +version = "0.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" + +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + [[package]] name = "smallvec" version = "1.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67" +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys 0.61.2", +] + [[package]] name = "spin" version = "0.9.8" @@ -988,6 +1905,12 @@ dependencies = [ "der", ] +[[package]] +name = "stable_deref_trait" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596" + [[package]] name = "strum" version = "0.27.1" @@ -1038,6 +1961,44 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "synstructure" +version = "0.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "901704edd0dfe137f1987838ee4f259e4e063c31371bdb423f7ae38ec6f77f02" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "system-configuration" +version = "0.7.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a13f3d0daba03132c0aa9767f98351b3488edc2c100cda2d2ec2b04f3d8d3c8b" +dependencies = [ + "bitflags", + "core-foundation 0.9.4", + "system-configuration-sys", +] + +[[package]] +name = "system-configuration-sys" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e1d1b10ced5ca923a1fcb8d03e96b8d3268065d724548c0211415ff6ac6bac4" +dependencies = [ + "core-foundation-sys", + "libc", +] + +[[package]] +name = "tagptr" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b2093cf4c8eb1e67749a6762251bc9cd836b6fc171623bd0a9d324d37af2417" + [[package]] name = "target-lexicon" version = "0.13.5" @@ -1074,6 +2035,16 @@ dependencies = [ "once_cell", ] +[[package]] +name = "tinystr" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "abc840c238154d58f9e86b2792261edfde565a17696e1cda1a449de27979801a" +dependencies = [ + "displaydoc", + "zerovec", +] + [[package]] name = "tinyvec" version = "1.8.1" @@ -1089,6 +2060,68 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1f3ccbac311fea05f86f61904b462b55fb3df8837a366dfc601a0161d0532f20" +[[package]] +name = "tokio" +version = "1.54.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b52fd3255258d26fac7264ab5a76b53c13367136dc011d3d6ad4f8ed070396a3" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys 0.61.2", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] + +[[package]] +name = "tokio-rustls" +version = "0.26.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c9cc2678c2cdd569ef8215e2afd7954ada2ae20b4fdd2c5fe6139a3b02d105db" +dependencies = [ + "rustls", + "tokio", +] + +[[package]] +name = "tokio-stream" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tokio-util" +version = "0.7.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e464cf451ba96ebfc6f9b6542f17ee8b8956e33f1e40d9690624e59d7a7f8a4b" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "futures-util", + "libc", + "pin-project-lite", + "tokio", +] + [[package]] name = "tracing" version = "0.1.41" @@ -1162,6 +2195,50 @@ version = "1.0.14" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "adb9e6ca4f869e1180728b7950e35922a7fc6397f7b641499e8f3ef06e50dc83" +[[package]] +name = "untrusted" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ecb6da28b8a351d773b68d5825ac39017e680750f980f3a1a85cd8dd28a47c1" + +[[package]] +name = "url" +version = "2.5.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ff67a8a4397373c3ef660812acab3268222035010ab8680ec4215f38ba3d0eed" +dependencies = [ + "form_urlencoded", + "idna", + "percent-encoding", + "serde", +] + +[[package]] +name = "utf8_iter" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "af3bd9d3e571c26fcce9d033b87ea5ab061b8d2ffe9b7a5675d810d68f03edf0" +dependencies = [ + "utf_types", +] + +[[package]] +name = "utf_types" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3a3b425286980a367278453f3b6d28cdcb134b574e5a35824a6ff5eebad50bb6" + +[[package]] +name = "uuid" +version = "1.29.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a0e2d11620c50695fe12bbbfde96b3477fe606e7284c22d505c12ce51214514" +dependencies = [ + "getrandom 0.4.3", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "valuable" version = "0.1.0" @@ -1174,6 +2251,16 @@ version = "0.9.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" +[[package]] +name = "walkdir" +version = "2.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29790946404f91d9c5d06f9874efddea1dc06c5efe94541a7d6863108e3a5e4b" +dependencies = [ + "same-file", + "winapi-util", +] + [[package]] name = "wasi" version = "0.11.0+wasi-snapshot-preview1" @@ -1237,6 +2324,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "widestring" +version = "1.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "72069c3113ab32ab29e5584db3c6ec55d416895e60715417b5b883a357c3e471" + [[package]] name = "winapi" version = "0.3.9" @@ -1253,12 +2346,167 @@ version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" +[[package]] +name = "winapi-util" +version = "0.1.11" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" +dependencies = [ + "windows-sys 0.61.2", +] + [[package]] name = "winapi-x86_64-pc-windows-gnu" version = "0.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "712e227841d057c1ee1cd2fb22fa7e5a5461ae8e48fa2ca79ec42cfc1931183f" +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-registry" +version = "0.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "02752bf7fbdcce7f2a27a742f798510f3e5ad88dbe84871e5168e2120c3d5720" +dependencies = [ + "windows-link", + "windows-result", + "windows-strings", +] + +[[package]] +name = "windows-result" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7781fa89eaf60850ac3d2da7af8e5242a5ea78d1a11c49bf2910bb5a73853eb5" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-strings" +version = "0.5.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7837d08f69c77cf6b07689544538e017c1bfcf57e34b4c0ff58e6c2cd3b37091" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-sys" +version = "0.52.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" +dependencies = [ + "windows-targets", +] + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + +[[package]] +name = "windows-targets" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" +dependencies = [ + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", +] + +[[package]] +name = "windows_aarch64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" + +[[package]] +name = "windows_aarch64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" + +[[package]] +name = "windows_i686_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" + +[[package]] +name = "windows_i686_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" + +[[package]] +name = "windows_i686_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" + +[[package]] +name = "windows_x86_64_gnu" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" + +[[package]] +name = "windows_x86_64_gnullvm" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" + +[[package]] +name = "windows_x86_64_msvc" +version = "0.52.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "writeable" +version = "0.6.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc" + +[[package]] +name = "yoke" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "417ef3fcf6c300b7dbd25c85531a746931c02ad536f3cf4786ed1e88223340b3" +dependencies = [ + "stable_deref_trait", + "yoke-derive", + "zerofrom", +] + +[[package]] +name = "yoke-derive" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ec8ebde2db3681e8c9980cc27822030e68752690ddfa9473e739aeb4dbde6d71" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", + "synstructure", +] + [[package]] name = "zerocopy" version = "0.8.27" @@ -1279,8 +2527,62 @@ dependencies = [ "syn 2.0.96", ] +[[package]] +name = "zerofrom" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0ec05a11813ea801ff6d75110ad09cd0824ddba17dfe17128ea0d5f68e6c5272" +dependencies = [ + "zerofrom-derive", +] + +[[package]] +name = "zerofrom-derive" +version = "0.1.8" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f75b4683f6c7f45248d4d64056a24298c6281e0993356d7d1b4a1a962ef10d4a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", + "synstructure", +] + [[package]] name = "zeroize" version = "1.9.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e13c156562582aa81c60cb29407084cdb54c4164760106ab78e6c5b0858cf64e" + +[[package]] +name = "zerotrie" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f" +dependencies = [ + "displaydoc", + "yoke", + "zerofrom", +] + +[[package]] +name = "zerovec" +version = "0.11.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f58a2799cfe9a1670c7ea8edbd52ef8edb93cb37a7f83bf31ba9e35d9dae6a40" +dependencies = [ + "yoke", + "zerofrom", + "zerovec-derive", +] + +[[package]] +name = "zerovec-derive" +version = "0.11.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3743ddbf47738d929cd1ae2e199d40d0162979dfe31aef8ae646d2bf00bf97a" +dependencies = [ + "proc-macro2", + "quote", + "syn 3.0.9", +] diff --git a/Cargo.toml b/Cargo.toml index a227cce7..2f65d562 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,5 +13,22 @@ doc = false [dependencies] pyo3 = { version = "0.30.1", features = ["extension-module"] } +pyo3-async-runtimes = { version = "0.30", features = ["tokio-runtime"] } +tracing = "0.1" tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] } -restate-sdk-shared-core = { version = "8.2.0", features = ["request_identity", "sha2_random_seed", "rust_crypto"] } +restate-sdk-shared-core = { version = "8.2.0", features = ["request_identity", "sha2_random_seed", "rust_crypto", "tunnel"] } + +# ASGI bridge for the tunnel +bytes = "1.10" +http = "1.3" +http-body = "1.0" +http-body-util = "0.1" +hyper = { version = "1.8", default-features = false } +percent-encoding = "2.3" +tokio = { version = "1.49", default-features = false, features = ["sync", "rt-multi-thread"] } +tokio-stream = { version = "0.1", default-features = false } +tokio-util = { version = "0.7", default-features = false } + +# Until restate-sdk-shared-core with the `cloud` and `tunnel` features is released. +[patch.crates-io] +restate-sdk-shared-core = { git = "https://github.com/restatedev/sdk-shared-core", branch = "cloud-exp-revamp" } diff --git a/README.md b/README.md index e2b6e256..73dd2cb3 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,34 @@ To use this SDK, add the dependency to your project: pip install restate_sdk ``` +## Connecting to Restate Cloud + +An application can connect outbound to Restate Cloud instead of exposing a port. Copy a connect key +from the Cloud UI (Development → Configure SDK), pick a name for the deployment, and run: + +```shell +RESTATE_CONNECT_KEY=rstcloudv1_... python -m restate connect app:app --name greeter +``` + +where `app:app` is the module and attribute holding the application returned by `restate.app(...)`. +Then register the deployment as `connect://greeter`. The same from code, running until stopped: + +```python +if __name__ == "__main__": + asyncio.run(restate.connect(app, name="greeter")) +``` + +To do something while connected, use the context manager: + +```python +async with restate.with_connection(app, name="greeter") as connection: + print(connection.info.deployment_url) + await connection.wait() +``` + +Clients can read an ingress key the same way: `restate.create_client()` without an URL uses +`RESTATE_INGRESS_KEY` (Development → Configure Client). + ## Versions The compatibility with Restate is described in the following table. sdk-python versions < 0.10 are deprecated and no longer supported. diff --git a/python/restate/__init__.py b/python/restate/__init__.py index 3b27a446..0ca0d250 100644 --- a/python/restate/__init__.py +++ b/python/restate/__init__.py @@ -43,6 +43,7 @@ from .asyncio import as_completed, gather, wait_completed, select from .endpoint import app +from .connection import connect, with_connection, ConnectError from .logging import getLogger, RestateLoggingFilter @@ -83,7 +84,7 @@ def test_harness( @asynccontextmanager async def create_client( - ingress: str, + ingress: typing.Optional[str] = None, headers: typing.Optional[dict] = None, journal_value_codec: typing.Optional[JournalValueCodec] = None, ) -> typing.AsyncGenerator[RestateClient, None]: @@ -112,6 +113,9 @@ async def create_client( "JournalValueCodec", "JournalValueCodecProvider", "app", + "connect", + "with_connection", + "ConnectError", "create_test_harness", "test_harness", "HarnessEnvironment", diff --git a/python/restate/__main__.py b/python/restate/__main__.py new file mode 100644 index 00000000..15d2bd29 --- /dev/null +++ b/python/restate/__main__.py @@ -0,0 +1,93 @@ +# +# Copyright (c) 2023-2026 - Restate Software, Inc., Restate GmbH +# +# This file is part of the Restate SDK for Python, +# which is released under the MIT license. +# +# You can find a copy of the license in file LICENSE in the root +# directory of this repository or package, or at +# https://github.com/restatedev/sdk-typescript/blob/main/LICENSE +# +""" +``python -m restate``: command-line entry points. + + python -m restate connect app:app --name greeter + +serves the ASGI application ``app`` of module ``app`` through an outbound +connection to Restate Cloud, until SIGINT or SIGTERM. +""" + +import argparse +import asyncio +import importlib +import os +import sys +import typing + +from restate.connection import ConnectError, connect + +EXIT_OK = 0 +EXIT_FAILURE = 1 +EXIT_USAGE = 2 + + +def _load_app(spec: str) -> typing.Any: + """Import ``module[:attribute]`` (attribute defaults to ``app``), like hypercorn.""" + module_name, _, attribute = spec.partition(":") + if not module_name: + raise ValueError(f"invalid application {spec!r}: expected module:attribute") + attribute = attribute or "app" + try: + module = importlib.import_module(module_name) + except ImportError as error: + raise ValueError(f"cannot import module {module_name!r}: {error}") from error + try: + return getattr(module, attribute) + except AttributeError as error: + raise ValueError(f"module {module_name!r} has no attribute {attribute!r}") from error + + +def main(argv: typing.Optional[typing.Sequence[str]] = None) -> int: + """Entry point of ``python -m restate``.""" + parser = argparse.ArgumentParser(prog="python -m restate", description="Restate SDK command line.") + commands = parser.add_subparsers(dest="command", required=True) + connect_parser = commands.add_parser( + "connect", + help="serve an application through an outbound connection to Restate Cloud", + description=( + "Serve the ASGI application through an outbound connection to Restate Cloud, so it needs " + "no inbound port. The connect key comes from the Cloud UI (Development → Configure SDK)." + ), + ) + connect_parser.add_argument( + "app", + help="the application as module:attribute, for example app:app (attribute defaults to app)", + ) + connect_parser.add_argument( + "--name", + help="the deployment name, registered as connect:// (default: $RESTATE_CONNECT_NAME)", + ) + connect_parser.add_argument( + "--key", + help="the connect key (default: $RESTATE_CONNECT_KEY; prefer the variable over the flag)", + ) + args = parser.parse_args(argv) + + # Resolve the application from the working directory, like a server would. + sys.path.insert(0, os.getcwd()) + try: + app = _load_app(args.app) + except ValueError as error: + parser.error(str(error)) + return EXIT_USAGE # pragma: no cover - parser.error exits + + try: + asyncio.run(connect(app, name=args.name, key=args.key)) + except ConnectError as error: + print(f"error: {error}", file=sys.stderr, flush=True) + return EXIT_FAILURE + return EXIT_OK + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python/restate/client.py b/python/restate/client.py index 35ed2382..bedb61bf 100644 --- a/python/restate/client.py +++ b/python/restate/client.py @@ -481,9 +481,33 @@ async def workflow_send( return RestateClientSendHandle(send.get("invocationId", ""), 200) +def resolve_ingress( + ingress: typing.Optional[str], headers: typing.Optional[dict] +) -> typing.Tuple[str, typing.Optional[dict]]: + """ + The ingress URL and default headers: the given ones, or the URL and bearer token of + ``RESTATE_INGRESS_KEY`` (an ingress key from the Cloud UI, Development → Configure Client). + """ + if ingress: + return ingress, headers + # pylint: disable=import-error,no-name-in-module,import-outside-toplevel + from restate._internal import INGRESS_KEY_URL, ConnectError, IngressKey + + key = IngressKey.from_env() + if key is None: + raise ConnectError( + "create_client(): ingress is required — pass the ingress URL, or set RESTATE_INGRESS_KEY " + f"with an ingress key from {INGRESS_KEY_URL}" + ) + merged = dict(headers or {}) + if not any(name.lower() == "authorization" for name in merged): + merged["Authorization"] = key.bearer() + return key.ingress_url(), merged + + @asynccontextmanager async def create_client( - ingress: str, + ingress: typing.Optional[str] = None, headers: typing.Optional[dict] = None, journal_value_codec: typing.Optional[JournalValueCodec] = None, ) -> typing.AsyncGenerator[RestateClient, None]: @@ -491,11 +515,14 @@ async def create_client( Create a new Restate client. Args: - ingress: The base URL of the Restate ingress. + ingress: The base URL of the Restate ingress. When omitted, the URL and the bearer + token come from ``RESTATE_INGRESS_KEY``, an ingress key from the Cloud UI + (Development → Configure Client). headers: Optional default headers to send with every request. journal_value_codec: Optional journal value codec. When set, request bodies are encoded and call responses decoded through it. It must match the codec configured on the endpoint. NOTE: This is experimental and may change in future releases. """ + ingress, headers = resolve_ingress(ingress, headers) async with httpx.AsyncClient(base_url=ingress, headers=headers, http2=True) as http_client: yield Client(http_client, headers, journal_codec=journal_value_codec) diff --git a/python/restate/connection.py b/python/restate/connection.py new file mode 100644 index 00000000..9ae615e9 --- /dev/null +++ b/python/restate/connection.py @@ -0,0 +1,202 @@ +# +# Copyright (c) 2023-2026 - Restate Software, Inc., Restate GmbH +# +# This file is part of the Restate SDK for Python, +# which is released under the MIT license. +# +# You can find a copy of the license in file LICENSE in the root +# directory of this repository or package, or at +# https://github.com/restatedev/sdk-typescript/blob/main/LICENSE +# +""" +Connect a Restate application to Restate Cloud without exposing a port. + +The application dials Restate Cloud outbound; invocations are forwarded over +that connection into the regular ASGI application. + +:func:`restate.connect` is the one-liner (``asyncio.run(restate.connect(app))``): +it handles the signals and runs until the connection stops. +:func:`restate.with_connection` is the context manager for programs that want +to do something while connected. +""" + +import asyncio +import signal +import sys +import typing +from contextlib import asynccontextmanager + +from restate.server_types import RestateAppT + +# pylint: disable=import-error,no-name-in-module +from restate._internal import ConnectError, TunnelEngine, TunnelEngineOptions, TunnelInfo + +__all__ = ["connect", "with_connection", "Connection", "ConnectError", "TunnelInfo"] + +# What the first signal gives in-flight invocations before the second one (or the +# deadline) cuts the connections. +SIGNAL_SHUTDOWN_GRACE_SECONDS = 120.0 + +# How long leaving a ``with_connection()`` block waits for in-flight invocations. +EXIT_SHUTDOWN_GRACE_SECONDS = 10.0 + + +class Connection: + """ + The live connection :func:`with_connection` yields. Not meant to be constructed. + """ + + def __init__(self, engine: TunnelEngine, options: TunnelEngineOptions, info: TunnelInfo) -> None: + self._engine = engine + self._options = options + self._info = info + + @property + def name(self) -> str: + """The deployment name: register it with Restate as ``connect://``.""" + return self._options.name + + @property + def registration_identifier(self) -> str: + """``connect://``, what the CLI and the Cloud UI accept when registering.""" + return self._options.registration_identifier + + @property + def info(self) -> TunnelInfo: + """Metadata from the first successful handshake (``deployment_url``, ``proxy_url``, …).""" + return self._info + + async def wait(self) -> None: + """ + Wait until the connection stops serving: returns after :meth:`shutdown`, raises + :class:`ConnectError` when the connection failed for good (rejected credentials, + an exhausted reconnect policy, …). + """ + await self._engine.terminated() + + async def shutdown(self, grace: float = 0.0) -> None: + """ + Stop serving. New invocations are refused and Restate Cloud is told; in-flight + invocations get ``grace`` seconds to finish before the connections are cut. + ``0`` cuts them at once. Never raises: a connection that already failed reports + its error through :meth:`wait`, not here. + """ + try: + await self._engine.shutdown(grace) + except ConnectError: + pass + + +@asynccontextmanager +async def with_connection( + app: RestateAppT, + *, + name: typing.Optional[str] = None, + key: typing.Optional[str] = None, +) -> typing.AsyncIterator[Connection]: + """ + Serve ``app`` through an outbound connection to Restate Cloud while inside the block. + + Entering waits for the first successful handshake. Leaving drains in-flight + invocations for up to ten seconds, then cuts the connections; call + :meth:`Connection.shutdown` first to leave faster. No signal handling: that is the + caller's business here (see :func:`connect` for the batteries-included form). + + Args: + app: the ASGI application returned by :func:`restate.app`. + name: the deployment name, shared by the replicas of one deployment. It is what + gets registered with Restate (``connect://``). Defaults to + ``RESTATE_CONNECT_NAME``. + key: the connect key from the Cloud UI (Development → Configure SDK). Defaults to + ``RESTATE_CONNECT_KEY``. Under the Restate operator, its ``RESTATE_INPROC_*`` + variables are used instead. + + Raises: + ConnectError: the configuration is incomplete or invalid, or Restate Cloud + rejected the connection. + + Example:: + + async with restate.with_connection(app, name="greeter") as connection: + print(connection.info.deployment_url) + await connection.wait() + """ + options = TunnelEngineOptions.resolve(name=name, key=key) + engine = TunnelEngine(app, options) + try: + info = await engine.ready() + except BaseException: + await engine.shutdown(0.0) + raise + connection = Connection(engine, options, info) + try: + yield connection + finally: + await connection.shutdown(EXIT_SHUTDOWN_GRACE_SECONDS) + + +async def connect( + app: RestateAppT, + *, + name: typing.Optional[str] = None, + key: typing.Optional[str] = None, +) -> None: + """ + Serve ``app`` through an outbound connection to Restate Cloud, until stopped. + + The first SIGINT or SIGTERM drains in-flight invocations for up to two minutes, a + second one stops at once. Returns when the connection has stopped. + + Args: + app: the ASGI application returned by :func:`restate.app`. + name: the deployment name, registered with Restate as ``connect://``. + Defaults to ``RESTATE_CONNECT_NAME``. + key: the connect key from the Cloud UI (Development → Configure SDK). Defaults to + ``RESTATE_CONNECT_KEY``. + + Raises: + ConnectError: the configuration is incomplete or invalid, or Restate Cloud + rejected the connection. + + Example:: + + app = restate.app(services=[greeter]) + + if __name__ == "__main__": + asyncio.run(restate.connect(app, name="greeter")) + """ + # pylint: disable=import-outside-toplevel + from restate.server import install_shutdown_hook + + loop = asyncio.get_running_loop() + async with with_connection(app, name=name, key=key) as connection: + signals_seen = 0 + + def on_signal() -> None: + nonlocal signals_seen + signals_seen += 1 + if signals_seen == 1: + print( + f"Shutting down: waiting up to {SIGNAL_SHUTDOWN_GRACE_SECONDS:.0f}s for in-flight " + "invocations (send the signal again to stop now)", + file=sys.stderr, + flush=True, + ) + loop.create_task(connection.shutdown(SIGNAL_SHUTDOWN_GRACE_SECONDS)) + elif signals_seen == 2: + print("Stopping now", file=sys.stderr, flush=True) + loop.create_task(connection.shutdown(0.0)) + + for signum in (signal.SIGTERM, signal.SIGINT): + try: + install_shutdown_hook(loop, on_signal, signum) + except (NotImplementedError, RuntimeError, ValueError): + pass # Windows or non-main thread: no signal handling, stop through shutdown() + + print( + f"Connected to Restate Cloud as {connection.name}. Register this deployment as " + f"{connection.registration_identifier}", + file=sys.stderr, + flush=True, + ) + await connection.wait() diff --git a/python/restate/server.py b/python/restate/server.py index 84d3c7ce..b3c77bf4 100644 --- a/python/restate/server.py +++ b/python/restate/server.py @@ -14,7 +14,8 @@ import functools import logging import signal -from typing import Dict, Optional, Set, TypedDict, Literal +import weakref +from typing import Callable, Dict, List, Optional, Set, TypedDict, Literal from restate.discovery import compute_discovery_json from restate.endpoint import Endpoint @@ -29,6 +30,39 @@ logger = logging.getLogger(__name__) +# Shutdown hooks, keyed by event loop and signal. asyncio keeps one handler per signal, so +# everything that wants to react to SIGTERM (the ASGI app, `python -m restate connect`) +# registers a hook here and a single dispatcher is installed per loop. +_shutdown_hooks: "weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, Dict[int, List[Callable[[], None]]]]" = ( + weakref.WeakKeyDictionary() +) + + +def install_shutdown_hook( + loop: asyncio.AbstractEventLoop, hook: Callable[[], None], signum: int = signal.SIGTERM +) -> None: + """ + Run ``hook`` when ``signum`` (default SIGTERM) is received while ``loop`` runs. + + Raises what ``loop.add_signal_handler`` raises when signals cannot be handled here + (Windows, or a non-main thread). + """ + hooks = _shutdown_hooks.setdefault(loop, {}) + if signum not in hooks: + registered: List[Callable[[], None]] = [] + + def dispatch() -> None: + for registered_hook in list(registered): + try: + registered_hook() + except Exception: # pylint: disable=broad-except + logger.exception("Shutdown hook failed") + + loop.add_signal_handler(signum, dispatch) + hooks[signum] = registered + hooks[signum].append(hook) + + X_RESTATE_SERVER = header_to_binary([("x-restate-server", f"restate-sdk-python/{SDK_VERSION}")]) @@ -261,7 +295,7 @@ async def app(scope: Scope, receive: Receive, send: Send): if not sigterm_installed: loop = asyncio.get_running_loop() try: - loop.add_signal_handler(signal.SIGTERM, _on_sigterm) + install_shutdown_hook(loop, _on_sigterm, signal.SIGTERM) except (NotImplementedError, RuntimeError, ValueError): pass # Windows or non-main thread sigterm_installed = True diff --git a/src/asgi.rs b/src/asgi.rs new file mode 100644 index 00000000..b7945f46 --- /dev/null +++ b/src/asgi.rs @@ -0,0 +1,694 @@ +//! The ASGI bridge: a hyper `Service` that dispatches every request the +//! tunnel engine forwards into the SDK's unchanged ASGI-3 application. +//! +//! One request = one `app(scope, receive, send)` task on the Python event +//! loop. `receive` pulls one HTTP/2 DATA frame per call and, after the request +//! body ended, blocks until the stream closes (`http.disconnect`). `send` +//! streams `http.response.body` chunks to hyper with backpressure and supports +//! the `http.response.trailers` extension. + +use std::convert::Infallible; +use std::future::Future; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex}; +use std::task::{Context, Poll}; + +use bytes::Bytes; +use http::header::{HeaderName, HeaderValue}; +use http::{HeaderMap, Request, Response, StatusCode}; +use http_body::{Body, Frame, SizeHint}; +use http_body_util::{BodyExt, Full, StreamBody}; +use hyper::body::Incoming; +use hyper::service::Service; +use pyo3::exceptions::{PyRuntimeError, PyTypeError}; +use pyo3::prelude::*; +use pyo3::types::{PyBytes, PyDict, PyList, PyTuple}; +use pyo3_async_runtimes::TaskLocals; +use restate_sdk_shared_core::tunnel::{BoxError, TunnelBody}; +use tokio::sync::{mpsc, oneshot}; +use tokio_stream::wrappers::ReceiverStream; +use tokio_util::sync::CancellationToken; + +/// How many response frames may sit between the app and hyper. One keeps the +/// app's `send` awaiting until hyper has taken the previous chunk. +const RESPONSE_FRAME_BUFFER: usize = 1; + +/// The ASGI app plus the event loop it runs on. +#[derive(Clone)] +pub struct AsgiService { + app: Arc>, + locals: TaskLocals, +} + +impl AsgiService { + pub fn new(app: Py, locals: TaskLocals) -> Self { + Self { + app: Arc::new(app), + locals, + } + } +} + +impl Service> for AsgiService { + type Response = Response; + type Error = Infallible; + type Future = Pin> + Send>>; + + fn call(&self, request: Request) -> Self::Future { + let app = Arc::clone(&self.app); + let locals = self.locals.clone(); + Box::pin(async move { Ok(handle_request(app, locals, request).await) }) + } +} + +async fn handle_request( + app: Arc>, + locals: TaskLocals, + request: Request, +) -> Response { + let (parts, body) = request.into_parts(); + let closed = CancellationToken::new(); + // Until the response head exists, this future is the only owner of the stream: + // hyper drops it on a reset, a lost connection or a force shutdown. Dropping it + // must still turn `receive()` into `http.disconnect`. + let closed_guard = closed.clone().drop_guard(); + let (head_tx, head_rx) = oneshot::channel::(); + let (frame_tx, frame_rx) = + mpsc::channel::, BoxError>>(RESPONSE_FRAME_BUFFER); + let abort = Arc::new(Mutex::new(None)); + + let receive = Arc::new(ReceiveState { + body: tokio::sync::Mutex::new(Some(body)), + ended: AtomicBool::new(false), + closed: closed.clone(), + }); + let send = Arc::new(SendState { + inner: Mutex::new(SendInner { + stage: Stage::AwaitingStart, + head: Some(head_tx), + frames: Some(frame_tx), + pending_trailers: HeaderMap::new(), + }), + abort: Arc::clone(&abort), + closed: closed.clone(), + }); + + // Building the scope and starting the coroutine takes the GIL: do it off the + // runtime's worker threads, which hyper needs for the connection itself. + let app_future = tokio::task::spawn_blocking({ + let receive = Arc::clone(&receive); + let send = Arc::clone(&send); + let locals = locals.clone(); + move || { + Python::attach(|py| -> PyResult<_> { + let scope = build_scope(py, &parts)?; + let receive = Py::new( + py, + AsgiReceive { + state: receive, + locals: locals.clone(), + }, + )?; + let send = Py::new( + py, + AsgiSend { + state: send, + locals: locals.clone(), + }, + )?; + let coroutine = app.bind(py).call1((scope, receive, send))?; + pyo3_async_runtimes::into_future_with_locals(&locals, coroutine) + }) + } + }) + .await; + let app_future = match app_future { + Ok(Ok(future)) => future, + Ok(Err(error)) => { + log_app_error(&error, "The ASGI application could not be started"); + return internal_error(); + } + Err(join_error) => { + tracing::error!(error = %join_error, "The ASGI application could not be started"); + return internal_error(); + } + }; + + // The app keeps running on the Python loop after the response head is + // available; this task observes how it ends. + tokio::spawn({ + let send = Arc::clone(&send); + async move { + match app_future.await { + Ok(_) => { + tracing::debug!("ASGI application task finished"); + send.app_finished(None) + } + Err(error) => { + let cancelled = Python::attach(|py| { + error.is_instance_of::(py) + }); + if cancelled { + // The event loop is going away (asyncio.run exit): not an app failure. + tracing::debug!("ASGI application task cancelled"); + } else { + log_app_error(&error, "Exception in the ASGI application"); + } + send.app_finished(Some(error.to_string())); + } + } + } + }); + + let head = match head_rx.await { + Ok(head) => head, + Err(_) => return internal_error(), + }; + // From here on the response body owns the stream's fate. + closed_guard.disarm(); + let mut response = Response::new( + ClosingBody { + inner: StreamBody::new(ReceiverStream::new(frame_rx)), + abort, + closed, + } + .boxed_unsync(), + ); + *response.status_mut() = head.status; + *response.headers_mut() = head.headers; + response +} + +fn log_app_error(error: &PyErr, message: &str) { + Python::attach(|py| { + let traceback = error + .traceback(py) + .and_then(|traceback| traceback.format().ok()) + .unwrap_or_default(); + tracing::error!(error = %error, "{message}\n{traceback}"); + }); +} + +fn internal_error() -> Response { + let mut response = Response::new( + Full::new(Bytes::from_static(b"Internal Server Error")) + .map_err(|never| match never {}) + .boxed_unsync(), + ); + *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR; + response +} + +fn build_scope<'py>(py: Python<'py>, parts: &http::request::Parts) -> PyResult> { + let scope = PyDict::new(py); + scope.set_item("type", "http")?; + let asgi = PyDict::new(py); + asgi.set_item("version", "3.0")?; + asgi.set_item("spec_version", "2.3")?; + scope.set_item("asgi", asgi)?; + scope.set_item("http_version", "2")?; + scope.set_item("method", parts.method.as_str())?; + scope.set_item("scheme", "https")?; + let raw_path = parts.uri.path(); + scope.set_item( + "path", + percent_encoding::percent_decode_str(raw_path) + .decode_utf8_lossy() + .as_ref(), + )?; + scope.set_item("raw_path", PyBytes::new(py, raw_path.as_bytes()))?; + scope.set_item( + "query_string", + PyBytes::new(py, parts.uri.query().unwrap_or("").as_bytes()), + )?; + scope.set_item("root_path", "")?; + let headers = PyList::empty(py); + for (name, value) in &parts.headers { + headers.append(PyTuple::new( + py, + [ + PyBytes::new(py, name.as_str().as_bytes()), + PyBytes::new(py, value.as_bytes()), + ], + )?)?; + } + scope.set_item("headers", headers)?; + scope.set_item("client", py.None())?; + scope.set_item("server", py.None())?; + let extensions = PyDict::new(py); + extensions.set_item("http.response.trailers", PyDict::new(py))?; + scope.set_item("extensions", extensions)?; + Ok(scope) +} + +// ───────────────────────────── receive ───────────────────────────── + +struct ReceiveState { + body: tokio::sync::Mutex>, + ended: AtomicBool, + closed: CancellationToken, +} + +/// What `receive()` resolves to; converted to the ASGI dict where the Python future is resolved. +enum ReceiveEvent { + Request { body: Bytes, more_body: bool }, + Disconnect, +} + +impl<'py> IntoPyObject<'py> for ReceiveEvent { + type Target = PyDict; + type Output = Bound<'py, PyDict>; + type Error = PyErr; + + fn into_pyobject(self, py: Python<'py>) -> Result { + let event = PyDict::new(py); + match self { + Self::Request { body, more_body } => { + event.set_item("type", "http.request")?; + event.set_item("body", PyBytes::new(py, &body))?; + event.set_item("more_body", more_body)?; + } + Self::Disconnect => { + event.set_item("type", "http.disconnect")?; + } + } + Ok(event) + } +} + +impl ReceiveState { + async fn next_event(&self) -> PyResult { + if !self.ended.load(Ordering::Acquire) { + let mut guard = self.body.lock().await; + while let Some(body) = guard.as_mut() { + let frame = tokio::select! { + biased; + _ = self.closed.cancelled() => { + *guard = None; + self.ended.store(true, Ordering::Release); + return disconnect_event(); + } + frame = body.frame() => frame, + }; + match frame { + Some(Ok(frame)) => match frame.into_data() { + Ok(data) => return request_event(data, true), + // Request trailers carry nothing for the app; the next + // frame is the end of the body. + Err(_) => continue, + }, + Some(Err(_)) => { + // The stream was reset by the peer. + *guard = None; + self.ended.store(true, Ordering::Release); + self.closed.cancel(); + return disconnect_event(); + } + None => { + *guard = None; + self.ended.store(true, Ordering::Release); + return request_event(Bytes::new(), false); + } + } + } + } + self.closed.cancelled().await; + disconnect_event() + } +} + +fn request_event(body: Bytes, more_body: bool) -> PyResult { + Ok(ReceiveEvent::Request { body, more_body }) +} + +fn disconnect_event() -> PyResult { + Ok(ReceiveEvent::Disconnect) +} + +/// The ASGI `receive` callable of one request. +#[pyclass(name = "AsgiReceive", frozen)] +struct AsgiReceive { + state: Arc, + locals: TaskLocals, +} + +#[pymethods] +impl AsgiReceive { + fn __call__<'py>(&self, py: Python<'py>) -> PyResult> { + let state = Arc::clone(&self.state); + pyo3_async_runtimes::tokio::future_into_py_with_locals( + py, + self.locals.clone(), + async move { state.next_event().await }, + ) + } +} + +// ───────────────────────────── send ───────────────────────────── + +struct ResponseHead { + status: StatusCode, + headers: HeaderMap, +} + +#[derive(Clone, Copy, PartialEq, Eq)] +enum Stage { + AwaitingStart, + Body { trailers: bool }, + Trailers, + Done, +} + +struct SendInner { + stage: Stage, + head: Option>, + frames: Option, BoxError>>>, + pending_trailers: HeaderMap, +} + +struct SendState { + inner: Mutex, + /// Set when the app ends without completing the response: the body then fails + /// (a stream reset) instead of ending cleanly, even if a frame is still buffered. + abort: Arc>>, + closed: CancellationToken, +} + +enum Message { + Start { + status: StatusCode, + headers: HeaderMap, + trailers: bool, + }, + Body { + body: Bytes, + more_body: bool, + }, + Trailers { + headers: HeaderMap, + more_trailers: bool, + }, +} + +impl Message { + fn parse(message: &Bound<'_, PyDict>) -> PyResult { + let kind: String = message + .get_item("type")? + .ok_or_else(|| PyTypeError::new_err("ASGI message has no 'type'"))? + .extract()?; + match kind.as_str() { + "http.response.start" => { + let status: u16 = message + .get_item("status")? + .ok_or_else(|| PyTypeError::new_err("http.response.start has no 'status'"))? + .extract()?; + let status = StatusCode::from_u16(status).map_err(|_| { + PyRuntimeError::new_err(format!("invalid HTTP status {status}")) + })?; + let headers = match message.get_item("headers")? { + Some(headers) => parse_headers(&headers)?, + None => HeaderMap::new(), + }; + let trailers = match message.get_item("trailers")? { + Some(value) => value.extract()?, + None => false, + }; + Ok(Self::Start { + status, + headers, + trailers, + }) + } + "http.response.body" => { + let body = match message.get_item("body")? { + Some(body) => Bytes::copy_from_slice(body.cast::()?.as_bytes()), + None => Bytes::new(), + }; + let more_body = match message.get_item("more_body")? { + Some(value) => value.extract()?, + None => false, + }; + Ok(Self::Body { body, more_body }) + } + "http.response.trailers" => { + let headers = match message.get_item("headers")? { + Some(headers) => parse_headers(&headers)?, + None => HeaderMap::new(), + }; + let more_trailers = match message.get_item("more_trailers")? { + Some(value) => value.extract()?, + None => false, + }; + Ok(Self::Trailers { + headers, + more_trailers, + }) + } + other => Err(PyRuntimeError::new_err(format!( + "Unexpected ASGI message type '{other}'" + ))), + } + } + + fn kind(&self) -> &'static str { + match self { + Self::Start { .. } => "http.response.start", + Self::Body { .. } => "http.response.body", + Self::Trailers { .. } => "http.response.trailers", + } + } +} + +fn parse_headers(headers: &Bound<'_, PyAny>) -> PyResult { + let mut map = HeaderMap::new(); + for pair in headers.try_iter()? { + let pair = pair?; + let name = pair.get_item(0)?; + let value = pair.get_item(1)?; + let name = HeaderName::from_bytes(name.cast::()?.as_bytes()) + .map_err(|error| PyRuntimeError::new_err(format!("invalid header name: {error}")))?; + let value = HeaderValue::from_bytes(value.cast::()?.as_bytes()) + .map_err(|error| PyRuntimeError::new_err(format!("invalid header value: {error}")))?; + map.append(name, value); + } + Ok(map) +} + +impl SendState { + async fn send(&self, message: Message) -> PyResult<()> { + // The message must fit the protocol state even after a disconnection; what + // is valid is then delivered or, with nobody left to deliver to, dropped so + // the app can finish. + let step = { + let mut inner = self.inner.lock().expect("send state lock poisoned"); + inner.step(message)? + }; + if self.closed.is_cancelled() { + return Ok(()); + } + match step { + Step::Nothing => Ok(()), + Step::Frame { + sender, + frame, + last, + } => { + // Backpressure: wait until hyper took the previous frame. A closed + // receiver means the stream is gone; the app may finish quietly. + if let Ok(permit) = sender.reserve().await { + permit.send(Ok(frame)); + } + if last { + drop(sender); + } + Ok(()) + } + } + } + + fn app_finished(&self, error: Option) { + let mut inner = self.inner.lock().expect("send state lock poisoned"); + match inner.stage { + Stage::AwaitingStart => { + // The app returned (or failed) without a response. + if let Some(head) = inner.head.take() { + let _ = head.send(ResponseHead { + status: StatusCode::INTERNAL_SERVER_ERROR, + headers: HeaderMap::new(), + }); + } + inner.frames.take(); + inner.stage = Stage::Done; + } + Stage::Body { .. } | Stage::Trailers => { + // An incomplete response: reset the stream rather than ending it cleanly. + // Recorded beside the channel, so a frame still buffered in it cannot + // turn the failure into a clean end of stream. + let reason = error.unwrap_or_else(|| { + "the ASGI application finished without completing the response".to_owned() + }); + *self.abort.lock().expect("abort lock poisoned") = Some(reason); + inner.frames.take(); + inner.stage = Stage::Done; + } + Stage::Done => {} + } + } +} + +enum Step { + Nothing, + Frame { + sender: mpsc::Sender, BoxError>>, + frame: Frame, + last: bool, + }, +} + +impl SendInner { + fn step(&mut self, message: Message) -> PyResult { + match (self.stage, message) { + ( + Stage::AwaitingStart, + Message::Start { + status, + headers, + trailers, + }, + ) => { + if let Some(head) = self.head.take() { + let _ = head.send(ResponseHead { status, headers }); + } + self.stage = Stage::Body { trailers }; + Ok(Step::Nothing) + } + (Stage::Body { trailers }, Message::Body { body, more_body }) => { + let Some(sender) = self.frames.clone() else { + return Ok(Step::Nothing); + }; + let last = !more_body && !trailers; + if !more_body { + self.stage = if trailers { + Stage::Trailers + } else { + Stage::Done + }; + } + if last { + self.frames.take(); + } + Ok(Step::Frame { + sender, + frame: Frame::data(body), + last, + }) + } + ( + Stage::Trailers, + Message::Trailers { + headers, + more_trailers, + }, + ) => { + // By reference: iterating an owned map names only the first value of each name. + for (name, value) in &headers { + self.pending_trailers.append(name, value.clone()); + } + if more_trailers { + return Ok(Step::Nothing); + } + let Some(sender) = self.frames.take() else { + return Ok(Step::Nothing); + }; + self.stage = Stage::Done; + Ok(Step::Frame { + sender, + frame: Frame::trailers(std::mem::take(&mut self.pending_trailers)), + last: true, + }) + } + (stage, message) => { + let expected = match stage { + Stage::AwaitingStart => "'http.response.start'", + Stage::Body { .. } => "'http.response.body'", + Stage::Trailers => "'http.response.trailers'", + Stage::Done => "no further message (the response is complete)", + }; + Err(PyRuntimeError::new_err(format!( + "Unexpected ASGI message '{}', expected {expected}", + message.kind() + ))) + } + } + } +} + +/// The ASGI `send` callable of one request. +#[pyclass(name = "AsgiSend", frozen)] +struct AsgiSend { + state: Arc, + locals: TaskLocals, +} + +#[pymethods] +impl AsgiSend { + fn __call__<'py>( + &self, + py: Python<'py>, + message: &Bound<'py, PyDict>, + ) -> PyResult> { + let message = Message::parse(message)?; + let state = Arc::clone(&self.state); + pyo3_async_runtimes::tokio::future_into_py_with_locals( + py, + self.locals.clone(), + async move { state.send(message).await }, + ) + } +} + +/// The response body handed to hyper. Ending or dropping it (hyper drops the +/// body on a stream reset) marks the stream closed, which turns `receive` into +/// `http.disconnect` and makes later `send`s no-ops. +struct ClosingBody { + inner: StreamBody, BoxError>>>, + abort: Arc>>, + closed: CancellationToken, +} + +impl Body for ClosingBody { + type Data = Bytes; + type Error = BoxError; + + fn poll_frame( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll, Self::Error>>> { + let this = self.get_mut(); + let mut result = Pin::new(&mut this.inner).poll_frame(cx); + if matches!(result, Poll::Ready(None)) { + if let Some(reason) = this.abort.lock().expect("abort lock poisoned").take() { + result = Poll::Ready(Some(Err(reason.into()))); + } + } + if matches!(result, Poll::Ready(None) | Poll::Ready(Some(Err(_)))) { + this.closed.cancel(); + } + result + } + + fn is_end_stream(&self) -> bool { + self.inner.is_end_stream() + } + + fn size_hint(&self) -> SizeHint { + self.inner.size_hint() + } +} + +impl Drop for ClosingBody { + fn drop(&mut self) { + self.closed.cancel(); + } +} diff --git a/src/lib.rs b/src/lib.rs index 12297367..57ecc9e4 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,6 @@ +mod asgi; +mod tunnel; + use pyo3::create_exception; use pyo3::prelude::*; use pyo3::types::{PyBytes, PyNone, PyString}; @@ -1030,9 +1033,15 @@ For further info on error handling, refer to https://docs.restate.dev/develop/py fn _internal(m: &Bound<'_, PyModule>) -> PyResult<()> { use tracing_subscriber::EnvFilter; - tracing_subscriber::fmt() - .with_env_filter(EnvFilter::from_env("RESTATE_CORE_LOG")) - .init(); + // RESTATE_CORE_LOG controls everything when set. When it is not, the Restate Cloud + // tunnel still reports its lifecycle (the registration hint, connection changes). + let filter = match std::env::var("RESTATE_CORE_LOG") { + Ok(directives) => EnvFilter::new(directives), + Err(_) => EnvFilter::new( + "restate_sdk_shared_core::tunnel=info,restate_sdk_shared_core::cloud=warn,restate_sdk_python_core=warn", + ), + }; + tracing_subscriber::fmt().with_env_filter(filter).init(); m.add_class::()?; m.add_class::()?; @@ -1051,6 +1060,12 @@ fn _internal(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add_class::()?; + m.add("ConnectError", m.py().get_type::())?; + m.add("INGRESS_KEY_URL", tunnel::INGRESS_KEY_HELP_URL)?; m.add("VMException", m.py().get_type::())?; m.add( diff --git a/src/tunnel.rs b/src/tunnel.rs new file mode 100644 index 00000000..42ef2045 --- /dev/null +++ b/src/tunnel.rs @@ -0,0 +1,328 @@ +//! Python bindings of the Restate Cloud tunnel engine and keys. + +use std::time::Duration; + +use pyo3::create_exception; +use pyo3::exceptions::{PyException, PyRuntimeError, PyValueError}; +use pyo3::prelude::*; +use pyo3::types::PyDict; +use restate_sdk_shared_core::cloud_keys::{IngressKey, KeyError, INGRESS_KEY_URL}; +use restate_sdk_shared_core::tunnel::{ + self, ConfigSource, ConnectOptions, ReconnectPolicy, Servers, Tuning, TunnelEngine, + TunnelEngineOptions, +}; + +use crate::asgi::AsgiService; + +create_exception!( + _internal, + ConnectError, + PyException, + "Raised when the deployment cannot connect to Restate Cloud: a configuration problem, \ + rejected credentials, or an unrecoverable connection failure." +); + +fn connect_error(error: impl std::fmt::Display) -> PyErr { + ConnectError::new_err(error.to_string()) +} + +/// Everything the engine needs, resolved from code and the environment. +#[pyclass(name = "TunnelEngineOptions", frozen)] +pub struct PyTunnelEngineOptions { + inner: TunnelEngineOptions, +} + +#[pymethods] +impl PyTunnelEngineOptions { + /// Resolve the options. Every argument is optional and falls back to the environment + /// (`RESTATE_CONNECT_NAME`, `RESTATE_CONNECT_KEY`, or the operator's `RESTATE_INPROC_*`). + #[staticmethod] + #[pyo3(signature = (name=None, key=None, servers=None, servers_srv=None, worker_id=None, tuning=None))] + fn resolve( + name: Option, + key: Option, + servers: Option>, + servers_srv: Option, + worker_id: Option, + tuning: Option<&Bound<'_, PyDict>>, + ) -> PyResult { + if servers.is_some() && servers_srv.is_some() { + return Err(PyValueError::new_err( + "specify at most one of servers and servers_srv", + )); + } + let servers = match (servers, servers_srv) { + (Some(servers), _) => Some(Servers::Explicit(servers)), + (None, Some(srv)) => Some(Servers::Srv(srv)), + (None, None) => None, + }; + let tuning = match tuning { + Some(tuning) => parse_tuning(tuning)?, + None => Tuning::default(), + }; + let inner = TunnelEngineOptions::resolve(ConnectOptions { + name, + key, + token: None, + servers, + worker_id, + tuning, + }) + .map_err(connect_error)?; + Ok(Self { inner }) + } + + #[getter] + fn name(&self) -> &str { + &self.inner.name + } + + #[getter] + fn environment_id(&self) -> &str { + &self.inner.environment_id + } + + #[getter] + fn region(&self) -> &str { + &self.inner.region + } + + /// Where the coordinates came from: "code", "connect_key" or "operator". + #[getter] + fn source(&self) -> &'static str { + match self.inner.source { + ConfigSource::Code => "code", + ConfigSource::ConnectKey => "connect_key", + ConfigSource::Operator => "operator", + } + } + + /// `connect://`: what to register with Restate. + #[getter] + fn registration_identifier(&self) -> String { + self.inner.registration_identifier() + } + + /// The Cloud UI link that opens the register dialog pre-filled. + #[getter] + fn registration_deep_link(&self) -> String { + self.inner.registration_deep_link() + } + + fn __repr__(&self) -> String { + format!("{:?}", self.inner) + } +} + +fn parse_tuning(tuning: &Bound<'_, PyDict>) -> PyResult { + let mut result = Tuning::default(); + let mut reconnect = ReconnectPolicy::default(); + for (key, value) in tuning.iter() { + let key: String = key.extract()?; + match key.as_str() { + "connect_timeout" => result.connect_timeout = seconds(&value)?, + "handshake_timeout" => result.handshake_timeout = seconds(&value)?, + "ping_interval" => result.ping_interval = seconds(&value)?, + "ping_timeout" => result.ping_timeout = seconds(&value)?, + "server_drain_grace" => result.server_drain_grace = seconds(&value)?, + "max_concurrent_streams" => result.max_concurrent_streams = value.extract()?, + "reconnect_initial_interval" => reconnect.initial_interval = seconds(&value)?, + "reconnect_max_interval" => reconnect.max_interval = seconds(&value)?, + "reconnect_factor" => reconnect.factor = value.extract()?, + "reconnect_max_attempts" => reconnect.max_attempts = value.extract()?, + "reconnect_max_duration" => { + reconnect.max_duration = if value.is_none() { + None + } else { + Some(seconds(&value)?) + } + } + other => { + return Err(PyValueError::new_err(format!( + "unknown tuning option {other:?}" + ))) + } + } + } + result.reconnect = reconnect; + Ok(result) +} + +fn seconds(value: &Bound<'_, PyAny>) -> PyResult { + let seconds: f64 = value.extract()?; + if !seconds.is_finite() || seconds < 0.0 { + return Err(PyValueError::new_err(format!( + "durations are non-negative seconds, got {seconds}" + ))); + } + Ok(Duration::from_secs_f64(seconds)) +} + +/// Metadata learned from a successful Cloud handshake. +#[pyclass(name = "TunnelInfo", frozen, get_all)] +#[derive(Clone)] +pub struct PyTunnelInfo { + name: String, + proxy_url: String, + tunnel_url: String, + deployment_url: String, +} + +impl From for PyTunnelInfo { + fn from(info: tunnel::TunnelInfo) -> Self { + Self { + name: info.name, + proxy_url: info.proxy_url, + tunnel_url: info.tunnel_url, + deployment_url: info.deployment_url, + } + } +} + +#[pymethods] +impl PyTunnelInfo { + fn __repr__(&self) -> String { + format!( + "TunnelInfo(name={:?}, proxy_url={:?}, tunnel_url={:?}, deployment_url={:?})", + self.name, self.proxy_url, self.tunnel_url, self.deployment_url + ) + } +} + +/// A running tunnel engine serving an ASGI application. Must be created inside a +/// running asyncio event loop: the application is called on that loop. +/// +/// Dropping the last reference cuts every connection at once. +#[pyclass(name = "TunnelEngine", frozen)] +pub struct PyTunnelEngine { + engine: TunnelEngine, +} + +#[pymethods] +impl PyTunnelEngine { + #[new] + fn new(py: Python<'_>, app: Py, options: &PyTunnelEngineOptions) -> PyResult { + let locals = pyo3_async_runtimes::tokio::get_current_locals(py).map_err(|_| { + PyRuntimeError::new_err( + "TunnelEngine must be created inside a running asyncio event loop", + ) + })?; + let service = AsgiService::new(app, locals); + let runtime = pyo3_async_runtimes::tokio::get_runtime(); + let engine = { + let _guard = runtime.enter(); + TunnelEngine::start(service, options.inner.clone()) + } + .map_err(connect_error)?; + Ok(Self { engine }) + } + + /// Await the first successful handshake; returns the `TunnelInfo`. + /// Raises `ConnectError` if the engine stops first. + fn ready<'py>(&self, py: Python<'py>) -> PyResult> { + let engine = self.engine.clone(); + pyo3_async_runtimes::tokio::future_into_py(py, async move { + engine + .ready() + .await + .map(PyTunnelInfo::from) + .map_err(connect_error) + }) + } + + /// Await the engine's end: returns after a shutdown, raises `ConnectError` after a + /// fatal error. + fn terminated<'py>(&self, py: Python<'py>) -> PyResult> { + let engine = self.engine.clone(); + pyo3_async_runtimes::tokio::future_into_py(py, async move { + engine.terminated().await.map_err(connect_error) + }) + } + + /// Stop serving: refuse new invocations, let the in-flight ones finish, and cut + /// whatever is still running after `grace_seconds`. Zero cuts everything at once. + /// Teardown begins before the returned awaitable is awaited. + fn shutdown<'py>(&self, py: Python<'py>, grace_seconds: f64) -> PyResult> { + if !grace_seconds.is_finite() || grace_seconds < 0.0 { + return Err(PyValueError::new_err( + "grace_seconds must be a non-negative number of seconds", + )); + } + let shutdown = self.engine.shutdown(Duration::from_secs_f64(grace_seconds)); + pyo3_async_runtimes::tokio::future_into_py(py, async move { + shutdown.await.map_err(connect_error) + }) + } +} + +/// A parsed `rstcloudv1_…` ingress key. +#[pyclass(name = "IngressKey", frozen, get_all)] +pub struct PyIngressKey { + environment_id: String, + api_key: String, + region: String, +} + +#[pymethods] +impl PyIngressKey { + /// Parse an ingress key. Raises `ConnectError` with the text to show the user. + #[staticmethod] + #[pyo3(signature = (value, what="the ingress key"))] + fn parse(value: &str, what: &str) -> PyResult { + IngressKey::parse(value) + .map(Self::from) + .map_err(|error| ConnectError::new_err(key_error(what, &error))) + } + + /// Read and parse `RESTATE_INGRESS_KEY`; `None` when it is unset or empty. + #[staticmethod] + fn from_env() -> PyResult> { + IngressKey::from_env() + .map(|key| key.map(Self::from)) + .map_err(|error| ConnectError::new_err(error.to_string())) + } + + /// `https://.env.:8080` + fn ingress_url(&self) -> String { + self.key().ingress_url() + } + + /// The `Authorization` header value. + fn bearer(&self) -> String { + self.key().bearer() + } + + fn __repr__(&self) -> String { + format!( + "IngressKey(environment_id={:?}, region={:?}, api_key=[REDACTED])", + self.environment_id, self.region + ) + } +} + +impl PyIngressKey { + fn key(&self) -> IngressKey { + IngressKey { + environment_id: self.environment_id.clone(), + api_key: self.api_key.clone(), + region: self.region.clone(), + } + } +} + +impl From for PyIngressKey { + fn from(key: IngressKey) -> Self { + Self { + environment_id: key.environment_id, + api_key: key.api_key, + region: key.region, + } + } +} + +/// The Cloud UI page where ingress keys are created. +pub const INGRESS_KEY_HELP_URL: &str = INGRESS_KEY_URL; + +fn key_error(what: &str, error: &KeyError) -> String { + format!("{what} {error} — copy it again from {INGRESS_KEY_URL}") +} diff --git a/tests/connect.py b/tests/connect.py new file mode 100644 index 00000000..a1588cb6 --- /dev/null +++ b/tests/connect.py @@ -0,0 +1,904 @@ +# +# Copyright (c) 2023-2026 - Restate Software, Inc., Restate GmbH +# +# This file is part of the Restate SDK for Python, +# which is released under the MIT license. +# +# You can find a copy of the license in file LICENSE in the root +# directory of this repository or package, or at +# https://github.com/restatedev/sdk-typescript/blob/main/LICENSE +# +""" +Tests for `restate.connect()` and `python -m restate connect`: the Python layer, the +ASGI bridge and the engine, against a fake Restate Cloud tunnel peer speaking +plaintext HTTP/2 on localhost (role reversed: the peer is the HTTP/2 client). +""" + +import asyncio +import base64 +import hashlib +import json +import sys +import time +import typing + +import pytest +from h2.config import H2Configuration +from h2.connection import ConnectionState, H2Connection +from h2.events import ConnectionTerminated, DataReceived, ResponseReceived, StreamEnded, StreamReset, TrailersReceived +from h2.exceptions import ProtocolError + +import restate +import restate._internal as _internal # pylint: disable=import-error,no-name-in-module +from restate.client import resolve_ingress +from restate.__main__ import _load_app + +pytestmark = [pytest.mark.anyio] + + +@pytest.fixture(scope="session") +def anyio_backend(): + return "asyncio" + + +# ───────────────────────────── ed25519 (RFC 8032, pure Python, tests only) ───────────────────────────── + +_P = 2**255 - 19 +_Q = 2**252 + 27742317777372353535851937790883648493 + + +def _inv(x: int) -> int: + return pow(x, _P - 2, _P) + + +_D = (-121665 * _inv(121666)) % _P +_I = pow(2, (_P - 1) // 4, _P) + + +def _xrecover(y: int) -> int: + xx = (y * y - 1) * _inv(_D * y * y + 1) + x = pow(xx, (_P + 3) // 8, _P) + if (x * x - xx) % _P != 0: + x = (x * _I) % _P + if x % 2 != 0: + x = _P - x + return x + + +_BY = 4 * _inv(5) % _P +_B = (_xrecover(_BY), _BY) + + +def _add(p: typing.Tuple[int, int], q: typing.Tuple[int, int]) -> typing.Tuple[int, int]: + x1, y1 = p + x2, y2 = q + x3 = (x1 * y2 + x2 * y1) * _inv(1 + _D * x1 * x2 * y1 * y2) + y3 = (y1 * y2 + x1 * x2) * _inv(1 - _D * x1 * x2 * y1 * y2) + return x3 % _P, y3 % _P + + +def _mul(p: typing.Tuple[int, int], e: int) -> typing.Tuple[int, int]: + result = (0, 1) + addend = p + while e: + if e & 1: + result = _add(result, addend) + addend = _add(addend, addend) + e >>= 1 + return result + + +def _encode_point(p: typing.Tuple[int, int]) -> bytes: + x, y = p + return ((y & ((1 << 255) - 1)) | ((x & 1) << 255)).to_bytes(32, "little") + + +def _clamp(h: bytes) -> int: + a = int.from_bytes(h[:32], "little") + return (a & ((1 << 254) - 8)) | (1 << 254) + + +def ed25519_public_key(seed: bytes) -> bytes: + return _encode_point(_mul(_B, _clamp(hashlib.sha512(seed).digest()))) + + +def ed25519_sign(message: bytes, seed: bytes) -> bytes: + h = hashlib.sha512(seed).digest() + a = _clamp(h) + public_key = _encode_point(_mul(_B, a)) + r = int.from_bytes(hashlib.sha512(h[32:] + message).digest(), "little") % _Q + encoded_r = _encode_point(_mul(_B, r)) + k = int.from_bytes(hashlib.sha512(encoded_r + public_key + message).digest(), "little") + s = (r + k * a) % _Q + return encoded_r + s.to_bytes(32, "little") + + +# ───────────────────────────── keys ───────────────────────────── + +_BASE58 = "123456789ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz" + + +def base58_encode(data: bytes) -> str: + number = int.from_bytes(data, "big") + digits = "" + while number: + number, remainder = divmod(number, 58) + digits = _BASE58[remainder] + digits + leading = len(data) - len(data.lstrip(b"\0")) + return "1" * leading + digits + + +IDENTITY_SEED = bytes([0x2A] * 32) +SIGNING_PUBLIC_KEY = "publickeyv1_" + base58_encode(ed25519_public_key(IDENTITY_SEED)) +ENVIRONMENT_ID = "env_test123" +REGION = "eu.restate.cloud" +API_KEY = "key_test.secret" +TUNNEL_NAME = "greeter-test" + + +def cloud_key(query: str) -> str: + return "rstcloudv1_" + base58_encode(query.encode()) + + +def connect_key(api_key: typing.Optional[str] = API_KEY) -> str: + query = f"envid={ENVIRONMENT_ID}" + if api_key: + query += f"&apikey={api_key}" + query += f"®ion={REGION}&signingkey={SIGNING_PUBLIC_KEY}" + return cloud_key(query) + + +def ingress_key() -> str: + return cloud_key(f"envid={ENVIRONMENT_ID}&apikey={API_KEY}®ion={REGION}") + + +def b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode() + + +def signed_identity(audience: str, seed: bytes = IDENTITY_SEED) -> str: + now = int(time.time()) + header = {"alg": "EdDSA", "typ": "JWT", "kid": "publickeyv1_" + base58_encode(ed25519_public_key(seed))} + claims = {"aud": audience, "nbf": now - 60, "iat": now, "exp": now + 60} + signing_input = f"{b64url(json.dumps(header).encode())}.{b64url(json.dumps(claims).encode())}" + return f"{signing_input}.{b64url(ed25519_sign(signing_input.encode(), seed))}" + + +def identity_headers(audience: str) -> typing.List[typing.Tuple[str, str]]: + return [ + ("accept", "application/vnd.restate.endpointmanifest.v2+json"), + ("x-restate-signature-scheme", "v1"), + ("x-restate-jwt-v1", signed_identity(audience)), + ] + + +# ───────────────────────────── the fake Cloud peer ───────────────────────────── + +TEST_TIMEOUT = 5.0 + + +class Response: + def __init__(self) -> None: + self.stream_id = 0 + self.headers: typing.Dict[str, str] = {} + self.trailers: typing.Dict[str, str] = {} + self.trailer_pairs: typing.List[typing.Tuple[str, str]] = [] + self.body = bytearray() + self.ended: asyncio.Future[None] = asyncio.get_running_loop().create_future() + self.started: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + @property + def status(self) -> int: + return int(self.headers[":status"]) + + +class FakeCloud: + """ + Accepts the SDK's outbound socket, runs HTTP/2 as the client, opens + `GET /_/start-tunnel` and keeps its request body open until `establish()` + sends the trailers. Then forwards requests over the same session. + """ + + def __init__(self) -> None: + self.port = 0 + self._server: typing.Optional[asyncio.AbstractServer] = None + self._conn: typing.Optional[H2Connection] = None + self._writer: typing.Optional[asyncio.StreamWriter] = None + self._responses: typing.Dict[int, Response] = {} + self._handshake_stream = 0 + self._handshake: asyncio.Future[Response] = asyncio.get_running_loop().create_future() + self._closed: asyncio.Future[None] = asyncio.get_running_loop().create_future() + self._goaway: asyncio.Future[None] = asyncio.get_running_loop().create_future() + self._reader_task: typing.Optional[asyncio.Task[None]] = None + + async def __aenter__(self) -> "FakeCloud": + self._server = await asyncio.start_server(self._accept, "127.0.0.1", 0) + self.port = self._server.sockets[0].getsockname()[1] + return self + + async def __aexit__(self, *_: object) -> None: + if self._reader_task is not None: + self._reader_task.cancel() + try: + await self._reader_task + except (asyncio.CancelledError, Exception): # pylint: disable=broad-except + pass + if self._writer is not None: + self._writer.close() + if self._server is not None: + self._server.close() + await asyncio.wait_for(self._server.wait_closed(), TEST_TIMEOUT) + + @property + def servers(self) -> typing.List[str]: + return [f"http://127.0.0.1:{self.port}"] + + async def _accept(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + if self._conn is not None: + writer.close() + return + conn = H2Connection(H2Configuration(client_side=True, header_encoding="utf-8")) + conn.initiate_connection() + self._conn = conn + self._writer = writer + stream_id = conn.get_next_available_stream_id() + self._handshake_stream = stream_id + response = Response() + self._responses[stream_id] = response + conn.send_headers( + stream_id, + [(":method", "GET"), (":path", "/_/start-tunnel"), (":scheme", "http"), (":authority", "fake")], + end_stream=False, + ) + await self._flush() + self._reader_task = asyncio.create_task(self._read(reader, writer)) + + async def _read(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None: + assert self._conn is not None + try: + while True: + data = await reader.read(65536) + if not data: + break + if self._goaway.done(): + # python-h2 refuses every frame once it has received a GOAWAY, unlike + # a real peer, which keeps draining its streams. Keep the socket open + # and let the SDK decide when the connection ends. + continue + try: + events = self._conn.receive_data(data) + except ProtocolError: + # A GOAWAY followed by more frames in the same read: python-h2 raises + # on the frame after the GOAWAY and discards the events before it. + if self._conn.state_machine.state != ConnectionState.CLOSED: + raise + if not self._goaway.done(): + self._goaway.set_result(None) + continue + for event in events: + self._dispatch(event) + await self._flush() + finally: + writer.close() + if not self._closed.done(): + self._closed.set_result(None) + for response in self._responses.values(): + if not response.ended.done(): + response.ended.set_exception(ConnectionError("session closed")) + if not response.started.done(): + response.started.set_exception(ConnectionError("session closed")) + + def _dispatch(self, event: typing.Any) -> None: + assert self._conn is not None + if isinstance(event, ResponseReceived): + response = self._responses[event.stream_id] + response.headers = typing.cast(typing.Dict[str, str], dict(event.headers)) + if not response.started.done(): + response.started.set_result(None) + if event.stream_id == self._handshake_stream and not self._handshake.done(): + self._handshake.set_result(response) + elif isinstance(event, DataReceived): + self._responses[event.stream_id].body.extend(event.data) + self._conn.acknowledge_received_data(event.flow_controlled_length, event.stream_id) + elif isinstance(event, TrailersReceived): + pairs = typing.cast(typing.List[typing.Tuple[str, str]], list(event.headers)) + self._responses[event.stream_id].trailers = dict(pairs) + self._responses[event.stream_id].trailer_pairs = pairs + elif isinstance(event, StreamEnded): + response = self._responses[event.stream_id] + if not response.ended.done(): + response.ended.set_result(None) + elif isinstance(event, StreamReset): + reset = self._responses.get(event.stream_id) + if reset is not None and not reset.ended.done(): + reset.ended.set_exception(ConnectionError(f"stream reset: {event.error_code}")) + elif isinstance(event, ConnectionTerminated): + if not self._goaway.done(): + self._goaway.set_result(None) + + async def _flush(self) -> None: + assert self._conn is not None and self._writer is not None + data = self._conn.data_to_send() + if data: + self._writer.write(data) + await self._writer.drain() + + async def handshake(self) -> Response: + """The SDK's answer to `GET /_/start-tunnel`: its credentials.""" + return await asyncio.wait_for(self._handshake, TEST_TIMEOUT) + + async def establish(self, name: str = TUNNEL_NAME, status: str = "ok") -> None: + """Send the handshake trailers: the verdict.""" + assert self._conn is not None + trailers = [("tunnel-status", status)] + if status == "ok": + trailers += [ + ("tunnel-name", name), + ("proxy-url", f"https://proxy.example/{ENVIRONMENT_ID}/{name}"), + ("tunnel-url", f"https://tunnel.example/{name}"), + ] + self._conn.send_headers(self._handshake_stream, trailers, end_stream=True) + await self._flush() + + async def request( + self, + path: str, + headers: typing.Optional[typing.List[typing.Tuple[str, str]]] = None, + body: bytes = b"", + method: str = "GET", + wait: bool = True, + ) -> Response: + """Forward a request over the session; `path` is the raw forwarded path.""" + assert self._conn is not None + stream_id = self._conn.get_next_available_stream_id() + response = Response() + response.stream_id = stream_id + self._responses[stream_id] = response + self._conn.send_headers( + stream_id, + [(":method", method), (":path", path), (":scheme", "http"), (":authority", "fake")] + (headers or []), + end_stream=not body, + ) + if body: + self._conn.send_data(stream_id, body, end_stream=True) + await self._flush() + if wait: + await asyncio.wait_for(response.ended, TEST_TIMEOUT) + return response + + async def reset(self, stream_id: int) -> None: + """Cancel a forwarded stream (RST_STREAM), as a timed-out invocation does.""" + assert self._conn is not None + self._conn.reset_stream(stream_id) + await self._flush() + + async def wait_closed(self) -> None: + await asyncio.wait_for(self._closed, TEST_TIMEOUT) + + async def wait_goaway(self) -> None: + """The SDK told this peer to stop opening streams (graceful shutdown).""" + await asyncio.wait_for(self._goaway, TEST_TIMEOUT) + + +# ───────────────────────────── helpers ───────────────────────────── + +greeter = restate.Service("Greeter") + + +@greeter.handler() +async def greet(ctx: restate.Context, name: str) -> str: + return f"Hello {name}" + + +def options_for(cloud: FakeCloud, **kwargs: typing.Any) -> typing.Any: + return _internal.TunnelEngineOptions.resolve( + name=kwargs.pop("name", TUNNEL_NAME), + key=kwargs.pop("key", connect_key()), + servers=cloud.servers, + worker_id="worker-test", + **kwargs, + ) + + +async def established(cloud: FakeCloud, app: typing.Any = None) -> typing.Tuple[typing.Any, typing.Any]: + app = app or restate.app(services=[greeter]) + engine = _internal.TunnelEngine(app, options_for(cloud)) + ready = asyncio.ensure_future(engine.ready()) + handshake = await cloud.handshake() + assert handshake.status == 200 + assert handshake.headers["authorization"] == f"Bearer {API_KEY}" + assert handshake.headers["environment-id"] == ENVIRONMENT_ID + assert handshake.headers["tunnel-name"] == TUNNEL_NAME + assert handshake.headers["tunnel-worker-id"] == "worker-test" + await asyncio.sleep(0) + assert not ready.done(), "ready completed before the handshake trailers" + await cloud.establish() + info = await asyncio.wait_for(ready, TEST_TIMEOUT) + return engine, info + + +# ───────────────────────────── configuration ───────────────────────────── + + +def test_options_resolve_from_arguments_and_errors(monkeypatch: pytest.MonkeyPatch): + for name in ("RESTATE_CONNECT_KEY", "RESTATE_CONNECT_NAME", "RESTATE_INPROC_ENVIRONMENT_ID"): + monkeypatch.delenv(name, raising=False) + + options = _internal.TunnelEngineOptions.resolve(name="greeter", key=connect_key()) + assert options.name == "greeter" + assert options.environment_id == ENVIRONMENT_ID + assert options.region == REGION + assert options.source == "code" + assert options.registration_identifier == "connect://greeter" + assert options.registration_deep_link.endswith("register-deployment=connect://greeter") + assert "secret" not in repr(options) + + with pytest.raises(_internal.ConnectError, match="needs a name"): + _internal.TunnelEngineOptions.resolve(key=connect_key()) + with pytest.raises(_internal.ConnectError, match="RESTATE_CONNECT_KEY"): + _internal.TunnelEngineOptions.resolve(name="greeter") + with pytest.raises(_internal.ConnectError, match="the key option is not a Restate Cloud key"): + _internal.TunnelEngineOptions.resolve(name="greeter", key="key_nope") + with pytest.raises(_internal.ConnectError, match="carries no API key"): + _internal.TunnelEngineOptions.resolve(name="greeter", key=connect_key(api_key=None)) + with pytest.raises(_internal.ConnectError, match="carries no signing key"): + _internal.TunnelEngineOptions.resolve(name="greeter", key=ingress_key()) + with pytest.raises(ValueError, match="unknown tuning option"): + _internal.TunnelEngineOptions.resolve(name="greeter", key=connect_key(), tuning={"nope": 1}) + + monkeypatch.setenv("RESTATE_CONNECT_KEY", connect_key()) + monkeypatch.setenv("RESTATE_CONNECT_NAME", "from-env") + options = _internal.TunnelEngineOptions.resolve() + assert options.name == "from-env" + assert options.source == "connect_key" + + +def test_ingress_key_and_client_resolution(monkeypatch: pytest.MonkeyPatch): + key = _internal.IngressKey.parse(ingress_key()) + assert key.environment_id == ENVIRONMENT_ID + assert key.ingress_url() == f"https://test123.env.{REGION}:8080" + assert key.bearer() == f"Bearer {API_KEY}" + assert "secret" not in repr(key) + with pytest.raises(_internal.ConnectError, match="RESTATE_INGRESS_KEY is not a Restate Cloud key"): + _internal.IngressKey.parse("nope", "RESTATE_INGRESS_KEY") + + monkeypatch.delenv("RESTATE_INGRESS_KEY", raising=False) + assert _internal.IngressKey.from_env() is None + assert resolve_ingress("http://localhost:8080", None) == ("http://localhost:8080", None) + with pytest.raises(_internal.ConnectError, match="RESTATE_INGRESS_KEY"): + resolve_ingress(None, None) + + monkeypatch.setenv("RESTATE_INGRESS_KEY", ingress_key()) + url, headers = resolve_ingress(None, {"x-extra": "1"}) + assert url == f"https://test123.env.{REGION}:8080" + assert headers == {"x-extra": "1", "Authorization": f"Bearer {API_KEY}"} + _, headers = resolve_ingress(None, {"authorization": "Bearer mine"}) + assert headers == {"authorization": "Bearer mine"} + + +def test_load_app_resolves_module_and_attribute(): + assert _load_app("restate:app") is restate.app + assert _load_app("restate.endpoint") is restate.app + with pytest.raises(ValueError, match="no attribute"): + _load_app("restate.connection:missing") + with pytest.raises(ValueError, match="cannot import"): + _load_app("definitely_not_a_module") + + +async def test_engine_requires_a_running_loop(): + options = _internal.TunnelEngineOptions.resolve(name="greeter", key=connect_key(), servers=["http://127.0.0.1:1"]) + + def outside() -> None: + with pytest.raises(RuntimeError, match="event loop"): + _internal.TunnelEngine(restate.app(services=[]), options) + + await asyncio.get_running_loop().run_in_executor(None, outside) + + +# ───────────────────────────── the engine through the ASGI bridge ───────────────────────────── + + +async def test_handshake_then_forwarded_discovery(): + async with FakeCloud() as cloud: + engine, info = await established(cloud) + assert info.name == TUNNEL_NAME + assert info.deployment_url == f"https://proxy.example:9080/{ENVIRONMENT_ID}/{TUNNEL_NAME}/http/in-process/9080/" + + response = await cloud.request("/http/h/9080/discover?x=1", identity_headers("/discover")) + assert response.status == 200 + assert response.headers["content-type"].startswith("application/vnd.restate.endpointmanifest") + assert response.headers["x-restate-server"].startswith("restate-sdk-python/") + manifest = json.loads(bytes(response.body)) + assert manifest["protocolMode"] == "BIDI_STREAM" + assert [service["name"] for service in manifest["services"]] == ["Greeter"] + + # Answered by the engine, unsigned. + assert (await cloud.request("/_/health")).status == 200 + # Identity is verified by the engine before the app sees the request. + assert (await cloud.request("/http/h/9080/discover", [("accept", "application/json")])).status == 401 + assert (await cloud.request("/http/h/9080/discover", identity_headers("/http/h/9080/discover"))).status == 401 + # The app's own routing after the rewrite. + assert (await cloud.request("/http/h/9080/restate/health", identity_headers("/restate/health"))).status == 200 + assert (await cloud.request("/http/h/9080/nope", identity_headers("/nope"))).status == 404 + + await engine.shutdown(0.0) + await cloud.wait_closed() + await engine.terminated() + + +async def test_invocation_reaches_the_handler_route(caplog: pytest.LogCaptureFixture): + async with FakeCloud() as cloud: + engine, _ = await established(cloud) + # An invocation without the service-protocol content type stops at the SDK's media + # type check, which raises inside the app: the request reached the handler route + # through the bridge, and an app exception is a 500 (as under any ASGI server). + response = await cloud.request( + "/http/h/9080/invoke/Greeter/greet", + identity_headers("/invoke/Greeter/greet") + [("content-type", "application/json")], + body=b'"world"', + method="POST", + ) + assert response.status == 500 + assert any("415" in record.getMessage() or "415" in str(record.exc_info) for record in caplog.records) + await engine.shutdown(0.0) + + +async def test_app_exceptions_become_500_and_resets(): + calls: typing.List[str] = [] + + async def failing_app(scope: typing.Any, receive: typing.Any, send: typing.Any) -> None: + calls.append(scope["path"]) + assert scope["type"] == "http" + assert scope["http_version"] == "2" + assert scope["scheme"] == "https" + assert "http.response.trailers" in scope["extensions"] + assert (b"x-restate-signature-scheme", b"v1") in scope["headers"] + event = await receive() + assert event == {"type": "http.request", "body": b"", "more_body": False} + if scope["path"] == "/my path": + assert scope["query_string"] == b"a=1" + assert scope["raw_path"] == b"/my%20path" + raise RuntimeError("boom before start") + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"partial", "more_body": True}) + raise RuntimeError("boom after start") + + async with FakeCloud() as cloud: + engine, _ = await established(cloud, failing_app) + # The identity audience is the raw forwarded path, as Restate signs it. + response = await cloud.request("/http/h/9080/my%20path?a=1", identity_headers("/my%20path")) + assert response.status == 500 + + response = await cloud.request("/http/h/9080/other", identity_headers("/other"), wait=False) + await asyncio.wait_for(response.started, TEST_TIMEOUT) + assert response.status == 200 + with pytest.raises(ConnectionError, match="stream reset"): + await asyncio.wait_for(response.ended, TEST_TIMEOUT) + assert calls == ["/my path", "/other"] + await engine.shutdown(0.0) + + +async def test_send_validation_and_trailers(): + errors: typing.List[str] = [] + + async def expect_error(coro: typing.Awaitable[None]) -> None: + try: + await coro + errors.append("no error") + except RuntimeError as error: + errors.append(str(error)) + + async def app(scope: typing.Any, receive: typing.Any, send: typing.Any) -> None: + if scope["path"] == "/bad-order": + await expect_error(send({"type": "http.response.body", "body": b"x"})) + await send({"type": "http.response.start", "status": 204, "headers": []}) + await expect_error(send({"type": "http.response.start", "status": 200, "headers": []})) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + # Even after the stream completed, the protocol is still enforced. + await asyncio.sleep(0.05) + await expect_error(send({"type": "http.response.body", "body": b""})) + return + if scope["path"] == "/trailers": + await send({"type": "http.response.start", "status": 200, "headers": [(b"x-a", b"1")], "trailers": True}) + for chunk in (b"one", b"two", b"three"): + await send({"type": "http.response.body", "body": chunk, "more_body": True}) + await send({"type": "http.response.body", "body": b"", "more_body": False}) + await send({"type": "http.response.trailers", "headers": [(b"x-t", b"1")], "more_trailers": True}) + await send({"type": "http.response.trailers", "headers": [(b"x-u", b"2"), (b"x-t", b"3"), (b"x-t", b"4")]}) + # The response is complete: receive reports the disconnect. + assert (await receive())["type"] == "http.disconnect" + return + raise AssertionError(scope["path"]) + + async with FakeCloud() as cloud: + engine, _ = await established(cloud, app) + response = await cloud.request("/http/h/9080/bad-order", identity_headers("/bad-order")) + assert response.status == 204 + await asyncio.sleep(0.2) + assert [e.split(",")[0] for e in errors] == [ + "Unexpected ASGI message 'http.response.body'", + "Unexpected ASGI message 'http.response.start'", + "Unexpected ASGI message 'http.response.body'", + ], errors + assert "no further message" in errors[2] + + response = await cloud.request("/http/h/9080/trailers", identity_headers("/trailers")) + assert response.status == 200 + assert response.headers["x-a"] == "1" + assert bytes(response.body) == b"onetwothree" + assert response.trailer_pairs == [("x-t", "1"), ("x-t", "3"), ("x-t", "4"), ("x-u", "2")] + await engine.shutdown(0.0) + + +async def test_request_body_is_streamed_and_disconnect_follows_completion(): + seen: typing.List[typing.Any] = [] + + async def app(scope: typing.Any, receive: typing.Any, send: typing.Any) -> None: + while True: + event = await receive() + seen.append(event) + if event["type"] == "http.request" and not event["more_body"]: + break + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"ok", "more_body": False}) + seen.append(await receive()) + # After the disconnect, sends are accepted and dropped. + await send({"type": "http.response.body", "body": b"late"}) + + async with FakeCloud() as cloud: + engine, _ = await established(cloud, app) + response = await cloud.request("/http/h/9080/echo", identity_headers("/echo"), body=b"payload", method="POST") + assert response.status == 200 + assert bytes(response.body) == b"ok" + await asyncio.sleep(0.1) + assert seen[0] == {"type": "http.request", "body": b"payload", "more_body": True} + assert seen[1] == {"type": "http.request", "body": b"", "more_body": False} + assert seen[2] == {"type": "http.disconnect"} + await engine.shutdown(0.0) + + +async def test_reset_before_the_response_head_delivers_disconnect(): + # A stream cancelled while the app has consumed the request but not yet started the + # response (hyper drops the request future): receive() must still report the disconnect. + outcome: asyncio.Future[typing.Any] = asyncio.get_running_loop().create_future() + consumed = asyncio.Event() + proceed = asyncio.Event() + + async def app(scope: typing.Any, receive: typing.Any, send: typing.Any) -> None: + assert (await receive())["more_body"] is False + consumed.set() + await proceed.wait() + try: + event = await asyncio.wait_for(receive(), TEST_TIMEOUT) + outcome.set_result(event) + except BaseException as error: # pylint: disable=broad-except + outcome.set_exception(error) + raise + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"late", "more_body": False}) + + async with FakeCloud() as cloud: + engine, _ = await established(cloud, app) + response = await cloud.request("/http/h/9080/slow-start", identity_headers("/slow-start"), wait=False) + await asyncio.wait_for(consumed.wait(), TEST_TIMEOUT) + await cloud.reset(response.stream_id) + await asyncio.sleep(0.1) + proceed.set() + assert await asyncio.wait_for(outcome, TEST_TIMEOUT) == {"type": "http.disconnect"} + # The late response is dropped silently and the engine is unaffected. + assert (await cloud.request("/_/health")).status == 200 + await engine.shutdown(0.0) + + +class StallingCloud(FakeCloud): + """A peer that never opens its flow-control window: the SDK's body frames queue up.""" + + def __init__(self) -> None: + super().__init__() + self.pending: typing.List[typing.Tuple[int, int]] = [] + + async def __aenter__(self) -> "StallingCloud": + await super().__aenter__() + return self + + def _dispatch(self, event: typing.Any) -> None: + if isinstance(event, DataReceived): + self._responses[event.stream_id].body.extend(event.data) + self.pending.append((event.flow_controlled_length, event.stream_id)) + return + super()._dispatch(event) + + async def open_window(self) -> None: + assert self._conn is not None + for length, stream_id in self.pending: + self._conn.acknowledge_received_data(length, stream_id) + self.pending.clear() + await self._flush() + + +async def test_app_failure_with_a_buffered_frame_resets_the_stream(): + # The peer stalls; the app has frames queued in hyper and in the bridge when it + # raises. The stream must end with a reset, never as a complete response. + raised = asyncio.Event() + + async def app(scope: typing.Any, receive: typing.Any, send: typing.Any) -> None: + await send({"type": "http.response.start", "status": 200, "headers": []}) + await send({"type": "http.response.body", "body": b"a" * 70000, "more_body": True}) + await send({"type": "http.response.body", "body": b"b" * 10, "more_body": True}) + await send({"type": "http.response.body", "body": b"c" * 10, "more_body": True}) + raised.set() + raise RuntimeError("boom with a buffered frame") + + async with StallingCloud() as cloud: + engine, _ = await established(cloud, app) + response = await cloud.request("/http/h/9080/stall", identity_headers("/stall"), wait=False) + await asyncio.wait_for(raised.wait(), TEST_TIMEOUT) + await asyncio.sleep(0.2) + await cloud.open_window() + with pytest.raises(ConnectionError, match="stream reset"): + await asyncio.wait_for(response.ended, TEST_TIMEOUT) + await engine.shutdown(0.0) + + +async def test_graceful_shutdown_waits_for_in_flight_invocations(): + # The drain protocol itself (sentinel for raced streams, flushing, the deadline) is + # covered by the engine's own tests; python-h2 cannot speak after a GOAWAY, so here + # only the observable contract is checked: shutdown waits for the in-flight + # invocation, tells the peer to stop, and the grace deadline bounds the wait. + release = asyncio.Event() + completed = asyncio.Event() + + async def app(scope: typing.Any, receive: typing.Any, send: typing.Any) -> None: + await receive() + await send({"type": "http.response.start", "status": 200, "headers": []}) + await release.wait() + await send({"type": "http.response.body", "body": b"done", "more_body": False}) + completed.set() + + async with FakeCloud() as cloud: + engine, _ = await established(cloud, app) + in_flight = await cloud.request("/http/h/9080/slow", identity_headers("/slow"), wait=False) + await asyncio.wait_for(in_flight.started, TEST_TIMEOUT) + + started_at = time.monotonic() + shutdown = asyncio.ensure_future(engine.shutdown(1.5)) + await cloud.wait_goaway() + await asyncio.sleep(0.2) + assert not shutdown.done(), "shutdown completed while an invocation was in flight" + + release.set() + await asyncio.wait_for(completed.wait(), TEST_TIMEOUT) + await asyncio.wait_for(shutdown, TEST_TIMEOUT) + # The peer never acknowledged the shutdown ping, so the deadline ended the wait. + assert 1.0 <= time.monotonic() - started_at <= 4.0 + await cloud.wait_closed() + await engine.terminated() + + +async def test_rejected_credentials_are_fatal(): + async with FakeCloud() as cloud: + engine = _internal.TunnelEngine(restate.app(services=[greeter]), options_for(cloud)) + ready = asyncio.ensure_future(engine.ready()) + await cloud.handshake() + await cloud.establish(status="unauthorized") + with pytest.raises(_internal.ConnectError, match="unauthorized"): + await asyncio.wait_for(ready, TEST_TIMEOUT) + with pytest.raises(_internal.ConnectError, match="unauthorized"): + await engine.terminated() + + +async def test_dropping_the_engine_closes_the_connection(): + async with FakeCloud() as cloud: + engine, _ = await established(cloud) + del engine + await cloud.wait_closed() + + +# ───────────────────────────── restate.connect() and Connection ───────────────────────────── + + +class OptionsWithServers: + """Stands in for `TunnelEngineOptions` in `restate.connection`, pointing at the fake peer.""" + + def __init__(self, cloud: FakeCloud, **tuning: typing.Any) -> None: + self._cloud = cloud + self._tuning = tuning + + def resolve(self, **kwargs: typing.Any) -> typing.Any: + return _internal.TunnelEngineOptions.resolve( + servers=self._cloud.servers, worker_id="worker-test", tuning=self._tuning or None, **kwargs + ) + + +async def test_with_connection(monkeypatch: pytest.MonkeyPatch): + async with FakeCloud() as cloud: + monkeypatch.setattr(sys.modules["restate.connection"], "TunnelEngineOptions", OptionsWithServers(cloud)) + + async def run() -> None: + async with restate.with_connection( + restate.app(services=[greeter]), name=TUNNEL_NAME, key=connect_key() + ) as conn: + assert conn.name == TUNNEL_NAME + assert conn.registration_identifier == f"connect://{TUNNEL_NAME}" + assert conn.info.name == TUNNEL_NAME + response = await cloud.request("/http/h/9080/discover", identity_headers("/discover")) + assert response.status == 200 + waiter = asyncio.ensure_future(conn.wait()) + await asyncio.sleep(0.05) + assert not waiter.done() + await conn.shutdown(0.0) + await asyncio.wait_for(waiter, TEST_TIMEOUT) + # Leaving the block shuts down again: a no-op that never raises. + + task = asyncio.ensure_future(run()) + await cloud.handshake() + await cloud.establish() + await asyncio.wait_for(task, TEST_TIMEOUT) + await cloud.wait_closed() + + +async def test_with_connection_leaves_the_block_with_a_drain(monkeypatch: pytest.MonkeyPatch): + async with FakeCloud() as cloud: + monkeypatch.setattr(sys.modules["restate.connection"], "TunnelEngineOptions", OptionsWithServers(cloud)) + + async def run() -> None: + async with restate.with_connection(restate.app(services=[greeter]), name=TUNNEL_NAME, key=connect_key()): + pass + + task = asyncio.ensure_future(run()) + await cloud.handshake() + await cloud.establish() + # Leaving the block drains: the peer is told to stop opening streams. + await cloud.wait_goaway() + assert not task.done() + # python-h2 cannot finish the drain handshake; closing the peer ends the connection. + await asyncio.wait_for(task, TEST_TIMEOUT) + + +async def test_connection_reports_configuration_errors(monkeypatch: pytest.MonkeyPatch): + monkeypatch.delenv("RESTATE_CONNECT_KEY", raising=False) + monkeypatch.delenv("RESTATE_INPROC_ENVIRONMENT_ID", raising=False) + app = restate.app(services=[greeter]) + with pytest.raises(restate.ConnectError, match="RESTATE_CONNECT_KEY"): + async with restate.with_connection(app, name="greeter"): + pass + with pytest.raises(restate.ConnectError, match="RESTATE_CONNECT_KEY"): + await restate.connect(app, name="greeter") + + +async def test_with_connection_fails_fast_after_a_fatal_error(monkeypatch: pytest.MonkeyPatch): + async with FakeCloud() as cloud: + monkeypatch.setattr(sys.modules["restate.connection"], "TunnelEngineOptions", OptionsWithServers(cloud)) + + async def run() -> None: + async with restate.with_connection(restate.app(services=[greeter]), name=TUNNEL_NAME, key=connect_key()): + raise AssertionError("must not enter") + + task = asyncio.ensure_future(run()) + await cloud.handshake() + await cloud.establish(status="unauthorized") + with pytest.raises(restate.ConnectError, match="unauthorized"): + await asyncio.wait_for(task, TEST_TIMEOUT) + + +async def test_connect_runs_until_the_connection_stops(monkeypatch: pytest.MonkeyPatch): + # The fake peer going away (with a reconnect policy that gives up at once) is what + # stops connect() here, as a ConnectError; signals are covered manually. + async with FakeCloud() as cloud: + monkeypatch.setattr( + sys.modules["restate.connection"], + "TunnelEngineOptions", + OptionsWithServers(cloud, reconnect_max_attempts=1), + ) + task = asyncio.ensure_future( + restate.connect(restate.app(services=[greeter]), name=TUNNEL_NAME, key=connect_key()) + ) + await cloud.handshake() + await cloud.establish() + response = await cloud.request("/http/h/9080/discover", identity_headers("/discover")) + assert response.status == 200 + assert not task.done() + + # Leaving the block closes the peer: the engine gives up and connect() raises. + with pytest.raises(restate.ConnectError, match="max_attempts"): + await asyncio.wait_for(task, TEST_TIMEOUT)