Interruptible execution
View as MarkdownStop long-running work safely when a durable interrupt arrives.
Use this pattern for work that repeats for a long time, such as batch processing. To wait for an external system, use Polling instead. 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 WorkA and WorkB 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 WorkA. It waits between units of work, completes when the interrupt signal arrives, and otherwise schedules its next loop iteration. WorkB follows the same pattern. The runnable examples also include Flow registration and the interrupt RPC.
class WorkA(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("WorkA completed")
return graceful_complete()
print(
f"[{context.flow_id}][{context.step_execution_id}]: "
f"Doing job {input.progress}"
)
return go_to(
WorkA,
WorkJobParametersInput(input.job_upper_bound, input.progress + 1),
)
Example: examples/python/dex_examples/patterns/interruptible/interruptible_execution_flow.py