Dex TypeScript SDK 1.44.0
    Preparing search index...

    Class Join<T>

    Synchronizes parallel branches: a join Step waits until branches of the same round report results.

    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 report a result. A Join is a ChannelMap named Join.name. Each round uses its own ChannelMap instance. Each branch publishes one message: the result value T itself.

    Declare a Join once, at module level, and register it in the Flow's getPersistenceSchema().channels. Its name must not match another Channel, ChannelMap, Join, Attribute, or Stream of the Flow. Dex Web shows it as a ChannelMap.

    Use it in four places:

    Rules and traps:

    • Each branch must report exactly one result per round, including on its recovery path. A branch that never reports keeps an all-branches join waiting forever. A branch that reports twice is counted twice. Add a Timer, such as Wait.anyOf(join.forAllBranches(round), Timer.byDuration(...)), when a branch may never report.
    • A branch passes its JoinRound only to endBranch or reportBranchResult. It must not wait on a join Condition. Only the join Step that the fork started waits on a round; two waiting Steps would split the results between them.
    • A Step or retry attempt forks one Join at most once. A second Join.fork call in the same attempt throws. Use two Joins, or Join.forkWithRoundKey with distinct keys, for two independent joins.
    • An RPC has no Step execution ID, so fork throws there. Use forkWithRoundKey in an RPC.
    • The receivedCount passed to forRemainingBranches must equal the number of results the join consumed: the count of forBranches, 1 for forFirstBranch, or branchCount for forAllBranches.
    • Results that arrive after the join fired stay in the round's ChannelMap instance until a reset consumes them or the Flow ends. They never satisfy another round. In a loop, always reset.
    • To cancel the remaining branches when the join fires, return the join decision with withCancelingSiblingSteps(decision, BranchA, BranchB). fork puts every branch and the join Step in one decision, so they are siblings. Sibling cancellation does not reach later Steps of a multi-Step branch; use withCancelingSteps there when only one round runs at a time. Steps started by an RPC are siblings of every Step started by any call of that RPC, so for a round forked by Join.forkWithRoundKey in an RPC, cancel the branches with withCancelingSteps.
    • A canceled branch reports nothing, so an all-branches join keeps waiting. Cancel the join Step together with the branches, add a Timer, or report a substitute result for the canceled branch from the canceling RPC with Join.reportBranchResult.
    • The same holds for a reset: forRemainingBranches waits for results that canceled branches never report. A cancelling join's round therefore resets only when an RPC reports a substitute result for each canceled branch. Without that, do not reset the round; its late results stay until the Flow ends.
    • When Dex cancels a join Step after its wait consumed results but before its execute commits, what happens to those results depends on the server version. A server that requeues consumed messages on cancellation returns them to the front of the round's instance, unless the canceling decision closes the Flow unconditionally. An older server drops them, and the round can no longer fire. See the server change log.
    import {
      Join,
      StepList,
      StepMovement,
      Wait,
      goToMany,
      gracefulComplete,
      joinRoundCodec,
      jsonCodec,
      stringCodec,
      type Context,
      type Flow,
      type JoinRound,
      type PersistenceSchema,
      type Step,
      type StepDecision,
    } from "@superdurable/dex";

    const quoteReceived = new Join("quoteReceived", stringCodec);

    interface QuoteRequest {
      readonly requiredQuotes: number;
      readonly roundNumber: number;
      readonly roundCount: number;
      readonly previousRound?: JoinRound;
    }

    interface QuoteJoinInput {
      readonly request: QuoteRequest;
      readonly round: JoinRound;
    }

    // Split: waits for the previous round's reset, then starts two branches and the join Step.
    class RequestQuotes implements Step<QuoteRequest> {
      public getStepType(): string {
        return "RequestQuotes";
      }

      public waitFor(_context: Context, request: QuoteRequest): Wait {
        if (request.previousRound === undefined) {
          return Wait.skipImmediately();
        }
        // The join consumed requiredQuotes results; consume the rest of that round.
        return Wait.until(
          quoteReceived.forRemainingBranches(request.previousRound, request.requiredQuotes),
        );
      }

      public execute(context: Context, request: QuoteRequest): StepDecision {
        if (request.previousRound !== undefined) {
          const lateQuotes = quoteReceived.remainingBranchResults(context, request.previousRound);
          console.log(`ignored ${lateQuotes.length} late quotes`);
        }
        return goToMany(...quoteReceived.fork(
          context,
          [
            (round) => StepMovement.of(QuoteFromSupplierA, round),
            (round) => StepMovement.of(QuoteFromSupplierB, round),
          ],
          (round) => StepMovement.of(AcceptQuotes, { request, round }),
        ));
      }
    }

    // Branches: each reports exactly one result and ends.
    class QuoteFromSupplierA implements Step<JoinRound> {
      public readonly inputCodec = joinRoundCodec;

      public getStepType(): string {
        return "QuoteFromSupplierA";
      }

      public execute(context: Context, round: JoinRound): StepDecision {
        return quoteReceived.endBranch(context, round, "A");
      }
    }

    class QuoteFromSupplierB implements Step<JoinRound> {
      public readonly inputCodec = joinRoundCodec;

      public getStepType(): string {
        return "QuoteFromSupplierB";
      }

      public execute(context: Context, round: JoinRound): StepDecision {
        return quoteReceived.endBranch(context, round, "B");
      }
    }

    // Join: continues after the first requiredQuotes results, then starts the next round.
    class AcceptQuotes implements Step<QuoteJoinInput> {
      public getStepType(): string {
        return "AcceptQuotes";
      }

      public waitFor(_context: Context, input: QuoteJoinInput): Wait {
        return Wait.until(quoteReceived.forBranches(input.round, input.request.requiredQuotes));
      }

      public execute(context: Context, input: QuoteJoinInput): StepDecision {
        const acceptedQuotes = quoteReceived.branchResults(context, input.round);
        const placeOrder = StepMovement.of(PlaceOrder, acceptedQuotes);
        if (input.request.roundNumber >= input.request.roundCount) {
          return goToMany(placeOrder);
        }
        const nextRequest: QuoteRequest = {
          ...input.request,
          roundNumber: input.request.roundNumber + 1,
          previousRound: input.round,
        };
        return goToMany(placeOrder, StepMovement.of(RequestQuotes, nextRequest));
      }
    }

    class PlaceOrder implements Step<readonly string[]> {
      public readonly inputCodec = jsonCodec<readonly string[]>();

      public getStepType(): string {
        return "PlaceOrder";
      }

      public execute(_context: Context, quotes: readonly string[]): StepDecision {
        return gracefulComplete(quotes);
      }
    }

    export class QuoteFlow implements Flow<QuoteRequest> {
      public getFlowType(): string {
        return "QuoteFlow";
      }

      public getSteps() {
        return StepList.startStep(new RequestQuotes()).otherSteps(
          new QuoteFromSupplierA(),
          new QuoteFromSupplierB(),
          new AcceptQuotes(),
          new PlaceOrder(),
        );
      }

      public getPersistenceSchema(): PersistenceSchema {
        return { channels: [quoteReceived] };
      }
    }

    For a runtime number of branches, map the inputs to branch builders, as the Join.forkWithRoundKey example does.

    1.44.0

    Type Parameters

    • T

      Result type that each branch reports.

    Index

    Constructors

    constructor

    Properties

    channelMap codec name

    Methods

    branchResults endBranch forAllBranches forBranches forFirstBranch fork forkWithRoundKey forRemainingBranches remainingBranchResults reportBranchResult

    Constructors

    • Creates a Join definition for a PersistenceSchema.

      Type Parameters

      • T

      Parameters

      • name: string

        Durable name of the Join and of its ChannelMap. It must be non-empty, must not contain /, and must be unique among the Flow's persistence definitions. Changing it while Flows are running loses their pending results.

      • codec: Codec<T>

        Codec of the branch result type. Changing the encoding while Flows are running can make pending results undecodable.

      Returns Join<T>

      When name is empty, contains only whitespace, or contains /.

    Properties

    channelMap: ChannelMap<T>

    The ChannelMap that carries this Join's branch results, named Join.name.

    Use it to read a round's pending results with the existing ChannelMap API. The instance key of a round is its JoinRound.roundKey. For example, an RPC selects the round with join.channelMap.loadMessages(round.roundKey) in its load options, then reads join.channelMap.pendingMessages(context, round.roundKey) or join.channelMap.size(context, round.roundKey). Pending results are the results that no join or reset has consumed, such as late results after a first-branch join.

    This is the exact definition the Flow registers when its PersistenceSchema lists the Join, so every API that checks for a registered ChannelMap accepts it.

    Never publish to it or delete its messages directly. Report results only with Join.endBranch or Join.reportBranchResult. A direct publication or deletion breaks the rule that each branch adds exactly one result per round, so joins and resets count wrong.

    class QuoteFlow implements Flow<never> {
      // getFlowType, getSteps, and getPersistenceSchema register the Steps and quoteReceived.

      @rpc()
      public getLateQuotes(context: Context, round: JoinRound): RPCResult<readonly string[]> {
        const messages = quoteReceived.channelMap.pendingMessages(context, round.roundKey);
        return { output: messages.map((message) => message.value) };
      }
    }

    const lateQuotes = await client.invokeRPCWithOptions(quoteFlow.getLateQuotes, "quote-42", round, {
      loadChannelMapInstances: [quoteReceived.channelMap.loadMessages(round.roundKey)],
    });

    1.44.0

    codec: Codec<T>

    Codec of the branch result type. Changing the encoding while Flows are running can make pending results undecodable.

    name: string

    Durable name of the Join and of its ChannelMap. It must be non-empty, must not contain /, and must be unique among the Flow's persistence definitions. Changing it while Flows are running loses their pending results.

    Methods

    • Returns the results that the join Step's satisfied Condition consumed, in commit order.

      Call it in the join Step's execute. It returns an empty array when the round's Condition was not the one that made the Wait ready, such as when a Timer won a Wait.anyOf.

      Parameters

      • context: Context

        Context of the join Step's execute call.

      • round: JoinRound

        Round that the join Step waited on.

      Returns readonly T[]

      The consumed results in the order Dex committed them.

      When round has an invalid key or the Join is not registered by this Flow.

      When round.branchCount is not a non-negative safe integer.

    • Reports this branch's result and ends the branch.

      Return it from the execute of a branch's last Step. It stages the result like Join.reportBranchResult and returns a dead-end decision. Dex commits the result only when this execute call succeeds, so a retried attempt does not report twice.

      Parameters

      • context: Context

        Context of the current Step execute call.

      • round: JoinRound

        Round that the fork passed to this branch.

      • result: T

        Result of this branch.

      Returns StepDecision

      A dead-end decision.

      When round has an invalid key or the Join is not registered by this Flow.

      When round.branchCount is not a non-negative safe integer.

    • Creates a Condition that waits for the results of every branch of a round.

      Return it from the join Step's waitFor. It consumes exactly round.branchCount results. A round with 0 branches is ready immediately. Combine it with a Timer through Wait.anyOf when a branch may never report. See Join for an example.

      Parameters

      • round: JoinRound

        Round that the fork passed to the join Step.

      • OptionalconditionId: string

        Optional stable Condition ID, required inside Wait.anyCombinationOf.

      Returns Condition

      A Channel Condition on the round's ChannelMap instance.

      When round has an invalid key or conditionId is empty.

      When round.branchCount is not a non-negative safe integer.

    • Creates a Condition that waits for count results of a round.

      Return it from the join Step's waitFor for a partial join. It consumes exactly count results in the order Dex committed them. The reset of this round uses count as its receivedCount. See Join for an example.

      Parameters

      • round: JoinRound

        Round that the fork passed to the join Step.

      • count: number

        Number of results to wait for, from 1 to round.branchCount.

      • OptionalconditionId: string

        Optional stable Condition ID, required inside Wait.anyCombinationOf.

      Returns Condition

      A Channel Condition on the round's ChannelMap instance.

      When round has an invalid key or conditionId is empty.

      When count is not an integer from 1 to round.branchCount.

    • Creates a Condition that waits for the first result of a round.

      Return it from the join Step's waitFor to continue with the first branch that reports. It consumes exactly one result, the earliest one Dex committed. Put the branch's identity in the result when the join must know which branch won. The reset of this round uses a receivedCount of 1. See Join for an example.

      It behaves like forBranches(round, 1). A round with 0 branches has no first result and would wait forever, so this method throws for it, which fails the waitFor attempt.

      Parameters

      • round: JoinRound

        Round that the fork passed to the join Step.

      • OptionalconditionId: string

        Optional stable Condition ID, required inside Wait.anyCombinationOf.

      Returns Condition

      A Channel Condition on the round's ChannelMap instance.

      When round has an invalid key or conditionId is empty.

      When round.branchCount is 0 or not a non-negative safe integer.

    • Starts one round from a Step: creates its JoinRound and the branch and join movements.

      Call it in the execute of the split Step and return the movements, for example goToMany(...join.fork(context, branchBuilders, joinBuilder)). Add more movements to the same decision when needed. The round key is the current Step execution ID, which stays the same across retries. The branch count is branchBuilders.length, so the application never counts branches. An empty list creates a round with 0 branches, whose Join.forAllBranches Condition is ready immediately.

      The fork calls each branch builder in order, then the join builder, and returns their movements in that order. It does not contact Dex. See Join for a complete example.

      Parameters

      • context: Context

        Context of the current Step waitFor or execute call.

      • branchBuilders: readonly ((round: JoinRound) => StepMovement<any>)[]

        One function per branch. Each receives the round and returns the movement that starts that branch.

      • joinBuilder: (round: JoinRound) => StepMovement<any>

        Receives the round and returns the movement of the join Step.

      Returns readonly StepMovement<any>[]

      The branch movements followed by the join movement.

      When context belongs to an RPC, which has no Step execution ID. Use Join.forkWithRoundKey there.

      When the same attempt already forked this Join.

    • Starts one round with an application-supplied round key.

      Use it in an RPC, which has no Step execution ID, or when the application already has a unique key per round. It behaves like Join.fork otherwise. Return the movements as the RPC's nextSteps or in a Step decision.

      The key must be unique per round of this Join within the Flow run. Reusing a key mixes results of different rounds. Steps started by an RPC are siblings of every Step started by any call of the same RPC, so prefer withCancelingSteps over withCancelingSiblingSteps for these rounds. When you migrate a hand-written ChannelMap join to a Join with the same name, pass the old instance key to keep running Flows working.

      Parameters

      • roundKey: string

        ChannelMap instance key of the round. It must be non-empty and must not contain /.

      • branchBuilders: readonly ((round: JoinRound) => StepMovement<any>)[]

        One function per branch. Each receives the round and returns the movement that starts that branch.

      • joinBuilder: (round: JoinRound) => StepMovement<any>

        Receives the round and returns the movement of the join Step.

      Returns readonly StepMovement<any>[]

      The branch movements followed by the join movement.

      import {
        Join,
        StepList,
        StepMovement,
        Wait,
        gracefulComplete,
        joinRoundCodec,
        rpc,
        stringCodec,
        type Context,
        type Flow,
        type JoinRound,
        type PersistenceSchema,
        type RPCResult,
        type Step,
        type StepDecision,
      } from "@superdurable/dex";

      const reviewCompleted = new Join("reviewCompleted", stringCodec);

      interface ReviewRequest {
        readonly requestId: string;
        readonly reviewers: readonly string[];
      }

      interface ReviewerInput {
        readonly reviewer: string;
        readonly round: JoinRound;
      }

      class ReviewByReviewer implements Step<ReviewerInput> {
        public getStepType(): string {
          return "ReviewByReviewer";
        }

        public execute(context: Context, input: ReviewerInput): StepDecision {
          return reviewCompleted.endBranch(context, input.round, `${input.reviewer}: approved`);
        }
      }

      class CollectReviews implements Step<JoinRound> {
        public readonly inputCodec = joinRoundCodec;

        public getStepType(): string {
          return "CollectReviews";
        }

        public waitFor(_context: Context, round: JoinRound): Wait {
          return Wait.until(reviewCompleted.forAllBranches(round));
        }

        public execute(context: Context, round: JoinRound): StepDecision {
          return gracefulComplete(reviewCompleted.branchResults(context, round));
        }
      }

      export class ReviewFlow implements Flow<never> {
        public getFlowType(): string {
          return "ReviewFlow";
        }

        public getSteps() {
          return StepList.withoutStartStep().otherSteps(new ReviewByReviewer(), new CollectReviews());
        }

        public getPersistenceSchema(): PersistenceSchema {
          return { channels: [reviewCompleted] };
        }

        @rpc()
        public requestReviews(_context: Context, request: ReviewRequest): RPCResult<void> {
          return {
            output: undefined,
            nextSteps: reviewCompleted.forkWithRoundKey(
              `review-${request.requestId}`,
              request.reviewers.map((reviewer) => (round: JoinRound) =>
                StepMovement.of(ReviewByReviewer, { reviewer, round })),
              (round) => StepMovement.of(CollectReviews, round),
            ),
          };
        }
      }

      When roundKey is empty, contains only whitespace, or contains /.

    • Creates a Condition that consumes the remaining results of a round, which resets it.

      Return it from the waitFor of the next round's split Step, with the previous round from that Step's input. It waits for the round.branchCount - receivedCount results that the join did not consume. The next round therefore starts only after every branch of the previous round has reported, and the previous round leaves no messages behind. When nothing remains, it is ready immediately. See Join for an example.

      A canceled branch never reports, so this Condition never becomes ready for a round whose branches were canceled. Reset such a round only after an RPC reports a substitute result for each canceled branch with Join.reportBranchResult. Never add canceled branches to receivedCount.

      Parameters

      • round: JoinRound

        Previous round, as the join Step received it.

      • receivedCount: number

        Number of results the join consumed: the count of forBranches, 1 for forFirstBranch, or round.branchCount for forAllBranches. It must be from 0 to round.branchCount.

      • OptionalconditionId: string

        Optional stable Condition ID, required inside Wait.anyCombinationOf.

      Returns Condition

      A Channel Condition on the round's ChannelMap instance.

      When round has an invalid key or conditionId is empty.

      When receivedCount is not an integer from 0 to round.branchCount.

    • Returns the late results that Join.forRemainingBranches consumed, in commit order.

      Call it in the execute of the Step that waited for the reset. These results arrived after the join had fired, so they did not affect the join.

      Parameters

      • context: Context

        Context of the resetting Step's execute call.

      • round: JoinRound

        Previous round that the Step waited on.

      Returns readonly T[]

      The consumed late results in the order Dex committed them.

      When round has an invalid key or the Join is not registered by this Flow.

      When round.branchCount is not a non-negative safe integer.

    • Reports this branch's result without ending the branch.

      Use it when the branch continues after it reports, or from an RPC that reports a substitute result for a branch it canceled. The result is staged with the current handler result and committed only when that handler succeeds. Each branch must report exactly once per round. See Join for an example.

      Parameters

      • context: Context

        Context of the current Step or RPC call.

      • round: JoinRound

        Round that the fork passed to this branch.

      • result: T

        Result of this branch.

      Returns void

      When round has an invalid key or the Join is not registered by this Flow.

      When round.branchCount is not a non-negative safe integer.

    Settings

    Member Visibility

    On This Page

    Constructors
    Properties
    Methods