跳到主要内容

为什么是 Dex？

Dex 是 Durable Execution(D-Ex) 平台。

Dex 是结构化编程，只有少数几个概念作为 primitives。你用 Dex 写 Flow，里面填普通代码：durable 的 Step、Attribute、RPC，以及用 Channel 和 Timer 表达的 durable condition。然后跑托管 Flow 的 Worker。Client 调 Dex Server 去 start 并和 Flow 实例交互。Dex Server 把 Step / RPC 调用任务派发给你的 Worker。

一个 Step 可以可选地 WaitFor 某个 condition（Channel、Timer，或两者），并且是 durable 的。condition 满足后调用 Execute。没有 WaitFor 时，直接调用 Execute。

Execute 返回 decision，用来决定下一步或完成这条 Flow。在 Step 和 RPC 里，用户代码可以读写 Attribute，也可以 publish durable 的 Channel 消息，用来接外部事件，或在同一条 Flow 的线程之间通信。

订单处理作为一个例子​

还没看过 什么是 Durable Execution？ 的话，下面用那一页的 order-processing 例子。

这条 Flow 会先给买家扣款，再等卖家批准（带提醒），然后发货；发货 API 的 retry 用尽就退款。

Definition graph

OrderProcessingFlow

Valid

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

  • ChargeStep 调支付 API，Step 级 retry。成功后原子进入 ShipStep。
  • ShipStep 等待卖家 Channel 或 Timer。Timer 先到：发提醒，继续等。批准到了：发货。发货 retry 用尽：进 RefundStep。
  • RPC approve 往卖家 Channel 发消息。
  • 店面 start handler 用 Client 的 wait-for-Step-completion API 等 ChargeStep 完成——不是自己轮询状态。
  • Attribute order-status 可检索，客服能搜卡住的订单。
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))

例子: examples/python/dex_examples/products/order-processing

前端要等「已付款」。GET /products/order-processing/start 先启动 Flow，等 ChargeStep 完成后再返回：

        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)

例子: examples/python/dex_examples/products/order-processing

Step 切换是原子的。Timer 和 Channel 等待能熬过 Worker 重启。Retry 和补偿写在 Step 上。

和传统 Durable Execution 差在哪​

传统 Durable Execution 靠 replay 事件历史来跑 workflow 函数。副作用放在 activity 里。你必须把业务逻辑拆成 workflow code 和 activity code。两边是完全不同的 paradigm。workflow 代码必须 deterministic：replay 的函数里不能多走时钟、随机数或 IO。Event history 有 size 上限。跑得久就要 ContinueAsNew。业务数据塞进这段 history，又贵又慢。

Replay 这套 Durable Execution 必须提供大量 API，才能模拟出一套 programming model。这些 API 引入了大量 implicit contracts 和 interfaces。理解和维护都非常复杂。

Dex 不要求你写一段可 replay 的 workflow 函数。你声明 Step 和切换。Worker handler 就是普通代码。等待、RPC、locking 写在 Flow 模型里，不是从 history primitive 自己拼回来。Attribute 数据放在 blob store 里，不进 event history。也可以 sync 到你已有的数据库。

接下来​