跳到主要内容

Step options

本页介绍用于深度定制你的 Step type 的 StepOptions 进阶功能。基础功能见 Basic Step 页面。

Retry After​

如果要深度定制 retry behavior，使用 RetryAfter。它覆盖 RetryPolicy 的 retry schedule。

from dex._grpc_errors import retry_after

if context.attempt < ready_after_attempt:
    raise retry_after(7, RuntimeError(f"not ready on attempt {context.attempt}"))

例子: examples/python/dex_examples/primitives/custom_retry/custom_retry_flow.py

Durability​

SYNC durability 会将 Step completion 同步记录到 Dex Server，并持久化到 storage。

ASYNC durability 会异步完成这件事。每隔几秒，Dex Server 会 flush 累积的 Step completion 并持久化。它减少了 backend database 和 server 的大量开销，因此吞吐量高得多。

SYNC 是默认值。Dex 按这个顺序确定 durability：StepOptions 中的 method override、Flow 的 FlowConfig，最后是 SYNC。

SYNC 只使用 regular Temporal activity。默认 method attempt timeout 是两小时，默认 heartbeat timeout 是一分钟。

ASYNC 先作为 Temporal local activity 执行，最多七秒和三次 attempt；如果仍然失败，再 fallback 到 regular activity。用户设置的更短 retry duration 或更少 attempt 仍然生效。Local activity 阶段忽略 method timeout 和 heartbeat 设置。Fallback 到 regular activity 后，Dex 使用剩余 retry budget，并采用同样的两小时 method attempt 和一分钟 heartbeat 默认值。

两种 durability mode 的默认 retry total duration 都是四小时。显式设置的 retry duration 可以更短或更长。

def get_step_options(self) -> StepOptions:
    return StepOptions(execute_durability=StepDurability.SYNC)

def get_step_options(self) -> StepOptions:
    return StepOptions(execute_durability=StepDurability.ASYNC)

例子: examples/python/dex_examples/primitives/durability/durability_flow.py

Heartbeat​

在 regular activity attempt 中，Worker 必须在 HeartbeatTimeout 到期前发送 heartbeat 或 Stream message。换句话说，发送 Stream message 就是 implicit heartbeat。

HeartbeatTimeout 默认是一分钟。默认允许的最小值是 10 秒。

例如，Execute method timeout 为 60 秒、HeartbeatTimeout 为 10 秒时，Dex 大约 10 秒后就能 retry 这个 Step，而不必等满 60 秒。

Heartbeat 可以持久化 checkpoint value。下一次 regular attempt 会收到最新值，并从那里继续。显式发送不带 value 的 heartbeat 会清除之前的 checkpoint。Stream message 产生的 implicit heartbeat 会保留最新的显式状态，包括已经显式清空的状态。

ASYNC durability 的 local 阶段忽略 heartbeat output，也不会把 checkpoint 传给 fallback regular activity。Local 阶段仍会发送 Stream message。

def get_step_options(self) -> StepOptions:
    return StepOptions(
        execute_method_timeout=timedelta(seconds=60),
        heartbeat_timeout=timedelta(seconds=10),
        execute_retry=RetryPolicy(maximum_attempts=3),
    )

async def execute(
    self, context: AsyncContext, batches: int
) -> StepDecision:
    completed_batches = context.get_last_heartbeat_value(int) or 0
    for batch in range(completed_batches, batches):
        if context.is_cancellation_requested():
            return dead_end()
        await asyncio.sleep(2)
        await context.heartbeat(batch + 1)
    return graceful_complete("processed")

例子: examples/python/dex_examples/primitives/heartbeat/heartbeat_flow.py

Heartbeat 有助于传递 cancellation。Step invocation 只会在完成、失败或记录 heartbeat 时被 cancel。

选择性加载状态​

每次 WaitFor、Execute 和 timeout handler 调用都会收到所有普通 Attribute，以及每个 Channel 和 ChannelMap instance 的 size metadata。除非对应方法显式选择，否则不会加载 AttributeMap value 和 pending Channel message。

StepOptions 为 WaitFor 和 Execute 提供独立的 selection。方法需要枚举当前 instance 时，选择整个 AttributeMap 或 ChannelMap；已经知道 key 时，只选择精确的 map instance。Attribute lock 只控制并发，不会加载 AttributeMap value。

两个方法收到独立的 snapshot。获胜的 Wait condition 消费消息之后，Execute 才读取自己的 snapshot，因此不会再次看到已经消费的消息。同一个逻辑方法调用的 retry 会复用首次 snapshot。

下面的 Step 只为 Execute 加载 Queued 的 pending message，并在同一个成功的 handler response 中删除第一条消息。

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

例子: examples/python/dex_examples/primitives/channel/channel_flow.py

读取未选择的 AttributeMap value 或 pending Channel message 时，SDK 会返回已有的 not-loaded error。已选择但为空的 Channel 会返回空消息列表。不加载消息 body 也仍然可以读取 size 和 map key。

Attribute locks​

并发 Step 和 RPC invocation 可以读取和写入同一个 Attribute。这可能导致 race condition。要避免它，使用 Attribute lock。Dex Server 会确保同一时间只有一个 Step execution 或 RPC handling 能 acquire 这个 lock。

WaitFor 与 Execute 有独立的 lock set。通过返回 StepOptions 设置它们。

更多细节见 Attribute locking。

动态提供 StepOptions​

Step implementation 返回的 StepOptions 默认适用于每一次 Step execution。有时为某一次 Step execution 动态 override 它们会很有用。

让 StepDecision 进入下一个 Steps 时，提供 StepOptions override。

options = StepOptions(
    wait_for_retry=RetryPolicy(maximum_attempts=2),
    wait_for_failure=WaitForFailurePolicy.PROCEED,
)
return go_to_many(StepMovement.of(OverrideSecondStep, output, options))

例子: examples/python/dex_examples/primitives/options_override/options_override_flow.py