Polling
View as MarkdownUse Polling when a Flow waits for an external system to become ready or to finish. Examples are a cloud build, a Kubernetes rollout, a provisioning job, or the status of a third-party API.
One Step owns the whole wait. Its Execute calls the external API, sleeps, and calls again until the work reaches a terminal state.
Business signals inside Dex, such as Channel messages and RPCs, are not polling. Wait for them with WaitFor. Business timers, such as a TTL, a reminder, or an inactivity deadline, also use WaitFor; see Durable Timer.
Shape
loop:
status = call the external API with its own short timeout (for example 10s)
transient error -> return the error; the retry policy runs the Step again
terminal status -> return the next decision (success or a business failure)
report progress -> write a Stream message when the status changes,
otherwise record a heartbeat with the checkpoint
sleep(interval)
"Not ready yet" is a normal branch of the loop. It is never an error.
A plain sleep is enough. The loop does not need to watch for cancellation. The external call has its own timeout, and after Dex cancels an attempt, the next call fails fast.
Interval and timeouts
Keep each round inside the heartbeat window:
interval + call timeout <= HeartbeatTimeout - 10s
The example uses a one-minute HeartbeatTimeout and a 10-second call timeout. Its interval can be at most 40 seconds. See Heartbeat for the default and the minimum.
Choose the interval for the wait. Use a few seconds when a user is watching, such as the readiness of something they just started. Use tens of seconds for long jobs such as builds.
Bound the wait
Use one maximum-wait value for both ExecuteMethodTimeout and TotalDuration. Dex enforces them, so the Step does not compute a deadline. Neither the checkpoint nor the Step input carries one. Give ExecuteRetry a few attempts with backoff for transient errors and Worker crashes. See Timeouts and total duration.
TotalDuration only decides whether Dex retries after a failure, so the last attempt can end after it. With Workers available, the total wait is at most about TotalDuration plus one ExecuteMethodTimeout.
The polling Step can use either durability. See Step options.
When the wait expires
When ExecuteMethodTimeout expires, Dex fails the attempt with a timeout. If TotalDuration leaves no time for another retry, the Step fails. Route that failure with the ExecuteFailure option to a short Step that records the business failure. See Execute Failure Recovery.
The failure record does not include the last observed status. The progress Stream and the logs show it. A business condition that is not a fixed duration, such as a lease the Flow does not own, still belongs in the loop.
Checkpoint and progress
A new attempt resumes from the last checkpoint, which GetLastHeartbeatValue returns. In the example, the checkpoint is the last reported job status, so a retry does not repeat a progress message.
Write a Stream message when the observed status changes, and record a heartbeat otherwise. Unchanged rounds should not fill the Stream. Stream messages are best effort, so consumers recover from durable state, such as Attributes, a read RPC, or the Flow result. See Heartbeat and Stream.
Locks
Do not hold ExecuteLockAttributes for the whole wait. Return the result from the polling Step, and commit anything that needs a lock in a short following Step.
Example
PollingFlow waits for a fake external job. StartJob starts the job with the Flow ID as an idempotent job ID. AwaitJob polls the job status every two seconds. It writes a JobProgress Stream message when the status changes and completes with the job result. When the ten-minute maximum wait expires, AwaitJob moves to RecordJobWaitFailure. That Step fails the Flow with the recovery error.
The fake job reports QUEUED, then RUNNING, and succeeds six seconds after it starts. Call the complete-job endpoint to finish it early.
# Keep POLL_INTERVAL + JOB_CALL_TIMEOUT <= heartbeat_timeout - 10s.
# MAX_JOB_WAIT bounds one attempt and all retries; Dex enforces it.
POLL_INTERVAL = timedelta(seconds=2)
JOB_CALL_TIMEOUT = timedelta(seconds=10)
MAX_JOB_WAIT = timedelta(minutes=10)
JOB_PROGRESS = Stream("JobProgress", JobStatus, 1 << 20)
class RecordJobWaitFailure(Step[str]):
def execute(self, context: Context, job_id: str) -> StepDecision:
failure = context.recovery_error
assert failure is not None
return force_fail(
f"waiting for job {job_id} failed: {failure.error_type}: {failure.detail}"
)
class AwaitJob(Step[str]):
def __init__(self, jobs: FakeJobService) -> None:
self.jobs = jobs
def get_step_options(self) -> StepOptions:
return StepOptions(
execute_method_timeout=MAX_JOB_WAIT,
heartbeat_timeout=timedelta(minutes=1),
execute_retry=RetryPolicy(
initial_interval=timedelta(seconds=1),
backoff_coefficient=2.0,
maximum_attempts=3,
total_duration=MAX_JOB_WAIT,
),
).on_execute_failure_proceed_to(RecordJobWaitFailure)
async def execute(self, context: AsyncContext, job_id: str) -> StepDecision:
reported_state = context.get_last_heartbeat_value(JobState)
while True:
status = await self.get_job_status(job_id)
if status.state is JobState.SUCCEEDED:
return graceful_complete(status.result)
if status.state is JobState.FAILED:
return force_fail(f"job {job_id} failed")
if status.state != reported_state:
JOB_PROGRESS.write(context, status)
reported_state = status.state
else:
await context.heartbeat(reported_state)
await asyncio.sleep(POLL_INTERVAL.total_seconds())
async def get_job_status(self, job_id: str) -> JobStatus:
return await asyncio.wait_for(
self.jobs.get_job_status(job_id), JOB_CALL_TIMEOUT.total_seconds()
)
Example: examples/python/dex_examples/patterns/polling/polling_flow.py
Anti-patterns
- A Step that waits for a Timer and moves to itself to poll again.
- Using the retry policy, backoff, or RetryAfter as the polling loop, where "not ready" is returned as an error.
- A long loop that never records a heartbeat or writes progress.
- An external call without its own timeout.
- A hand-written deadline that duplicates ExecuteMethodTimeout or TotalDuration.
- Iterating over pages by passing a page token to a new run.