Skip to main content

Wait for Attribute Match

Use WaitForAttributeMatch when a caller needs to refresh after durable business state changes. The application owns the Attribute and its meaning. The wait observes only that Attribute's value, not run IDs or Flow history.

A common pattern is an integer revision Attribute initialized to zero. Every Step and RPC that advances the represented state declares the same Attribute lock. Inside that lock, it reads the revision, adds one, writes the business state, and then writes the new revision.

The caller waits for revision greater than its last observed value. The wait returns the current matched revision. The caller then fetches the new state with its Describe or read RPC, or reads the Attributes directly with GetAttributes.

Revision is a watermark, not an event stream. If state advances quickly from 0 to 1 to 2 to 3, one wait for a value greater than 0 may return 3. Callers must not expect to observe every intermediate value.

Leave RequestTimeout at 0 for an ordinary durable wait. Set it explicitly at an admission boundary. Each Flow allows a limited number of concurrent waits. See Choose the timeout boundary for details.

Advance revision under a lock​

The Job Posting example serializes its update RPC with one Attribute lock and advances UpdateVersion in the same invocation that writes the state.

@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)

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

Wait, then refresh​

The application passes only the Flow ID. After the wait returns a revision greater than the caller's watermark, it invokes the 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)

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

String and Boolean Attributes support equal and not-equal matches. Integer and floating-point Attributes support all six comparison operators. Missing Attributes never match. Object, bytes, null, blob-backed, non-finite floating-point, cross-type, and invalid ordering comparisons are rejected.

The wait works only while the Flow is active.