Channel

Channel is a durable first-in, first-out (FIFO) queue inside one Flow execution. Steps and RPCs append typed messages. A Step consumes messages by returning a Channel condition from WaitFor.

A Channel keeps messages in arrival order. A message can be consumed by one matching wait, once. It is not copied to other waiting Steps.

A Channel is scoped to one Flow execution. Messages never cross between Flow executions.

Flow execution

STEPExampleStepA

RPCExampleRpc

publish messages

publish messages

CHANNELrequests_queueFIFO queue · each message is consumed once

consume

STEPExampleStepB

WaitForrequests_queue.for_one()

Executeprocess request

Persistence schema

A Channel has a stable name and a message type. Define it and add it to the Flow persistence schema.

This example has an ApprovalMessages Channel. Its start Step waits for either one approval or a Timer. The publishApprovalMessage RPC publishes the approval message. The enqueueChannelMessage RPC lets applications publish another queued message.

@dataclass(frozen=True)
class QueuedMessageReference:
    message_id: str


class ChannelWaitStep(Step[int]):
    def __init__(
        self,
        approval_messages: Channel[str],
        queued_messages: Channel[str],
    ) -> None:
        self.approval_messages = approval_messages
        self.queued_messages = queued_messages

    def get_step_options(self) -> StepOptions:
        return StepOptions(execute_load_channels=(self.queued_messages,))

    def wait_for(self, context: Context, input: int) -> Wait:
        return Wait.any_of(
            self.approval_messages.for_one(),
            Timer.by_duration(timedelta(seconds=input)),
        )

    def execute(self, context: Context, input: int) -> StepDecision:
        pending_queued_messages = self.queued_messages.pending_messages(context)
        if pending_queued_messages:
            self.queued_messages.delete(context, pending_queued_messages[0].message_id)
            return graceful_complete(pending_queued_messages[0].value)
        if context.has_timer_fired():
            return graceful_complete("approval timed out")
        approval_message_values = self.approval_messages.results(context)
        return graceful_complete(approval_message_values[0])


class ChannelFlow(Flow[int]):
    approval_messages = Channel("ApprovalMessages", str)
    queued_messages = Channel("QueuedMessages", str)
    prioritized_messages = Channel("PrioritizedMessages", str)

    def __init__(self) -> None:
        self.wait_for_approval = ChannelWaitStep(self.approval_messages, self.queued_messages)

    def get_steps(self) -> StepList[int]:
        return StepList.start_step(self.wait_for_approval)

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(self.approval_messages, self.queued_messages, self.prioritized_messages)

    @rpc
    def publish_approval_message(self, context: Context) -> None:
        self.approval_messages.publish(context, "approved")

    @rpc(is_transactional=True, load_channels=(queued_messages,))
    def move_queued_message_to_prioritized_messages(
        self, context: Context, queued_message: QueuedMessageReference
    ) -> None:
        message_to_prioritize = self.queued_messages.find_pending_message(
            context, queued_message.message_id
        )
        self.queued_messages.delete(context, queued_message.message_id)
        if message_to_prioritize is not None:
            self.prioritized_messages.publish(context, message_to_prioritize.value)

Example: examples/python/dex_examples/primitives/channel/channel_flow.py

The Timer branch returns before reading Channel results. If the approval wins, the Channel condition has consumed one message, so the first result is present.

Wait and consume

Returning a Channel condition from WaitFor waits for the Channel. When the condition is met, it consumes messages from the Channel. Dex Server keeps messages until a waiting Step wins them, then removes the selected messages from the queue in FIFO order and passes them to Execute as Channel results.

Condition	When it proceeds	What it consumes
ForOne	At least one message is queued.	Exactly one message.
ForN	At least the requested number is queued.	Exactly that number of messages.
AtLeast	At least the requested number is queued.	Every message currently available.
AtMost	Immediately, including when the queue is empty.	Up to the requested number of currently queued messages.
AtLeastAtMost	At least the lower bound is queued.	From the lower bound up to the upper bound of currently queued messages.

AtMost does not wait on its own for messages to accumulate. When the surrounding Wait completes, it consumes up to the limit from the queue at that time. To wait for at least one message and consume up to a limit, use AtLeastAtMost with a lower bound of 1.

Publish messages

Inside a Step or RPC, call the Channel publish method with the handler Context. Dex appends the message when it accepts that handler result.

Code outside a Worker invokes a Flow RPC that publishes with the handler Context. This makes the Flow the boundary for validation, authorization, locking, and related state changes.

Messages can arrive before a Step starts waiting. Dex Server keeps them in the Channel until a later wait consumes them.

Manage pending messages

Dex assigns a UUIDv7 message ID when it accepts each publication. An RPC can list every pending message in FIFO order and delete one by ID. Listing does not consume messages. Only a message that is still pending can be deleted; a concurrent Step consumption or deletion returns the Channel-message-not-found error.

This is queue state, not Flow history. After a Step consumes a message, it disappears from the Channel even though the publication and consumption remain observable in Flow history until retention removes them.

Every Step method, timeout handler, and RPC always receives Channel size metadata. It must explicitly load pending messages before reading their IDs or Values. Deleting a message whose ID is already known does not require loading it. StepOptions selects state separately for WaitFor and Execute; FlowTimeoutHandlerOptions selects state for the timeout handler. Loading a Channel creates one invocation snapshot and does not consume messages. A loaded empty Channel returns an empty list; reading pending messages without loading the Channel is a usage error.

Treat every pending-message snapshot read inside a Step or RPC as potentially stale. Other Steps and RPCs can consume, delete, or publish messages while the handler is running. Transactional execution validates selected deletions and commits the handler's writes atomically, but it does not lock the whole Channel snapshot. Read and write pending messages directly only when the operation explicitly tolerates that concurrency. If a decision requires the queue to remain unchanged, every cooperating Step and RPC writer must use the same Attribute lock.

A Step or timeout handler can stage deletion with its other side effects. Dex applies successful handler effects in this order: Attribute writes, best-effort Channel deletions, then Channel publications, followed by the Step decision. If another operation already consumed or deleted a selected message, deletion is a no-op and the remaining effects still apply. StepDecision describes only control flow; it does not contain deletions.

An RPC can combine deletion with other durable writes. The move example loads QueuedMessages, finds the original Value by message ID, then stages its deletion and republishes that Value. The caller sends only the ID. Enable transactional execution when a missing message must abort every write; see Transactional reads and writes.

@rpc(is_transactional=True, load_channels=(queued_messages,))
def move_queued_message_to_prioritized_messages(
    self, context: Context, queued_message: QueuedMessageReference
) -> None:
    message_to_prioritize = self.queued_messages.find_pending_message(
        context, queued_message.message_id
    )
    self.queued_messages.delete(context, queued_message.message_id)
    if message_to_prioritize is not None:
        self.prioritized_messages.publish(context, message_to_prioritize.value)

Example: examples/python/dex_examples/primitives/channel/channel_flow.py

ChannelMap

Use ChannelMap when one Flow has an unbounded set of independent queues, such as one Channel per order, customer, or job. A ChannelMap has one declared name and one value type. Each instance key has its own FIFO queue.

RPCs can always read the current ChannelMap keys and sizes. Load the entire ChannelMap when the handler needs every instance's pending messages. Load exact instances when it needs only known keys. Instance keys cannot contain / because Dex uses that character between the map name and instance key.