From 9a92ae58b6cb713169e2ba554b8ea2be07b5bb0b Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Tue, 10 Nov 2026 19:41:34 -0800 Subject: [PATCH 1/3] Rust examples: use the Join helper for the WCP joins and the Await pattern Bump examples/rust to dex-sdk 1.44.0. Migrate the parallel-split, multi-choice, thread, multiple-instance, cancelling, and reset partial join Flows and AwaitParallelStepsFlow from hand-written Channel counts to Join. The reset partial joins drop their reset Steps: the next round's split resets the previous round and records its late results in an Attribute that the tests read. --- examples/rust/Cargo.lock | 28 +- examples/rust/Cargo.toml | 2 +- examples/rust/README.md | 2 +- .../patterns/parallel/parallel_step_flows.rs | 49 ++- examples/rust/src/workflow_patterns.rs | 16 +- .../workflow_patterns/and_join_and_threads.rs | 66 +-- .../workflow_patterns/basic_control_flow.rs | 55 ++- .../multi_choice_and_merges.rs | 66 ++- .../multiple_instances/instance_creation.rs | 78 ++-- .../instance_partial_joins.rs | 243 ++++++----- .../src/workflow_patterns/partial_joins.rs | 390 ++++++++++-------- examples/rust/tests/workflow_patterns.rs | 40 +- 12 files changed, 593 insertions(+), 442 deletions(-) diff --git a/examples/rust/Cargo.lock b/examples/rust/Cargo.lock index 09fb3bec8d..64cc7147dd 100644 --- a/examples/rust/Cargo.lock +++ b/examples/rust/Cargo.lock @@ -247,9 +247,9 @@ dependencies = [ [[package]] name = "dex-blob-cache" -version = "1.31.0" +version = "1.44.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e7c22ba1ca2ce2ea71465088d3eacd085c9395a7a5daa98ba8ad6bbea35b0cc4" +checksum = "689ff0171fd27bcf58678a749a95de0bd3eb543de3d29512423c426ff0d6203e" dependencies = [ "crc32c", "sha2", @@ -274,9 +274,9 @@ dependencies = [ [[package]] name = "dex-protocol" -version = "1.31.0" +version = "1.44.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "593a66c639fe264152db07853f5317087135155bd651cdbd64ad07ed8a4d3adb" +checksum = "f61fa1a80ef42df8741a783c406a8a98fcd02b3cb9cd3db2cd40b97b2bf328f8" dependencies = [ "prost", "prost-types", @@ -288,9 +288,9 @@ dependencies = [ [[package]] name = "dex-sdk" -version = "1.31.0" +version = "1.44.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e92ec761f9e56361a24e314c1adbcfff0ccb8ec3d9f720ec40900fa6f9dfbec8" +checksum = "785b561629e880a7a2ecfd43b46ccd8140224f85891532e7b01da7e61a5e9c82" dependencies = [ "bytes", "dex-blob-cache", @@ -301,6 +301,8 @@ dependencies = [ "prost", "prost-types", "reqwest", + "rustls", + "rustls-native-certs", "serde", "serde_json", "tokio", @@ -1474,6 +1476,7 @@ version = "0.23.43" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0283386ce02abc0151e1761d08802dfe86c173b0b494af5cbc086574e453da06" dependencies = [ + "log", "once_cell", "ring", "rustls-pki-types", @@ -1482,6 +1485,18 @@ dependencies = [ "zeroize", ] +[[package]] +name = "rustls-native-certs" +version = "0.8.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4a073f5dc7469f984c52ad2752b63b0807745133b6de880b7b64c1ac4c48aec4" +dependencies = [ + "openssl-probe", + "rustls-pki-types", + "schannel", + "security-framework", +] + [[package]] name = "rustls-pki-types" version = "1.15.1" @@ -1918,6 +1933,7 @@ dependencies = [ "socket2", "sync_wrapper", "tokio", + "tokio-rustls", "tokio-stream", "tower", "tower-layer", diff --git a/examples/rust/Cargo.toml b/examples/rust/Cargo.toml index 569f6f6fb6..4ad873fcaf 100644 --- a/examples/rust/Cargo.toml +++ b/examples/rust/Cargo.toml @@ -6,7 +6,7 @@ publish = false rust-version = "1.97" [dependencies] -dex-sdk = "=1.31.0" +dex-sdk = "=1.44.0" fastrand = "2.3" serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" diff --git a/examples/rust/README.md b/examples/rust/README.md index a695aa7215..17f3fe141d 100644 --- a/examples/rust/README.md +++ b/examples/rust/README.md @@ -87,7 +87,7 @@ remain split so their orchestration boundaries are visible. | Manual recovery | [`ManualRecoveryFlow`](src/patterns/intervention/flow.rs) | Exhausted retry recovery and manual Channel decision | | Static parallel Steps | [`StaticParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | A fixed set of concurrent Step types | | Dynamic parallel Steps | [`DynamicParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | A runtime-sized set of one Step type | -| Await parallel Steps | [`AwaitParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | Wait for every worker's Channel message | +| Await parallel Steps | [`AwaitParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | Join every worker's result | | First-win parallel Steps | [`FirstWinParallelStepsFlow`](src/patterns/parallel/parallel_step_flows.rs) | Keep the first result and cancel siblings | | Basic SubFlow fan-out | [`BasicParentFlow`](src/patterns/parallel_subflows/flow.rs) | Wait for half of the SubFlows and cancel the remaining IDs | | Long-lived SubFlow parent | [`AdvancedLongLiveParentFlow`](src/patterns/parallel_subflows/flow.rs) | Bounded workers, request Channel, stop Attribute, and SubFlow loop | diff --git a/examples/rust/src/patterns/parallel/parallel_step_flows.rs b/examples/rust/src/patterns/parallel/parallel_step_flows.rs index ead17a43e9..f0f15a21b3 100644 --- a/examples/rust/src/patterns/parallel/parallel_step_flows.rs +++ b/examples/rust/src/patterns/parallel/parallel_step_flows.rs @@ -13,9 +13,10 @@ // limitations under the License. use dex_sdk::{ - Channel, Context, Flow, HandlerResult, PersistenceSchema, Step, StepDecision, StepList, + Context, Flow, HandlerResult, Join, JoinRound, PersistenceSchema, Step, StepDecision, StepList, StepMovement, Wait, }; +use serde::{Deserialize, Serialize}; use std::{sync::LazyLock, thread, time::Duration}; #[derive(Default)] @@ -120,7 +121,7 @@ impl Flow for AwaitParallelStepsFlow { .and(&self.await_step) } fn persistence(&self) -> PersistenceSchema { - PersistenceSchema::new().channel(&COMPLETE) + PersistenceSchema::new().join(&COMPLETE) } } #[derive(Default)] @@ -130,40 +131,58 @@ impl Step for AwaitInit { fn step_type(&self) -> &'static str { "InitStep" } - fn execute(&self, _: &mut Context, count: usize) -> HandlerResult { - let mut movements = vec![StepMovement::to(&Await, count)]; - movements.extend((0..count).map(|index| StepMovement::to(&AwaitWork, index))); + fn execute(&self, context: &mut Context, count: usize) -> HandlerResult { + let branches = (0..count) + .map(|index| { + Box::new(move |round: &JoinRound| { + StepMovement::to( + &AwaitWork, + WorkItem { + index, + round: round.clone(), + }, + ) + }) as Box StepMovement> + }) + .collect(); + let movements = + COMPLETE.fork(context, branches, |round| StepMovement::to(&Await, round))?; Ok(StepDecision::go_to_many(movements)) } } +#[derive(Clone, Debug, Deserialize, Serialize)] +pub struct WorkItem { + pub index: usize, + pub round: JoinRound, +} #[derive(Default)] struct AwaitWork; impl Step for AwaitWork { - type Input = usize; + type Input = WorkItem; fn step_type(&self) -> &'static str { "DoWork" } - fn execute(&self, context: &mut Context, _: usize) -> HandlerResult { + fn execute(&self, context: &mut Context, item: WorkItem) -> HandlerResult { thread::sleep(Duration::from_millis(fastrand::u64(50..500))); - COMPLETE.publish(context, ())?; - Ok(StepDecision::dead_end()) + COMPLETE.end_branch(context, &item.round, item.index) } } #[derive(Default)] struct Await; impl Step for Await { - type Input = usize; + type Input = JoinRound; fn step_type(&self) -> &'static str { "Await" } - fn wait_for(&self, _: &mut Context, count: usize) -> HandlerResult { - Ok(Wait::until(COMPLETE.for_n(count))) + fn wait_for(&self, _: &mut Context, round: JoinRound) -> HandlerResult { + Ok(Wait::until(COMPLETE.for_all_branches(&round))) } - fn execute(&self, _: &mut Context, count: usize) -> HandlerResult { - Ok(StepDecision::graceful_complete(count)) + fn execute(&self, context: &mut Context, round: JoinRound) -> HandlerResult { + let results = COMPLETE.branch_results(context, &round)?; + Ok(StepDecision::graceful_complete(results.len())) } } -static COMPLETE: LazyLock> = LazyLock::new(|| Channel::new("parallel-complete")); +static COMPLETE: LazyLock> = LazyLock::new(|| Join::new("parallel-complete")); #[derive(Default)] pub struct FirstWinParallelStepsFlow { diff --git a/examples/rust/src/workflow_patterns.rs b/examples/rust/src/workflow_patterns.rs index d9a920a906..3397179040 100644 --- a/examples/rust/src/workflow_patterns.rs +++ b/examples/rust/src/workflow_patterns.rs @@ -31,7 +31,9 @@ pub mod triggers; use std::sync::Arc; use std::time::{SystemTime, UNIX_EPOCH}; -use dex_sdk::{Client, HandlerError, HandlerResult, Registry, SdkResult}; +use dex_sdk::{ + Client, HandlerError, HandlerResult, JoinRound, Registry, SdkResult, Step, StepMovement, +}; use serde::{Deserialize, Serialize}; /// The time one task held a shared resource, in Unix epoch milliseconds. @@ -116,9 +118,15 @@ pub(crate) fn sorted(mut values: Vec) -> Vec { values } -/// Returns the ChannelMap instance key that holds the signals of `round`. -pub(crate) fn round_instance(round: u32) -> String { - format!("round-{round}") +/// Builds one branch movement of a Join round. +pub(crate) type JoinBranchBuilder = Box StepMovement>; + +/// Returns a branch builder that starts `step` with the round as its input. +pub(crate) fn round_branch(step: BranchStep) -> JoinBranchBuilder +where + BranchStep: Step + 'static, +{ + Box::new(move |round: &JoinRound| StepMovement::to(&step, round.clone())) } pub(crate) fn current_unix_millis() -> HandlerResult { diff --git a/examples/rust/src/workflow_patterns/and_join_and_threads.rs b/examples/rust/src/workflow_patterns/and_join_and_threads.rs index f6b32d8b4d..098f718468 100644 --- a/examples/rust/src/workflow_patterns/and_join_and_threads.rs +++ b/examples/rust/src/workflow_patterns/and_join_and_threads.rs @@ -18,12 +18,12 @@ use std::sync::LazyLock; use dex_sdk::{ - Channel, Context, Flow, HandlerError, HandlerResult, PersistenceSchema, Rpc, RpcList, Step, - StepDecision, StepList, StepMovement, Wait, + Channel, Context, Flow, HandlerError, HandlerResult, Join, JoinRound, PersistenceSchema, Rpc, + RpcList, Step, StepDecision, StepList, StepMovement, Wait, }; use serde::{Deserialize, Serialize}; -use super::sorted; +use super::{JoinBranchBuilder, sorted}; static INVOICE_RECEIVED: LazyLock> = LazyLock::new(|| Channel::new("invoiceReceived")); @@ -147,10 +147,18 @@ fn single_condition_result(context: &Context, channel: &Channel) -> Hand .ok_or_else(|| HandlerError::new("MissingChannelMessage", "the join consumed no message")) } -static PARCEL_PACKED: LazyLock> = LazyLock::new(|| Channel::new("parcelPacked")); +static PARCEL_PACKED: LazyLock> = LazyLock::new(|| Join::new("parcelPacked")); -/// WCP-42 Thread Split and WCP-41 Thread Merge: GoToMany starts several -/// executions of one Step type, and the merge waits for one message each. +/// One parcel of a shipment, packed in its own thread. +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct ParcelPacking { + pub index: usize, + pub round: JoinRound, +} + +/// WCP-42 Thread Split and WCP-41 Thread Merge: a Join forks several +/// executions of one Step type, and the merge waits for one result each. #[derive(Default)] pub struct ThreadSplitMergeFlow { split_shipment: SplitShipment, @@ -168,7 +176,7 @@ impl Flow for ThreadSplitMergeFlow { } fn persistence(&self) -> PersistenceSchema { - PersistenceSchema::new().channel(&PARCEL_PACKED) + PersistenceSchema::new().join(&PARCEL_PACKED) } } @@ -180,13 +188,26 @@ impl Step for SplitShipment { fn execute( &self, - _context: &mut Context, + context: &mut Context, parcel_count: Self::Input, ) -> HandlerResult { - let parcels = (0..parcel_count).map(|index| StepMovement::to(&PackParcel, index)); - Ok(StepDecision::go_to_many( - parcels.chain([StepMovement::to(&MergeParcels, parcel_count)]), - )) + let parcels = (0..parcel_count) + .map(|index| { + Box::new(move |round: &JoinRound| { + StepMovement::to( + &PackParcel, + ParcelPacking { + index, + round: round.clone(), + }, + ) + }) as JoinBranchBuilder + }) + .collect(); + let movements = PARCEL_PACKED.fork(context, parcels, |round| { + StepMovement::to(&MergeParcels, round) + })?; + Ok(StepDecision::go_to_many(movements)) } } @@ -194,11 +215,10 @@ impl Step for SplitShipment { struct PackParcel; impl Step for PackParcel { - type Input = usize; + type Input = ParcelPacking; - fn execute(&self, context: &mut Context, index: Self::Input) -> HandlerResult { - PARCEL_PACKED.publish(context, index)?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, parcel: Self::Input) -> HandlerResult { + PARCEL_PACKED.end_branch(context, &parcel.round, parcel.index) } } @@ -206,18 +226,14 @@ impl Step for PackParcel { struct MergeParcels; impl Step for MergeParcels { - type Input = usize; + type Input = JoinRound; - fn wait_for(&self, _context: &mut Context, parcel_count: Self::Input) -> HandlerResult { - Ok(Wait::until(PARCEL_PACKED.for_n(parcel_count))) + fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { + Ok(Wait::until(PARCEL_PACKED.for_all_branches(&round))) } - fn execute( - &self, - context: &mut Context, - _parcel_count: Self::Input, - ) -> HandlerResult { - let packed_parcels = PARCEL_PACKED.condition_results(context)?; + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + let packed_parcels = PARCEL_PACKED.branch_results(context, &round)?; Ok(StepDecision::graceful_complete(sorted(packed_parcels))) } } diff --git a/examples/rust/src/workflow_patterns/basic_control_flow.rs b/examples/rust/src/workflow_patterns/basic_control_flow.rs index e92e722efc..c2accd59c3 100644 --- a/examples/rust/src/workflow_patterns/basic_control_flow.rs +++ b/examples/rust/src/workflow_patterns/basic_control_flow.rs @@ -18,11 +18,11 @@ use std::sync::LazyLock; use dex_sdk::{ - Channel, Context, Flow, HandlerResult, PersistenceSchema, Rpc, RpcList, Step, StepDecision, - StepList, StepMovement, Wait, + Channel, Context, Flow, HandlerResult, Join, JoinRound, PersistenceSchema, Rpc, RpcList, Step, + StepDecision, StepList, StepMovement, Wait, }; -use super::{extend_trail, sorted}; +use super::{extend_trail, round_branch, sorted}; /// Claims at or above this amount need a manual review. pub const MANUAL_REVIEW_CLAIM_AMOUNT: i64 = 1000; @@ -87,17 +87,15 @@ impl Step for ShipOrder { } } -static FULFILLMENT_BRANCH_COMPLETED: LazyLock> = - LazyLock::new(|| Channel::new("fulfillmentBranchCompleted")); +static FULFILLMENT_BRANCH_COMPLETED: LazyLock> = + LazyLock::new(|| Join::new("fulfillmentBranchCompleted")); static PAYMENT_APPROVED: LazyLock> = LazyLock::new(|| Channel::new("paymentApproved")); pub const APPROVE_PAYMENT: Rpc<(), ()> = Rpc::new("approve_payment"); -const FULFILLMENT_BRANCH_COUNT: usize = 2; - -/// WCP-2 Parallel Split and WCP-3 Synchronization: GoToMany starts both -/// branches, and the join waits for one Channel message per branch. +/// WCP-2 Parallel Split and WCP-3 Synchronization: a Join forks both +/// branches, and the join waits for one result per branch. #[derive(Default)] pub struct ParallelSplitSynchronizationFlow { split_order: SplitOrder, @@ -124,7 +122,7 @@ impl Flow for ParallelSplitSynchronizationFlow { fn persistence(&self) -> PersistenceSchema { PersistenceSchema::new() - .channel(&FULFILLMENT_BRANCH_COMPLETED) + .join(&FULFILLMENT_BRANCH_COMPLETED) .channel(&PAYMENT_APPROVED) } @@ -139,12 +137,13 @@ struct SplitOrder; impl Step for SplitOrder { type Input = (); - fn execute(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult { - Ok(StepDecision::go_to_many([ - StepMovement::to(&PackItems, ()), - StepMovement::to(&ChargePayment, ()), - StepMovement::to(&SynchronizeFulfillment, ()), - ])) + fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { + let movements = FULFILLMENT_BRANCH_COMPLETED.fork( + context, + vec![round_branch(PackItems), round_branch(ChargePayment)], + |round| StepMovement::to(&SynchronizeFulfillment, round), + )?; + Ok(StepDecision::go_to_many(movements)) } } @@ -152,11 +151,10 @@ impl Step for SplitOrder { pub struct PackItems; impl Step for PackItems { - type Input = (); + type Input = JoinRound; - fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { - FULFILLMENT_BRANCH_COMPLETED.publish(context, self.step_type().to_string())?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + FULFILLMENT_BRANCH_COMPLETED.end_branch(context, &round, self.step_type().to_string()) } } @@ -164,15 +162,14 @@ impl Step for PackItems { struct ChargePayment; impl Step for ChargePayment { - type Input = (); + type Input = JoinRound; fn wait_for(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult { Ok(Wait::until(PAYMENT_APPROVED.for_one())) } - fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { - FULFILLMENT_BRANCH_COMPLETED.publish(context, self.step_type().to_string())?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + FULFILLMENT_BRANCH_COMPLETED.end_branch(context, &round, self.step_type().to_string()) } } @@ -180,16 +177,16 @@ impl Step for ChargePayment { struct SynchronizeFulfillment; impl Step for SynchronizeFulfillment { - type Input = (); + type Input = JoinRound; - fn wait_for(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult { + fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { Ok(Wait::until( - FULFILLMENT_BRANCH_COMPLETED.for_n(FULFILLMENT_BRANCH_COUNT), + FULFILLMENT_BRANCH_COMPLETED.for_all_branches(&round), )) } - fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { - let branches = FULFILLMENT_BRANCH_COMPLETED.condition_results(context)?; + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + let branches = FULFILLMENT_BRANCH_COMPLETED.branch_results(context, &round)?; Ok(StepDecision::graceful_complete(sorted(branches))) } } diff --git a/examples/rust/src/workflow_patterns/multi_choice_and_merges.rs b/examples/rust/src/workflow_patterns/multi_choice_and_merges.rs index 7d41664e00..fcbf9edd80 100644 --- a/examples/rust/src/workflow_patterns/multi_choice_and_merges.rs +++ b/examples/rust/src/workflow_patterns/multi_choice_and_merges.rs @@ -19,12 +19,12 @@ use std::sync::LazyLock; use dex_sdk::{ - Attribute, Channel, Context, Flow, HandlerError, HandlerResult, PersistenceSchema, Step, - StepDecision, StepList, StepMovement, StepOptions, Value, Wait, + Attribute, Channel, Context, Flow, HandlerError, HandlerResult, Join, JoinRound, + PersistenceSchema, Step, StepDecision, StepList, StepMovement, StepOptions, Value, Wait, }; use serde::{Deserialize, Serialize}; -use super::sorted; +use super::{round_branch, sorted}; /// The notification branches a caller selects. #[derive(Clone, Debug, Default, Deserialize, Serialize)] @@ -35,12 +35,11 @@ pub struct NotificationChannelSelection { pub send_push: bool, } -static NOTIFICATION_SENT: LazyLock> = - LazyLock::new(|| Channel::new("notificationSent")); +static NOTIFICATION_SENT: LazyLock> = LazyLock::new(|| Join::new("notificationSent")); /// WCP-6 Multi-Choice, WCP-7 Structured Synchronizing Merge, and WCP-37 -/// Local Synchronizing Merge: the diverging decision passes the selected -/// branch count to the join Step. +/// Local Synchronizing Merge: the diverging decision forks the selected +/// branches, and the join waits for each of them. #[derive(Default)] pub struct MultiChoiceSynchronizingMergeFlow { select_notification_channels: SelectNotificationChannels, @@ -62,7 +61,7 @@ impl Flow for MultiChoiceSynchronizingMergeFlow { } fn persistence(&self) -> PersistenceSchema { - PersistenceSchema::new().channel(&NOTIFICATION_SENT) + PersistenceSchema::new().join(&NOTIFICATION_SENT) } } @@ -74,26 +73,26 @@ impl Step for SelectNotificationChannels { fn execute( &self, - _context: &mut Context, + context: &mut Context, selection: Self::Input, ) -> HandlerResult { let mut branches = Vec::new(); if selection.send_email { - branches.push(StepMovement::to(&SendEmail, ())); + branches.push(round_branch(SendEmail)); } if selection.send_sms { - branches.push(StepMovement::to(&SendSms, ())); + branches.push(round_branch(SendSms)); } if selection.send_push { - branches.push(StepMovement::to(&SendPush, ())); + branches.push(round_branch(SendPush)); } if branches.is_empty() { return Ok(StepDecision::graceful_complete(Vec::::new())); } - let merge = StepMovement::to(&SynchronizingMerge, branches.len()); - Ok(StepDecision::go_to_many( - std::iter::once(merge).chain(branches), - )) + let movements = NOTIFICATION_SENT.fork(context, branches, |round| { + StepMovement::to(&SynchronizingMerge, round) + })?; + Ok(StepDecision::go_to_many(movements)) } } @@ -101,11 +100,10 @@ impl Step for SelectNotificationChannels { struct SendEmail; impl Step for SendEmail { - type Input = (); + type Input = JoinRound; - fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { - NOTIFICATION_SENT.publish(context, self.step_type().to_string())?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + NOTIFICATION_SENT.end_branch(context, &round, self.step_type().to_string()) } } @@ -113,11 +111,10 @@ impl Step for SendEmail { struct SendSms; impl Step for SendSms { - type Input = (); + type Input = JoinRound; - fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { - NOTIFICATION_SENT.publish(context, self.step_type().to_string())?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + NOTIFICATION_SENT.end_branch(context, &round, self.step_type().to_string()) } } @@ -125,11 +122,10 @@ impl Step for SendSms { struct SendPush; impl Step for SendPush { - type Input = (); + type Input = JoinRound; - fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { - NOTIFICATION_SENT.publish(context, self.step_type().to_string())?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + NOTIFICATION_SENT.end_branch(context, &round, self.step_type().to_string()) } } @@ -137,18 +133,14 @@ impl Step for SendPush { struct SynchronizingMerge; impl Step for SynchronizingMerge { - type Input = usize; + type Input = JoinRound; - fn wait_for(&self, _context: &mut Context, branch_count: Self::Input) -> HandlerResult { - Ok(Wait::until(NOTIFICATION_SENT.for_n(branch_count))) + fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { + Ok(Wait::until(NOTIFICATION_SENT.for_all_branches(&round))) } - fn execute( - &self, - context: &mut Context, - _branch_count: Self::Input, - ) -> HandlerResult { - let notifications = NOTIFICATION_SENT.condition_results(context)?; + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + let notifications = NOTIFICATION_SENT.branch_results(context, &round)?; Ok(StepDecision::graceful_complete(sorted(notifications))) } } diff --git a/examples/rust/src/workflow_patterns/multiple_instances/instance_creation.rs b/examples/rust/src/workflow_patterns/multiple_instances/instance_creation.rs index ffb46da21b..d07ff3301f 100644 --- a/examples/rust/src/workflow_patterns/multiple_instances/instance_creation.rs +++ b/examples/rust/src/workflow_patterns/multiple_instances/instance_creation.rs @@ -16,11 +16,12 @@ use std::sync::LazyLock; -use crate::workflow_patterns::sorted; +use crate::workflow_patterns::{JoinBranchBuilder, sorted}; use dex_sdk::{ - Attribute, AttributeMap, Channel, Context, Flow, HandlerResult, PersistenceSchema, Rpc, - RpcList, RpcResult, Step, StepDecision, StepList, StepMovement, Wait, + Attribute, AttributeMap, Channel, Context, Flow, HandlerResult, Join, JoinRound, + PersistenceSchema, Rpc, RpcList, RpcResult, Step, StepDecision, StepList, StepMovement, Wait, }; +use serde::{Deserialize, Serialize}; static INVITATION_RELEASED: LazyLock> = LazyLock::new(|| Channel::new("invitationReleased")); @@ -119,8 +120,7 @@ pub static STARTED_REVIEWERS: LazyLock> = LazyLock::new(|| AttributeMap::new("startedReviewers")); static REVIEW_GATE_OPENED: LazyLock> = LazyLock::new(|| Channel::new("reviewGateOpened")); -static REVIEW_COMPLETED: LazyLock> = - LazyLock::new(|| Channel::new("reviewCompleted")); +static REVIEW_COMPLETED: LazyLock> = LazyLock::new(|| Join::new("reviewCompleted")); pub const GET_STARTED_DESIGN_TIME_REVIEW_COUNT: Rpc<(), usize> = Rpc::new("get_started_design_time_review_count"); @@ -189,10 +189,8 @@ struct StartDesignTimeReviews; impl Step for StartDesignTimeReviews { type Input = (); - fn execute(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult { - Ok(start_reviews( - DESIGN_TIME_REVIEWERS.map(str::to_string).to_vec(), - )) + fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { + start_reviews(context, DESIGN_TIME_REVIEWERS.map(str::to_string).to_vec()) } } @@ -249,10 +247,10 @@ impl Step for StartRunTimeReviews { fn execute( &self, - _context: &mut Context, + context: &mut Context, reviewers: Self::Input, ) -> HandlerResult { - Ok(start_reviews(reviewers)) + start_reviews(context, reviewers) } } @@ -260,7 +258,7 @@ fn review_persistence() -> PersistenceSchema { PersistenceSchema::new() .attribute_map(&STARTED_REVIEWERS) .channel(&REVIEW_GATE_OPENED) - .channel(&REVIEW_COMPLETED) + .join(&REVIEW_COMPLETED) } fn count_started_reviews(context: &Context) -> HandlerResult> { @@ -274,30 +272,50 @@ fn open_review_gate(context: &mut Context, count: usize) -> HandlerResult<()> { Ok(()) } -/// Starts one ReviewDocument per reviewer and the join that waits for all. -fn start_reviews(reviewers: Vec) -> StepDecision { - let collect_reviews = StepMovement::to(&CollectReviews, reviewers.len()); +/// One review instance of a round. +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct DocumentReview { + pub reviewer: String, + pub round: JoinRound, +} + +/// Forks one ReviewDocument per reviewer and the join that waits for all. +fn start_reviews(context: &mut Context, reviewers: Vec) -> HandlerResult { let reviews = reviewers .into_iter() - .map(|reviewer| StepMovement::to(&ReviewDocument, reviewer)); - StepDecision::go_to_many(reviews.chain([collect_reviews])) + .map(|reviewer| { + Box::new(move |round: &JoinRound| { + StepMovement::to( + &ReviewDocument, + DocumentReview { + reviewer, + round: round.clone(), + }, + ) + }) as JoinBranchBuilder + }) + .collect(); + let movements = REVIEW_COMPLETED.fork(context, reviews, |round| { + StepMovement::to(&CollectReviews, round) + })?; + Ok(StepDecision::go_to_many(movements)) } #[derive(Default)] struct ReviewDocument; impl Step for ReviewDocument { - type Input = String; + type Input = DocumentReview; /// Marks the instance started, then holds it at the concurrency gate. - fn wait_for(&self, context: &mut Context, reviewer: Self::Input) -> HandlerResult { - STARTED_REVIEWERS.set(context, &reviewer, true)?; + fn wait_for(&self, context: &mut Context, review: Self::Input) -> HandlerResult { + STARTED_REVIEWERS.set(context, &review.reviewer, true)?; Ok(Wait::until(REVIEW_GATE_OPENED.for_one())) } - fn execute(&self, context: &mut Context, reviewer: Self::Input) -> HandlerResult { - REVIEW_COMPLETED.publish(context, reviewer)?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, review: Self::Input) -> HandlerResult { + REVIEW_COMPLETED.end_branch(context, &review.round, review.reviewer) } } @@ -305,18 +323,14 @@ impl Step for ReviewDocument { struct CollectReviews; impl Step for CollectReviews { - type Input = usize; + type Input = JoinRound; - fn wait_for(&self, _context: &mut Context, review_count: Self::Input) -> HandlerResult { - Ok(Wait::until(REVIEW_COMPLETED.for_n(review_count))) + fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { + Ok(Wait::until(REVIEW_COMPLETED.for_all_branches(&round))) } - fn execute( - &self, - context: &mut Context, - _review_count: Self::Input, - ) -> HandlerResult { - let reviewers = REVIEW_COMPLETED.condition_results(context)?; + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + let reviewers = REVIEW_COMPLETED.branch_results(context, &round)?; Ok(StepDecision::graceful_complete(sorted(reviewers))) } } diff --git a/examples/rust/src/workflow_patterns/multiple_instances/instance_partial_joins.rs b/examples/rust/src/workflow_patterns/multiple_instances/instance_partial_joins.rs index 907792ad1a..0d56c1573d 100644 --- a/examples/rust/src/workflow_patterns/multiple_instances/instance_partial_joins.rs +++ b/examples/rust/src/workflow_patterns/multiple_instances/instance_partial_joins.rs @@ -17,12 +17,13 @@ use std::sync::LazyLock; use dex_sdk::{ - Attribute, AttributeMap, Channel, ChannelMap, Context, Flow, HandlerResult, PersistenceSchema, - Rpc, RpcList, RpcResult, Step, StepDecision, StepList, StepMovement, StepOptions, Wait, + Attribute, AttributeMap, Channel, Context, Flow, HandlerResult, Join, JoinRound, + PersistenceSchema, Rpc, RpcList, RpcResult, Step, StepDecision, StepList, StepMovement, + StepOptions, Wait, }; use serde::{Deserialize, Serialize}; -use crate::workflow_patterns::{round_instance, sorted}; +use crate::workflow_patterns::{JoinBranchBuilder, sorted}; /// One round of the static partial join. #[derive(Clone, Debug, Deserialize, Serialize)] @@ -32,16 +33,49 @@ pub struct InspectionRound { pub required_inspections: usize, pub round: u32, pub round_count: u32, + /// The Join round to reset before this round starts; `None` in round 1. + pub previous_round: Option, +} + +/// The input of the join Step of one inspection round. +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct InspectionRoundJoin { + pub inspection_round: InspectionRound, + pub join_round: JoinRound, } /// One inspection instance of a round. #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] pub struct UnitInspection { - pub round: u32, + pub round: JoinRound, pub unit_index: usize, } +/// Returns one inspection branch builder per unit index below `inspection_count`. +fn unit_inspection_branches( + step: InspectionStep, + inspection_count: usize, +) -> Vec +where + InspectionStep: Step + Copy + 'static, +{ + (0..inspection_count) + .map(|unit_index| { + Box::new(move |round: &JoinRound| { + StepMovement::to( + &step, + UnitInspection { + round: round.clone(), + unit_index, + }, + ) + }) as JoinBranchBuilder + }) + .collect() +} + /// The unit indexes that one Step of a round consumed. #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] @@ -52,21 +86,21 @@ pub struct InspectedUnits { static INSPECTION_RELEASED: LazyLock> = LazyLock::new(|| Channel::new("inspectionReleased")); -static UNIT_INSPECTED: LazyLock> = - LazyLock::new(|| ChannelMap::new("unitInspected")); +static UNIT_INSPECTED: LazyLock> = LazyLock::new(|| Join::new("unitInspected")); +pub static IGNORED_INSPECTIONS: LazyLock> = + LazyLock::new(|| Attribute::new("ignoredInspections")); pub const RELEASE_INSPECTIONS: Rpc = Rpc::new("release_inspections"); /// WCP-34 Static Partial Join for Multiple Instances: the join fires after -/// the required instances of a round, and a reset Step absorbs the rest -/// before the next round re-enables the join. +/// the required instances of a round, and the next round's split resets +/// the rest before the join re-enables. #[derive(Default)] pub struct StaticPartialJoinForMultipleInstancesFlow { start_inspections: StartInspections, inspect_unit: InspectUnit, accept_inspections: AcceptInspections, approve_batch: ApproveBatch, - reset_inspection_join: ResetInspectionJoin, } impl StaticPartialJoinForMultipleInstancesFlow { @@ -86,13 +120,13 @@ impl Flow for StaticPartialJoinForMultipleInstancesFlow { .and(&self.inspect_unit) .and(&self.accept_inspections) .and(&self.approve_batch) - .and(&self.reset_inspection_join) } fn persistence(&self) -> PersistenceSchema { PersistenceSchema::new() .channel(&INSPECTION_RELEASED) - .channel_map(&UNIT_INSPECTED) + .join(&UNIT_INSPECTED) + .attribute(&IGNORED_INSPECTIONS) } fn rpcs(&self) -> RpcList { @@ -106,27 +140,47 @@ pub struct StartInspections; impl Step for StartInspections { type Input = InspectionRound; - fn execute(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { + /// Resets the previous round: waits for the instances its join did not accept. + fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { + match &round.previous_round { + None => Ok(Wait::skip_immediately()), + Some(previous_round) => Ok(Wait::until( + UNIT_INSPECTED.for_remaining_branches(previous_round, round.required_inspections), + )), + } + } + + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { if round.required_inspections > round.inspection_count { return Ok(StepDecision::force_fail( "required inspections exceed the inspection count", )); } - let inspections = (0..round.inspection_count).map(|unit_index| { + if let Some(previous_round) = &round.previous_round { + let late_units = UNIT_INSPECTED.remaining_branch_results(context, previous_round)?; + IGNORED_INSPECTIONS.set( + context, + InspectedUnits { + round: round.round - 1, + unit_indexes: sorted(late_units), + }, + )?; + } + let inspections = unit_inspection_branches(InspectUnit, round.inspection_count); + let movements = UNIT_INSPECTED.fork(context, inspections, |join_round| { StepMovement::to( - &InspectUnit, - UnitInspection { - round: round.round, - unit_index, + &AcceptInspections, + InspectionRoundJoin { + inspection_round: round, + join_round, }, ) - }); - let join = StepMovement::to(&AcceptInspections, round.clone()); - Ok(StepDecision::go_to_many(inspections.chain([join]))) + })?; + Ok(StepDecision::go_to_many(movements)) } } -#[derive(Default)] +#[derive(Clone, Copy, Default)] struct InspectUnit; impl Step for InspectUnit { @@ -141,12 +195,7 @@ impl Step for InspectUnit { context: &mut Context, inspection: Self::Input, ) -> HandlerResult { - UNIT_INSPECTED.publish( - context, - &round_instance(inspection.round), - inspection.unit_index, - )?; - Ok(StepDecision::dead_end()) + UNIT_INSPECTED.end_branch(context, &inspection.round, inspection.unit_index) } } @@ -154,26 +203,33 @@ impl Step for InspectUnit { struct AcceptInspections; impl Step for AcceptInspections { - type Input = InspectionRound; + type Input = InspectionRoundJoin; - fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { - Ok(Wait::until(UNIT_INSPECTED.for_n( - &round_instance(round.round), - round.required_inspections, + fn wait_for(&self, _context: &mut Context, input: Self::Input) -> HandlerResult { + Ok(Wait::until(UNIT_INSPECTED.for_branches( + &input.join_round, + input.inspection_round.required_inspections, ))) } - fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult { + let round = input.inspection_round; let inspected_units = InspectedUnits { round: round.round, - unit_indexes: sorted( - UNIT_INSPECTED.condition_results(context, &round_instance(round.round))?, - ), + unit_indexes: sorted(UNIT_INSPECTED.branch_results(context, &input.join_round)?), }; - Ok(StepDecision::go_to_many([ - StepMovement::to(&ApproveBatch, inspected_units), - StepMovement::to(&ResetInspectionJoin, round), - ])) + let mut movements = vec![StepMovement::to(&ApproveBatch, inspected_units)]; + if round.round < round.round_count { + movements.push(StepMovement::to( + &StartInspections, + InspectionRound { + round: round.round + 1, + previous_round: Some(input.join_round), + ..round + }, + )); + } + Ok(StepDecision::go_to_many(movements)) } } @@ -192,39 +248,6 @@ impl Step for ApproveBatch { } } -#[derive(Default)] -struct ResetInspectionJoin; - -impl Step for ResetInspectionJoin { - type Input = InspectionRound; - - fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { - let remaining_inspections = round.inspection_count - round.required_inspections; - Ok(Wait::until(UNIT_INSPECTED.for_n( - &round_instance(round.round), - remaining_inspections, - ))) - } - - fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - let unit_indexes = - sorted(UNIT_INSPECTED.condition_results(context, &round_instance(round.round))?); - if round.round < round.round_count { - return Ok(StepDecision::go_to( - &StartInspections, - InspectionRound { - round: round.round + 1, - ..round - }, - )); - } - Ok(StepDecision::graceful_complete(InspectedUnits { - round: round.round, - unit_indexes, - })) - } -} - /// How many inspection instances to start, and how many complete the join. #[derive(Clone, Debug, Deserialize, Serialize)] #[serde(rename_all = "camelCase")] @@ -235,8 +258,8 @@ pub struct InspectionPlan { static QUORUM_INSPECTION_RELEASED: LazyLock> = LazyLock::new(|| Channel::new("quorumInspectionReleased")); -static QUORUM_UNIT_INSPECTED: LazyLock> = - LazyLock::new(|| Channel::new("quorumUnitInspected")); +static QUORUM_UNIT_INSPECTED: LazyLock> = + LazyLock::new(|| Join::new("quorumUnitInspected")); static QUORUM_CERTIFIED: LazyLock> = LazyLock::new(|| Channel::new("quorumCertified")); @@ -245,8 +268,16 @@ pub const GET_QUEUED_QUORUM_RELEASE_COUNT: Rpc<(), usize> = Rpc::new("get_queued_quorum_release_count"); pub const CERTIFY_QUORUM: Rpc<(), ()> = Rpc::new("certify_quorum"); +/// The input of the join Step of the quorum inspections. +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct QuorumInspectionJoin { + pub required_inspections: usize, + pub join_round: JoinRound, +} + /// WCP-35 Cancelling Partial Join for Multiple Instances: the join cancels -/// the instances that have not reported. +/// the sibling instances that have not reported. #[derive(Default)] pub struct CancellingPartialJoinForMultipleInstancesFlow { start_quorum_inspections: StartQuorumInspections, @@ -289,7 +320,7 @@ impl Flow for CancellingPartialJoinForMultipleInstancesFlow { fn persistence(&self) -> PersistenceSchema { PersistenceSchema::new() .channel(&QUORUM_INSPECTION_RELEASED) - .channel(&QUORUM_UNIT_INSPECTED) + .join(&QUORUM_UNIT_INSPECTED) .channel(&QUORUM_CERTIFIED) } @@ -310,28 +341,37 @@ struct StartQuorumInspections; impl Step for StartQuorumInspections { type Input = InspectionPlan; - fn execute(&self, _context: &mut Context, plan: Self::Input) -> HandlerResult { - let inspections = - (0..plan.inspection_count).map(|index| StepMovement::to(&InspectQuorumUnit, index)); - Ok(StepDecision::go_to_many(inspections.chain([ - StepMovement::to(&AcceptQuorumInspections, plan.required_inspections), - ]))) + fn execute(&self, context: &mut Context, plan: Self::Input) -> HandlerResult { + let inspections = unit_inspection_branches(InspectQuorumUnit, plan.inspection_count); + let movements = QUORUM_UNIT_INSPECTED.fork(context, inspections, |join_round| { + StepMovement::to( + &AcceptQuorumInspections, + QuorumInspectionJoin { + required_inspections: plan.required_inspections, + join_round, + }, + ) + })?; + Ok(StepDecision::go_to_many(movements)) } } -#[derive(Default)] +#[derive(Clone, Copy, Default)] struct InspectQuorumUnit; impl Step for InspectQuorumUnit { - type Input = usize; + type Input = UnitInspection; - fn wait_for(&self, _context: &mut Context, _index: Self::Input) -> HandlerResult { + fn wait_for(&self, _context: &mut Context, _inspection: Self::Input) -> HandlerResult { Ok(Wait::until(QUORUM_INSPECTION_RELEASED.for_one())) } - fn execute(&self, context: &mut Context, index: Self::Input) -> HandlerResult { - QUORUM_UNIT_INSPECTED.publish(context, index)?; - Ok(StepDecision::dead_end()) + fn execute( + &self, + context: &mut Context, + inspection: Self::Input, + ) -> HandlerResult { + QUORUM_UNIT_INSPECTED.end_branch(context, &inspection.round, inspection.unit_index) } } @@ -339,26 +379,19 @@ impl Step for InspectQuorumUnit { pub struct AcceptQuorumInspections; impl Step for AcceptQuorumInspections { - type Input = usize; + type Input = QuorumInspectionJoin; - fn wait_for( - &self, - _context: &mut Context, - required_inspections: Self::Input, - ) -> HandlerResult { - Ok(Wait::until( - QUORUM_UNIT_INSPECTED.for_n(required_inspections), - )) + fn wait_for(&self, _context: &mut Context, input: Self::Input) -> HandlerResult { + Ok(Wait::until(QUORUM_UNIT_INSPECTED.for_branches( + &input.join_round, + input.required_inspections, + ))) } - fn execute( - &self, - context: &mut Context, - _required_inspections: Self::Input, - ) -> HandlerResult { - let inspected_units = QUORUM_UNIT_INSPECTED.condition_results(context)?; + fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult { + let inspected_units = QUORUM_UNIT_INSPECTED.branch_results(context, &input.join_round)?; Ok(StepDecision::go_to(&CertifyQuorum, sorted(inspected_units)) - .cancel_step(&InspectQuorumUnit)) + .cancel_sibling_step(&InspectQuorumUnit)) } } diff --git a/examples/rust/src/workflow_patterns/partial_joins.rs b/examples/rust/src/workflow_patterns/partial_joins.rs index 732f1a2680..ba5b58b4f3 100644 --- a/examples/rust/src/workflow_patterns/partial_joins.rs +++ b/examples/rust/src/workflow_patterns/partial_joins.rs @@ -20,12 +20,12 @@ use std::sync::LazyLock; use dex_sdk::{ - Channel, ChannelMap, Context, Flow, HandlerError, HandlerResult, PersistenceSchema, Rpc, - RpcList, RpcResult, Step, StepDecision, StepList, StepMovement, Wait, + Attribute, Channel, Context, Flow, HandlerError, HandlerResult, Join, JoinRound, + PersistenceSchema, Rpc, RpcList, RpcResult, Step, StepDecision, StepList, StepMovement, Wait, }; use serde::{Deserialize, Serialize}; -use super::{round_instance, sorted}; +use super::{round_branch, sorted}; const SUPPLIER_COUNT: usize = 3; const CARRIER_COUNT: usize = 3; @@ -38,8 +38,9 @@ static SUPPLIER_B_RELEASED: LazyLock> = LazyLock::new(|| Channel::new("supplierBReleased")); static SUPPLIER_C_RELEASED: LazyLock> = LazyLock::new(|| Channel::new("supplierCReleased")); -static QUOTE_RECEIVED: LazyLock> = - LazyLock::new(|| ChannelMap::new("quoteReceived")); +static QUOTE_RECEIVED: LazyLock> = LazyLock::new(|| Join::new("quoteReceived")); +pub static IGNORED_QUOTES: LazyLock> = + LazyLock::new(|| Attribute::new("ignoredQuotes")); pub const RELEASE_SUPPLIER_QUOTE: Rpc = Rpc::new("release_supplier_quote"); @@ -50,6 +51,16 @@ pub struct QuoteRound { pub required_quotes: usize, pub round: u32, pub round_count: u32, + /// The Join round to reset before this round starts; `None` in round 1. + pub previous_round: Option, +} + +/// The input of the join Step of one quote round. +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct QuoteRoundJoin { + pub quote_round: QuoteRound, + pub join_round: JoinRound, } /// The quotes the join accepted in one round. @@ -60,7 +71,7 @@ pub struct AcceptedQuotes { pub accepted_quotes: Vec, } -/// The quotes the reset Step absorbed in the last round. +/// The late quotes of a round, read by the next round's split Step. #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] #[serde(rename_all = "camelCase")] pub struct IgnoredQuotes { @@ -70,7 +81,7 @@ pub struct IgnoredQuotes { /// WCP-9 Structured Discriminator (`required_quotes` = 1) and WCP-30 /// Structured Partial Join (`required_quotes` = 2). Quotes are keyed by -/// round, and a reset Step absorbs the rest before the join re-enables. +/// Join round, and the next round's split resets the rest before the join re-enables. #[derive(Default)] pub struct StructuredPartialJoinFlow { request_quotes: RequestQuotes, @@ -79,7 +90,6 @@ pub struct StructuredPartialJoinFlow { quote_from_supplier_c: QuoteFromSupplierC, accept_quotes: AcceptQuotes, place_order: PlaceOrder, - reset_quote_join: ResetQuoteJoin, } impl StructuredPartialJoinFlow { @@ -110,7 +120,6 @@ impl Flow for StructuredPartialJoinFlow { .and(&self.quote_from_supplier_c) .and(&self.accept_quotes) .and(&self.place_order) - .and(&self.reset_quote_join) } fn persistence(&self) -> PersistenceSchema { @@ -118,7 +127,8 @@ impl Flow for StructuredPartialJoinFlow { .channel(&SUPPLIER_A_RELEASED) .channel(&SUPPLIER_B_RELEASED) .channel(&SUPPLIER_C_RELEASED) - .channel_map("E_RECEIVED) + .join("E_RECEIVED) + .attribute(&IGNORED_QUOTES) } fn rpcs(&self) -> RpcList { @@ -132,19 +142,51 @@ pub struct RequestQuotes; impl Step for RequestQuotes { type Input = QuoteRound; - fn execute(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { + /// Resets the previous round: waits for the quotes its join did not accept. + fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { + match &round.previous_round { + None => Ok(Wait::skip_immediately()), + Some(previous_round) => Ok(Wait::until( + QUOTE_RECEIVED.for_remaining_branches(previous_round, round.required_quotes), + )), + } + } + + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { if !(1..=SUPPLIER_COUNT).contains(&round.required_quotes) { return Ok(StepDecision::force_fail(format!( "required quotes must be between 1 and {SUPPLIER_COUNT}, got {}", round.required_quotes ))); } - Ok(StepDecision::go_to_many([ - StepMovement::to(&QuoteFromSupplierA, round.round), - StepMovement::to(&QuoteFromSupplierB, round.round), - StepMovement::to(&QuoteFromSupplierC, round.round), - StepMovement::to(&AcceptQuotes, round), - ])) + if let Some(previous_round) = &round.previous_round { + let late_quotes = QUOTE_RECEIVED.remaining_branch_results(context, previous_round)?; + IGNORED_QUOTES.set( + context, + IgnoredQuotes { + round: round.round - 1, + ignored_quotes: sorted(late_quotes), + }, + )?; + } + let movements = QUOTE_RECEIVED.fork( + context, + vec![ + round_branch(QuoteFromSupplierA), + round_branch(QuoteFromSupplierB), + round_branch(QuoteFromSupplierC), + ], + |join_round| { + StepMovement::to( + &AcceptQuotes, + QuoteRoundJoin { + quote_round: round, + join_round, + }, + ) + }, + )?; + Ok(StepDecision::go_to_many(movements)) } } @@ -152,15 +194,14 @@ impl Step for RequestQuotes { pub struct QuoteFromSupplierA; impl Step for QuoteFromSupplierA { - type Input = u32; + type Input = JoinRound; fn wait_for(&self, _context: &mut Context, _round: Self::Input) -> HandlerResult { Ok(Wait::until(SUPPLIER_A_RELEASED.for_one())) } fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - QUOTE_RECEIVED.publish(context, &round_instance(round), "A".to_string())?; - Ok(StepDecision::dead_end()) + QUOTE_RECEIVED.end_branch(context, &round, "A".to_string()) } } @@ -168,15 +209,14 @@ impl Step for QuoteFromSupplierA { struct QuoteFromSupplierB; impl Step for QuoteFromSupplierB { - type Input = u32; + type Input = JoinRound; fn wait_for(&self, _context: &mut Context, _round: Self::Input) -> HandlerResult { Ok(Wait::until(SUPPLIER_B_RELEASED.for_one())) } fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - QUOTE_RECEIVED.publish(context, &round_instance(round), "B".to_string())?; - Ok(StepDecision::dead_end()) + QUOTE_RECEIVED.end_branch(context, &round, "B".to_string()) } } @@ -184,15 +224,14 @@ impl Step for QuoteFromSupplierB { pub struct QuoteFromSupplierC; impl Step for QuoteFromSupplierC { - type Input = u32; + type Input = JoinRound; fn wait_for(&self, _context: &mut Context, _round: Self::Input) -> HandlerResult { Ok(Wait::until(SUPPLIER_C_RELEASED.for_one())) } fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - QUOTE_RECEIVED.publish(context, &round_instance(round), "C".to_string())?; - Ok(StepDecision::dead_end()) + QUOTE_RECEIVED.end_branch(context, &round, "C".to_string()) } } @@ -200,25 +239,36 @@ impl Step for QuoteFromSupplierC { struct AcceptQuotes; impl Step for AcceptQuotes { - type Input = QuoteRound; + type Input = QuoteRoundJoin; - fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { - Ok(Wait::until(QUOTE_RECEIVED.for_n( - &round_instance(round.round), - round.required_quotes, + fn wait_for(&self, _context: &mut Context, input: Self::Input) -> HandlerResult { + Ok(Wait::until(QUOTE_RECEIVED.for_branches( + &input.join_round, + input.quote_round.required_quotes, ))) } - fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - let quotes = QUOTE_RECEIVED.condition_results(context, &round_instance(round.round))?; - let accepted_quotes = AcceptedQuotes { - round: round.round, - accepted_quotes: sorted(quotes), - }; - Ok(StepDecision::go_to_many([ - StepMovement::to(&PlaceOrder, accepted_quotes), - StepMovement::to(&ResetQuoteJoin, round), - ])) + fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult { + let quotes = QUOTE_RECEIVED.branch_results(context, &input.join_round)?; + let round = input.quote_round; + let mut movements = vec![StepMovement::to( + &PlaceOrder, + AcceptedQuotes { + round: round.round, + accepted_quotes: sorted(quotes), + }, + )]; + if round.round < round.round_count { + movements.push(StepMovement::to( + &RequestQuotes, + QuoteRound { + round: round.round + 1, + previous_round: Some(input.join_round), + ..round + }, + )); + } + Ok(StepDecision::go_to_many(movements)) } } @@ -237,42 +287,10 @@ impl Step for PlaceOrder { } } -#[derive(Default)] -struct ResetQuoteJoin; - -impl Step for ResetQuoteJoin { - type Input = QuoteRound; - - fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { - Ok(Wait::until(QUOTE_RECEIVED.for_n( - &round_instance(round.round), - SUPPLIER_COUNT - round.required_quotes, - ))) - } - - fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - let quotes = QUOTE_RECEIVED.condition_results(context, &round_instance(round.round))?; - if round.round < round.round_count { - return Ok(StepDecision::go_to( - &RequestQuotes, - QuoteRound { - round: round.round + 1, - ..round - }, - )); - } - Ok(StepDecision::graceful_complete(IgnoredQuotes { - round: round.round, - ignored_quotes: sorted(quotes), - })) - } -} - static BRANCH_A_SIGNAL: LazyLock> = LazyLock::new(|| Channel::new("branchASignal")); static BRANCH_B_SIGNAL: LazyLock> = LazyLock::new(|| Channel::new("branchBSignal")); static BRANCH_C_SIGNAL: LazyLock> = LazyLock::new(|| Channel::new("branchCSignal")); -static BRANCH_COMPLETED: LazyLock> = - LazyLock::new(|| ChannelMap::new("branchCompleted")); +static BRANCH_COMPLETED: LazyLock> = LazyLock::new(|| Join::new("branchCompleted")); pub const SEND_BRANCH_SIGNAL: Rpc = Rpc::new("send_branch_signal"); pub const GET_QUEUED_BRANCH_SIGNAL_COUNT: Rpc<(), usize> = @@ -284,6 +302,16 @@ pub const GET_QUEUED_BRANCH_SIGNAL_COUNT: Rpc<(), usize> = pub struct BlockingJoinRound { pub required_signals: usize, pub round: u32, + /// The Join round to reset before this round starts; `None` in round 1. + pub previous_round: Option, +} + +/// The input of the join Step of one blocking round. +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct BlockingRoundJoin { + pub blocking_round: BlockingJoinRound, + pub join_round: JoinRound, } /// The branches that fired the blocking join in one round. @@ -296,7 +324,7 @@ pub struct FiredBranchesRound { /// WCP-28 Blocking Discriminator (`required_signals` = 1) and WCP-31 /// Blocking Partial Join (`required_signals` = 2). No Step waits on a -/// next-round enablement until the current round resets. +/// next-round enablement until the next round's split resets the current round. #[derive(Default)] pub struct BlockingPartialJoinFlow { start_blocking_round: StartBlockingRound, @@ -305,7 +333,6 @@ pub struct BlockingPartialJoinFlow { signal_from_branch_c: SignalFromBranchC, blocking_join: BlockingJoin, process_round: ProcessRound, - reset_blocking_join: ResetBlockingJoin, } impl BlockingPartialJoinFlow { @@ -340,7 +367,6 @@ impl Flow for BlockingPartialJoinFlow { .and(&self.signal_from_branch_c) .and(&self.blocking_join) .and(&self.process_round) - .and(&self.reset_blocking_join) } fn persistence(&self) -> PersistenceSchema { @@ -348,7 +374,7 @@ impl Flow for BlockingPartialJoinFlow { .channel(&BRANCH_A_SIGNAL) .channel(&BRANCH_B_SIGNAL) .channel(&BRANCH_C_SIGNAL) - .channel_map(&BRANCH_COMPLETED) + .join(&BRANCH_COMPLETED) } fn rpcs(&self) -> RpcList { @@ -370,19 +396,41 @@ pub struct StartBlockingRound; impl Step for StartBlockingRound { type Input = BlockingJoinRound; - fn execute(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { + /// Resets the previous round: waits for the branches that did not fire its join. + fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { + match &round.previous_round { + None => Ok(Wait::skip_immediately()), + Some(previous_round) => Ok(Wait::until( + BRANCH_COMPLETED.for_remaining_branches(previous_round, round.required_signals), + )), + } + } + + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { if !(1..BLOCKING_BRANCH_COUNT).contains(&round.required_signals) { return Ok(StepDecision::force_fail(format!( "required signals must be 1 or 2, got {}", round.required_signals ))); } - Ok(StepDecision::go_to_many([ - StepMovement::to(&SignalFromBranchA, round.round), - StepMovement::to(&SignalFromBranchB, round.round), - StepMovement::to(&SignalFromBranchC, round.round), - StepMovement::to(&BlockingJoin, round), - ])) + let movements = BRANCH_COMPLETED.fork( + context, + vec![ + round_branch(SignalFromBranchA), + round_branch(SignalFromBranchB), + round_branch(SignalFromBranchC), + ], + |join_round| { + StepMovement::to( + &BlockingJoin, + BlockingRoundJoin { + blocking_round: round, + join_round, + }, + ) + }, + )?; + Ok(StepDecision::go_to_many(movements)) } } @@ -390,15 +438,14 @@ impl Step for StartBlockingRound { struct SignalFromBranchA; impl Step for SignalFromBranchA { - type Input = u32; + type Input = JoinRound; fn wait_for(&self, _context: &mut Context, _round: Self::Input) -> HandlerResult { Ok(Wait::until(BRANCH_A_SIGNAL.for_one())) } fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - BRANCH_COMPLETED.publish(context, &round_instance(round), "A".to_string())?; - Ok(StepDecision::dead_end()) + BRANCH_COMPLETED.end_branch(context, &round, "A".to_string()) } } @@ -406,15 +453,14 @@ impl Step for SignalFromBranchA { struct SignalFromBranchB; impl Step for SignalFromBranchB { - type Input = u32; + type Input = JoinRound; fn wait_for(&self, _context: &mut Context, _round: Self::Input) -> HandlerResult { Ok(Wait::until(BRANCH_B_SIGNAL.for_one())) } fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - BRANCH_COMPLETED.publish(context, &round_instance(round), "B".to_string())?; - Ok(StepDecision::dead_end()) + BRANCH_COMPLETED.end_branch(context, &round, "B".to_string()) } } @@ -422,15 +468,14 @@ impl Step for SignalFromBranchB { struct SignalFromBranchC; impl Step for SignalFromBranchC { - type Input = u32; + type Input = JoinRound; fn wait_for(&self, _context: &mut Context, _round: Self::Input) -> HandlerResult { Ok(Wait::until(BRANCH_C_SIGNAL.for_one())) } fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - BRANCH_COMPLETED.publish(context, &round_instance(round), "C".to_string())?; - Ok(StepDecision::dead_end()) + BRANCH_COMPLETED.end_branch(context, &round, "C".to_string()) } } @@ -438,27 +483,36 @@ impl Step for SignalFromBranchC { struct BlockingJoin; impl Step for BlockingJoin { - type Input = BlockingJoinRound; + type Input = BlockingRoundJoin; - fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { - Ok(Wait::until(BRANCH_COMPLETED.for_n( - &round_instance(round.round), - round.required_signals, + fn wait_for(&self, _context: &mut Context, input: Self::Input) -> HandlerResult { + Ok(Wait::until(BRANCH_COMPLETED.for_branches( + &input.join_round, + input.blocking_round.required_signals, ))) } - fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - let branches = BRANCH_COMPLETED.condition_results(context, &round_instance(round.round))?; - Ok(StepDecision::go_to_many([ - StepMovement::to( - &ProcessRound, - FiredBranchesRound { - round: round.round, - branches: sorted(branches), + fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult { + let branches = BRANCH_COMPLETED.branch_results(context, &input.join_round)?; + let round = input.blocking_round; + let mut movements = vec![StepMovement::to( + &ProcessRound, + FiredBranchesRound { + round: round.round, + branches: sorted(branches), + }, + )]; + if round.round < BLOCKING_JOIN_ROUND_COUNT { + movements.push(StepMovement::to( + &StartBlockingRound, + BlockingJoinRound { + round: round.round + 1, + previous_round: Some(input.join_round), + ..round }, - ), - StepMovement::to(&ResetBlockingJoin, round), - ])) + )); + } + Ok(StepDecision::go_to_many(movements)) } } @@ -477,40 +531,13 @@ impl Step for ProcessRound { } } -#[derive(Default)] -struct ResetBlockingJoin; - -impl Step for ResetBlockingJoin { - type Input = BlockingJoinRound; - - fn wait_for(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { - Ok(Wait::until(BRANCH_COMPLETED.for_n( - &round_instance(round.round), - BLOCKING_BRANCH_COUNT - round.required_signals, - ))) - } - - fn execute(&self, _context: &mut Context, round: Self::Input) -> HandlerResult { - if round.round < BLOCKING_JOIN_ROUND_COUNT { - return Ok(StepDecision::go_to( - &StartBlockingRound, - BlockingJoinRound { - round: round.round + 1, - ..round - }, - )); - } - Ok(StepDecision::graceful_complete(round.round)) - } -} - static CARRIER_A_RELEASED: LazyLock> = LazyLock::new(|| Channel::new("carrierAReleased")); static CARRIER_B_RELEASED: LazyLock> = LazyLock::new(|| Channel::new("carrierBReleased")); static CARRIER_C_RELEASED: LazyLock> = LazyLock::new(|| Channel::new("carrierCReleased")); -static BID_RECEIVED: LazyLock> = LazyLock::new(|| Channel::new("bidReceived")); +static BID_RECEIVED: LazyLock> = LazyLock::new(|| Join::new("bidReceived")); static CARRIER_BOOKING_CONFIRMED: LazyLock> = LazyLock::new(|| Channel::new("carrierBookingConfirmed")); @@ -519,9 +546,17 @@ pub const GET_QUEUED_CARRIER_RELEASE_COUNT: Rpc<(), usize> = Rpc::new("get_queued_carrier_release_count"); pub const CONFIRM_CARRIER_BOOKING: Rpc<(), ()> = Rpc::new("confirm_carrier_booking"); +/// The input of the join Step of the carrier bids. +#[derive(Clone, Debug, Deserialize, Serialize)] +#[serde(rename_all = "camelCase")] +pub struct CarrierBidJoin { + pub required_bids: usize, + pub join_round: JoinRound, +} + /// WCP-29 Cancelling Discriminator (`required_bids` = 1) and WCP-32 /// Cancelling Partial Join (`required_bids` = 2): the join cancels the -/// branches that have not reported. +/// sibling branches that have not reported. #[derive(Default)] pub struct CancellingPartialJoinFlow { request_bids: RequestBids, @@ -575,7 +610,7 @@ impl Flow for CancellingPartialJoinFlow { .channel(&CARRIER_A_RELEASED) .channel(&CARRIER_B_RELEASED) .channel(&CARRIER_C_RELEASED) - .channel(&BID_RECEIVED) + .join(&BID_RECEIVED) .channel(&CARRIER_BOOKING_CONFIRMED) } @@ -601,7 +636,7 @@ impl Step for RequestBids { fn execute( &self, - _context: &mut Context, + context: &mut Context, required_bids: Self::Input, ) -> HandlerResult { if !(1..=CARRIER_COUNT).contains(&required_bids) { @@ -609,12 +644,24 @@ impl Step for RequestBids { "required bids must be between 1 and {CARRIER_COUNT}, got {required_bids}" ))); } - Ok(StepDecision::go_to_many([ - StepMovement::to(&BidFromCarrierA, ()), - StepMovement::to(&BidFromCarrierB, ()), - StepMovement::to(&BidFromCarrierC, ()), - StepMovement::to(&AcceptBids, required_bids), - ])) + let movements = BID_RECEIVED.fork( + context, + vec![ + round_branch(BidFromCarrierA), + round_branch(BidFromCarrierB), + round_branch(BidFromCarrierC), + ], + |join_round| { + StepMovement::to( + &AcceptBids, + CarrierBidJoin { + required_bids, + join_round, + }, + ) + }, + )?; + Ok(StepDecision::go_to_many(movements)) } } @@ -622,15 +669,14 @@ impl Step for RequestBids { pub struct BidFromCarrierA; impl Step for BidFromCarrierA { - type Input = (); + type Input = JoinRound; - fn wait_for(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult { + fn wait_for(&self, _context: &mut Context, _round: Self::Input) -> HandlerResult { Ok(Wait::until(CARRIER_A_RELEASED.for_one())) } - fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { - BID_RECEIVED.publish(context, "A".to_string())?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + BID_RECEIVED.end_branch(context, &round, "A".to_string()) } } @@ -638,15 +684,14 @@ impl Step for BidFromCarrierA { struct BidFromCarrierB; impl Step for BidFromCarrierB { - type Input = (); + type Input = JoinRound; - fn wait_for(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult { + fn wait_for(&self, _context: &mut Context, _round: Self::Input) -> HandlerResult { Ok(Wait::until(CARRIER_B_RELEASED.for_one())) } - fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { - BID_RECEIVED.publish(context, "B".to_string())?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + BID_RECEIVED.end_branch(context, &round, "B".to_string()) } } @@ -654,15 +699,14 @@ impl Step for BidFromCarrierB { struct BidFromCarrierC; impl Step for BidFromCarrierC { - type Input = (); + type Input = JoinRound; - fn wait_for(&self, _context: &mut Context, _input: Self::Input) -> HandlerResult { + fn wait_for(&self, _context: &mut Context, _round: Self::Input) -> HandlerResult { Ok(Wait::until(CARRIER_C_RELEASED.for_one())) } - fn execute(&self, context: &mut Context, _input: Self::Input) -> HandlerResult { - BID_RECEIVED.publish(context, "C".to_string())?; - Ok(StepDecision::dead_end()) + fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { + BID_RECEIVED.end_branch(context, &round, "C".to_string()) } } @@ -670,23 +714,21 @@ impl Step for BidFromCarrierC { pub struct AcceptBids; impl Step for AcceptBids { - type Input = usize; + type Input = CarrierBidJoin; - fn wait_for(&self, _context: &mut Context, required_bids: Self::Input) -> HandlerResult { - Ok(Wait::until(BID_RECEIVED.for_n(required_bids))) + fn wait_for(&self, _context: &mut Context, input: Self::Input) -> HandlerResult { + Ok(Wait::until( + BID_RECEIVED.for_branches(&input.join_round, input.required_bids), + )) } - fn execute( - &self, - context: &mut Context, - _required_bids: Self::Input, - ) -> HandlerResult { - let accepted_bids = BID_RECEIVED.condition_results(context)?; + fn execute(&self, context: &mut Context, input: Self::Input) -> HandlerResult { + let accepted_bids = BID_RECEIVED.branch_results(context, &input.join_round)?; Ok( StepDecision::go_to(&ConfirmCarrierBooking, sorted(accepted_bids)) - .cancel_step(&BidFromCarrierA) - .cancel_step(&BidFromCarrierB) - .cancel_step(&BidFromCarrierC), + .cancel_sibling_step(&BidFromCarrierA) + .cancel_sibling_step(&BidFromCarrierB) + .cancel_sibling_step(&BidFromCarrierC), ) } } diff --git a/examples/rust/tests/workflow_patterns.rs b/examples/rust/tests/workflow_patterns.rs index f6e50e95f8..f6f51f80a2 100644 --- a/examples/rust/tests/workflow_patterns.rs +++ b/examples/rust/tests/workflow_patterns.rs @@ -55,8 +55,8 @@ use dex_examples_rust::workflow_patterns::multiple_instances::{ DesignTimeMultipleInstancesFlow, DynamicPartialJoinForMultipleInstancesFlow, GET_QUEUED_QUORUM_RELEASE_COUNT, GET_STARTED_ANALYSIS_COUNT, GET_STARTED_DESIGN_TIME_REVIEW_COUNT, GET_STARTED_EVALUATION_COUNT, - GET_STARTED_RUN_TIME_REVIEW_COUNT, IS_BID_ADMISSION_CLOSED, InspectedUnits, InspectionPlan, - InspectionRound, MultipleInstancesWithoutSynchronizationFlow, + GET_STARTED_RUN_TIME_REVIEW_COUNT, IGNORED_INSPECTIONS, IS_BID_ADMISSION_CLOSED, + InspectedUnits, InspectionPlan, InspectionRound, MultipleInstancesWithoutSynchronizationFlow, NoPriorKnowledgeMultipleInstancesFlow, OPEN_BID_GATE, OPEN_DESIGN_TIME_REVIEW_GATE, OPEN_RUN_TIME_REVIEW_GATE, OPEN_SAMPLE_GATE, RELEASE_INSPECTIONS, RELEASE_INVITATIONS, RELEASE_QUORUM_INSPECTIONS, RunTimeMultipleInstancesFlow, SIGN_CONTRACT, StartInspections, @@ -65,9 +65,10 @@ use dex_examples_rust::workflow_patterns::multiple_instances::{ use dex_examples_rust::workflow_patterns::partial_joins::{ AcceptBids, AcceptedQuotes, BidFromCarrierA, BlockingJoinRound, BlockingPartialJoinFlow, CONFIRM_CARRIER_BOOKING, CancellingPartialJoinFlow, FiredBranchesRound, - GET_QUEUED_BRANCH_SIGNAL_COUNT, GET_QUEUED_CARRIER_RELEASE_COUNT, IgnoredQuotes, PlaceOrder, - ProcessRound, QuoteFromSupplierA, QuoteFromSupplierC, QuoteRound, RELEASE_CARRIER_BID, - RELEASE_SUPPLIER_QUOTE, RequestQuotes, SEND_BRANCH_SIGNAL, StructuredPartialJoinFlow, + GET_QUEUED_BRANCH_SIGNAL_COUNT, GET_QUEUED_CARRIER_RELEASE_COUNT, IGNORED_QUOTES, + IgnoredQuotes, PlaceOrder, ProcessRound, QuoteFromSupplierA, QuoteFromSupplierC, QuoteRound, + RELEASE_CARRIER_BID, RELEASE_SUPPLIER_QUOTE, RequestQuotes, SEND_BRANCH_SIGNAL, + StructuredPartialJoinFlow, }; use dex_examples_rust::workflow_patterns::state_based::{ ACCEPT_OFFER, APPROVE_RELEASE, CriticalSectionFlow, DECLINE_OFFER, DeferredChoiceFlow, @@ -227,6 +228,15 @@ impl WorkflowPatternsEnvironment { assert_eq!(observed, expected); } + /// Reads a committed Attribute that the Flow must have set. + fn read_attribute(&self, flow_id: &str, attribute: &Attribute) -> T { + let (value,) = self + .client + .get_attributes(flow_id, (attribute,)) + .unwrap_or_else(|error| panic!("read an Attribute of {flow_id}: {error}")); + value.unwrap_or_else(|| panic!("{flow_id} has no value for the Attribute")) + } + fn flow_status(&self, flow_id: &str) -> FlowStatus { self.client .describe_flow(flow_id) @@ -529,6 +539,7 @@ fn first_quote_round(required_quotes: usize) -> QuoteRound { required_quotes, round: 1, round_count: 2, + previous_round: None, } } @@ -560,8 +571,8 @@ fn wcp09_structured_discriminator_fires_once_per_round() { [accepted_quotes(1, &["A"]), accepted_quotes(2, &["B"])] ); assert_eq!( - single_output_of::(&result, "ResetQuoteJoin"), - ignored_quotes(2, &["A", "C"]) + environment.read_attribute(&flow_id, &IGNORED_QUOTES), + ignored_quotes(1, &["B", "C"]) ); } @@ -598,8 +609,8 @@ fn wcp30_structured_partial_join_fires_once_per_round() { ] ); assert_eq!( - single_output_of::(&result, "ResetQuoteJoin"), - ignored_quotes(2, &["B"]) + environment.read_attribute(&flow_id, &IGNORED_QUOTES), + ignored_quotes(1, &["C"]) ); } @@ -620,6 +631,7 @@ fn wcp28_blocking_discriminator_blocks_a_next_round_signal_until_the_reset() { BlockingJoinRound { required_signals: 1, round: 1, + previous_round: None, }, ); let signal = |branch: &str| { @@ -655,6 +667,7 @@ fn wcp31_blocking_partial_join_blocks_next_round_signals_until_the_reset() { BlockingJoinRound { required_signals: 2, round: 1, + previous_round: None, }, ); let signal = |branch: &str| { @@ -884,6 +897,7 @@ fn wcp34_static_partial_join_for_instances_fires_once_per_round() { required_inspections: 2, round: 1, round_count: 2, + previous_round: None, }, ); environment.invoke(&flow_id, RELEASE_INSPECTIONS, 2); @@ -905,16 +919,16 @@ fn wcp34_static_partial_join_for_instances_fires_once_per_round() { assert_eq!(units.len(), 2); assert!(units.is_subset(&BTreeSet::from([0, 1, 2]))); } - let ignored: InspectedUnits = single_output_of(&result, "ResetInspectionJoin"); - assert_eq!(ignored.round, 2); + let ignored = environment.read_attribute(&flow_id, &IGNORED_INSPECTIONS); + assert_eq!(ignored.round, 1); assert_eq!(ignored.unit_indexes.len(), 1); - let round_two_units: BTreeSet = batches[1] + let round_one_units: BTreeSet = batches[0] .unit_indexes .iter() .chain(&ignored.unit_indexes) .copied() .collect(); - assert_eq!(round_two_units, BTreeSet::from([0, 1, 2])); + assert_eq!(round_one_units, BTreeSet::from([0, 1, 2])); } #[test] From a346d5ac0888c874c47897a63ebdf9d577d3afc6 Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Tue, 10 Nov 2026 19:50:27 -0800 Subject: [PATCH 2/3] Rust examples: match the 1.44 SDK startup error in the pull Worker test dex-sdk 1.44 reports a failed GetServerInfo as a typed service error for the get_server_info operation instead of a message containing "GetServerInfo failed". Assert the operation instead. --- examples/rust/tests/dex_integration.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/examples/rust/tests/dex_integration.rs b/examples/rust/tests/dex_integration.rs index 0262f068c9..2419807959 100644 --- a/examples/rust/tests/dex_integration.rs +++ b/examples/rust/tests/dex_integration.rs @@ -230,7 +230,10 @@ fn pull_worker_stops_when_startup_fails() { } Err(failure) => failure, }; - assert!(failure.to_string().contains("GetServerInfo failed")); + let Some(SdkError::Service { service }) = failure.downcast_ref::() else { + panic!("pull Worker startup failed with an unexpected error: {failure:?}"); + }; + assert_eq!(service.operation(), "get_server_info", "{failure:?}"); assert!(started_at.elapsed() < Duration::from_secs(45)); cache.close().expect("close Rust pull timeout cache"); } From b62ac2d0bb9c45cc4c8261665b8756bdc62ebdca Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Tue, 10 Nov 2026 19:58:28 -0800 Subject: [PATCH 3/3] Rust examples: validate partial join round inputs before the Join reset Fail the Flow when a split's round is below 1, when previous_round is missing after round 1 or present in round 1, or when required_inspections is outside 1..=inspection_count. Note that the last round's late results stay pending until the Flow closes. --- .../multiple_instances/instance_partial_joins.rs | 16 ++++++++++++---- .../rust/src/workflow_patterns/partial_joins.rs | 8 ++++++++ 2 files changed, 20 insertions(+), 4 deletions(-) diff --git a/examples/rust/src/workflow_patterns/multiple_instances/instance_partial_joins.rs b/examples/rust/src/workflow_patterns/multiple_instances/instance_partial_joins.rs index 0d56c1573d..572e57980e 100644 --- a/examples/rust/src/workflow_patterns/multiple_instances/instance_partial_joins.rs +++ b/examples/rust/src/workflow_patterns/multiple_instances/instance_partial_joins.rs @@ -151,10 +151,17 @@ impl Step for StartInspections { } fn execute(&self, context: &mut Context, round: Self::Input) -> HandlerResult { - if round.required_inspections > round.inspection_count { - return Ok(StepDecision::force_fail( - "required inspections exceed the inspection count", - )); + if !(1..=round.inspection_count).contains(&round.required_inspections) { + return Ok(StepDecision::force_fail(format!( + "required inspections must be between 1 and {}, got {}", + round.inspection_count, round.required_inspections + ))); + } + if round.round < 1 || round.previous_round.is_some() != (round.round > 1) { + return Ok(StepDecision::force_fail(format!( + "round must be at least 1 and carry a previous round exactly after round 1, got round {}", + round.round + ))); } if let Some(previous_round) = &round.previous_round { let late_units = UNIT_INSPECTED.remaining_branch_results(context, previous_round)?; @@ -229,6 +236,7 @@ impl Step for AcceptInspections { }, )); } + // The last round's late inspections stay pending until the Flow closes. Ok(StepDecision::go_to_many(movements)) } } diff --git a/examples/rust/src/workflow_patterns/partial_joins.rs b/examples/rust/src/workflow_patterns/partial_joins.rs index ba5b58b4f3..bbf78278c7 100644 --- a/examples/rust/src/workflow_patterns/partial_joins.rs +++ b/examples/rust/src/workflow_patterns/partial_joins.rs @@ -159,6 +159,12 @@ impl Step for RequestQuotes { round.required_quotes ))); } + if round.round < 1 || round.previous_round.is_some() != (round.round > 1) { + return Ok(StepDecision::force_fail(format!( + "round must be at least 1 and carry a previous round exactly after round 1, got round {}", + round.round + ))); + } if let Some(previous_round) = &round.previous_round { let late_quotes = QUOTE_RECEIVED.remaining_branch_results(context, previous_round)?; IGNORED_QUOTES.set( @@ -268,6 +274,7 @@ impl Step for AcceptQuotes { }, )); } + // The last round's late quotes stay pending until the Flow closes. Ok(StepDecision::go_to_many(movements)) } } @@ -512,6 +519,7 @@ impl Step for BlockingJoin { }, )); } + // The last round's late signals stay pending until the Flow closes. Ok(StepDecision::go_to_many(movements)) } }