跳到主要内容

等待 Attribute 匹配

当调用方需要在 durable 业务状态变化后刷新时，使用 WaitForAttributeMatch。Attribute 及其语义由应用拥有。应用不需要观察 workflow history、run ID 或 Temporal event ID。

常见做法是定义一个初始值为 0 的整数 revision Attribute。所有推进相关状态的 Step 和 RPC 都声明同一个 Attribute lock。在锁内读取 revision、加一、写入业务状态，再写入新的 revision。

调用方等待 revision 大于上次观察到的值。等待会返回当前命中的 revision。随后，调用方调用自己的 Describe 或 read RPC 读取新状态。

Revision 是 watermark，不是事件流。如果状态快速地从 0 变成 1、2、3，一次等待大于 0 的调用可能直接返回 3。调用方不能假设会观察到每个中间值。

Dex 会从完整的 Attribute 条件派生 Request ID。在同一 active Flow 上重复相同条件会共享一个已接受的 durable Update，包括跨 Continue-As-New 的情况。普通等待应让 MaximumWaitTime 保持为 0。只有当动态 predicate、大量并发 consumer、被放弃的调用方或永远不会命中的条件可能耗尽该 Flow 的 in-flight 容量时，才考虑正值。过期会释放 slot，但继续等待会创建新的 -N Update generation。容量和成本的取舍请参阅选择 durable handler 的生命周期。

在锁内推进 revision​

Job Posting 示例使用同一个 Attribute lock 串行化 update RPC，并在写入状态的同一次 invocation 中推进 UpdateVersion。

@rpc(lock_attributes=(update_posting_lock.lock(),))
def update(self, context: Context, input: JobInfo) -> RPCResult[int]:
    version = self.update_version.get(context) + 1
    self.title.set(context, input.title or "")
    self.job_description.set(context, input.description or "")
    self.last_update_time_millis.set(context, int(time.time() * 1000))
    if input.notes is not None:
        self.notes.set(context, input.notes)
    self.update_version.set(context, version)
    update = PostingUpdate(
        version,
        f"{context.flow_id}:{version}",
        input,
    )
    self.linkedin_posting_updates.publish(context, update)
    self.indeed_posting_updates.publish(context, update)
    return RPCResult(version)

例子: examples/python/dex_examples/products/job-post/job_post_flow.py

等待后刷新​

应用只传入 Flow ID。等待返回大于调用方 watermark 的 revision 后，再调用 read RPC。

revision = await app_state.client.wait_for_attribute_match(
    flow_id,
    app_state.job_post.update_version,
    AttributeMatch.greater_than(required_int_query("lastRevision")),
    WaitForAttributeOptions(),
)
job_info = await app_state.client.invoke_rpc(app_state.job_post.get, flow_id)

例子: examples/python/dex_examples/products/job-post/controller.py

String 和 Boolean Attribute 支持 equal 与 not-equal。整数和浮点 Attribute 支持全部六种比较。缺失的 Attribute 永远不命中。object、bytes、null、blob-backed、非有限浮点、跨类型比较和无效的 ordering 比较会被拒绝。

该等待只支持 Temporal，并且只适用于 active Flow。它会自动跟随 Continue-As-New。Cadence 返回 Unimplemented。

相关内容​