Interruptible execution
Stop long-running work safely when a durable interrupt arrives.
Use this pattern for work that repeats for a long time, such as polling or batch processing. A user can request an interrupt while the work is running. Each branch finishes its current unit of work, sees the durable signal before its next loop iteration, and then completes safely.
InterruptibleFlow starts WorkAStep and WorkBStep in parallel. Each Step checks the durable interruptSignal before scheduling its next unit of work. The interrupt RPC sets that signal, letting either branch finish the Flow gracefully.
Core implementation
Each sample shows WorkAStep. It waits between units of work, completes when the interrupt signal arrives, and otherwise schedules its next loop iteration. WorkBStep follows the same pattern. The runnable examples also include Flow registration and the interrupt RPC.
class WorkAStep(Step[WorkJobParametersInput]):
def __init__(self, interrupt_signal: Attribute[str]) -> None:
self.interrupt_signal = interrupt_signal
def wait_for(self, context: Context, input: WorkJobParametersInput) -> Wait:
return Wait.until(Timer.by_duration(timedelta(seconds=2)))
def execute(
self,
context: Context,
input: WorkJobParametersInput,
) -> StepDecision:
if (self.interrupt_signal.get(context) or "") == INTERRUPT_VALUE:
print("A: Interrupted!")
return graceful_complete()
if input.progress > input.job_upper_bound:
print("WorkAStep completed")
return graceful_complete()
print(
f"[{context.flow_id}][{context.step_execution_id}]: "
f"Doing job {input.progress}"
)
return go_to(
WorkAStep,
WorkJobParametersInput(input.job_upper_bound, input.progress + 1),
)
Example: examples/python/dex_examples/patterns/interruptible/interruptible_execution_flow.py