Skip to main content

Why Dex?

Dex is a Durable Execution(D-Ex) platform.

Dex is structural programming with only a few concepts as primitives. You use Dex to write a Flow filled with ordinary code: durable Steps, Attributes, RPCs, and durable conditions using Channels and Timers. Then you run Workers hosting your Flow. The Client calls Dex Server to start and interact with Flow instances. Dex Server dispatches Step and RPC invocation tasks to your Workers.

A Step can optionally WaitFor some condition (Channel, Timer, or both) durably. When the condition is met, Execute is invoked. If there is no WaitFor, Execute is invoked directly.

Execute returns a decision for the next Steps or for completing the Flow. Within the Steps and RPCs, user code can read or write Attributes, and/or publish durable Channel messages as communication from external events or within threads of a Flow.

Order processing as an example​

If you have not read What is Durable Execution?, the example below is the order-processing flow from that page.

The flow will charge the buyer, then wait for seller approval (with reminders), then ship the item, and refund if ship API call retries exhaust.

Definition graph

OrderProcessingFlow

Valid

python · examples/python/dex_examples/products/order-processing/order_processing_flow.py

  • ChargeStep calls the payment API with Step-level retry. On success, move to ShipStep atomically.
  • ShipStep waits for a seller Channel or a Timer. Timer first: send a reminder and keep waiting. Approval: ship. Exhausted ship retries: RefundStep.
  • RPC approve publishes the seller Channel.
  • The storefront start handler waits for ChargeStep with the Client wait-for-Step-completion API — not a homemade status poll.
  • Attribute order-status is indexed so support can search stuck orders.
order_status = Attribute(
    "order-status",
    str,
    AttributeIndex(IndexType.KEYWORD),
)
seller_ok = Channel[str]("seller-ok", str)


class ChargeStep(Step[OrderRequest]):
    def __init__(self, service: MyDependencyService, ship: ShipStep) -> None:
        self.service = service
        self.ship = ship

    def get_step_options(self) -> StepOptions:
        return StepOptions(
            execute_retry=RetryPolicy(
                total_duration=timedelta(hours=1),
            )
        )

    def execute(self, context: Context, input: OrderRequest) -> StepDecision:
        self.service.charge_user(input.email, input.customer_id, input.amount)
        order_status.set(context, "charged")
        return go_to(ShipStep, input)


class ShipStep(Step[OrderRequest]):
    def __init__(self, service: MyDependencyService, refund: RefundStep) -> None:
        self.service = service
        self.refund = refund

    def get_step_options(self) -> StepOptions:
        return StepOptions(
            execute_retry=RetryPolicy(
                total_duration=timedelta(hours=1),
            )
        ).on_execute_failure_proceed_to(
            RefundStep,
            StepOptions(
                execute_retry=RetryPolicy(
                    total_duration=timedelta(hours=1),
                )
            ),
        )

    def wait_for(self, context: Context, input: OrderRequest) -> Wait:
        return Wait.any_of(
            seller_ok.for_one(),
            Timer.by_duration(timedelta(hours=24)),
        )

    def execute(self, context: Context, input: OrderRequest) -> StepDecision:
        if context.has_timer_fired():
            self.service.send_email(
                input.email,
                "Reminder: approve shipment",
                "Please approve or provide a tracking number.",
            )
            return go_to(ShipStep, input)
        self.service.ship_item(input.order_id)
        order_status.set(context, "shipped")
        return graceful_complete(f"shipped:{input.order_id}")


class RefundStep(Step[OrderRequest]):
    def __init__(self, service: MyDependencyService) -> None:
        self.service = service

    def execute(self, context: Context, input: OrderRequest) -> StepDecision:
        self.service.update_external_system(f"refund {input.order_id}")
        order_status.set(context, "refunded")
        return graceful_complete(f"refunded:{input.order_id}")


class OrderProcessingFlow(Flow[OrderRequest]):
    def __init__(self, service: MyDependencyService) -> None:
        self.service = service
        self.refund = RefundStep(service)
        self.ship = ShipStep(service, self.refund)
        self.charge = ChargeStep(service, self.ship)

    def get_steps(self) -> StepList[OrderRequest]:
        return StepList.start_step(self.charge).other_steps(self.ship, self.refund)

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(order_status, seller_ok)

    @rpc
    def approve(self, context: Context, _note: str) -> RPCResult[str]:
        seller_ok.publish(context, "approved")
        return RPCResult("ok")

    @rpc
    def describe(self, context: Context) -> RPCResult[str]:
        return RPCResult(order_status.get(context))

Example: examples/python/dex_examples/products/order-processing

Frontend wait for “payment received”. GET /products/order-processing/start starts the Flow, then waits for ChargeStep before it returns:

        run_id = await app_state.client.start_flow(
            app_state.order_processing,
            flow_id,
            request,
            start_options(),
        )
        await app_state.client.wait_for_step_completion(
            flow_id,
            CHARGE_STEP,
            CHARGE_WAIT_TIMEOUT,
        )
        return started_flow(flow_id, run_id)

Example: examples/python/dex_examples/products/order-processing

Step transitions are atomic. Timers and Channel waits survive Worker restarts. Retries and compensation are declared on the Step.

How this differs from traditional Durable Execution​

Traditional Durable Execution runs workflow functions by replaying event history. Side effects go in activities. You must split business logic into workflow code and activity code. Those two are completely different paradigms. The workflow code must stay deterministic: no extra clocks, random, or IO in the replayed function. Event history has a size constraint. Long-running executions must ContinueAsNew. That history is also a poor store for application data: large payloads are expensive and slow.

Replay-based Durable Execution must expose a large set of APIs to simulate a programming model. Those APIs introduce many implicit contracts and interfaces. They are hard to understand and maintain.

Dex does not ask you to write a replayable workflow function. You declare Steps and movements. Worker handlers are ordinary code. Waits, RPC, and locking are part of the Flow model, not something you reconstruct from history primitives. Attribute data lives in a blob store, not in event history. You can sync it to a database you already run.

Next steps​