diff --git a/Cargo.lock b/Cargo.lock index 4fc1f75..d9d8b86 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -23,33 +23,12 @@ version = "1.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ace50bade8e6234aa140d9a2f552bbee1db4d353f69b8217bc503490fc1a9f26" -[[package]] -name = "base16ct" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4c7f02d4ea65f2c1853089ffd8d2787bdbc63de2f0d29dedbcf8ccdfa0ccd4cf" - [[package]] name = "base64" version = "0.22.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "72b3254f16251a8381aa12e40e3c4d2f0199f8c6508fbecb9d91f575e0fbb8c6" -[[package]] -name = "base64ct" -version = "1.8.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" - -[[package]] -name = "block-buffer" -version = "0.10.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71" -dependencies = [ - "generic-array", -] - [[package]] name = "block-buffer" version = "0.12.0" @@ -96,12 +75,6 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "baf1de4339761588bc0619e3cbc0120ee582ebb74b53b4efbf79117bd2da40fd" -[[package]] -name = "const-oid" -version = "0.9.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c2459377285ad874054d797f3ccebf984978aa39129f6eafde5cdc8315b612f8" - [[package]] name = "const-oid" version = "0.10.2" @@ -110,35 +83,13 @@ checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" [[package]] name = "cpufeatures" -version = "0.2.16" +version = "0.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "16b80225097f2e5ae4e7179dd2266824648f3e2f49d9134d584b76389d31c4c3" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" dependencies = [ "libc", ] -[[package]] -name = "crypto-bigint" -version = "0.5.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76" -dependencies = [ - "generic-array", - "rand_core", - "subtle", - "zeroize", -] - -[[package]] -name = "crypto-common" -version = "0.1.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3" -dependencies = [ - "generic-array", - "typenum", -] - [[package]] name = "crypto-common" version = "0.2.1" @@ -148,103 +99,15 @@ dependencies = [ "hybrid-array", ] -[[package]] -name = "curve25519-dalek" -version = "4.1.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "97fb8b7c4503de7d6ae7b42ab72a5a59857b4c937ec27a3d4539dba95b5ab2be" -dependencies = [ - "cfg-if", - "cpufeatures", - "curve25519-dalek-derive", - "digest 0.10.7", - "fiat-crypto", - "rustc_version", - "subtle", - "zeroize", -] - -[[package]] -name = "curve25519-dalek-derive" -version = "0.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f46882e17999c6cc590af592290432be3bce0428cb0d5f8b6715e4dc7b383eb3" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "der" -version = "0.7.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7c1832837b905bbfb5101e07cc24c8deddf52f93225eee6ead5f4d63d53ddcb" -dependencies = [ - "const-oid 0.9.6", - "pem-rfc7468", - "zeroize", -] - -[[package]] -name = "digest" -version = "0.10.7" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292" -dependencies = [ - "block-buffer 0.10.4", - "const-oid 0.9.6", - "crypto-common 0.1.6", - "subtle", -] - [[package]] name = "digest" version = "0.11.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4850db49bf08e663084f7fb5c87d202ef91a3907271aff24a94eb97ff039153c" dependencies = [ - "block-buffer 0.12.0", - "const-oid 0.10.2", - "crypto-common 0.2.1", -] - -[[package]] -name = "ecdsa" -version = "0.16.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ee27f32b5c5292967d2d4a9d7f1e0b0aed2c15daded5a60300e4abb9d8020bca" -dependencies = [ - "der", - "digest 0.10.7", - "elliptic-curve", - "rfc6979", - "signature", - "spki", -] - -[[package]] -name = "ed25519" -version = "2.2.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "115531babc129696a58c64a4fef0a8bf9e9698629fb97e9e40767d235cfbcd53" -dependencies = [ - "pkcs8", - "signature", -] - -[[package]] -name = "ed25519-dalek" -version = "2.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "70e796c081cee67dc755e1a36a0a172b897fab85fc3f6bc48307991f64e4eca9" -dependencies = [ - "curve25519-dalek", - "ed25519", - "serde", - "sha2 0.10.9", - "subtle", - "zeroize", + "block-buffer", + "const-oid", + "crypto-common", ] [[package]] @@ -253,54 +116,6 @@ version = "1.13.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "60b1af1c220855b6ceac025d3f6ecdd2b7c4894bfe9cd9bda4fbb4bc7c0d4cf0" -[[package]] -name = "elliptic-curve" -version = "0.13.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b5e6043086bf7973472e0c7dff2142ea0b680d30e18d9cc40f267efbf222bd47" -dependencies = [ - "base16ct", - "crypto-bigint", - "digest 0.10.7", - "ff", - "generic-array", - "group", - "hkdf", - "pem-rfc7468", - "pkcs8", - "rand_core", - "sec1", - "subtle", - "zeroize", -] - -[[package]] -name = "ff" -version = "0.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c0b50bfb653653f9ca9095b427bed08ab8d75a137839d9ad64eb11810d5b6393" -dependencies = [ - "rand_core", - "subtle", -] - -[[package]] -name = "fiat-crypto" -version = "0.2.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" - -[[package]] -name = "generic-array" -version = "0.14.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2" -dependencies = [ - "typenum", - "version_check", - "zeroize", -] - [[package]] name = "getrandom" version = "0.2.15" @@ -312,41 +127,12 @@ dependencies = [ "wasi", ] -[[package]] -name = "group" -version = "0.13.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f0f9ef7462f7c099f518d754361858f86d8a07af53ba9af0fe635bbccb151a63" -dependencies = [ - "ff", - "rand_core", - "subtle", -] - [[package]] name = "heck" version = "0.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" -[[package]] -name = "hkdf" -version = "0.12.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7b5f8eb2ad728638ea2c7d47a21db23b7b58a72ed6a38256b8a1849f15fbbdf7" -dependencies = [ - "hmac", -] - -[[package]] -name = "hmac" -version = "0.12.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6c49c37c09c17a53d937dfbb742eb3a961d65a994e6bcdcf37e7399d0cc8ab5e" -dependencies = [ - "digest 0.10.7", -] - [[package]] name = "hybrid-array" version = "0.4.8" @@ -394,17 +180,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0529410abe238729a60b108898784df8984c87f6054c9c4fcacc47e4803c1ce1" dependencies = [ "base64", - "ed25519-dalek", "getrandom", - "hmac", "js-sys", - "p256", - "p384", - "rand", - "rsa", "serde", "serde_json", - "sha2 0.10.9", "signature", ] @@ -413,9 +192,6 @@ name = "lazy_static" version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe" -dependencies = [ - "spin", -] [[package]] name = "libc" @@ -423,12 +199,6 @@ version = "0.2.169" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b5aba8db14291edd000dfcc4d620c7ebfb122c613afb886ca8803fa4e128a20a" -[[package]] -name = "libm" -version = "0.2.16" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" - [[package]] name = "log" version = "0.4.25" @@ -469,52 +239,6 @@ dependencies = [ "winapi", ] -[[package]] -name = "num-bigint-dig" -version = "0.8.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e661dda6640fad38e827a6d4a310ff4763082116fe217f279885c97f511bb0b7" -dependencies = [ - "lazy_static", - "libm", - "num-integer", - "num-iter", - "num-traits", - "rand", - "smallvec", - "zeroize", -] - -[[package]] -name = "num-integer" -version = "0.1.46" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7969661fd2958a5cb096e56c8e1ad0444ac2bbcd0061bd28660485a44879858f" -dependencies = [ - "num-traits", -] - -[[package]] -name = "num-iter" -version = "0.1.45" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1429034a0490724d0075ebb2bc9e875d6503c3cf69e235a8941aa757d83ef5bf" -dependencies = [ - "autocfg", - "num-integer", - "num-traits", -] - -[[package]] -name = "num-traits" -version = "0.2.19" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "071dfc062690e90b734c0b2273ce72ad0ffa95f0c74596bc250dcfd960262841" -dependencies = [ - "autocfg", - "libm", -] - [[package]] name = "once_cell" version = "1.20.2" @@ -527,96 +251,24 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b15813163c1d831bf4a13c3610c05c0d03b39feb07f7e09fa234dac9b15aaf39" -[[package]] -name = "p256" -version = "0.13.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c9863ad85fa8f4460f9c48cb909d38a0d689dba1f6f6988a5e3e0d31071bcd4b" -dependencies = [ - "ecdsa", - "elliptic-curve", - "primeorder", - "sha2 0.10.9", -] - -[[package]] -name = "p384" -version = "0.13.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fe42f1670a52a47d448f14b6a5c61dd78fce51856e68edaa38f7ae3a46b8d6b6" -dependencies = [ - "ecdsa", - "elliptic-curve", - "primeorder", - "sha2 0.10.9", -] - [[package]] name = "pastey" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b867cad97c0791bbd3aaa6472142568c6c9e8f71937e98379f584cfb0cf35bec" -[[package]] -name = "pem-rfc7468" -version = "0.7.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88b39c9bfcfc231068454382784bb460aae594343fb030d46e9f50a645418412" -dependencies = [ - "base64ct", -] - [[package]] name = "pin-project-lite" version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" -[[package]] -name = "pkcs1" -version = "0.7.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f" -dependencies = [ - "der", - "pkcs8", - "spki", -] - -[[package]] -name = "pkcs8" -version = "0.10.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f950b2377845cebe5cf8b5165cb3cc1a5e0fa5cfa3e1f7f55707d8fd82e0a7b7" -dependencies = [ - "der", - "spki", -] - [[package]] name = "portable-atomic" version = "1.10.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "280dc24453071f1b63954171985a0b0d30058d287960968b9b2aca264c8d4ee6" -[[package]] -name = "ppv-lite86" -version = "0.2.21" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "85eae3c4ed2f50dcfe72643da4befc30deadb458a9b590d720cde2f2b1e97da9" -dependencies = [ - "zerocopy", -] - -[[package]] -name = "primeorder" -version = "0.13.6" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "353e1ca18966c16d9deb1c69278edbc5f194139612772bd9537af60ac231e1e6" -dependencies = [ - "elliptic-curve", -] - [[package]] name = "proc-macro2" version = "1.0.93" @@ -720,27 +372,6 @@ dependencies = [ "proc-macro2", ] -[[package]] -name = "rand" -version = "0.8.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404" -dependencies = [ - "libc", - "rand_chacha", - "rand_core", -] - -[[package]] -name = "rand_chacha" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e6c10a63a0fa32252be49d21e7709d4d4baf8d231c2dbce1eaa8141b9b127d88" -dependencies = [ - "ppv-lite86", - "rand_core", -] - [[package]] name = "rand_core" version = "0.6.4" @@ -805,9 +436,8 @@ dependencies = [ [[package]] name = "restate-sdk-shared-core" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5f0417331c92f9ef4a14dcbd40759cb4008221a3538affda05f907923c574119" +version = "0.10.0" +source = "git+https://github.com/restatedev/sdk-shared-core.git?rev=d8a42ecceab6e7874138b6316e128a09f2de76d1#d8a42ecceab6e7874138b6316e128a09f2de76d1" dependencies = [ "base64", "bs58", @@ -817,51 +447,12 @@ dependencies = [ "pastey", "prost", "serde", - "sha2 0.11.0-rc.5", + "sha2", "strum", "thiserror", "tracing", ] -[[package]] -name = "rfc6979" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f8dd2a808d456c4a54e300a23e9f5a67e122c3024119acbfd73e3bf664491cb2" -dependencies = [ - "hmac", - "subtle", -] - -[[package]] -name = "rsa" -version = "0.9.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d" -dependencies = [ - "const-oid 0.9.6", - "digest 0.10.7", - "num-bigint-dig", - "num-integer", - "num-traits", - "pkcs1", - "pkcs8", - "rand_core", - "signature", - "spki", - "subtle", - "zeroize", -] - -[[package]] -name = "rustc_version" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cfcb3a22ef46e85b45de6ee7e79d063319ebb6594faafcf1c225ea92ab6e9b92" -dependencies = [ - "semver", -] - [[package]] name = "rustversion" version = "1.0.19" @@ -874,26 +465,6 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f3cb5ba0dc43242ce17de99c180e96db90b235b8a9fdc9543c96d2209116bd9f" -[[package]] -name = "sec1" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc" -dependencies = [ - "base16ct", - "der", - "generic-array", - "pkcs8", - "subtle", - "zeroize", -] - -[[package]] -name = "semver" -version = "1.0.27" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d767eb0aabc880b29956c35734170f26ed551a859dbd361d140cdbeca61ab1e2" - [[package]] name = "serde" version = "1.0.228" @@ -938,24 +509,13 @@ dependencies = [ [[package]] name = "sha2" -version = "0.10.9" +version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +checksum = "446ba717509524cb3f22f17ecc096f10f4822d76ab5c0b9822c5f9c284e825f4" dependencies = [ "cfg-if", "cpufeatures", - "digest 0.10.7", -] - -[[package]] -name = "sha2" -version = "0.11.0-rc.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7c5f3b1e2dc8aad28310d8410bd4d7e180eca65fca176c52ab00d364475d0024" -dependencies = [ - "cfg-if", - "cpufeatures", - "digest 0.11.2", + "digest", ] [[package]] @@ -973,7 +533,6 @@ version = "2.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "77549399552de45a898a580c1b41d445bf730df867cc44e6c0233bbc4b8329de" dependencies = [ - "digest 0.10.7", "rand_core", ] @@ -983,22 +542,6 @@ version = "1.13.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3c5e1a9a646d36c3599cd173a41282daf47c44583ad367b8e6837255952e5c67" -[[package]] -name = "spin" -version = "0.9.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6980e8d7511241f8acf4aebddbb1ff938df5eebe98691418c4468d0b72a96a67" - -[[package]] -name = "spki" -version = "0.7.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d91ed6c858b01f942cd56b37a94b3e0a1798290327d1236e4d9cf4eaca44d29d" -dependencies = [ - "base64ct", - "der", -] - [[package]] name = "strum" version = "0.27.1" @@ -1021,12 +564,6 @@ dependencies = [ "syn", ] -[[package]] -name = "subtle" -version = "2.6.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292" - [[package]] name = "syn" version = "2.0.96" @@ -1174,12 +711,6 @@ version = "0.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d" -[[package]] -name = "version_check" -version = "0.9.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0b928f33d975fc6ad9f86c8f283853ad26bdd5b10b7f1542aa2fa15e2289105a" - [[package]] name = "wasi" version = "0.11.0+wasi-snapshot-preview1" @@ -1264,29 +795,3 @@ 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 = "zerocopy" -version = "0.8.27" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0894878a5fa3edfd6da3f88c4805f4c8558e2b996227a3d864f47fe11e38282c" -dependencies = [ - "zerocopy-derive", -] - -[[package]] -name = "zerocopy-derive" -version = "0.8.27" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88d2b8d9c68ad2b9e4340d7832716a4d21a22a1154777ad56ea55c51a9cf3831" -dependencies = [ - "proc-macro2", - "quote", - "syn", -] - -[[package]] -name = "zeroize" -version = "1.8.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0" diff --git a/Cargo.toml b/Cargo.toml index 7e4b467..7abc1c4 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,4 +14,4 @@ doc = false [dependencies] pyo3 = { version = "0.25.1", features = ["extension-module"] } tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] } -restate-sdk-shared-core = { version = "=0.9.0", features = ["request_identity", "sha2_random_seed"] } +restate-sdk-shared-core = { git = "https://github.com/restatedev/sdk-shared-core.git", rev = "d8a42ecceab6e7874138b6316e128a09f2de76d1", features = ["request_identity", "sha2_random_seed"] } diff --git a/python/restate/asyncio.py b/python/restate/asyncio.py index 7c8d18f..8fa5bf5 100644 --- a/python/restate/asyncio.py +++ b/python/restate/asyncio.py @@ -16,6 +16,11 @@ from restate.exceptions import TerminalError from restate.context import RestateDurableFuture from restate.server_context import ServerDurableFuture, ServerInvocationContext +from restate.vm import ( + AllCompletedUnresolvedFuture, + FirstCompletedUnresolvedFuture, + SingleUnresolvedFuture, +) async def gather(*futures: RestateDurableFuture[Any]) -> List[RestateDurableFuture[Any]]: @@ -24,9 +29,28 @@ async def gather(*futures: RestateDurableFuture[Any]) -> List[RestateDurableFutu Returns a list of all futures. """ - async for _ in as_completed(*futures): - pass - return list(futures) + context: ServerInvocationContext | None = None + handles: List[int] = [] + futures_list = list(futures) + + if not futures_list: + return [] + for f in futures_list: + if not isinstance(f, ServerDurableFuture): + raise TerminalError("All futures must SDK created futures.") + if context is None: + context = f.context + elif context is not f.context: + raise TerminalError("All futures must be created by the same SDK context.") + if not f.is_completed(): + handles.append(f.handle) + + if handles: + assert context is not None + await context.create_poll_or_cancel_coroutine( + AllCompletedUnresolvedFuture([SingleUnresolvedFuture(h) for h in handles]) + ) + return futures_list async def select(**kws: RestateDurableFuture[Any]) -> List[Any]: @@ -118,7 +142,9 @@ async def wait_completed( completed = [] uncompleted = [] assert context is not None - await context.create_poll_or_cancel_coroutine(handles) + await context.create_poll_or_cancel_coroutine( + FirstCompletedUnresolvedFuture([SingleUnresolvedFuture(h) for h in handles]) + ) for index, handle in enumerate(handles): future = futures[index] diff --git a/python/restate/discovery.py b/python/restate/discovery.py index 2c90c65..b6a0d95 100644 --- a/python/restate/discovery.py +++ b/python/restate/discovery.py @@ -429,4 +429,4 @@ def compute_discovery(endpoint: RestateEndpoint, discovered_as: typing.Literal[" protocol_mode = PROTOCOL_MODES[endpoint.protocol] else: protocol_mode = PROTOCOL_MODES[discovered_as] - return Endpoint(protocolMode=protocol_mode, minProtocolVersion=5, maxProtocolVersion=6, services=services) + return Endpoint(protocolMode=protocol_mode, minProtocolVersion=5, maxProtocolVersion=7, services=services) diff --git a/python/restate/server_context.py b/python/restate/server_context.py index a7de250..721b29b 100644 --- a/python/restate/server_context.py +++ b/python/restate/server_context.py @@ -61,9 +61,10 @@ from restate.vm import ( DoProgressAnyCompleted, DoProgressCancelSignalReceived, - DoProgressReadFromInput, + DoProgressWaitExternalProgress, DoProgressExecuteRun, - DoWaitPendingRun, + SingleUnresolvedFuture, + UnresolvedFuture, ) logger = logging.getLogger(__name__) @@ -193,7 +194,7 @@ def __init__(self, context: "ServerInvocationContext", handle: int) -> None: async def coro() -> str: if not context.vm.is_completed(handle): - await context.create_poll_or_cancel_coroutine([handle]) + await context.create_poll_or_cancel_coroutine(SingleUnresolvedFuture(handle)) invocation_id = await context.must_take_notification(handle) return typing.cast(str, invocation_id) @@ -235,7 +236,7 @@ def resolve(self, value: Any) -> Awaitable[None]: async def await_point(): if not self.server_context.vm.is_completed(handle): - await self.server_context.create_poll_or_cancel_coroutine([handle]) + await self.server_context.create_poll_or_cancel_coroutine(SingleUnresolvedFuture(handle)) await self.server_context.must_take_notification(handle) return ServerDurableFuture(self.server_context, handle, await_point) @@ -248,7 +249,7 @@ def reject(self, message: str, code: int = 500) -> Awaitable[None]: async def await_point(): if not self.server_context.vm.is_completed(handle): - await self.server_context.create_poll_or_cancel_coroutine([handle]) + await self.server_context.create_poll_or_cancel_coroutine(SingleUnresolvedFuture(handle)) await self.server_context.must_take_notification(handle) return ServerDurableFuture(self.server_context, handle, await_point) @@ -527,11 +528,11 @@ async def must_take_notification(self, handle): raise TerminalError(res.message, res.code) return res - async def create_poll_or_cancel_coroutine(self, handles: typing.List[int]) -> None: - """Create a coroutine to poll the handle.""" + async def create_poll_or_cancel_coroutine(self, unresolved_future: UnresolvedFuture) -> None: + """Create a coroutine to poll the unresolved future.""" while True: await self.take_and_send_output() - do_progress_response = self.vm.do_progress(handles) + do_progress_response = self.vm.do_progress(unresolved_future) if isinstance(do_progress_response, BaseException): logger.exception("Exception in do_progress", exc_info=do_progress_response) raise SdkInternalException() from do_progress_response @@ -556,7 +557,7 @@ async def wrapper(f): task = asyncio.create_task(wrapper(fn)) self.tasks.add(task) continue - if isinstance(do_progress_response, (DoWaitPendingRun, DoProgressReadFromInput)): + if isinstance(do_progress_response, DoProgressWaitExternalProgress): chunk = await self.receive() if chunk.get("type") == "restate.run_completed": continue @@ -574,7 +575,7 @@ def _create_fetch_result_coroutine(self, handle: int, serde: Serde[T] | None = N async def fetch_result(): if not self.vm.is_completed(handle): - await self.create_poll_or_cancel_coroutine([handle]) + await self.create_poll_or_cancel_coroutine(SingleUnresolvedFuture(handle)) res = await self.must_take_notification(handle) if res is None or serde is None: return res @@ -593,7 +594,7 @@ def create_sleep_future(self, handle: int) -> ServerDurableSleepFuture: async def transform(): if not self.vm.is_completed(handle): - await self.create_poll_or_cancel_coroutine([handle]) + await self.create_poll_or_cancel_coroutine(SingleUnresolvedFuture(handle)) await self.must_take_notification(handle) return ServerDurableSleepFuture(self, handle, transform) @@ -605,7 +606,7 @@ def create_call_future( async def inv_id_factory(): if not self.vm.is_completed(invocation_id_handle): - await self.create_poll_or_cancel_coroutine([invocation_id_handle]) + await self.create_poll_or_cancel_coroutine(SingleUnresolvedFuture(invocation_id_handle)) return await self.must_take_notification(invocation_id_handle) return ServerCallDurableFuture(self, handle, self._create_fetch_result_coroutine(handle, serde), inv_id_factory) diff --git a/python/restate/vm.py b/python/restate/vm.py index 3250a0a..2efaa8b 100644 --- a/python/restate/vm.py +++ b/python/restate/vm.py @@ -15,10 +15,10 @@ # pylint: disable=E1101,R0917 # pylint: disable=too-many-arguments # pylint: disable=too-few-public-methods -from typing import Optional +from typing import List, Optional, Union from datetime import timedelta -from dataclasses import dataclass +from dataclasses import dataclass, field import typing from restate._internal import ( PyVM, @@ -30,10 +30,10 @@ PyStateKeys, PyExponentialRetryConfig, PyDoProgressAnyCompleted, - PyDoProgressReadFromInput, + PyDoProgressWaitExternalProgress, PyDoProgressExecuteRun, - PyDoWaitForPendingRun, PyDoProgressCancelSignalReceived, + PyUnresolvedFuture, CANCEL_NOTIFICATION_HANDLE, ) # pylint: disable=import-error,no-name-in-module,line-too-long @@ -105,9 +105,10 @@ class DoProgressAnyCompleted: """ -class DoProgressReadFromInput: +class DoProgressWaitExternalProgress: """ - Represents a notification that the input needs to be read. + Represents a notification that external progress is required + (either new input from the server or a pending run proposal). """ @@ -128,26 +129,57 @@ class DoProgressCancelSignalReceived: """ -class DoWaitPendingRun: - """ - Represents a notification that a run is pending - """ - - DO_PROGRESS_ANY_COMPLETED = DoProgressAnyCompleted() -DO_PROGRESS_READ_FROM_INPUT = DoProgressReadFromInput() +DO_PROGRESS_WAIT_EXTERNAL_PROGRESS = DoProgressWaitExternalProgress() DO_PROGRESS_CANCEL_SIGNAL_RECEIVED = DoProgressCancelSignalReceived() -DO_WAIT_PENDING_RUN = DoWaitPendingRun() DoProgressResult = typing.Union[ DoProgressAnyCompleted, - DoProgressReadFromInput, + DoProgressWaitExternalProgress, DoProgressExecuteRun, DoProgressCancelSignalReceived, - DoWaitPendingRun, ] +@dataclass(frozen=True) +class SingleUnresolvedFuture: + """A single leaf handle.""" + + handle: int + + +@dataclass(frozen=True) +class FirstCompletedUnresolvedFuture: + """first child to complete (success or failure) wins.""" + + children: List["UnresolvedFuture"] = field(default_factory=list) + + +@dataclass(frozen=True) +class AllCompletedUnresolvedFuture: + """wait for all children to complete.""" + + children: List["UnresolvedFuture"] = field(default_factory=list) + + +UnresolvedFuture = Union[ + SingleUnresolvedFuture, + FirstCompletedUnresolvedFuture, + AllCompletedUnresolvedFuture, +] + + +def _unresolved_future_to_pyo3(uf: UnresolvedFuture) -> PyUnresolvedFuture: + """Recursively convert a Python-side UnresolvedFuture dataclass to its PyO3 pyclass.""" + if isinstance(uf, SingleUnresolvedFuture): + return PyUnresolvedFuture.single(uf.handle) + if isinstance(uf, FirstCompletedUnresolvedFuture): + return PyUnresolvedFuture.first_completed([_unresolved_future_to_pyo3(c) for c in uf.children]) + if isinstance(uf, AllCompletedUnresolvedFuture): + return PyUnresolvedFuture.all_completed([_unresolved_future_to_pyo3(c) for c in uf.children]) + raise TypeError(f"Unknown UnresolvedFuture variant: {type(uf).__name__}") + + # pylint: disable=too-many-public-methods class VMWrapper: """ @@ -195,24 +227,22 @@ def is_completed(self, handle: int) -> bool: return self.vm.is_completed(handle) # pylint: disable=R0911 - def do_progress(self, handles: list[int]) -> typing.Union[DoProgressResult, Exception, Suspended]: + def do_progress(self, unresolved_future: UnresolvedFuture) -> typing.Union[DoProgressResult, Exception, Suspended]: """Do progress with notifications.""" try: - result = self.vm.do_progress(handles) + result = self.vm.do_progress(_unresolved_future_to_pyo3(unresolved_future)) except VMException as e: return e if isinstance(result, PySuspended): return SUSPENDED if isinstance(result, PyDoProgressAnyCompleted): return DO_PROGRESS_ANY_COMPLETED - if isinstance(result, PyDoProgressReadFromInput): - return DO_PROGRESS_READ_FROM_INPUT + if isinstance(result, PyDoProgressWaitExternalProgress): + return DO_PROGRESS_WAIT_EXTERNAL_PROGRESS if isinstance(result, PyDoProgressExecuteRun): return DoProgressExecuteRun(result.handle) if isinstance(result, PyDoProgressCancelSignalReceived): return DO_PROGRESS_CANCEL_SIGNAL_RECEIVED - if isinstance(result, PyDoWaitForPendingRun): - return DO_WAIT_PENDING_RUN return ValueError(f"Unknown progress type: {result}") def take_notification(self, handle: int) -> typing.Union[NotificationType, Exception, Suspended]: @@ -343,9 +373,8 @@ def sys_call( headers: typing.Optional[typing.List[typing.Tuple[str, str]]] = None, ): """Call a service""" - if headers: - headers = [PyHeader(key=h[0], value=h[1]) for h in headers] - return self.vm.sys_call(service, handler, parameter, key, idempotency_key, headers) + py_headers = [PyHeader(key=h[0], value=h[1]) for h in headers] if headers else None + return self.vm.sys_call(service, handler, parameter, key, idempotency_key, py_headers) # pylint: disable=too-many-arguments def sys_send( @@ -362,9 +391,8 @@ def sys_send( send an invocation to a service, and return the handle to the promise that will resolve with the invocation id """ - if headers: - headers = [PyHeader(key=h[0], value=h[1]) for h in headers] - return self.vm.sys_send(service, handler, parameter, key, delay, idempotency_key, headers) + py_headers = [PyHeader(key=h[0], value=h[1]) for h in headers] if headers else None + return self.vm.sys_send(service, handler, parameter, key, delay, idempotency_key, py_headers) def sys_run(self, name: str) -> int: """ @@ -391,17 +419,11 @@ def sys_reject_awakeable(self, name: str, failure: Failure): py_failure = PyFailure(failure.code, failure.message) self.vm.sys_complete_awakeable_failure(name, py_failure) - def propose_run_completion_success(self, handle: int, output: bytes) -> int: + def propose_run_completion_success(self, handle: int, output: bytes) -> None: """ - Exit a side effect - - Args: - output: The output of the side effect. - - Returns: - handle + Exit a side effect with a success value. """ - return self.vm.propose_run_completion_success(handle, output) + self.vm.propose_run_completion_success(handle, output) def sys_get_promise(self, name: str) -> int: """Returns the promise handle""" @@ -420,16 +442,12 @@ def sys_complete_promise_failure(self, name: str, failure: Failure) -> int: res = PyFailure(failure.code, failure.message) return self.vm.sys_complete_promise_failure(name, res) - def propose_run_completion_failure(self, handle: int, output: Failure) -> int: + def propose_run_completion_failure(self, handle: int, output: Failure) -> None: """ - Exit a side effect - - Args: - name: The name of the side effect. - output: The output of the side effect. + Exit a side effect with a terminal failure. """ res = PyFailure(output.code, output.message) - return self.vm.propose_run_completion_failure(handle, res) + self.vm.propose_run_completion_failure(handle, res) # pylint: disable=line-too-long def propose_run_completion_transient( diff --git a/src/lib.rs b/src/lib.rs index 78b4dd4..e582919 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -3,9 +3,9 @@ use pyo3::prelude::*; use pyo3::types::{PyBytes, PyNone, PyString}; use restate_sdk_shared_core::fmt::{set_error_formatter, ErrorFormatter}; use restate_sdk_shared_core::{ - CallHandle, CoreVM, DoProgressResponse, Error, Header, IdentityVerifier, Input, NonEmptyValue, + AwaitResponse, CallHandle, CoreVM, Error, Header, IdentityVerifier, Input, NonEmptyValue, NotificationHandle, ResponseHead, RetryPolicy, RunExitResult, TakeOutputResult, Target, - TerminalFailure, VMOptions, Value, CANCEL_NOTIFICATION_HANDLE, VM, + TerminalFailure, UnresolvedFuture, VMOptions, Value, CANCEL_NOTIFICATION_HANDLE, VM, }; use std::fmt; use std::time::{Duration, SystemTime}; @@ -245,7 +245,7 @@ impl From for PyInput { } #[pyclass] -struct PyDoProgressReadFromInput; +struct PyDoProgressWaitExternalProgress; #[pyclass] struct PyDoProgressAnyCompleted; @@ -260,7 +260,38 @@ struct PyDoProgressExecuteRun { struct PyDoProgressCancelSignalReceived; #[pyclass] -struct PyDoWaitForPendingRun; +#[derive(Clone)] +pub struct PyUnresolvedFuture { + inner: UnresolvedFuture, +} + +#[pymethods] +impl PyUnresolvedFuture { + #[staticmethod] + fn single(handle: PyNotificationHandle) -> Self { + PyUnresolvedFuture { + inner: UnresolvedFuture::Single(NotificationHandle::from(handle)), + } + } + + #[staticmethod] + fn first_completed(children: Vec>) -> Self { + PyUnresolvedFuture { + inner: UnresolvedFuture::FirstCompleted( + children.into_iter().map(|c| c.inner.clone()).collect(), + ), + } + } + + #[staticmethod] + fn all_completed(children: Vec>) -> Self { + PyUnresolvedFuture { + inner: UnresolvedFuture::AllCompleted( + children.into_iter().map(|c| c.inner.clone()).collect(), + ), + } + } +} #[pyclass] pub struct PyCallHandle { @@ -362,41 +393,33 @@ impl PyVM { self_.vm.is_completed(handle.into()) } - fn do_progress( - mut self_: PyRefMut<'_, Self>, - any_handle: Vec, - ) -> PyResult> { - let res = self_.vm.do_progress( - any_handle - .into_iter() - .map(NotificationHandle::from) - .collect(), - ); + fn do_progress<'py>( + mut self_: PyRefMut<'py, Self>, + unresolved_future: PyRef<'py, PyUnresolvedFuture>, + ) -> PyResult> { + let res = self_.vm.do_await(unresolved_future.inner.clone()); let py = self_.py(); match res { Err(e) if e.is_suspended_error() => Ok(Bound::new(py, PySuspended)?.into_any()), Err(e) => Err(PyVMError::from(e))?, - Ok(DoProgressResponse::AnyCompleted) => { + Ok(AwaitResponse::AnyCompleted) => { Ok(Bound::new(py, PyDoProgressAnyCompleted)?.into_any()) } - Ok(DoProgressResponse::ReadFromInput) => { - Ok(Bound::new(py, PyDoProgressReadFromInput)?.into_any()) + Ok(AwaitResponse::WaitingExternalProgress { .. }) => { + Ok(Bound::new(py, PyDoProgressWaitExternalProgress)?.into_any()) } - Ok(DoProgressResponse::ExecuteRun(handle)) => Ok(Bound::new( + Ok(AwaitResponse::ExecuteRun(handle)) => Ok(Bound::new( py, PyDoProgressExecuteRun { handle: handle.into(), }, )? .into_any()), - Ok(DoProgressResponse::CancelSignalReceived) => { + Ok(AwaitResponse::CancelSignalReceived) => { Ok(Bound::new(py, PyDoProgressCancelSignalReceived)?.into_any()) } - Ok(DoProgressResponse::WaitingPendingRun) => { - Ok(Bound::new(py, PyDoWaitForPendingRun)?.into_any()) - } } } @@ -890,10 +913,10 @@ 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_class::()?; + m.add_class::()?; m.add_class::()?; m.add("VMException", m.py().get_type::())?; diff --git a/tests/disconnect_hotloop.py b/tests/disconnect_hotloop.py index f81a724..46a7b57 100644 --- a/tests/disconnect_hotloop.py +++ b/tests/disconnect_hotloop.py @@ -105,17 +105,17 @@ async def mock_receive() -> ASGIReceiveEvent: async def test_empty_body_frames_do_not_cause_hotloop(): """ - When the VM returns DoProgressReadFromInput and the chunk has body=b'', + When the VM returns DoProgressWaitExternalProgress and the chunk has body=b'', notify_input should NOT be called (it would cause a tight loop). The loop should exit via DisconnectedException when http.disconnect arrives. """ from restate.server_context import ServerInvocationContext, DisconnectedException - from restate.vm import DoProgressReadFromInput + from restate.vm import DoProgressWaitExternalProgress, SingleUnresolvedFuture # Build a minimal mock context vm = MagicMock() vm.take_output.return_value = None - vm.do_progress.return_value = DoProgressReadFromInput() + vm.do_progress.return_value = DoProgressWaitExternalProgress() handler = MagicMock() invocation = MagicMock() @@ -149,7 +149,7 @@ async def mock_receive() -> ASGIReceiveEvent: try: with pytest.raises(DisconnectedException): await asyncio.wait_for( - ctx.create_poll_or_cancel_coroutine([0]), + ctx.create_poll_or_cancel_coroutine(SingleUnresolvedFuture(0)), timeout=2.0, )