Struct Join 

pub struct Join<T> { /* private fields */ }

Synchronizes the parallel branches of one Flow by counting the results they report.

Use a Join when one Step starts several branches and a later Step must continue after all of them, the first of them, or any n of them have finished. The Join produces the branch count and the round key, so the application keeps no counter and no round number.

A round has three roles:

1. The split Step calls Self::fork in execute. It passes one builder per branch and one builder for the join Step. Each builder receives the round’s JoinRound and returns a StepMovement. Return the movements with StepDecision::go_to_many.
2. Each branch reports exactly one result. A single-Step branch returns Self::end_branch. A branch that continues after reporting calls Self::report_branch_result.
3. The join Step waits in wait_for with Self::for_all_branches, Self::for_first_branch, or Self::for_branches. Its execute reads the results with Self::branch_results.

A Join is a ChannelMap of T with the same name. Each round uses its own instance, keyed by JoinRound::round_key. A branch result is stored as the plain T value, so Dex Web, pending-message reads, and Channel-map loads show and read the Join like any Channel map. Results of another round never satisfy this round’s condition. The conditions are ordinary Condition values. Combine them with crate::Wait::all_of, crate::Wait::any_of, or crate::Wait::any_combination_of, for example with a crate::Timer as a deadline.

After a first-of-m or n-of-m join, the other branches still report. To reuse the Join in a loop, reset the finished round, unless a cancelling join finished it. Pass its JoinRound to the next round’s split Step. That Step waits in wait_for with Self::for_remaining_branches and the count the join consumed. Its execute can read the late results with Self::remaining_branch_results. The next round therefore cannot start before the previous round’s branches have all reported. Without a reset, late results stay pending in the round’s instance until the Flow closes. They never count toward another round.

Add the Join to the Flow with crate::PersistenceSchema::join. Declare it as a module-level std::sync::LazyLock static. The name follows the Channel-map rules: it is stable, it contains no /, and it differs from every other persistence definition of the Flow.

Rules and traps

• Each branch reports exactly one result per round. A branch that never reports keeps an all-of join waiting forever. A branch that reports twice counts twice. When an execute failure routes a branch to a recovery Step, that Step must report a result that represents the failure.
• A multi-Step branch passes its JoinRound down to the Step that reports. A branch passes the round only to Self::end_branch or Self::report_branch_result. It never waits on a join condition.
• Only the join Step that the fork started waits for a round. Two Steps that wait for the same round split its results between them.
• The received_count of Self::for_remaining_branches equals the number of results the join consumed: count after Self::for_branches, and 1 after Self::for_first_branch.
• A reset waits for every branch that has not reported yet, so it never completes while a canceled branch has not reported. Never reset the round of a cancelling join. The next round forks a new round without the reset wait, as the first round does. Without a reset, the round’s late results stay pending until the Flow closes.
• One invocation forks one Join at most once with Self::fork. Use a second Join, or Self::fork_with_round_key with distinct keys, for two independent joins.
• An RPC has no Step execution ID, so it must use Self::fork_with_round_key.
• A cancelling join adds StepDecision::cancel_sibling_step for each branch Step type to the join Step’s decision. The fork starts the branches and the join Step from one decision, so they are siblings, and only this round’s branches are selected. A later Step of a multi-Step branch is not a sibling. Select it with StepDecision::cancel_step, which is correct only while one round is active. Steps started by an RPC count as siblings of every Step started by any call of that RPC. For rounds that an RPC forks, also use StepDecision::cancel_step with one active round.
• A canceled branch never reports, and no other code can safely report for it. Dex decides at commit time whether a cancellation takes effect, so the code that cancels a branch cannot know whether the branch already reported. If it did, the cancellation does nothing, and an extra report counts that branch twice, so an all-of join can complete while another branch has not reported. To keep an all-of join from waiting forever for a canceled branch, cancel the join Step together with its branches, or wait on crate::Wait::any_of with the join condition and a crate::Timer and handle the timeout in execute.
• When a join Step is canceled after its wait matched, what happens to the results it consumed depends on the server version. A server that requeues them puts them back at the front of the round’s instance, unless the canceling decision closes the Flow. On a server that drops them, a new join Step for that round never completes. See the server-zd change log at https://docs.superdurable.io/reference/server/changelog.

Examples

A structured partial join that loops: each round asks three suppliers, continues after the first required_quotes quotes, and resets before the next round.

use std::sync::LazyLock;

use dex_sdk::{
    Context, Flow, HandlerResult, Join, JoinRound, PersistenceSchema, Step, StepDecision,
    StepList, StepMovement, Wait,
};
use serde::{Deserialize, Serialize};

static QUOTE_RECEIVED: LazyLock<Join<String>> = LazyLock::new(|| Join::new("quoteReceived"));

#[derive(Serialize, Deserialize)]
struct QuoteRequest {
    required_quotes: usize,
    round_number: u32,
    round_count: u32,
    previous_round: Option<JoinRound>,
}

#[derive(Serialize, Deserialize)]
struct SupplierQuote {
    supplier: String,
    round: JoinRound,
}

#[derive(Serialize, Deserialize)]
struct AcceptQuotesInput {
    request: QuoteRequest,
    round: JoinRound,
}

// Split: resets the previous round, then forks one branch per supplier and the join.
struct RequestQuotes;

impl Step for RequestQuotes {
    type Input = QuoteRequest;

    fn wait_for(&self, _context: &mut Context, request: QuoteRequest) -> HandlerResult<Wait> {
        match &request.previous_round {
            None => Ok(Wait::skip_immediately()),
            Some(previous_round) => Ok(Wait::until(
                QUOTE_RECEIVED.for_remaining_branches(previous_round, request.required_quotes),
            )),
        }
    }

    fn execute(
        &self,
        context: &mut Context,
        request: QuoteRequest,
    ) -> HandlerResult<StepDecision> {
        if let Some(previous_round) = &request.previous_round {
            let late_quotes = QUOTE_RECEIVED.remaining_branch_results(context, previous_round)?;
            context.record_event("lateQuotes", late_quotes)?;
        }
        let branch_builders = ["A", "B", "C"]
            .into_iter()
            .map(|supplier| {
                Box::new(move |round: &JoinRound| {
                    StepMovement::to(
                        &QuoteFromSupplier,
                        SupplierQuote {
                            supplier: supplier.to_string(),
                            round: round.clone(),
                        },
                    )
                }) as Box<dyn FnOnce(&JoinRound) -> StepMovement>
            })
            .collect();
        let movements = QUOTE_RECEIVED.fork(context, branch_builders, |round| {
            StepMovement::to(&AcceptQuotes, AcceptQuotesInput { request, round })
        })?;
        Ok(StepDecision::go_to_many(movements))
    }
}

// Branch: reports one result and ends.
struct QuoteFromSupplier;

impl Step for QuoteFromSupplier {
    type Input = SupplierQuote;

    fn execute(
        &self,
        context: &mut Context,
        quote: SupplierQuote,
    ) -> HandlerResult<StepDecision> {
        QUOTE_RECEIVED.end_branch(context, &quote.round, quote.supplier)
    }
}

// Join: continues after the required number of quotes, then starts the next round.
struct AcceptQuotes;

impl Step for AcceptQuotes {
    type Input = AcceptQuotesInput;

    fn wait_for(&self, _context: &mut Context, input: AcceptQuotesInput) -> HandlerResult<Wait> {
        Ok(Wait::until(
            QUOTE_RECEIVED.for_branches(&input.round, input.request.required_quotes),
        ))
    }

    fn execute(
        &self,
        context: &mut Context,
        input: AcceptQuotesInput,
    ) -> HandlerResult<StepDecision> {
        let accepted_quotes = QUOTE_RECEIVED.branch_results(context, &input.round)?;
        let request = input.request;
        if request.round_number == request.round_count {
            return Ok(StepDecision::graceful_complete(accepted_quotes));
        }
        Ok(StepDecision::go_to(
            &RequestQuotes,
            QuoteRequest {
                round_number: request.round_number + 1,
                previous_round: Some(input.round),
                ..request
            },
        ))
    }
}

struct QuoteFlow {
    request_quotes: RequestQuotes,
    quote_from_supplier: QuoteFromSupplier,
    accept_quotes: AcceptQuotes,
}

impl Flow for QuoteFlow {
    type StartInput = QuoteRequest;

    fn steps(&self) -> StepList<'_, QuoteRequest> {
        StepList::start(&self.request_quotes)
            .and(&self.quote_from_supplier)
            .and(&self.accept_quotes)
    }

    fn persistence(&self) -> PersistenceSchema {
        PersistenceSchema::new().join(&QUOTE_RECEIVED)
    }
}

Since 1.44.0.

Implementations

impl<T> Join<T>

pub fn new(name: impl Into<String>) -> Self

Defines a Join whose branch results use the Channel map named name.

The name is the durable identity of the Join. Keep it stable while any Flow that uses it is running. Slash is prohibited because it is a reserved character. A Join that replaces a hand-written Channel-map join keeps in-flight Flows working when it reuses that map’s name and value type and forks with Self::fork_with_round_key and the same instance keys.

Since 1.44.0.

pub fn fork<'builder>( &self, context: &mut Context, branch_builders: Vec<Box<dyn FnOnce(&JoinRound) -> StepMovement + 'builder>>, join_builder: impl FnOnce(JoinRound) -> StepMovement, ) -> HandlerResult<Vec<StepMovement>>

Starts one round from a split Step: builds every branch movement and the join movement.

The round key is the current Step execution ID, which stays the same across retries and time travel. The branch count is branch_builders.len(). The SDK calls each branch builder once, in order, with the round, and then calls join_builder with the same round. The returned movements are the branch movements in builder order followed by the join movement. Pass them to StepDecision::go_to_many, optionally with more movements. The call stages no durable change.

An empty branch_builders creates a round with zero branches. Its Self::for_all_branches condition completes immediately, so a multi-choice split that selects no branch needs no special case.

Each closure has its own type, so box every branch builder. Map a runtime list to builders to start one branch per element; see the example on Join.

Errors

Returns a HandlerError when the Context has no Step execution ID, as in an RPC; use Self::fork_with_round_key there. Also returns an error when this invocation already forked this Join with fork. Both errors are deterministic, so a retry fails the same way.

Since 1.44.0.

pub fn fork_with_round_key<'builder>( &self, round_key: &str, branch_builders: Vec<Box<dyn FnOnce(&JoinRound) -> StepMovement + 'builder>>, join_builder: impl FnOnce(JoinRound) -> StepMovement, ) -> HandlerResult<Vec<StepMovement>>

Starts one round with a caller-supplied round key; otherwise behaves like Self::fork.

Use it in an RPC, which has no Step execution ID, and add the returned movements to the crate::RpcResult with crate::RpcResult::then. Also use it to keep the instance keys of a hand-written Channel-map join, or to fork one Join twice in one invocation with distinct keys. The key must be unique among the Join’s active rounds; a reused key mixes results of two rounds. This method does not check for a repeated fork.

Errors

Returns a HandlerError when round_key is empty or contains /.

Since 1.44.0.

pub fn end_branch( &self, context: &mut Context, round: &JoinRound, result: T, ) -> HandlerResult<StepDecision>

where T: Value,

Reports this branch’s result and returns StepDecision::dead_end.

Call it as the return value of a branch’s last Step. The result is staged like any Channel publication: it becomes durable only when the handler returns successfully, together with the decision. A failed attempt reports nothing, and a retry does not report twice.

Errors

Returns the errors of Self::report_branch_result.

Since 1.44.0.

pub fn report_branch_result( &self, context: &mut Context, round: &JoinRound, result: T, ) -> HandlerResult<()>

where T: Value,

Reports this branch’s result without ending the branch.

Use it when the branch continues after reporting, from a Step’s wait_for or execute, for example when a multi-Step branch reports from a Step other than its last. An RPC can report a branch’s result when that result arrives through the RPC and no Step of that branch reports it, for example for a handoff branch in a round the RPC forked with Self::fork_with_round_key; its input then carries the round. Never report for a canceled branch; see the rules on Join. The result is staged and becomes durable only when the handler returns successfully. Results queue in the order their handlers commit.

Errors

Returns a HandlerError when the Join is not in the Flow’s persistence schema, when the round key is empty or contains /, or when result cannot be encoded.

Since 1.44.0.

pub fn for_all_branches(&self, round: &JoinRound) -> Condition

Creates a condition that consumes the results of every branch in the round.

It consumes exactly JoinRound::branch_count results. A round with zero branches completes immediately and consumes nothing.

An invalid round, with an empty key or a key containing /, makes the Step’s wait_for fail when the SDK sends the Wait.

Since 1.44.0.

pub fn for_first_branch(&self, round: &JoinRound) -> Condition

Creates a condition that consumes the first reported result of the round.

Use it for a discriminator. When several results are already queued, it consumes only the earliest one; the others remain for Self::for_remaining_branches with a received_count of 1. Put the branch identity into T when the join needs to know which branch won. It is Self::for_branches with a count of 1.

A round with zero branches, or an invalid round, makes the Step’s wait_for fail when the SDK sends the Wait.

Since 1.44.0.

pub fn for_branches(&self, round: &JoinRound, count: usize) -> Condition

Creates a condition that consumes the first count reported results of the round.

Use it for a partial join that continues after count of the round’s branches have reported. Pass the same count to Self::for_remaining_branches when a loop resets the round.

A count outside 1..=round.branch_count, or an invalid round, makes the Step’s wait_for fail when the SDK sends the Wait.

Since 1.44.0.

pub fn for_remaining_branches( &self, round: &JoinRound, received_count: usize, ) -> Condition

Creates a condition that consumes the results a finished round has not consumed yet.

Use it in the next round’s split Step wait_for to reset the Join. It waits for, and consumes, round.branch_count - received_count results. received_count is the number of results the join consumed: the count of Self::for_branches, or 1 for Self::for_first_branch. When nothing remains, it completes immediately.

received_count counts only results the join consumed. Never count a canceled branch, which does not report. The reset therefore waits for every canceled branch too, so never reset the round of a cancelling join; see the rules on Join.

A received_count greater than round.branch_count, or an invalid round, makes the Step’s wait_for fail when the SDK sends the Wait.

Since 1.44.0.

pub fn branch_results( &self, context: &Context, round: &JoinRound, ) -> HandlerResult<Vec<T>>

where T: Value,

Returns the results the join Step’s satisfied condition consumed, in report order.

Call it in the join Step’s execute. It returns an empty list when the round’s condition did not complete, for example when a Timer won a crate::Wait::any_of.

Errors

Returns a HandlerError when the Join is not in the Flow’s persistence schema, when the round key is invalid, or when a result cannot be decoded.

Since 1.44.0.

pub fn remaining_branch_results( &self, context: &Context, round: &JoinRound, ) -> HandlerResult<Vec<T>>

where T: Value,

Returns the late results that a reset with Self::for_remaining_branches consumed.

Call it in the execute of the Step that waited for the reset, usually the next round’s split Step. It reads the same consumed results as Self::branch_results; the separate name states at the call site that these results arrived after the round’s join.

Errors

Returns the errors of Self::branch_results.

Since 1.44.0.

pub fn channel_map(&self) -> &ChannelMap<T>

Returns the Channel map that holds this Join’s branch results.

Use it to read a round’s pending results with the existing Channel-map API. The instance key is the round’s JoinRound::round_key. For example, an RPC selects the round with crate::Rpc::load_channel_map or ChannelMap::load_messages, then calls ChannelMap::pending_messages or ChannelMap::size. The returned definition is the one that crate::PersistenceSchema::join registers, so every API that accepts a registered Channel map accepts it.

Never publish to it or delete from it directly. Report results only with Self::end_branch or Self::report_branch_result. A direct publication counts as a branch result and breaks the rule that each branch reports exactly once.

use std::sync::LazyLock;

use dex_sdk::{Context, HandlerResult, Join, Rpc, RpcResult};

static QUOTE_RECEIVED: LazyLock<Join<String>> = LazyLock::new(|| Join::new("quoteReceived"));

const GET_PENDING_QUOTES: Rpc<String, Vec<String>> = Rpc::new("get_pending_quotes");

// Register with .function(GET_PENDING_QUOTES.load_channel_map(QUOTE_RECEIVED.channel_map()), ...).
fn get_pending_quotes(context: &mut Context, round_key: String) -> HandlerResult<RpcResult<Vec<String>>> {
    let pending_quotes = QUOTE_RECEIVED
        .channel_map()
        .pending_messages(context, &round_key)?
        .into_iter()
        .map(|message| message.value)
        .collect();
    Ok(RpcResult::new(pending_quotes))
}

Since 1.44.0.

Trait Implementations

impl<T> Clone for Join<T>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.