diff --git a/CHANGELOG.md b/CHANGELOG.md index 90fb719..595e853 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -128,3 +128,21 @@ alone, so a read always moves on. - The Redis and memory Stores read no `to_offset` or `wake_only`. A read with one fails with `stream_unsupported`, since only the native backend serves them. +- The runtime is a plain object with no Core trait. A bridge builds it per client and calls + `StreamExtension::call` with a method name and proto bytes. An error has `code()` and + `details()`, the encoded `StreamErrorDetails`. The crate no longer depends on + `temporalio-sdk-core`, and it pins `temporalio-common` at sdk-rust Brook c7 `b565c433`. +- The handles of one Producer epoch on one Stream share one writer. Each `OpenWriter` adds a + handle, and each `CloseWriter` or the new `FreeWriter` releases one. `FreeWriter` and + `StreamExtension::free_writer` release with no wait, for a handle that lang drops. The last + handle drops the writer, so a process keeps no writer that no handle uses. +- Every writer in a process takes its batch `sequence` from one counter, so a writer opened again + after its close goes on above its earlier batches. The map of closed writers' sequences is + gone, and it no longer grows for the life of the process. +- `ids::named_client_producer_id` makes `client:`, a client Producer id that the caller + keeps across restarts. With the caller's epoch in `attempt`, a new incarnation fences the + earlier one. +- A client Producer's newer epoch fences its earlier epochs but writes no `Superseded` item and + withdraws nothing, in the Redis script, the memory Store and the reader rule. The withdrawal + rule stays for Workflow, Local Activity and Activity Producers. `testdata/withdrawals.json` has + the case. diff --git a/Cargo.lock b/Cargo.lock index cae8de3..7b77869 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,12 +2,6 @@ # It is not intended for manual editing. version = 4 -[[package]] -name = "adler2" -version = "2.0.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "320119579fcad9c21884f5c4861d16174d0e06250625266f50fe6898340abefa" - [[package]] name = "aho-corasick" version = "1.1.5" @@ -17,12 +11,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "anstyle" -version = "1.0.14" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" - [[package]] name = "anyhow" version = "1.0.104" @@ -354,21 +342,6 @@ dependencies = [ "cfg-if", ] -[[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-utils" -version = "0.8.23" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a31eee39dddec8330830986fcd7625edb5a24ec90ea038215273bbc3adb08ac6" - [[package]] name = "crypto-common" version = "0.1.7" @@ -457,62 +430,18 @@ dependencies = [ "syn 3.0.9", ] -[[package]] -name = "downcast" -version = "0.11.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1435fa1053d8b2fbbe9be7e97eca7f33d37b28409959813daefc1446a14247f1" - [[package]] name = "dunce" version = "1.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" -[[package]] -name = "dyn-clone" -version = "1.0.20" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d0881ea181b1df73ff77ffaaf9c7544ecc11e82fba9b5f27b262a3c73a332555" - [[package]] name = "either" version = "1.19.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0e9c71c2167ca323c882b99918929403426e2373ea17242ff5653e0d5e1058be" -[[package]] -name = "enum-iterator" -version = "2.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a4549325971814bda7a44061bf3fe7e487d447cba01e4220a4b454d630d7a016" -dependencies = [ - "enum-iterator-derive", -] - -[[package]] -name = "enum-iterator-derive" -version = "1.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "685adfa4d6f3d765a26bc5dbc936577de9abf756c1feeb3089b01dd395034842" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - -[[package]] -name = "enum_dispatch" -version = "0.3.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "aa18ce2bc66555b3218614519ac839ddb759a7d6720732f979ef8d13be147ecd" -dependencies = [ - "once_cell", - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "equivalent" version = "1.0.3" @@ -578,17 +507,6 @@ version = "0.5.7" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" -[[package]] -name = "flate2" -version = "1.1.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6e634e2e0ebac1ee034020da1ca582e17ffe4e0f5e985823721e168928136dcb" -dependencies = [ - "crc32fast", - "miniz_oxide", - "zlib-rs", -] - [[package]] name = "fnv" version = "1.0.7" @@ -601,12 +519,6 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" -[[package]] -name = "foldhash" -version = "0.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" - [[package]] name = "form_urlencoded" version = "1.2.2" @@ -616,15 +528,6 @@ dependencies = [ "percent-encoding", ] -[[package]] -name = "fragile" -version = "2.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8878864ba14bb86e818a412bfd6f18f9eabd4ec0f008a28e8f7eb61db532fcf9" -dependencies = [ - "futures-core", -] - [[package]] name = "fs_extra" version = "1.3.0" @@ -639,7 +542,6 @@ checksum = "9a31d2a3fbaaeb2af2368bbdd904aa8e812d3c04a1ee10d3171f52d556e5d0a3" dependencies = [ "futures-channel", "futures-core", - "futures-executor", "futures-io", "futures-sink", "futures-task", @@ -690,17 +592,6 @@ dependencies = [ "syn 3.0.9", ] -[[package]] -name = "futures-retry" -version = "0.6.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fde5a672a61f96552aa5ed9fd9c81c3fbdae4be9b1e205d6eaf17c83705adc0f" -dependencies = [ - "futures", - "pin-project-lite", - "tokio", -] - [[package]] name = "futures-sink" version = "0.3.34" @@ -719,7 +610,6 @@ version = "0.3.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" dependencies = [ - "futures-channel", "futures-core", "futures-io", "futures-macro", @@ -740,16 +630,6 @@ dependencies = [ "version_check", ] -[[package]] -name = "gethostname" -version = "1.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1bd49230192a3797a9a4d6abe9b3eed6f7fa4c8a8a4947977c6f80025f92cbd8" -dependencies = [ - "rustix", - "windows-link", -] - [[package]] name = "getrandom" version = "0.2.17" @@ -820,7 +700,7 @@ version = "0.15.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ - "foldhash 0.1.5", + "foldhash", ] [[package]] @@ -828,9 +708,6 @@ name = "hashbrown" version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a" -dependencies = [ - "foldhash 0.2.0", -] [[package]] name = "heck" @@ -1226,15 +1103,6 @@ version = "0.4.34" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" -[[package]] -name = "lru" -version = "0.18.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ef9ac18847474e638e3702b76c65d4eb93428471a74778ef0f1be711717f89b5" -dependencies = [ - "hashbrown 0.17.1", -] - [[package]] name = "lru-slab" version = "0.1.3" @@ -1262,16 +1130,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" -[[package]] -name = "miniz_oxide" -version = "0.9.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b63fbc4a50860e98e7b2aa7804ded1db5cbc3aff9193adaff57a6931bf7c4b4c" -dependencies = [ - "adler2", - "simd-adler32", -] - [[package]] name = "mio" version = "1.2.4" @@ -1283,32 +1141,6 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "mockall" -version = "0.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f58d964098a5f9c6b63d0798e5372fd04708193510a7af313c22e9f29b7b620b" -dependencies = [ - "cfg-if", - "downcast", - "fragile", - "mockall_derive", - "predicates", - "predicates-tree", -] - -[[package]] -name = "mockall_derive" -version = "0.14.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ca41ce716dda6a9be188b385aa78ee5260fc25cd3802cb2a8afdc6afbe6b6dbf" -dependencies = [ - "cfg-if", - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "multimap" version = "0.10.1" @@ -1325,15 +1157,6 @@ dependencies = [ "minimal-lexical", ] -[[package]] -name = "ntapi" -version = "0.4.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c3b335231dfd352ffb0f8017f3b6027a4917f7df785ea2143d8af2adc66980ae" -dependencies = [ - "winapi", -] - [[package]] name = "nu-ansi-term" version = "0.50.3" @@ -1352,25 +1175,6 @@ dependencies = [ "autocfg", ] -[[package]] -name = "objc2-core-foundation" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2a180dd8642fa45cdb7dd721cd4c11b1cadd4929ce112ebd8b9f5803cc79d536" -dependencies = [ - "bitflags", -] - -[[package]] -name = "objc2-io-kit" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33fafba39597d6dc1fb709123dfa8289d39406734be322956a69f0931c73bb15" -dependencies = [ - "libc", - "objc2-core-foundation", -] - [[package]] name = "once_cell" version = "1.21.4" @@ -1525,15 +1329,6 @@ dependencies = [ "indexmap", ] -[[package]] -name = "pid" -version = "4.0.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d7c931ef9756cd5e3fa3d395bfe09df4dfa6f0612c6ca8f6b12927d17ca34e36" -dependencies = [ - "num-traits", -] - [[package]] name = "pin-project" version = "1.1.13" @@ -1572,15 +1367,6 @@ version = "1.15.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" -[[package]] -name = "portable-atomic-util" -version = "0.2.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "10ab3eb7f3becc3a1cbc4f2c6f20267996cfc1a6467a873763411b136a122715" -dependencies = [ - "portable-atomic", -] - [[package]] name = "potential_utf" version = "0.1.6" @@ -1599,32 +1385,6 @@ dependencies = [ "zerocopy", ] -[[package]] -name = "predicates" -version = "3.1.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ada8f2932f28a27ee7b70dd6c1c39ea0675c55a36879ab92f3a715eaa1e63cfe" -dependencies = [ - "anstyle", - "predicates-core", -] - -[[package]] -name = "predicates-core" -version = "1.0.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "cad38746f3166b4031b1a0d39ad9f954dd291e7854fcc0eed52ee41a0b50d144" - -[[package]] -name = "predicates-tree" -version = "1.0.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d0de1b847b39c8131db0467e9df1ff60e6d0562ab8e9a16e568ad0fdb372e2f2" -dependencies = [ - "predicates-core", - "termtree", -] - [[package]] name = "prettyplease" version = "0.2.37" @@ -2002,20 +1762,15 @@ dependencies = [ "rustls", "rustls-pki-types", "rustls-platform-verifier", - "serde", - "serde_json", - "serde_urlencoded", "sync_wrapper", "tokio", "tokio-rustls", - "tokio-util", "tower", "tower-http", "tower-service", "url", "wasm-bindgen", "wasm-bindgen-futures", - "wasm-streams", "web-sys", ] @@ -2033,17 +1788,6 @@ dependencies = [ "windows-sys 0.52.0", ] -[[package]] -name = "ringbuf" -version = "0.5.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b58b631e416dd7d84549d7e6a50f767e111f9b03e30d04f40072f6618f386aaf" -dependencies = [ - "crossbeam-utils", - "portable-atomic", - "portable-atomic-util", -] - [[package]] name = "rustc-hash" version = "2.1.3" @@ -2257,18 +2001,6 @@ dependencies = [ "zmij", ] -[[package]] -name = "serde_urlencoded" -version = "0.7.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d3491c14715ca2294c4d6a88f15e84739788c1d030eed8c110436aafdaa2f3fd" -dependencies = [ - "form_urlencoded", - "itoa", - "ryu", - "serde", -] - [[package]] name = "sha1_smol" version = "1.0.1" @@ -2301,22 +2033,6 @@ version = "2.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8fadd59c855ef2080decdef8ff161eb6661b86933c9d82e5ba29dc602a55aba" -[[package]] -name = "signal-hook-registry" -version = "1.4.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c4db69cba1110affc0e9f7bcd48bbf87b3f4fc7c61fc9155afd4c469eb3d6c1b" -dependencies = [ - "errno", - "libc", -] - -[[package]] -name = "simd-adler32" -version = "0.3.10" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3a219298ac11a56ea9a6d2120044824d6f01aeb034955e7af7bc16858527deea" - [[package]] name = "simd_cesu8" version = "1.2.0" @@ -2333,27 +2049,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3a9fe34e3e7a50316060351f37187a3f546bce95496156754b601a5fa71b76e" -[[package]] -name = "siphasher" -version = "1.0.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "33f4fe9184a62d842c9ef383018f3306d8ba224fd9d836f56d7288308847c256" - [[package]] name = "slab" version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" -[[package]] -name = "slotmap" -version = "1.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bdd58c3c93c3d278ca835519292445cb4b0d4dc59ccfdf7ceadaab3f8aeb4038" -dependencies = [ - "version_check", -] - [[package]] name = "smallvec" version = "1.16.3" @@ -2430,20 +2131,6 @@ dependencies = [ "syn 3.0.9", ] -[[package]] -name = "sysinfo" -version = "0.38.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "92ab6a2f8bfe508deb3c6406578252e491d299cbbf3bc0529ecc3313aee4a52f" -dependencies = [ - "libc", - "memchr", - "ntapi", - "objc2-core-foundation", - "objc2-io-kit", - "windows", -] - [[package]] name = "tempfile" version = "3.27.0" @@ -2457,43 +2144,10 @@ dependencies = [ "windows-sys 0.61.2", ] -[[package]] -name = "temporalio-client" -version = "1.1.0" -source = "git+https://github.com/moetemp/sdk-rust?rev=32ba79d6207385417ef5c5219183e43e7a01f069#32ba79d6207385417ef5c5219183e43e7a01f069" -dependencies = [ - "anyhow", - "async-trait", - "backon", - "base64 0.23.1", - "bon", - "bytes", - "derive_more", - "dyn-clone", - "futures-retry", - "futures-util", - "http", - "http-body-util", - "hyper", - "hyper-util", - "parking_lot", - "rand 0.10.3", - "serde_json", - "temporalio-common", - "thiserror", - "tokio", - "tokio-rustls", - "tonic", - "tower", - "tracing", - "url", - "uuid", -] - [[package]] name = "temporalio-common" version = "1.1.0" -source = "git+https://github.com/moetemp/sdk-rust?rev=32ba79d6207385417ef5c5219183e43e7a01f069#32ba79d6207385417ef5c5219183e43e7a01f069" +source = "git+https://github.com/moetemp/sdk-rust?rev=b565c433547496f2c14a911549c288a1076e72b6#b565c433547496f2c14a911549c288a1076e72b6" dependencies = [ "anyhow", "async-trait", @@ -2502,13 +2156,11 @@ dependencies = [ "derive_more", "erased-serde", "futures", - "futures-channel", "opentelemetry-otlp", "parking_lot", "prost", "prost-types", "reqwest", - "ringbuf", "rustls", "serde", "serde_json", @@ -2526,7 +2178,7 @@ dependencies = [ [[package]] name = "temporalio-common-wasm" version = "1.1.0" -source = "git+https://github.com/moetemp/sdk-rust?rev=32ba79d6207385417ef5c5219183e43e7a01f069#32ba79d6207385417ef5c5219183e43e7a01f069" +source = "git+https://github.com/moetemp/sdk-rust?rev=b565c433547496f2c14a911549c288a1076e72b6#b565c433547496f2c14a911549c288a1076e72b6" dependencies = [ "anyhow", "async-trait", @@ -2549,20 +2201,10 @@ dependencies = [ "url", ] -[[package]] -name = "temporalio-macros" -version = "1.1.0" -source = "git+https://github.com/moetemp/sdk-rust?rev=32ba79d6207385417ef5c5219183e43e7a01f069#32ba79d6207385417ef5c5219183e43e7a01f069" -dependencies = [ - "proc-macro2", - "quote", - "syn 3.0.9", -] - [[package]] name = "temporalio-protos" version = "0.10.0" -source = "git+https://github.com/moetemp/sdk-rust?rev=32ba79d6207385417ef5c5219183e43e7a01f069#32ba79d6207385417ef5c5219183e43e7a01f069" +source = "git+https://github.com/moetemp/sdk-rust?rev=b565c433547496f2c14a911549c288a1076e72b6#b565c433547496f2c14a911549c288a1076e72b6" dependencies = [ "anyhow", "base64 0.23.1", @@ -2580,52 +2222,6 @@ dependencies = [ "tonic-prost-build", ] -[[package]] -name = "temporalio-sdk-core" -version = "0.10.0" -source = "git+https://github.com/moetemp/sdk-rust?rev=32ba79d6207385417ef5c5219183e43e7a01f069#32ba79d6207385417ef5c5219183e43e7a01f069" -dependencies = [ - "anyhow", - "async-trait", - "backon", - "bon", - "crossbeam-channel", - "crossbeam-utils", - "derive_more", - "enum-iterator", - "enum_dispatch", - "futures", - "futures-util", - "gethostname", - "itertools", - "lru", - "mockall", - "opentelemetry-otlp", - "parking_lot", - "pid", - "pin-project", - "prost", - "prost-wkt-types", - "rand 0.10.3", - "reqwest", - "serde", - "serde_json", - "siphasher", - "slotmap", - "sysinfo", - "temporalio-client", - "temporalio-common", - "temporalio-macros", - "thiserror", - "tokio", - "tokio-stream", - "tokio-util", - "tonic", - "tracing", - "url", - "uuid", -] - [[package]] name = "temporalio-streams" version = "0.1.0" @@ -2640,7 +2236,6 @@ dependencies = [ "serde_json", "sha2", "temporalio-common", - "temporalio-sdk-core", "thiserror", "tokio", "tokio-util", @@ -2649,12 +2244,6 @@ dependencies = [ "tracing", ] -[[package]] -name = "termtree" -version = "0.5.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8f50febec83f5ee1df3015341d8bd429f2d1cc62bcba7ea2076759d315084683" - [[package]] name = "thiserror" version = "2.0.21" @@ -2709,9 +2298,7 @@ dependencies = [ "bytes", "libc", "mio", - "parking_lot", "pin-project-lite", - "signal-hook-registry", "socket2", "tokio-macros", "windows-sys 0.61.2", @@ -2773,7 +2360,6 @@ dependencies = [ "async-trait", "base64 0.22.1", "bytes", - "flate2", "h2", "http", "http-body", @@ -3163,19 +2749,6 @@ dependencies = [ "unicode-ident", ] -[[package]] -name = "wasm-streams" -version = "0.5.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9d1ec4f6517c9e11ae630e200b2b65d193279042e28edd4a2cda233e46670bbb" -dependencies = [ - "futures-util", - "js-sys", - "wasm-bindgen", - "wasm-bindgen-futures", - "web-sys", -] - [[package]] name = "web-sys" version = "0.3.106" @@ -3205,22 +2778,6 @@ dependencies = [ "rustls-pki-types", ] -[[package]] -name = "winapi" -version = "0.3.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5c839a674fcd7a98952e593242ea400abe93992746761e38641405d28b00f419" -dependencies = [ - "winapi-i686-pc-windows-gnu", - "winapi-x86_64-pc-windows-gnu", -] - -[[package]] -name = "winapi-i686-pc-windows-gnu" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ac3b87c63620426dd9b991e5ce0329eff545bccbbb34f3be09ff6fb6ab51b7b6" - [[package]] name = "winapi-util" version = "0.1.11" @@ -3230,113 +2787,12 @@ 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" -version = "0.62.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "527fadee13e0c05939a6a05d5bd6eec6cd2e3dbd648b9f8e447c6518133d8580" -dependencies = [ - "windows-collections", - "windows-core", - "windows-future", - "windows-numerics", -] - -[[package]] -name = "windows-collections" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "23b2d95af1a8a14a3c7367e1ed4fc9c20e0a26e79551b1454d72583c97cc6610" -dependencies = [ - "windows-core", -] - -[[package]] -name = "windows-core" -version = "0.62.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8e83a14d34d0623b51dce9581199302a221863196a1dde71a7663a4c2be9deb" -dependencies = [ - "windows-implement", - "windows-interface", - "windows-link", - "windows-result", - "windows-strings", -] - -[[package]] -name = "windows-future" -version = "0.3.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e1d6f90251fe18a279739e78025bd6ddc52a7e22f921070ccdc67dde84c605cb" -dependencies = [ - "windows-core", - "windows-link", - "windows-threading", -] - -[[package]] -name = "windows-implement" -version = "0.60.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "053e2e040ab57b9dc951b72c264860db7eb3b0200ba345b4e4c3b14f67855ddf" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - -[[package]] -name = "windows-interface" -version = "0.59.3" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3f316c4a2570ba26bbec722032c4099d8c8bc095efccdc15688708623367e358" -dependencies = [ - "proc-macro2", - "quote", - "syn 2.0.119", -] - [[package]] name = "windows-link" version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" -[[package]] -name = "windows-numerics" -version = "0.3.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6e2e40844ac143cdb44aead537bbf727de9b044e107a0f1220392177d15b0f26" -dependencies = [ - "windows-core", - "windows-link", -] - -[[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" @@ -3371,15 +2827,6 @@ dependencies = [ "windows_x86_64_msvc", ] -[[package]] -name = "windows-threading" -version = "0.2.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3949bd5b99cafdf1c7ca86b43ca564028dfe27d66958f2470940f73d86d75b37" -dependencies = [ - "windows-link", -] - [[package]] name = "windows_aarch64_gnullvm" version = "0.52.6" @@ -3549,12 +2996,6 @@ dependencies = [ "syn 3.0.9", ] -[[package]] -name = "zlib-rs" -version = "0.6.8" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b268e58e7c693d7c271f93ffc4ba3b380412554231c85bf61ca7af91042a4112" - [[package]] name = "zmij" version = "1.0.23" diff --git a/Cargo.toml b/Cargo.toml index fcdc1b4..a36e00a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -20,12 +20,7 @@ tonic-prost-build = "0.14.6" # since a lang's `[patch]` key must match it byte for byte. [workspace.dependencies.temporalio-common] git = "https://github.com/moetemp/sdk-rust" -rev = "32ba79d6207385417ef5c5219183e43e7a01f069" -default-features = false - -[workspace.dependencies.temporalio-sdk-core] -git = "https://github.com/moetemp/sdk-rust" -rev = "32ba79d6207385417ef5c5219183e43e7a01f069" +rev = "b565c433547496f2c14a911549c288a1076e72b6" default-features = false [workspace.lints.rust] diff --git a/README.md b/README.md index 0460006..027ee3c 100644 --- a/README.md +++ b/README.md @@ -3,19 +3,21 @@ Internal and experimental. `temporalio-streams` is the Stream runtime for Temporal SDKs. It serves the Stream service of the -Brook API proposal on the customer's Redis. The SDKs link this crate into their bridges. Lang -calls it through the bridge's generic extension calls. A Workflow writes through a Local Activity -per call, on the same calls. Temporal Core holds no stream code. +Brook API proposal on the customer's Redis. The SDKs link this crate into their bridges. A bridge +builds the runtime per client, as a plain object, and calls `StreamExtension::call` with a method +name and proto bytes. A Workflow writes through a Local Activity per call, on the same calls. +Temporal Core holds no stream code and no hook for it. ## Stage pins -The crate depends on `temporalio-common` and `temporalio-sdk-core` from one sdk-rust commit. That -commit must equal the `sdk-core` submodule commit of the sdk-python stage that links the crate. +The crate depends on `temporalio-common` from one sdk-rust commit. That commit must equal the +`sdk-core` submodule commit of the sdk-python stage that links the crate. | Stage | sdk-streams branch | sdk-rust branch | |---|---|---| | Brook v3 | `moe/brook/v3/s1-protos-and-ids` and up | `moe/brook/v3/c4-local-activity-workflow-task` `3a9203ac` | -| Brook v4 | `moe/brook/v4/s20-orphan-local-activities` and up | `moe/brook/v4/c6-local-activity-task-history` `32ba79d6` | +| Brook v4 | `moe/brook/v4/s20-orphan-local-activities` to `s21` | `moe/brook/v4/c6-local-activity-task-history` `32ba79d6` | +| Brook v4 | `moe/brook/v4/s22-client-objects` and up | `moe/brook/v4/c7-no-worker-extension` `b565c433` | ## Linking from a lang bridge @@ -32,18 +34,19 @@ default-features = false [patch."https://github.com/moetemp/sdk-rust"] temporalio-common = { path = "./sdk-core/crates/common" } -temporalio-sdk-core = { path = "./sdk-core/crates/sdk-core" } ``` -The crate features `tls-ring` and `tls-aws-lc` forward to the sdk-rust crates. A FIPS build takes +The crate features `tls-ring` and `tls-aws-lc` forward to `temporalio-common`. A FIPS build takes `fips` with no default features. It turns on `tls-aws-lc` and the FIPS build of `aws-lc-rs`, which needs CMake and Go. ## Protos and shared vectors -- `protos/temporal/api/streamservice/v1/stream_service.proto` is the draft Stream service of the - API proposal. sdk-rust doesn't have it, so the crate generates it. Lang generates the same file. -- `protos/temporal/sdk/streams/v1/extension.proto` has the crate's own extension calls. +- `protos/temporal/api/streamservice/v1/stream_service.proto` is the Stream service, a byte for + byte copy of the api fork's file. sdk-rust doesn't have it, so the crate generates it. Lang + generates the same file. +- `protos/temporal/sdk/streams/v1/extension.proto` has the crate's own calls, such as the writer + calls. - `nexus/stream-service.wit` is the Stream service as a System Nexus service, backed by the proto above. Each SDK copies it and generates its bindings with nexgen. A test keeps every line at most 100 columns. @@ -55,8 +58,9 @@ needs CMake and Go. `scripts/sync-api-protos.sh` refreshes it from the pinned sdk-rust commit, and CI fails when the copy differs. - `testdata/*.json` are the shared vectors: Stream ids, Producer ids, Redis keys and meta fields, - offsets, epoch keys and the withdrawal rule. `testdata/gen_vectors.py` writes them from a - second implementation. Each SDK runs the same files. + offsets, epoch keys, the withdrawal rule and the batch digest of a dedupe entry. + `testdata/gen_vectors.py` writes them from a second implementation. Each SDK and the server run + the same files. ## Building and testing diff --git a/crates/streams/Cargo.toml b/crates/streams/Cargo.toml index e70fab1..ad063fc 100644 --- a/crates/streams/Cargo.toml +++ b/crates/streams/Cargo.toml @@ -15,8 +15,8 @@ categories = ["development-tools"] default = ["tls-ring"] # The TLS providers of the sdk-rust crates. A FIPS build takes `tls-aws-lc` with no default # features, so no crate in the build pulls `ring`. -tls-ring = ["temporalio-sdk-core/tls-ring", "temporalio-common/tls-ring"] -tls-aws-lc = ["temporalio-sdk-core/tls-aws-lc", "temporalio-common/tls-aws-lc"] +tls-ring = ["temporalio-common/tls-ring"] +tls-aws-lc = ["temporalio-common/tls-aws-lc"] # A FIPS build: `tls-aws-lc`, and `aws-lc-rs` in FIPS mode, so the shared rustls provider links # `aws-lc-fips-sys`. fips = ["tls-aws-lc", "dep:aws-lc-rs"] @@ -40,7 +40,6 @@ redis = { version = "1.7", default-features = false, features = [ ] } sha2 = "0.10" temporalio-common = { workspace = true } -temporalio-sdk-core = { workspace = true } thiserror = { workspace = true } tonic = { workspace = true } tokio = { version = "1.47", default-features = false, features = ["rt", "sync", "time"] } diff --git a/crates/streams/docs/redis.md b/crates/streams/docs/redis.md index 9f33ded..3a3df6c 100644 --- a/crates/streams/docs/redis.md +++ b/crates/streams/docs/redis.md @@ -49,6 +49,10 @@ Each entry has `r`, the encoded `temporal.api.streamservice.v1.StreamRecord` wit only `s` withdraws and has no Record. A reader gives a `SUPERSEDED` item for `s` just before the Record of the same entry. An offset is the entry id `-` packed as `ms << 20 | seq`. +The first batch of a Producer's newer epoch withdraws its earlier epochs' Records, for a +Workflow, a Local Activity and an Activity. A client's newer epoch only fences: it writes no `s`, +since a restarted client's earlier Records are real messages, not a failed attempt's. + ## Orphaned Local Activity Records A Local Activity batch that gives its run's last completed Workflow Task (`previous_completed_task`, diff --git a/crates/streams/lua/append.lua b/crates/streams/lua/append.lua index bc5a6c8..799ac4b 100644 --- a/crates/streams/lua/append.lua +++ b/crates/streams/lua/append.lua @@ -68,9 +68,10 @@ if refused then return refused end --- The first batch of a newer epoch withdraws the Records of the earlier ones. +-- The first batch of a newer epoch withdraws the Records of the earlier ones. A client's (4) +-- newer epoch only fences, since its earlier Records are real messages. local holds = redis.call('HGET', meta, 'h' .. suffix) -local flagged = holds ~= false and holds < epoch +local flagged = kind ~= '4' and holds ~= false and holds < epoch if withdraw_run(meta, withdrawn_run) then flagged = true end diff --git a/crates/streams/nexus/stream-writer.wit b/crates/streams/nexus/stream-writer.wit index a8b8233..1d2081a 100644 --- a/crates/streams/nexus/stream-writer.wit +++ b/crates/streams/nexus/stream-writer.wit @@ -1,6 +1,6 @@ -// Internal. Not a Temporal API, and no server serves it. The crate's Worker extension serves -// these calls in process, backed by the messages in ../protos/temporal/sdk/streams/v1/ -// extension.proto. Lang reaches them through an extension-call transport, from Activities and +// Internal. Not a Temporal API, and no server serves it. The crate's runtime object serves these +// calls in process, backed by the messages in ../protos/temporal/sdk/streams/v1/ +// extension.proto. Lang reaches them through the bridge's call transport, from Activities and // clients only. So the interface is client-only and has no endpoint, which keeps it out of the // System Nexus path. The writer keeps the Producer epoch and the sequence in the crate. This file // is the source. Each SDK copies it beside stream-service.wit and generates its bindings with @@ -62,6 +62,15 @@ interface stream-writer { last-offset: option, } + /// @nexus.proto "temporal.sdk.streams.v1.FreeWriterRequest" + record free-writer-request { + writer-id: option, + } + + /// @nexus.proto "temporal.sdk.streams.v1.FreeWriterResponse" + record free-writer-response { + } + /// @nexus.proto "temporal.sdk.streams.v1.CloseWritersRequest" record close-writers-request { } @@ -72,7 +81,8 @@ interface stream-writer { failed: option, } - // A writer never changes. Offsets are values that its methods return. + // A writer never changes. Offsets are values that its methods return. The handles of one + // Producer epoch on one Stream share one writer, and each close or free releases one handle. /// @nexus.doc "An open writer of one Producer. The crate keeps its epoch and sequence." resource writer { constructor(writer-id: u64); @@ -81,9 +91,13 @@ interface stream-writer { /// @nexus.operation "write-records" write-records: func(records: list) -> write-records-response; - /// @nexus.doc "Wait for a write in flight, then drop the writer. The Stream stays open." + /// @nexus.doc "Release the handle. The last one waits for a write, then drops the writer." /// @nexus.operation "close-writer" close: func() -> close-writer-response; + + /// @nexus.doc "Release the handle with no wait, for a handle that lang drops." + /// @nexus.operation "free-writer" + free: func() -> free-writer-response; } /// @nexus.proto "temporal.sdk.streams.v1.OpenWriterResponse" @@ -98,6 +112,9 @@ interface stream-writer { /// @nexus.operation name="CloseWriter" close-writer: func(request: close-writer-request) -> close-writer-response; + /// @nexus.operation name="FreeWriter" + free-writer: func(request: free-writer-request) -> free-writer-response; + /// @nexus.doc "Drop every open writer, for example at shutdown." /// @nexus.operation name="CloseWriters" close-writers: func(request: close-writers-request) -> close-writers-response; diff --git a/crates/streams/protos/temporal/sdk/streams/v1/extension.proto b/crates/streams/protos/temporal/sdk/streams/v1/extension.proto index a38cb91..c615904 100644 --- a/crates/streams/protos/temporal/sdk/streams/v1/extension.proto +++ b/crates/streams/protos/temporal/sdk/streams/v1/extension.proto @@ -1,8 +1,8 @@ syntax = "proto3"; -// The messages of the crate's own extension calls, next to the five Stream service operations. -// Lang sends them through the bridge's generic extension call, by the method name in each -// comment. Field numbers are part of the contract with every SDK. +// The messages of the crate's own calls, next to the five Stream service operations. Lang sends +// them through the bridge's call on the runtime object, by the method name in each comment. Field +// numbers are part of the contract with every SDK. package temporal.sdk.streams.v1; import "google/protobuf/duration.proto"; @@ -33,7 +33,10 @@ message RedisStreamConfig { // Method `OpenWriter`, asynchronous. A writer for an Activity or a client Producer, which keeps // its epoch and `sequence` order. A writer of an Activity attempt above 1 first sends an empty -// batch, so the new epoch withdraws the earlier attempt's Records. +// batch, so the new epoch withdraws the earlier attempt's Records. The handles of one Producer +// epoch on one Stream share one writer: each open adds a handle, and each `CloseWriter` or +// `FreeWriter` releases one. The batches of every writer in a process take their `sequence` from +// one counter, so a writer opened again after its close goes on above its earlier batches. message OpenWriterRequest { string namespace = 1; temporal.api.streamservice.v1.StreamRef stream = 2; @@ -71,8 +74,9 @@ message WriteRecordsResponse { int64 last_offset = 2; } -// Method `CloseWriter`, asynchronous. Waits for a write in flight, then drops the writer. It -// doesn't close the Stream. It fails with the writer's error if a write failed. +// Method `CloseWriter`, asynchronous. Releases one handle of the writer. The last handle waits for +// a write in flight, then drops the writer. It doesn't close the Stream. It fails with the +// writer's error if a write failed. message CloseWriterRequest { uint64 writer_id = 1; } @@ -82,6 +86,16 @@ message CloseWriterResponse { int64 last_offset = 1; } +// Method `FreeWriter`. Releases one handle of the writer with no wait, for a handle that lang +// drops, such as from a finalizer. The bridge may also call it with no runtime. A write in flight +// still ends. The writer's error, if any, is lost. +message FreeWriterRequest { + uint64 writer_id = 1; +} + +message FreeWriterResponse { +} + // Method `CloseWriters`, asynchronous. Drops every open writer, for example when the plugin's // client or Worker shuts down. message CloseWritersRequest { diff --git a/crates/streams/src/epoch.rs b/crates/streams/src/epoch.rs index 4be2607..bbf975d 100644 --- a/crates/streams/src/epoch.rs +++ b/crates/streams/src/epoch.rs @@ -85,11 +85,12 @@ pub fn timestamp(time: SystemTime) -> Timestamp { /// Whether a `Superseded` item withdraws an earlier Record. /// -/// For an Activity, a Local Activity or a client, it withdraws that Producer's Records with a -/// lower epoch. For a Workflow task t with the previous completed task c, it withdraws the -/// Records of the run and of its Local Activities whose task is in `failed_tasks` or strictly -/// between c and t, and those at c or at t with a lower epoch. A retry run withdraws every -/// Record whose Producer has the previous run. +/// For an Activity or a Local Activity, it withdraws that Producer's Records with a lower epoch. +/// For a Workflow task t with the previous completed task c, it withdraws the Records of the run +/// and of its Local Activities whose task is in `failed_tasks` or strictly between c and t, and +/// those at c or at t with a lower epoch. A retry run withdraws every Record whose Producer has +/// the previous run. A client's newer epoch fences the earlier ones but withdraws nothing, since +/// a restarted client's earlier Records are real messages, not a failed attempt's. pub fn withdraws(supersession: &Supersession, record: &StreamRecord) -> bool { let Some(producer) = record.producer.as_ref() else { return false; @@ -98,6 +99,9 @@ pub fn withdraws(supersession: &Supersession, record: &StreamRecord) -> bool { return true; } let kind = ProducerKind::try_from(supersession.producer_kind).unwrap_or_default(); + if kind == ProducerKind::Client { + return false; + } let (Some(epoch), Some(old)) = (supersession.epoch.as_ref(), producer.epoch.as_ref()) else { return false; }; diff --git a/crates/streams/src/error.rs b/crates/streams/src/error.rs index 72d568f..c5401f2 100644 --- a/crates/streams/src/error.rs +++ b/crates/streams/src/error.rs @@ -1,8 +1,7 @@ -//! The one error every call fails with, and its form on the extension boundary. +//! The one error every call fails with, and its form for lang. use crate::proto::StreamErrorDetails; use prost::Message; -use temporalio_sdk_core::ExtensionError; use tonic::Code; /// The result of every fallible call in the crate. @@ -114,15 +113,19 @@ impl StreamError { pub fn storage(message: impl Into) -> Self { Self::new(Reason::Storage, message) } -} -impl From for ExtensionError { - fn from(error: StreamError) -> Self { - let details = StreamErrorDetails { - reason: error.reason.token().to_string(), - }; - ExtensionError::new(error.reason.code(), error.message) - .with_details(details.encode_to_vec()) + /// The status code that lang raises the error with. + pub fn code(&self) -> Code { + self.reason.code() + } + + /// The encoded [StreamErrorDetails], which lang reads the reason token from, as from the + /// details of a gRPC status. + pub fn details(&self) -> Vec { + StreamErrorDetails { + reason: self.reason.token().to_string(), + } + .encode_to_vec() } } @@ -132,11 +135,9 @@ mod tests { #[test] fn an_error_carries_its_code_and_reason_to_lang() { - let error: ExtensionError = - StreamError::new(Reason::FormatTooNew, "format 2 is newer than 1").into(); - assert_eq!(error.code, Code::FailedPrecondition); - assert_eq!(error.message, "format 2 is newer than 1"); - let details = StreamErrorDetails::decode(error.details.as_slice()).unwrap(); + let error = StreamError::new(Reason::FormatTooNew, "format 2 is newer than 1"); + assert_eq!(error.code(), Code::FailedPrecondition); + let details = StreamErrorDetails::decode(error.details().as_slice()).unwrap(); assert_eq!(details.reason, "stream_format_too_new"); } } diff --git a/crates/streams/src/extension.rs b/crates/streams/src/extension.rs index 7cc2a6d..cb69d6c 100644 --- a/crates/streams/src/extension.rs +++ b/crates/streams/src/extension.rs @@ -1,14 +1,15 @@ -//! The Worker extension that lang's bridge hands to Core and calls by method name. +//! The Stream runtime as a plain object, which lang's bridge builds per client and calls by +//! method name. Core has no hook for it. use crate::{ - StreamError, StreamResult, + Reason, StreamError, StreamResult, memory::MemoryStore, metrics::Metrics, proto::{ AppendRecordsRequest, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, - CloseWritersRequest, CloseWritersResponse, DescribeStreamRequest, OpenWriterRequest, - OpenWriterResponse, ReadRecordsRequest, RedisStreamConfig, WriteRecordsRequest, - WriteRecordsResponse, + CloseWritersRequest, CloseWritersResponse, DescribeStreamRequest, FreeWriterRequest, + FreeWriterResponse, OpenWriterRequest, OpenWriterResponse, ReadRecordsRequest, + RedisStreamConfig, WriteRecordsRequest, WriteRecordsResponse, }, redis::{RedisStore, RedisStoreOptions}, service::StreamService, @@ -18,10 +19,9 @@ use crate::{ use prost::Message; use std::{sync::Arc, time::Duration}; use temporalio_common::telemetry::metrics::TemporalMeter; -use temporalio_sdk_core::{ExtensionError, WorkerExtension}; use tokio::sync::OnceCell; -/// The name of the extension, for logs and for a bridge that selects an extension by name. +/// The name of the Redis runtime, for logs and for a bridge that selects a runtime by name. pub const EXTENSION_NAME: &str = "temporalio-streams-redis"; /// What the extension needs besides the Store options. @@ -96,11 +96,11 @@ impl Shared { } } -/// The Stream runtime as a Worker extension, on Redis or, for tests, on the memory Store. +/// The Stream runtime on Redis or, for tests, on the memory Store. /// /// On Redis it connects on its first call that needs Redis, so a bridge can build it with no -/// runtime, and a Workflow that never publishes never needs Redis. One extension serves every -/// Worker and client of a plugin. +/// runtime, and a Workflow that never publishes never needs Redis. A bridge builds one for each +/// client of a plugin, and that client's Workers share it. pub struct StreamExtension { shared: Arc, } @@ -158,17 +158,31 @@ impl StreamExtension { Ok(Self::new(ExtensionOptions::from_config(config)?, meter)) } - /// The options the extension runs with. + /// The options the runtime runs with. pub fn options(&self) -> &ExtensionOptions { &self.shared.options } - /// Runs one asynchronous method on encoded input. - pub async fn dispatch_async( - &self, - method: &str, - input: &[u8], - ) -> Result, ExtensionError> { + /// A name that identifies the runtime in logs. + pub fn name(&self) -> &str { + self.shared.name + } + + /// Releases one handle of a writer with no wait, as the `FreeWriter` method does. A bridge + /// calls it from a finalizer, which has no runtime to await on. + pub fn free_writer(&self, writer_id: u64) -> StreamResult<()> { + self.shared.writers.free(writer_id) + } + + /// Runs one method on encoded input and answers the encoded output. The bridge raises an + /// error with [StreamError::code] and [StreamError::details]. + pub async fn call(&self, method: &str, input: &[u8]) -> StreamResult> { + // A free needs no Store, so it works also when Redis is down. + if method == "FreeWriter" { + let request = decode::(input)?; + self.free_writer(request.writer_id)?; + return Ok(FreeWriterResponse {}.encode_to_vec()); + } let service = self.shared.service().await?; let output = match method { "CreateStream" => service.create_stream(decode(input)?).await?.encode_to_vec(), @@ -219,7 +233,12 @@ impl StreamExtension { let last_offset = self.shared.writers.close(request.writer_id).await?; CloseWriterResponse { last_offset }.encode_to_vec() } - _ => return Err(ExtensionError::unknown_method(method)), + _ => { + return Err(StreamError::new( + Reason::Unsupported, + format!("Unknown method {method}"), + )); + } }; Ok(output) } @@ -285,17 +304,6 @@ pub(crate) fn decode(input: &[u8]) -> StreamResult { M::decode(input).map_err(|error| StreamError::invalid(format!("The request is bad: {error}"))) } -#[async_trait::async_trait] -impl WorkerExtension for StreamExtension { - fn name(&self) -> &str { - self.shared.name - } - - async fn call_async(&self, method: String, input: Vec) -> Result, ExtensionError> { - self.dispatch_async(&method, &input).await - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/streams/src/ids.rs b/crates/streams/src/ids.rs index 684d1cb..b9abb91 100644 --- a/crates/streams/src/ids.rs +++ b/crates/streams/src/ids.rs @@ -107,6 +107,12 @@ pub fn client_producer_id(identity: &str, process_key: &str) -> String { format!("client:{}", join(&[identity, process_key])) } +/// The Producer id that a client names itself, `client:`. The caller keeps the name and +/// raises the epoch across restarts, so a new incarnation fences the one before it. +pub fn named_client_producer_id(name: &str) -> String { + format!("client:{}", join(&[name])) +} + /// The id prefix of a Producer kind. pub(crate) fn kind_prefix(kind: ProducerKind) -> Option<&'static str> { match kind { @@ -135,7 +141,10 @@ pub fn check_producer(producer: &Producer) -> StreamResult<()> { let parts: Vec<_> = rest.split(':').collect(); let expected = match kind { ProducerKind::Workflow => 1, - ProducerKind::LocalActivity | ProducerKind::StandaloneActivity | ProducerKind::Client => 2, + ProducerKind::LocalActivity | ProducerKind::StandaloneActivity => 2, + // A named client has one part, and a process's client two. + ProducerKind::Client if parts.len() == 1 => 1, + ProducerKind::Client => 2, ProducerKind::Activity => 3, ProducerKind::Unspecified => unreachable!("refused above"), }; diff --git a/crates/streams/src/lib.rs b/crates/streams/src/lib.rs index 368778c..071201c 100644 --- a/crates/streams/src/lib.rs +++ b/crates/streams/src/lib.rs @@ -2,9 +2,9 @@ //! The Stream runtime of Temporal SDKs. //! -//! The crate serves the Stream service of the Brook API proposal on the customer's Redis. Lang -//! reaches it through the bridge's generic extension calls, with an operation name and proto -//! bytes. A Workflow makes each call from a Local Activity. +//! The crate serves the Stream service of the Brook API proposal on the customer's Redis. Lang's +//! bridge builds the runtime per client and calls it with an operation name and proto bytes. +//! Core has no hook for it. A Workflow makes each call from a Local Activity. pub mod digest; pub mod epoch; diff --git a/crates/streams/src/memory.rs b/crates/streams/src/memory.rs index fc74762..54c41bf 100644 --- a/crates/streams/src/memory.rs +++ b/crates/streams/src/memory.rs @@ -591,9 +591,11 @@ impl Store for MemoryStore { let records = batch.records.len() as u64; keys.check_ids_fit(now, records + 2)?; - let mut flagged = keys - .get(&format!("h{suffix}")) - .is_some_and(|holds| holds.as_str() < epoch.as_str()); + // A client's newer epoch only fences, as in `append.lua`. + let mut flagged = producer.kind != ProducerKind::Client as i32 + && keys + .get(&format!("h{suffix}")) + .is_some_and(|holds| holds.as_str() < epoch.as_str()); if keys.withdraw_run(&withdrawn_run) { flagged = true; } diff --git a/crates/streams/src/writer.rs b/crates/streams/src/writer.rs index 3e13840..49fe04c 100644 --- a/crates/streams/src/writer.rs +++ b/crates/streams/src/writer.rs @@ -2,6 +2,9 @@ //! offsets, which is what an awaited publish needs. A writer keeps one Producer epoch and one //! `sequence` order, so a retry of a batch goes in once. Lang buffers on its own, if it wants //! to, and writes each batch with one call. +//! +//! The handles of one Producer epoch on one Stream share one writer. Each open counts a handle, +//! and each close or free releases one, so a process keeps no writer that no handle uses. use crate::{ Reason, StreamError, StreamResult, @@ -18,7 +21,7 @@ use std::{ collections::HashMap, sync::{ Arc, - atomic::{AtomicU64, Ordering}, + atomic::{AtomicI64, AtomicU64, Ordering}, }, time::Duration, }; @@ -28,10 +31,17 @@ const DEFAULT_BATCH: usize = 1000; const DEFAULT_RETRY: Duration = Duration::from_secs(10); const FIRST_BACKOFF: Duration = Duration::from_millis(50); +/// The last `sequence` of any batch in this process. Every writer numbers its batches from it, +/// so a writer opened again after its close goes on above its earlier batches, with no record +/// of closed writers to keep. +static LAST_SEQUENCE: AtomicI64 = AtomicI64::new(0); + +fn next_sequence() -> i64 { + LAST_SEQUENCE.fetch_add(1, Ordering::Relaxed) + 1 +} + #[derive(Default)] struct State { - /// The last `sequence` the writer sent in its epoch. - sequence: i64, last_offset: i64, /// A write that failed for good. Every later call fails with it, since the batch may be in /// the Store under a `sequence` the writer can't reuse. @@ -91,14 +101,10 @@ impl Writer { while !records.is_empty() { let count = self.batch_len(records.iter()); let rest = records.split_off(count); - let sequence = { - let mut state = self.state.lock(); - if let Some(error) = &state.failed { - return Err(error.clone()); - } - state.sequence += 1; - state.sequence - }; + if let Some(error) = &self.state.lock().failed { + return Err(error.clone()); + } + let sequence = next_sequence(); let (batch_first, batch_last) = match self.send(records, sequence).await { Ok(offsets) => offsets, // The batch may be in the Store, so the writer can't go on in this epoch. @@ -168,24 +174,50 @@ fn check_sizes(producer: &Producer, records: &[AppendRecord]) -> StreamResult<() Ok(()) } -/// The open writers of one extension. +/// One open writer and the handles that share it. +struct Open { + writer: Arc, + key: String, + handles: usize, +} + #[derive(Default)] -pub(crate) struct Writers { - next: AtomicU64, - open: Mutex>>, +struct Table { + open: HashMap, /// The writer of each (namespace, Stream, Producer id, epoch), so a second open shares its /// sequence instead of reusing it. - by_producer: Mutex>, - /// The last `sequence` of each closed writer, so a reopen of the same epoch goes on from - /// it instead of sending a used `sequence` again. - closed_sequences: Mutex>, - keys: Mutex>, + by_producer: HashMap, +} + +impl Table { + /// Releases one handle of a writer, or every handle with `all`. Answers the writer when it + /// has no handle left, so the caller ends it. + fn release(&mut self, id: u64, all: bool) -> StreamResult>> { + let open = self + .open + .get_mut(&id) + .ok_or_else(|| StreamError::new(Reason::UnknownWriter, format!("No writer {id}")))?; + open.handles -= 1; + if open.handles > 0 && !all { + return Ok(None); + } + let open = self.open.remove(&id).expect("found above"); + self.by_producer.remove(&open.key); + Ok(Some(open.writer)) + } +} + +/// The open writers of one runtime. +#[derive(Default)] +pub(crate) struct Writers { + next: AtomicU64, + table: Mutex, } impl Writers { - /// Opens a writer, or finds the one this Producer epoch has on the Stream. A writer of an - /// Activity attempt above 1 first sends an empty batch, so the new epoch withdraws the - /// earlier attempt's Records also when it publishes nothing. + /// Opens a writer, or adds a handle to the one this Producer epoch has on the Stream. A + /// writer of an Activity attempt above 1 first sends an empty batch, so the new epoch + /// withdraws the earlier attempt's Records also when it publishes nothing. pub(crate) async fn open( &self, service: &Arc, @@ -210,8 +242,8 @@ impl Writers { producer.id, crate::epoch::epoch_key(&epoch) ); - if let Some(id) = self.by_producer.lock().get(&key) { - return Ok(*id); + if let Some(id) = self.share(&key) { + return Ok(id); } let writer = Arc::new(Writer { namespace: request.namespace, @@ -228,47 +260,55 @@ impl Writers { state: Mutex::default(), writing: tokio::sync::Mutex::new(()), }); - let reopened = self.closed_sequences.lock().get(&key).copied(); - if let Some(sequence) = reopened { - writer.state.lock().sequence = sequence; - } - let retry = reopened.is_none() - && matches!( - ProducerKind::try_from(writer.producer.kind).unwrap_or_default(), - ProducerKind::Activity - | ProducerKind::LocalActivity - | ProducerKind::StandaloneActivity - ) - && epoch.attempt > 1; + let retry = matches!( + ProducerKind::try_from(writer.producer.kind).unwrap_or_default(), + ProducerKind::Activity | ProducerKind::LocalActivity | ProducerKind::StandaloneActivity + ) && epoch.attempt > 1; if retry { - writer.state.lock().sequence = 1; - match writer.send(vec![], 1).await { + match writer.send(vec![], next_sequence()).await { // A Stream with no key has nothing to withdraw yet. Ok(_) => {} Err(error) if error.reason == Reason::NotFound => {} Err(error) => return Err(error), } } - let id = self.next.fetch_add(1, Ordering::Relaxed) + 1; - let mut by_producer = self.by_producer.lock(); - if let Some(id) = by_producer.get(&key) { - return Ok(*id); + let mut table = self.table.lock(); + if let Some(id) = table.by_producer.get(&key).copied() { + table.open.get_mut(&id).expect("indexed").handles += 1; + return Ok(id); } - by_producer.insert(key.clone(), id); - self.keys.lock().insert(id, key); - self.open.lock().insert(id, writer); + let id = self.next.fetch_add(1, Ordering::Relaxed) + 1; + table.by_producer.insert(key.clone(), id); + table.open.insert( + id, + Open { + writer, + key, + handles: 1, + }, + ); Ok(id) } + /// Adds a handle to the open writer of `key`, if there is one. + fn share(&self, key: &str) -> Option { + let mut table = self.table.lock(); + let id = table.by_producer.get(key).copied()?; + table.open.get_mut(&id).expect("indexed").handles += 1; + Some(id) + } + fn get(&self, id: u64) -> StreamResult> { - self.open + self.table .lock() + .open .get(&id) - .cloned() + .map(|open| open.writer.clone()) .ok_or_else(|| StreamError::new(Reason::UnknownWriter, format!("No writer {id}"))) } - /// Appends Records at once and answers the first and the last offset. A retry after a Store failure resends the same batch, so it goes in once. + /// Appends Records at once and answers the first and the last offset. A retry after a Store + /// failure resends the same batch, so it goes in once. pub(crate) async fn write( &self, id: u64, @@ -291,37 +331,54 @@ impl Writers { writer.write(records).await } - /// Closes every open writer. Answers how many had failed. + /// Closes every open writer, whatever its handles. Answers how many had failed. pub(crate) async fn close_all(&self) -> usize { - let ids: Vec = self.open.lock().keys().copied().collect(); + let ids: Vec = self.table.lock().open.keys().copied().collect(); let mut failed = 0; for id in ids { - if self.close(id).await.is_err() { + let released = self.table.lock().release(id, true); + let Ok(Some(writer)) = released else { + continue; + }; + if end(&writer).await.is_err() { failed += 1; } } failed } - /// Waits for a write in flight, then drops the writer. Answers its last offset, or the - /// error it failed with. + /// Releases one handle. The last one waits for a write in flight, then drops the writer. + /// Answers its last offset, or the error it failed with. pub(crate) async fn close(&self, id: u64) -> StreamResult { let writer = self.get(id)?; - writer.state.lock().closed = true; - let result = { - let _writing = writer.writing.lock().await; + if self.table.lock().release(id, false)?.is_none() { let state = writer.state.lock(); - match &state.failed { + return match &state.failed { Some(error) => Err(error.clone()), None => Ok(state.last_offset), - } - }; - self.open.lock().remove(&id); - if let Some(key) = self.keys.lock().remove(&id) { - self.by_producer.lock().remove(&key); - let sequence = writer.state.lock().sequence; - self.closed_sequences.lock().insert(key, sequence); + }; } - result + end(&writer).await + } + + /// Releases one handle with no wait, for a handle that lang drops. A write in flight still + /// ends, and a write that waits behind it is refused once no handle is left. + pub(crate) fn free(&self, id: u64) -> StreamResult<()> { + if let Some(writer) = self.table.lock().release(id, false)? { + writer.state.lock().closed = true; + } + Ok(()) + } +} + +/// Ends a writer that no handle uses: waits for a write in flight, then answers its last offset +/// or the error it failed with. +async fn end(writer: &Writer) -> StreamResult { + writer.state.lock().closed = true; + let _writing = writer.writing.lock().await; + let state = writer.state.lock(); + match &state.failed { + Some(error) => Err(error.clone()), + None => Ok(state.last_offset), } } diff --git a/crates/streams/testdata/gen_vectors.py b/crates/streams/testdata/gen_vectors.py index 3369fd0..69ca4e2 100644 --- a/crates/streams/testdata/gen_vectors.py +++ b/crates/streams/testdata/gen_vectors.py @@ -100,6 +100,7 @@ def ids(): "process_key": "p-1", "id": "client:" + join("1@host", "p-1"), }, + "named_client": {"name": "ingest:eu/1", "id": "client:" + join("ingest:eu/1")}, } return {"streams": cases, "producers": producers} @@ -205,6 +206,9 @@ def withdraws(supersession, record): producer = record["producer"] if supersession.get("withdraw_run") and producer.get("run_id") == supersession["withdraw_run"]: return True + # A client's newer epoch only fences. Its earlier Records are real messages. + if supersession["producer_kind"] == "client": + return False old, new = producer.get("epoch", {}), supersession.get("epoch", {}) if supersession["producer_kind"] != "workflow": return ( @@ -326,6 +330,10 @@ def completed(started, attempt=1, time=(100, 0)): case("an Activity retry withdraws attempt 1", s, act, True) other = dict(act, id="act:" + join(r1, "b", "s")) case("an Activity retry keeps another Activity", s, other, False) + # A client's newer epoch fences its earlier one and withdraws nothing. + named = {"kind": "client", "id": "client:" + join("ingest"), "epoch": {"attempt": 1}} + s = {"producer_kind": "client", "producer_id": named["id"], "epoch": {"attempt": 2}} + case("a client's newer epoch keeps its earlier Records", s, named, False) return cases diff --git a/crates/streams/testdata/ids.json b/crates/streams/testdata/ids.json index 0793666..c359a9e 100644 --- a/crates/streams/testdata/ids.json +++ b/crates/streams/testdata/ids.json @@ -56,6 +56,10 @@ "identity": "1@host", "process_key": "p-1", "id": "client:1%40host:p-1" + }, + "named_client": { + "name": "ingest:eu/1", + "id": "client:ingest%3Aeu%2F1" } } } diff --git a/crates/streams/testdata/withdrawals.json b/crates/streams/testdata/withdrawals.json index 057d84f..e37e85b 100644 --- a/crates/streams/testdata/withdrawals.json +++ b/crates/streams/testdata/withdrawals.json @@ -748,5 +748,25 @@ } }, "withdraws": false + }, + { + "name": "a client's newer epoch keeps its earlier Records", + "supersession": { + "producer_kind": "client", + "producer_id": "client:ingest", + "epoch": { + "attempt": 2 + } + }, + "record": { + "producer": { + "kind": "client", + "id": "client:ingest", + "epoch": { + "attempt": 1 + } + } + }, + "withdraws": false } ] diff --git a/crates/streams/tests/capabilities.rs b/crates/streams/tests/capabilities.rs index 280bef5..6eb56a3 100644 --- a/crates/streams/tests/capabilities.rs +++ b/crates/streams/tests/capabilities.rs @@ -8,9 +8,8 @@ use std::{ sync::Arc, time::{Duration, SystemTime, UNIX_EPOCH}, }; -use temporalio_sdk_core::{ExtensionError, WorkerExtension}; use temporalio_streams::{ - MemoryStreamExtension, StreamExtension, StreamResult, + MemoryStreamExtension, StreamError, StreamExtension, StreamResult, ids::{client_producer_id, default_stream_id}, memory::MemoryStore, proto::{ @@ -149,9 +148,9 @@ fn client() -> Producer { } /// The capability an error names, after it checked the code and the reason. -fn refused_capability(error: &ExtensionError) -> String { - assert_eq!(error.code, Code::Unimplemented, "{error:?}"); - let details = StreamErrorDetails::decode(error.details.as_slice()).unwrap(); +fn refused_capability(error: &StreamError) -> String { + assert_eq!(error.code(), Code::Unimplemented, "{error:?}"); + let details = StreamErrorDetails::decode(error.details().as_slice()).unwrap(); assert_eq!(details.reason, "stream_unsupported"); error .message @@ -169,14 +168,8 @@ impl Case { }) } - async fn call( - &self, - method: &str, - request: Req, - ) -> Result, ExtensionError> { - self.ext - .call_async(method.to_string(), request.encode_to_vec()) - .await + async fn call(&self, method: &str, request: Req) -> Result, StreamError> { + self.ext.call(method, &request.encode_to_vec()).await } async fn create(&self) { @@ -199,7 +192,7 @@ impl Case { } /// A create from a Workflow's call outside Workflow code, which names the owning run. - async fn create_owned(&self, withdraw_run: &str) -> Result, ExtensionError> { + async fn create_owned(&self, withdraw_run: &str) -> Result, StreamError> { let request = CreateStreamRequest { namespace: NAMESPACE.to_string(), stream_id: self.stream.clone(), diff --git a/crates/streams/tests/service.rs b/crates/streams/tests/service.rs index 05d7697..18521a8 100644 --- a/crates/streams/tests/service.rs +++ b/crates/streams/tests/service.rs @@ -2,6 +2,8 @@ mod support; +use support::Needs; + use prost::Message; use prost_types::Timestamp; use std::{ @@ -9,11 +11,10 @@ use std::{ time::{Duration, Instant}, }; use temporalio_common::protos::temporal::api::{common::v1::Payload, failure::v1::Failure}; -use temporalio_sdk_core::{ExtensionError, WorkerExtension}; use temporalio_streams::{ - MemoryStreamExtension, RedisStreamExtension, + MemoryStreamExtension, RedisStreamExtension, StreamError, epoch::withdraws, - ids::{activity_producer_id, client_producer_id}, + ids::{activity_producer_id, client_producer_id, named_client_producer_id}, proto::{ AppendRecord, AppendRecordsRequest, AppendRecordsResponse, CloseStatus, CloseStreamRequest, CloseStreamResponse, CreateStreamRequest, CreateStreamResponse, DescribeStreamRequest, @@ -28,6 +29,8 @@ struct Case { ext: Arc, namespace: String, stream: String, + /// What the backend gives, for a step that needs more than the case asked for. + gives: Needs, } static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); @@ -41,15 +44,18 @@ fn unique() -> String { format!("{now}-{n}") } -async fn cases() -> Vec { - support::extensions(|_| {}, true) +/// A fresh Stream on each backend that gives `needs`. +async fn cases_for(needs: Needs) -> Vec { + support::runtimes(needs) + .await .into_iter() - .map(|ext| { + .map(|runtime| { let unique = unique(); Case { - ext, - namespace: format!("ns-{unique}"), + ext: runtime.ext, + namespace: runtime.namespace.unwrap_or(format!("ns-{unique}")), stream: format!("s-{unique}"), + gives: runtime.gives, } }) .collect() @@ -59,15 +65,13 @@ async fn call( ext: &RedisStreamExtension, method: &str, request: Req, -) -> Result { - let bytes = ext - .call_async(method.to_string(), request.encode_to_vec()) - .await?; +) -> Result { + let bytes = ext.call(method, &request.encode_to_vec()).await?; Ok(Resp::decode(bytes.as_slice()).unwrap()) } -fn reason(error: &ExtensionError) -> String { - StreamErrorDetails::decode(error.details.as_slice()) +fn reason(error: &StreamError) -> String { + StreamErrorDetails::decode(error.details().as_slice()) .unwrap() .reason } @@ -84,6 +88,21 @@ fn client(attempt: i32) -> Producer { } } +/// A client Producer that a test names, at `attempt`. Every backend takes it as sent. +fn named(name: &str, attempt: i32) -> Producer { + Producer { + kind: ProducerKind::Client as i32, + id: named_client_producer_id(name), + epoch: Some(ProducerEpoch { + attempt, + ..Default::default() + }), + run_id: String::new(), + } +} + +/// An Activity Producer of a made-up run, at `attempt`. A higher attempt withdraws the lower +/// ones' Records. The server wants the Activity's task token, so only Redis and memory take it. fn activity(attempt: i32, scheduled: i64) -> Producer { Producer { kind: ProducerKind::Activity as i32, @@ -129,7 +148,7 @@ impl Case { async fn create( &self, retention: Option, - ) -> Result { + ) -> Result { call( &self.ext, "CreateStream", @@ -146,7 +165,7 @@ impl Case { async fn create_auto_close( &self, auto_close: Option, - ) -> Result { + ) -> Result { call( &self.ext, "CreateStream", @@ -165,7 +184,7 @@ impl Case { producer: Producer, sequence: i64, records: Vec, - ) -> Result { + ) -> Result { call( &self.ext, "AppendRecords", @@ -181,10 +200,7 @@ impl Case { .await } - async fn read( - &self, - request: ReadRecordsRequest, - ) -> Result { + async fn read(&self, request: ReadRecordsRequest) -> Result { call( &self.ext, "ReadRecords", @@ -197,7 +213,7 @@ impl Case { .await } - async fn describe(&self) -> Result { + async fn describe(&self) -> Result { call( &self.ext, "DescribeStream", @@ -209,7 +225,7 @@ impl Case { .await } - async fn close(&self, producer: Producer) -> Result { + async fn close(&self, producer: Producer) -> Result { call( &self.ext, "CloseStream", @@ -245,22 +261,25 @@ fn texts(response: &ReadRecordsResponse) -> Vec { .collect() } +/// A create is idempotent for one Retention, and another Retention fails with +/// `stream_already_exists`. #[tokio::test] async fn a_create_is_idempotent_for_one_retention() { - for case in cases().await { + for case in cases_for(Needs::CALLS).await { let created = case.create(None).await.unwrap(); assert!(created.created); assert_eq!(created.status, StreamStatus::Open as i32); assert!(!case.create(None).await.unwrap().created); let error = case.create(millis(1000)).await.unwrap_err(); - assert_eq!(error.code, Code::AlreadyExists); + assert_eq!(error.code(), Code::AlreadyExists); assert_eq!(reason(&error), "stream_already_exists"); } } +/// A Producer's id must match its kind, and a Workflow appends only through a Local Activity. #[tokio::test] async fn a_producer_must_match_its_kind_and_not_be_a_workflow() { - for case in cases().await { + for case in cases_for(Needs::CALLS).await { case.create(None).await.unwrap(); let wrong = Producer { kind: ProducerKind::Activity as i32, @@ -271,7 +290,7 @@ async fn a_producer_must_match_its_kind_and_not_be_a_workflow() { .await .unwrap_err(); assert_eq!( - (error.code, reason(&error).as_str()), + (error.code(), reason(&error).as_str()), (Code::InvalidArgument, "stream_invalid") ); let workflow = Producer { @@ -281,13 +300,15 @@ async fn a_producer_must_match_its_kind_and_not_be_a_workflow() { epoch: None, }; let error = case.append(workflow, 1, vec![]).await.unwrap_err(); - assert_eq!(error.code, Code::InvalidArgument); + assert_eq!(error.code(), Code::InvalidArgument); } } +/// A read is inclusive from its offset, a read from the end gives only new Records, and an +/// offset past the newest + 1 fails with `stream_offset`. #[tokio::test] async fn a_read_is_inclusive_from_its_offset_and_get_offset_reads_only_new_records() { - for case in cases().await { + for case in cases_for(Needs::CALLS).await { case.create(None).await.unwrap(); let first = case .append(client(1), 1, vec![record("a", "1"), record("b", "2")]) @@ -331,15 +352,16 @@ async fn a_read_is_inclusive_from_its_offset_and_get_offset_reads_only_new_recor .await .unwrap_err(); assert_eq!( - (error.code, reason(&error).as_str()), + (error.code(), reason(&error).as_str()), (Code::InvalidArgument, "stream_offset") ); } } +/// A Topic filter keeps the `Superseded` items, which pass every filter. #[tokio::test] async fn a_topic_filter_keeps_superseded_items() { - for case in cases().await { + for case in cases_for(Needs::UNCHECKED).await { case.create(None).await.unwrap(); case.append(activity(1, 10), 1, vec![record("a", "old")]) .await @@ -353,8 +375,11 @@ async fn a_topic_filter_keeps_superseded_items() { .unwrap(); let all = case.read(ReadRecordsRequest::default()).await.unwrap(); assert_eq!(texts(&all), ["old", "", "b1", "a1"]); + // The item comes just before the batch's first Record. Redis keeps both in one entry with + // one offset, and the server gives the item its own offset just below. let superseded = &all.records[1]; - assert_eq!(superseded.offset, all.records[2].offset); + assert!(all.records[0].offset < superseded.offset); + assert!(superseded.offset <= all.records[2].offset); let supersession = superseded.supersession.as_ref().unwrap(); assert_eq!(supersession.producer_id, activity(2, 20).id); let only_a = case @@ -368,12 +393,13 @@ async fn a_topic_filter_keeps_superseded_items() { } } +/// A read waits for the Stream's create only as long as `create_wait`. #[tokio::test] async fn a_read_waits_for_the_create_only_as_long_as_create_wait() { - for case in cases().await { + for case in cases_for(Needs::CALLS).await { let error = case.read(ReadRecordsRequest::default()).await.unwrap_err(); assert_eq!( - (error.code, reason(&error).as_str()), + (error.code(), reason(&error).as_str()), (Code::NotFound, "stream_not_found") ); let started = Instant::now(); @@ -384,7 +410,7 @@ async fn a_read_waits_for_the_create_only_as_long_as_create_wait() { }) .await .unwrap_err(); - assert_eq!(error.code, Code::NotFound); + assert_eq!(error.code(), Code::NotFound); assert!(started.elapsed() >= Duration::from_millis(200)); let ext = case.ext.clone(); let (namespace, stream) = (case.namespace.clone(), case.stream.clone()); @@ -409,9 +435,10 @@ async fn a_read_waits_for_the_create_only_as_long_as_create_wait() { } } +/// A quiet read ends empty after its wait, and a long poll wakes on an append. #[tokio::test] async fn a_quiet_read_ends_empty_after_its_wait_and_a_long_poll_wakes_on_append() { - for case in cases().await { + for case in cases_for(Needs::CALLS).await { case.create(None).await.unwrap(); let first = case .append(client(1), 1, vec![record("", "1")]) @@ -462,9 +489,11 @@ async fn a_quiet_read_ends_empty_after_its_wait_and_a_long_poll_wakes_on_append( } } +/// A close ends every read with its close Record and refuses later appends, and a create finds +/// the Stream closed. #[tokio::test] async fn a_close_ends_every_read_with_its_record() { - for case in cases().await { + for case in cases_for(Needs::CALLS).await { case.create(None).await.unwrap(); case.append(client(1), 1, vec![record("", "1")]) .await @@ -492,7 +521,7 @@ async fn a_close_ends_every_read_with_its_record() { .await .unwrap_err(); assert_eq!( - (error.code, reason(&error).as_str()), + (error.code(), reason(&error).as_str()), (Code::FailedPrecondition, "stream_closed") ); let created = case.create(None).await.unwrap(); @@ -500,9 +529,10 @@ async fn a_close_ends_every_read_with_its_record() { } } +/// A read from an offset that the Retention trimmed fails with `stream_expired`. #[tokio::test] async fn a_read_from_a_trimmed_offset_fails_as_expired() { - for case in cases().await { + for case in cases_for(Needs::SHORT_TIMES).await { case.create(millis(200)).await.unwrap(); // Each write keeps the client Stream one more Retention, so only the old Record goes. let old = case @@ -525,7 +555,7 @@ async fn a_read_from_a_trimmed_offset_fails_as_expired() { .await .unwrap_err(); assert_eq!( - (error.code, reason(&error).as_str()), + (error.code(), reason(&error).as_str()), (Code::NotFound, "stream_expired") ); let from_start = case.read(ReadRecordsRequest::default()).await.unwrap(); @@ -533,25 +563,19 @@ async fn a_read_from_a_trimmed_offset_fails_as_expired() { } } +/// An unknown method fails with UNIMPLEMENTED. #[tokio::test] async fn an_unknown_method_is_unimplemented() { - for case in cases().await { - let error = case - .ext - .call_async("Nope".to_string(), vec![]) - .await - .unwrap_err(); - assert_eq!(error.code, Code::Unimplemented); - assert_eq!( - case.ext.call("Nope", &[]).unwrap_err().code, - Code::Unimplemented - ); + for case in cases_for(Needs::CALLS).await { + let error = case.ext.call("Nope", &[]).await.unwrap_err(); + assert_eq!(error.code(), Code::Unimplemented); } } +/// A read of an open Stream that expired fails at once instead of waiting. #[tokio::test] async fn a_read_of_an_expired_open_stream_fails_instead_of_waiting() { - for case in cases().await { + for case in cases_for(Needs::SHORT_TIMES).await { case.create(millis(300)).await.unwrap(); case.append(client(1), 1, vec![record("", "1")]) .await @@ -575,9 +599,10 @@ async fn a_read_of_an_expired_open_stream_fails_instead_of_waiting() { } } +/// A prefix filter reads the Topics under it and keeps the `Superseded` items. #[tokio::test] async fn a_prefix_filter_reads_the_topics_under_it_and_keeps_superseded_items() { - for case in cases().await { + for case in cases_for(Needs::UNCHECKED).await { case.create(None).await.unwrap(); case.append(activity(1, 10), 1, vec![record("tool/a", "old")]) .await @@ -615,9 +640,10 @@ async fn a_prefix_filter_reads_the_topics_under_it_and_keeps_superseded_items() } } +/// A ref with the owner fields and one without reach the same Stream. #[tokio::test] async fn a_ref_with_the_owner_fields_and_one_without_reach_the_same_stream() { - for case in cases().await { + for case in cases_for(Needs::ID_KEYS).await { case.create(None).await.unwrap(); let owned = Some(StreamRef { id: case.stream.clone(), @@ -658,9 +684,10 @@ async fn a_ref_with_the_owner_fields_and_one_without_reach_the_same_stream() { } } +/// An auto close ends a long poll with its close Record and refuses later appends. #[tokio::test] async fn an_auto_close_ends_a_long_poll_and_refuses_later_appends() { - for case in cases().await { + for case in cases_for(Needs::SHORT_TIMES).await { let created = case.create_auto_close(millis(400)).await.unwrap(); assert!(created.created); assert!(case.create_auto_close(millis(400)).await.is_ok()); @@ -704,9 +731,10 @@ async fn an_auto_close_ends_a_long_poll_and_refuses_later_appends() { } } +/// A describe or a read applies a due auto close. #[tokio::test] async fn a_describe_or_a_read_applies_a_due_auto_close() { - for case in cases().await { + for case in cases_for(Needs::SHORT_TIMES).await { case.create_auto_close(millis(200)).await.unwrap(); tokio::time::sleep(Duration::from_millis(300)).await; let described = case.describe().await.unwrap(); @@ -716,7 +744,7 @@ async fn a_describe_or_a_read_applies_a_due_auto_close() { assert_eq!(texts(&read), [""]); assert_eq!(read.records[0].offset, described.last_offset); } - for case in cases().await { + for case in cases_for(Needs::SHORT_TIMES).await { case.create_auto_close(millis(200)).await.unwrap(); tokio::time::sleep(Duration::from_millis(300)).await; let read = case.read(ReadRecordsRequest::default()).await.unwrap(); @@ -725,6 +753,7 @@ async fn a_describe_or_a_read_applies_a_due_auto_close() { } } +/// A create answer names the configured backend, and the Store never reads it. #[tokio::test] async fn a_create_answer_names_the_backend_and_the_store_never_reads_it() { let mut configured = @@ -741,6 +770,7 @@ async fn a_create_answer_names_the_backend_and_the_store_never_reads_it() { ext, namespace: format!("ns-{unique}"), stream: format!("s-{unique}"), + gives: Needs::ALL, }; let created = case.create(None).await.unwrap(); let stream = created.stream.unwrap(); @@ -774,7 +804,7 @@ async fn a_create_answer_names_the_backend_and_the_store_never_reads_it() { assert_eq!(texts(&read), ["1", "2", "3"]); } // With no backend in the config, the ref has none, and lang reads it as the default. - for case in cases().await { + for case in cases_for(Needs::OWN_CONFIG).await { let created = case.create(None).await.unwrap(); assert_eq!(created.stream.unwrap().backend, ""); } @@ -808,9 +838,10 @@ fn workflow_owner(run: &str) -> Producer { } } +/// A retry run's Local Activity calls withdraw the previous run. #[tokio::test] async fn a_retry_run_s_local_activity_calls_withdraw_the_previous_run() { - for case in cases().await { + for case in cases_for(Needs::UNCHECKED).await { let (previous, retry) = (format!("{}-r1", case.stream), format!("{}-r2", case.stream)); let create = |run: &str, withdraw_run: &str| CreateStreamRequest { namespace: case.namespace.clone(), @@ -896,9 +927,10 @@ fn user_local_activity(run: &str, started: i64, seconds: i64) -> Producer { } } +/// A Local Activity rerun in a later task supersedes its first attempt. #[tokio::test] async fn a_rerun_local_activity_supersedes_its_first_attempt() { - for case in cases().await { + for case in cases_for(Needs::UNCHECKED).await { let run = format!("{}-run", case.stream); let _: CreateStreamResponse = call( &case.ext, @@ -984,9 +1016,10 @@ fn local_activity_write( } } +/// A Local Activity of a task that never completed is withdrawn by the run's next write. #[tokio::test] async fn a_local_activity_of_a_task_that_never_completed_is_withdrawn_by_the_next_write() { - for case in cases().await { + for case in cases_for(Needs::UNCHECKED).await { let run = format!("{}-run", case.stream); case.create(None).await.unwrap(); // Task [2] completed. Task [3] runs a Local Activity that writes, then task [3] fails. @@ -1049,9 +1082,10 @@ async fn a_local_activity_of_a_task_that_never_completed_is_withdrawn_by_the_nex } } +/// A failed task before the last completed one is withdrawn by its id. #[tokio::test] async fn a_failed_task_before_the_last_completed_one_is_withdrawn_by_its_id() { - for case in cases().await { + for case in cases_for(Needs::UNCHECKED).await { let run = format!("{}-run", case.stream); case.create(None).await.unwrap(); let _: AppendRecordsResponse = call( @@ -1073,9 +1107,10 @@ async fn a_failed_task_before_the_last_completed_one_is_withdrawn_by_its_id() { } } +/// A close Record gives its status, and a caller can't close as terminated or timed out. #[tokio::test] async fn a_close_record_gives_its_status_and_a_caller_cant_terminate() { - for case in cases().await { + for case in cases_for(Needs::CALLS).await { case.create(None).await.unwrap(); let close = |status: CloseStatus, failure: Option| CloseStreamRequest { namespace: case.namespace.clone(), @@ -1091,7 +1126,7 @@ async fn a_close_record_gives_its_status_and_a_caller_cant_terminate() { .await .unwrap_err(); assert_eq!( - (error.code, reason(&error).as_str()), + (error.code(), reason(&error).as_str()), (Code::InvalidArgument, "stream_invalid") ); } @@ -1110,9 +1145,11 @@ async fn a_close_record_gives_its_status_and_a_caller_cant_terminate() { } } +/// A last read gives the newest Records oldest first, skips withdrawn ones, and ends with the +/// close Record of a closed Stream. #[tokio::test] async fn a_last_read_gives_the_newest_records_oldest_first_without_withdrawn_ones() { - for case in cases().await { + for case in cases_for(Needs::CALLS).await { case.create(None).await.unwrap(); let last = |n: i32, topics: &[&str]| ReadRecordsRequest { last: n, @@ -1122,13 +1159,19 @@ async fn a_last_read_gives_the_newest_records_oldest_first_without_withdrawn_one assert!(case.read(last(5, &[])).await.unwrap().records.is_empty()); let records = vec![record("t", "c1"), record("u", "c2"), record("t", "c3")]; case.append(client(1), 1, records).await.unwrap(); - case.append(activity(1, 10), 1, vec![record("t", "a1")]) - .await - .unwrap(); - // The retry's first batch withdraws a1, so a1 doesn't count. - case.append(activity(2, 20), 1, vec![record("t", "a2")]) - .await - .unwrap(); + if case.gives.gives(Needs::UNCHECKED) { + case.append(activity(1, 10), 1, vec![record("t", "a1")]) + .await + .unwrap(); + // The retry's first batch withdraws a1, so a1 doesn't count. + case.append(activity(2, 20), 1, vec![record("t", "a2")]) + .await + .unwrap(); + } else { + case.append(named("w", 1), 1, vec![record("t", "a2")]) + .await + .unwrap(); + } assert_eq!( texts(&case.read(last(3, &[])).await.unwrap()), ["c2", "c3", "a2"] @@ -1153,9 +1196,10 @@ async fn a_last_read_gives_the_newest_records_oldest_first_without_withdrawn_one } } +/// A page keeps under `max_bytes`, and a larger Record still comes alone. #[tokio::test] async fn a_page_keeps_under_max_bytes_and_a_larger_record_comes_alone() { - for case in cases().await { + for case in cases_for(Needs::CALLS).await { case.create(None).await.unwrap(); let big = "x".repeat(1000); let records = vec![ diff --git a/crates/streams/tests/store.rs b/crates/streams/tests/store.rs index 25e4a65..ce36aa5 100644 --- a/crates/streams/tests/store.rs +++ b/crates/streams/tests/store.rs @@ -7,10 +7,12 @@ mod support; use prost::Message; use prost_types::Timestamp; use std::time::Duration; -use support::{Run, Target, digest, record, records_of, stores, supersession}; +use support::{Needs, Run, Target, digest, record, records_of, stores_for, supersession}; use temporalio_streams::{ Reason, - ids::{activity_producer_id, client_producer_id, workflow_producer_id}, + ids::{ + activity_producer_id, client_producer_id, named_client_producer_id, workflow_producer_id, + }, proto::{Producer, ProducerEpoch, ProducerKind, StreamRef}, store::{Batch, Capabilities, CreateOutcome, Owner, Withdrawal}, }; @@ -43,6 +45,14 @@ fn client(attempt: i32) -> Producer { } } +/// A second client Producer, which every backend takes as sent. +fn other_client() -> Producer { + Producer { + id: named_client_producer_id("other"), + ..client(1) + } +} + async fn append( target: &Target, producer: &Producer, @@ -69,9 +79,10 @@ async fn append( .await } +/// A create writes the key once, and another Retention is a conflict that keeps the stored one. #[tokio::test] async fn a_create_writes_the_key_once_and_keeps_its_retention() { - for target in stores().await { + for target in stores_for(Needs::STORE_ANSWERS).await { let none = Withdrawal::default(); assert_eq!( append(&target, &client(1), 1, &["x"]) @@ -124,20 +135,21 @@ async fn a_create_writes_the_key_once_and_keeps_its_retention() { } } +/// A resend with the last sequence gets the offsets of its first answer, also after another +/// Producer wrote, and a lower sequence is refused. #[tokio::test] async fn a_resend_gets_its_first_positions_and_a_lower_sequence_is_refused() { - for target in stores().await { + for target in stores_for(Needs::CALLS).await { target.create(Owner::Client).await; let p = client(1); let first = append(&target, &p, 1, &["p1", "p2"]).await.unwrap(); - assert!(!first.resend); // Another Producer writes between the batch and its resend. - append(&target, &activity("r", 1, 10), 1, &["q1"]) - .await - .unwrap(); + append(&target, &other_client(), 1, &["q1"]).await.unwrap(); let resend = append(&target, &p, 1, &["p1", "p2"]).await.unwrap(); - assert!(resend.resend); assert_eq!(resend.offsets, first.offsets); + if target.gives.gives(Needs::STORE_ANSWERS) { + assert!(!first.resend && resend.resend); + } append(&target, &p, 2, &["p3"]).await.unwrap(); assert_eq!( append(&target, &p, 1, &["p1"]).await.unwrap_err().reason, @@ -148,15 +160,20 @@ async fn a_resend_gets_its_first_positions_and_a_lower_sequence_is_refused() { } } +/// A resend with the last sequence and other content fails as `seq_divergent` and writes +/// nothing. With `RAW_KEYS`, a dedupe entry from before the digest takes any resend. #[tokio::test] async fn a_resend_with_other_content_fails_as_divergent() { - for target in stores().await { + for target in stores_for(Needs::CALLS).await { target.create(Owner::Client).await; let p = client(1); let first = append(&target, &p, 1, &["p1"]).await.unwrap(); let divergent = append(&target, &p, 1, &["other"]).await.unwrap_err(); assert_eq!(divergent.reason, Reason::SeqDivergent); assert_eq!(target.texts().await, ["p1"]); + if !target.gives.gives(Needs::RAW_KEYS) { + continue; + } // A dedupe entry written before the digest has none, so it takes any resend. let field = temporalio_streams::ids::fields::dedupe(ProducerKind::Client as i32, &p.id); let entry = target.meta(&field).await.unwrap(); @@ -168,9 +185,10 @@ async fn a_resend_with_other_content_fails_as_divergent() { } } +/// A lower epoch is fenced for each Producer kind. #[tokio::test] async fn a_lower_epoch_is_fenced_for_each_producer_kind() { - for target in stores().await { + for target in stores_for(Needs::UNCHECKED).await { target.create(Owner::Workflow).await; let run = Run::new("r1"); let la = run.local_activity("k", 8, 1, 2); @@ -191,14 +209,15 @@ async fn a_lower_epoch_is_fenced_for_each_producer_kind() { } } +/// A close refuses later appends but not a resend of an earlier batch, and a create finds the +/// Stream closed. With `STORE_ANSWERS`, a second close answers `already` with the same offset. #[tokio::test] async fn a_close_stops_appends_but_not_a_resend() { - for target in stores().await { + for target in stores_for(Needs::CALLS).await { target.create(Owner::Client).await; let p = client(1); let first = append(&target, &p, 1, &["a"]).await.unwrap(); let closed = target.close(&p).await.unwrap(); - assert!(!closed.already); assert_eq!( append(&target, &p, 2, &["b"]).await.unwrap_err().reason, Reason::Closed @@ -208,8 +227,10 @@ async fn a_close_stops_appends_but_not_a_resend() { first.offsets ); let again = target.close(&p).await.unwrap(); - assert!(again.already); - assert_eq!(again.offset, closed.offset); + if target.gives.gives(Needs::STORE_ANSWERS) { + assert!(!closed.already && again.already); + assert_eq!(again.offset, closed.offset); + } let created = target .store .create( @@ -226,9 +247,10 @@ async fn a_close_stops_appends_but_not_a_resend() { } } +/// A Stream with a newer format version refuses every write. #[tokio::test] async fn a_newer_format_refuses_every_write() { - for target in stores().await { + for target in stores_for(Needs::RAW_KEYS).await { target.create(Owner::Client).await; target.set_meta("v", "2").await; let error = append(&target, &client(1), 1, &["a"]).await.unwrap_err(); @@ -252,9 +274,10 @@ async fn a_newer_format_refuses_every_write() { } } +/// An expired Stream refuses writes, and its create finds it expired. #[tokio::test] async fn an_expired_stream_refuses_writes_and_its_create() { - for target in stores().await { + for target in stores_for(Needs::RAW_KEYS).await { target.create(Owner::Client).await; target.set_meta("exp", "1").await; let error = append(&target, &client(1), 1, &["a"]).await.unwrap_err(); @@ -275,9 +298,10 @@ async fn an_expired_stream_refuses_writes_and_its_create() { } } +/// An Activity retry flags only its first batch, which withdraws the earlier attempt. #[tokio::test] async fn an_activity_retry_flags_its_first_batch_only() { - for target in stores().await { + for target in stores_for(Needs::UNCHECKED.and(Needs::STORE_ANSWERS)).await { target.create(Owner::Workflow).await; let (a1, a2) = (activity("r1", 1, 10), activity("r1", 2, 20)); let other = Producer { @@ -292,9 +316,10 @@ async fn an_activity_retry_flags_its_first_batch_only() { } } +/// An empty batch of an Activity retry withdraws the earlier attempt's Records. #[tokio::test] async fn an_empty_batch_of_a_retry_withdraws_the_earlier_attempt() { - for target in stores().await { + for target in stores_for(Needs::UNCHECKED).await { target.create(Owner::Workflow).await; append(&target, &activity("r1", 1, 10), 1, &["x1"]) .await @@ -317,9 +342,10 @@ async fn an_empty_batch_of_a_retry_withdraws_the_earlier_attempt() { } } +/// A retry run's first write withdraws the previous run and fences its Producers. #[tokio::test] async fn a_retry_run_withdraws_and_fences_the_previous_run() { - for target in stores().await { + for target in stores_for(Needs::UNCHECKED).await { target.create(Owner::Workflow).await; let (r1, r2) = (Run::new("r1"), Run::new("r2")); target @@ -350,9 +376,10 @@ async fn a_retry_run_withdraws_and_fences_the_previous_run() { } } +/// An open Workflow Stream keeps the safety TTL, and its close gives it the Retention. #[tokio::test] async fn the_ttl_follows_the_owner_and_the_close() { - for target in stores().await { + for target in stores_for(Needs::RAW_KEYS).await { target.create(Owner::Workflow).await; let run = Run::new("r1"); target @@ -380,9 +407,11 @@ async fn the_ttl_follows_the_owner_and_the_close() { } } +/// A read hides Records past the Retention, and the next append trims them and keeps the +/// newest trimmed offset. #[tokio::test] async fn an_append_trims_by_retention_and_keeps_the_watermark() { - for target in stores().await { + for target in stores_for(Needs::RAW_KEYS.and(Needs::SHORT_TIMES)).await { let none = Withdrawal::default(); target .store @@ -418,9 +447,10 @@ async fn an_append_trims_by_retention_and_keeps_the_watermark() { } } +/// A wait for the next entry wakes on an append and ends at its deadline. #[tokio::test] async fn a_blocked_read_wakes_on_an_append_and_a_wait_for_replicas_ends() { - for target in stores().await { + for target in stores_for(Needs::RAW_KEYS).await { target.create(Owner::Client).await; let first = append(&target, &client(1), 1, &["a"]).await.unwrap(); let newest = first.offsets.unwrap().1; @@ -445,9 +475,10 @@ async fn a_blocked_read_wakes_on_an_append_and_a_wait_for_replicas_ends() { } } +/// A sequence below one is refused. #[tokio::test] async fn a_sequence_below_one_is_refused() { - for target in stores().await { + for target in stores_for(Needs::CALLS).await { target.create(Owner::Client).await; for sequence in [0, -1] { let error = append(&target, &client(1), sequence, &["x"]) @@ -458,9 +489,10 @@ async fn a_sequence_below_one_is_refused() { } } +/// A Workflow's create makes a Stream that an Activity created Workflow-owned. #[tokio::test] async fn a_workflow_create_takes_over_a_stream_an_activity_created() { - for target in stores().await { + for target in stores_for(Needs::RAW_KEYS).await { target.create(Owner::Client).await; target.create(Owner::Workflow).await; assert_eq!(target.meta("own").await.as_deref(), Some("wf")); @@ -472,9 +504,10 @@ async fn a_workflow_create_takes_over_a_stream_an_activity_created() { } } +/// A Producer with a lower epoch can't close the Stream. #[tokio::test] async fn a_stale_producer_cant_close_the_stream() { - for target in stores().await { + for target in stores_for(Needs::CALLS).await { target.create(Owner::Client).await; append(&target, &client(2), 1, &["a"]).await.unwrap(); assert_eq!( @@ -485,9 +518,11 @@ async fn a_stale_producer_cant_close_the_stream() { } } +/// Only the owner closes a Stream: its run chain for a Workflow-owned one, a client for a +/// client's. #[tokio::test] async fn only_the_owner_closes_a_stream() { - for target in stores().await { + for target in stores_for(Needs::UNCHECKED.and(Needs::STORE_ANSWERS)).await { let run = Run::new("r1"); let la = run.local_activity("k", 3, 1, 1); target.create(Owner::Client).await; @@ -540,9 +575,10 @@ async fn only_the_owner_closes_a_stream() { } } +/// An append whose entry ids would not fit an offset is refused before it writes. #[tokio::test] async fn an_append_whose_ids_would_not_fit_an_offset_is_refused_before_it_writes() { - for target in stores().await { + for target in stores_for(Needs::RAW_KEYS).await { target.create(Owner::Client).await; // An entry a minute ahead of the clock, with nearly the last 20-bit sequence. let ahead = std::time::SystemTime::now() @@ -559,6 +595,7 @@ async fn an_append_whose_ids_would_not_fit_an_offset_is_refused_before_it_writes } } +/// Every Store keeps the meta fields that the shared vectors name. #[tokio::test] async fn every_store_keeps_the_meta_fields_of_the_shared_vectors() { use temporalio_streams::ids::fields; @@ -570,7 +607,7 @@ async fn every_store_keeps_the_meta_fields_of_the_shared_vectors() { .unwrap(), ) .unwrap(); - for target in stores().await { + for target in stores_for(Needs::RAW_KEYS).await { target.create(Owner::Workflow).await; let producer = activity("run:1", 1, 10); append(&target, &producer, 1, &["a"]).await.unwrap(); @@ -591,9 +628,17 @@ async fn every_store_keeps_the_meta_fields_of_the_shared_vectors() { } } +/// An auto close closes a quiet Stream on its next write, with a close Record that has no +/// Producer and a `COMPLETED` status. Each append holds it off for one more period. #[tokio::test] async fn an_auto_close_closes_a_quiet_stream_on_its_next_write() { - for target in stores().await { + for target in stores_for( + Needs::RAW_KEYS + .and(Needs::SHORT_TIMES) + .and(Needs::STORE_ANSWERS), + ) + .await + { let none = Withdrawal::default(); let period = Some(Duration::from_millis(300)); let stream = target.stream_ref(); @@ -682,9 +727,10 @@ async fn an_auto_close_closes_a_quiet_stream_on_its_next_write() { } } +/// An auto close counts from the create, and each step that finds it due applies it once. #[tokio::test] async fn an_auto_close_counts_from_the_create_and_each_step_applies_it_once() { - for target in stores().await { + for target in stores_for(Needs::RAW_KEYS.and(Needs::SHORT_TIMES)).await { let none = Withdrawal::default(); target .store @@ -741,9 +787,10 @@ async fn an_auto_close_counts_from_the_create_and_each_step_applies_it_once() { } } +/// An auto close period below 1 ms rounds up to 1 ms, and a huge one doesn't overflow. #[tokio::test] async fn an_auto_close_period_at_the_edges_neither_overflows_nor_rounds_to_zero() { - for target in stores().await { + for target in stores_for(Needs::STORE_ANSWERS).await { let none = Withdrawal::default(); let created = target .store @@ -759,7 +806,7 @@ async fn an_auto_close_period_at_the_edges_neither_overflows_nor_rounds_to_zero( .unwrap(); assert_eq!(created.auto_close, Some(Duration::from_millis(1))); } - for target in stores().await { + for target in stores_for(Needs::STORE_ANSWERS).await { let none = Withdrawal::default(); target .store diff --git a/crates/streams/tests/support/mod.rs b/crates/streams/tests/support/mod.rs index 8ebb66e..7936051 100644 --- a/crates/streams/tests/support/mod.rs +++ b/crates/streams/tests/support/mod.rs @@ -30,12 +30,52 @@ pub(crate) use server::server_url; static NEXT: AtomicU64 = AtomicU64::new(0); +/// What a case needs beyond the Store's calls on its own Stream. A case runs on every backend +/// that gives all it needs, so one suite covers each backend as far as its guarantees go. Each +/// case's comment says its guarantee and what it needs. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct Needs(u8); + +impl Needs { + /// Only the Store's calls on the Stream and reads of its Records. + pub(crate) const CALLS: Self = Self(0); + /// The Store's own answers past the API: a conflict's stored Retention, `resend`, `flagged`, + /// and a close's offset and `already`. + pub(crate) const STORE_ANSWERS: Self = Self(1); + /// The keys under the calls: meta fields, TTLs and raw log entries, and the Store's log calls + /// `read`, `read_back` and `wait_after`. + pub(crate) const RAW_KEYS: Self = Self(1 << 1); + /// Producers, owners and runs as sent. The server checks them against its executions, so a + /// case with made-up runs, or an Activity with no task token, can't run there. + pub(crate) const UNCHECKED: Self = Self(1 << 2); + /// A Retention or an auto close of a few hundred ms. + pub(crate) const SHORT_TIMES: Self = Self(1 << 3); + /// The runtime built from the crate's own config, such as a backend name. + pub(crate) const OWN_CONFIG: Self = Self(1 << 4); + /// The Store keys on the id only, so the owner fields of a ref don't route a call. + pub(crate) const ID_KEYS: Self = Self(1 << 5); + /// Every need. Redis and memory give it. + pub(crate) const ALL: Self = Self(u8::MAX); + + /// Both needs. + pub(crate) const fn and(self, other: Self) -> Self { + Self(self.0 | other.0) + } + + /// Whether this set gives every need of `needs`. + pub(crate) const fn gives(self, needs: Self) -> bool { + self.0 & needs.0 == needs.0 + } +} + /// One Stream on one server, fresh for each case. pub(crate) struct Target { pub(crate) store: Arc, pub(crate) backend: Backend, pub(crate) namespace: String, pub(crate) stream: String, + /// What the backend gives. A case checks it before a step that needs more than it asked for. + pub(crate) gives: Needs, } /// The Store under a target, for the steps a test takes under the Store's calls. @@ -60,12 +100,19 @@ fn unique() -> String { } /// A Store on each server that is set, standalone first, then the cluster, then the memory -/// Store. +/// Store, for a case that needs all that they give. pub(crate) async fn stores() -> Vec { - stores_with(|options| options).await + stores_for(Needs::ALL).await +} + +/// The Stores of every backend that gives `needs`. +pub(crate) async fn stores_for(needs: Needs) -> Vec { + let mut targets = stores_with(|options| options).await; + targets.retain(|target| target.gives.gives(needs)); + targets } -/// Like [stores], with the options changed. +/// Like [stores], with the Redis options changed. pub(crate) async fn stores_with( change: impl Fn(RedisStoreOptions) -> RedisStoreOptions, ) -> Vec { @@ -91,6 +138,7 @@ pub(crate) async fn stores_with( }, namespace: format!("ns-{unique}"), stream: format!("wf:{unique}"), + gives: Needs::ALL, }); } // The memory Store runs the same cases, so both Stores keep one set of rules. @@ -102,6 +150,7 @@ pub(crate) async fn stores_with( backend: Backend::Memory(memory), namespace: format!("ns-{unique}"), stream: format!("wf:{unique}"), + gives: Needs::ALL, }); targets } @@ -459,6 +508,29 @@ fn node_of(slots: &redis::Value, slot: u16) -> Option { None } +/// One runtime under test, the namespace its cases use, and what its backend gives. +pub(crate) struct Runtime { + pub(crate) ext: Arc, + /// The namespace of a backend that has fixed ones, such as the server. `None` lets the case + /// make one of its own. + pub(crate) namespace: Option, + pub(crate) gives: Needs, +} + +/// A runtime on each backend that is set and gives `needs`, Redis and memory first. +pub(crate) async fn runtimes(needs: Needs) -> Vec { + let mut runtimes: Vec<_> = extensions(|_| {}, true) + .into_iter() + .map(|ext| Runtime { + ext, + namespace: None, + gives: Needs::ALL, + }) + .collect(); + runtimes.retain(|runtime| runtime.gives.gives(needs)); + runtimes +} + /// An extension on each Redis server that is set, with the config changed by `change`, and on /// the memory Store when `memory` is true. pub(crate) fn extensions( diff --git a/crates/streams/tests/vectors.rs b/crates/streams/tests/vectors.rs index 32a0305..eaa5667 100644 --- a/crates/streams/tests/vectors.rs +++ b/crates/streams/tests/vectors.rs @@ -108,6 +108,10 @@ fn producer_ids_match() { ids::client_producer_id(field("client", "identity"), field("client", "process_key")), id("client") ); + assert_eq!( + ids::named_client_producer_id(field("named_client", "name")), + id("named_client") + ); } #[test] diff --git a/crates/streams/tests/wit.rs b/crates/streams/tests/wit.rs index 448452a..2625458 100644 --- a/crates/streams/tests/wit.rs +++ b/crates/streams/tests/wit.rs @@ -60,7 +60,13 @@ fn the_wit_declares_the_stream_handle_with_its_methods() { fn the_writer_wit_is_client_only_and_names_each_writer_call() { assert!(WRITER_WIT.contains("/// @nexus.client-only\n")); assert!(!WRITER_WIT.contains("@nexus.endpoint")); - for operation in ["OpenWriter", "WriteRecords", "CloseWriter", "CloseWriters"] { + for operation in [ + "OpenWriter", + "WriteRecords", + "CloseWriter", + "FreeWriter", + "CloseWriters", + ] { assert!( WRITER_WIT.contains(&format!("@nexus.operation name=\"{operation}\"")), "{operation}" diff --git a/crates/streams/tests/writer.rs b/crates/streams/tests/writer.rs index f94f667..6ca2325 100644 --- a/crates/streams/tests/writer.rs +++ b/crates/streams/tests/writer.rs @@ -10,11 +10,13 @@ use std::{ time::{Duration, SystemTime, UNIX_EPOCH}, }; use temporalio_common::protos::temporal::api::common::v1::Payload; -use temporalio_sdk_core::{ExtensionError, WorkerExtension}; use temporalio_streams::{ MemoryStreamExtension, RedisStreamExtension, StreamError, StreamResult, epoch::withdraws, - ids::{activity_producer_id, client_producer_id, local_activity_producer_id}, + ids::{ + activity_producer_id, client_producer_id, local_activity_producer_id, + named_client_producer_id, + }, memory::MemoryStore, proto::{ AppendRecord, CloseStreamRequest, CloseWriterRequest, CloseWriterResponse, @@ -106,8 +108,8 @@ fn records(texts: &[&str]) -> Vec { .collect() } -fn reason(error: &ExtensionError) -> String { - StreamErrorDetails::decode(error.details.as_slice()) +fn reason(error: &StreamError) -> String { + StreamErrorDetails::decode(error.details().as_slice()) .unwrap() .reason } @@ -124,11 +126,8 @@ impl Case { &self, method: &str, request: Req, - ) -> Result { - let bytes = self - .ext - .call_async(method.to_string(), request.encode_to_vec()) - .await?; + ) -> Result { + let bytes = self.ext.call(method, &request.encode_to_vec()).await?; Ok(Resp::decode(bytes.as_slice()).unwrap()) } @@ -148,7 +147,7 @@ impl Case { } /// An awaited publish: the Records are in the Store when it returns. - async fn write(&self, writer_id: u64, texts: &[&str]) -> Result<(i64, i64), ExtensionError> { + async fn write(&self, writer_id: u64, texts: &[&str]) -> Result<(i64, i64), StreamError> { let written: WriteRecordsResponse = self .call( "WriteRecords", @@ -250,7 +249,7 @@ async fn a_failed_write_fails_every_later_write_and_the_close() { case.call("CloseStream", close).await.unwrap(); let error = case.write(writer, &["late"]).await.unwrap_err(); assert_eq!( - (error.code, reason(&error).as_str()), + (error.code(), reason(&error).as_str()), (Code::FailedPrecondition, "stream_closed") ); assert_eq!( @@ -287,7 +286,7 @@ async fn a_closed_writer_is_gone() { assert!(closed.last_offset > 0); let error = case.write(writer, &["x"]).await.unwrap_err(); assert_eq!( - (error.code, reason(&error).as_str()), + (error.code(), reason(&error).as_str()), (Code::NotFound, "stream_unknown_writer") ); } @@ -305,7 +304,7 @@ async fn a_write_splits_big_records_by_bytes_and_refuses_an_oversize_one() { let huge = "y".repeat(1_700_000); let error = case.write(writer, &[&huge]).await.unwrap_err(); assert_eq!( - (error.code, reason(&error).as_str()), + (error.code(), reason(&error).as_str()), (Code::InvalidArgument, "stream_too_large") ); // A refusal before the Store leaves the writer usable. @@ -329,6 +328,49 @@ async fn a_reopen_of_a_closed_writer_goes_on_with_its_sequence() { } } +#[tokio::test] +async fn the_handles_of_one_producer_share_a_writer_until_the_last_one_lets_go() { + for case in cases().await { + let a = case.open(client(), |_| {}).await; + let b = case.open(client(), |_| {}).await; + assert_eq!(a, b); + case.write(a, &["1"]).await.unwrap(); + let _: CloseWriterResponse = case + .call("CloseWriter", CloseWriterRequest { writer_id: a }) + .await + .unwrap(); + // The other handle still writes, on the same sequence. + case.write(b, &["2"]).await.unwrap(); + case.ext.free_writer(b).unwrap(); + let error = case.write(b, &["3"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_unknown_writer"); + assert_eq!(case.visible().await, ["1", "2"]); + } +} + +#[tokio::test] +async fn a_named_client_with_a_higher_epoch_fences_its_earlier_self() { + for case in cases().await { + let named = |attempt| Producer { + kind: ProducerKind::Client as i32, + id: named_client_producer_id("ingest-1"), + epoch: Some(ProducerEpoch { + attempt, + ..Default::default() + }), + run_id: String::new(), + }; + let old = case.open(named(1), |_| {}).await; + case.write(old, &["old"]).await.unwrap(); + let new = case.open(named(2), |_| {}).await; + case.write(new, &["new"]).await.unwrap(); + let error = case.write(old, &["late"]).await.unwrap_err(); + assert_eq!(reason(&error), "stream_fenced"); + // The earlier epoch's Records are real messages, so the new epoch withdraws none. + assert_eq!(case.visible().await, ["old", "new"]); + } +} + #[tokio::test] async fn closing_every_writer_drops_them_all() { for case in cases().await { @@ -352,13 +394,9 @@ async fn closing_every_writer_drops_them_all() { #[tokio::test] async fn the_removed_buffered_calls_are_unknown() { for case in cases().await { - let error = case.ext.call("Publish", &[]).unwrap_err(); + let error = case.ext.call("Publish", &[]).await.unwrap_err(); assert!(error.message.contains("Publish"), "{error:?}"); - let error = case - .ext - .call_async("FlushWriter".to_string(), vec![]) - .await - .unwrap_err(); + let error = case.ext.call("FlushWriter", &[]).await.unwrap_err(); assert!(error.message.contains("FlushWriter"), "{error:?}"); } }