Worker¶

Lifecycle and shutdown¶

Worker.run() and Worker.run_until() register through the worker protocol before polling. After registration succeeds, shutdown stops every poller and waits for accepted work within the shared shutdown_timeout deadline before removing that worker-plane registration. Work still running at the deadline is cancelled. Calling stop() more than once is safe and sends at most one deregistration request. No deregistration is sent if registration failed.

An external stop() also interrupts run_until(); the pending run_until() call ends with asyncio.CancelledError after its accepted workflow or activity work has drained or been cancelled. If an async poller, the query-poller thread, or accepted work cannot be stopped, stop() raises RuntimeError and leaves the registration active. Treat that as an incomplete shutdown requiring operator attention rather than continuing as though the worker stopped cleanly.

For a long-running process, coordinate shutdown with the same worker instance and await both stop() and the run task:

run_task = asyncio.create_task(worker.run())

try:
    await shutdown_requested.wait()
finally:
    try:
        await worker.stop()
    finally:
        await run_task

Deregistration authentication, protocol, and HTTP failures are raised from stop() and run(); a process must not treat those failures as a clean shutdown. When a worker-loop error and deregistration both fail, the loop error remains the raised exception and the deregistration error is available through its __cause__.

The worker-plane cleanup operation is separate from Client.deregister_worker(). That existing control-plane method remains an operator management action for retiring or recovering worker records and is not used by normal Worker shutdown.

worker ¶

Long-polling worker that runs workflow and activity tasks.

Worker registers itself with the server for a given task queue, then spawns poll loops for both workflow tasks and activity tasks. Each received task is dispatched to the registered workflow class or activity function, results are serialized, and success/failure commands are sent back to the server. Workers stop their pollers, drain in-flight tasks on shutdown up to a configurable shutdown_timeout, and remove successful worker registrations.

Most applications create one Worker per task queue and pass it the same Client used for control-plane calls, plus lists of workflow classes and activity callables registered via durable_workflow.workflow.defn and durable_workflow.activity.defn.

Worker ¶

Worker(client, *, task_queue, workflows=(), activities=(), worker_id=None, build_id=None, poll_timeout=35.0, max_concurrent_workflow_tasks=10, max_concurrent_activity_tasks=10, shutdown_timeout=30.0, heartbeat_interval=60.0, metrics=None, interceptors=(), external_storage=None, external_storage_threshold_bytes=None, external_storage_cache=None)

Polls workflow and activity tasks and dispatches them to Python callables.

run `async` ¶

run()

Register the worker and poll until stop() is called or the task is cancelled.

run_until `async` ¶

run_until(*, workflow_id, timeout=60.0, poll_interval=0.5)

Run this worker until a workflow reaches a terminal state.

This is intended for examples, smoke tests, and single-workflow scripts. Long-running workers should call run and coordinate shutdown from their process supervisor.

stop `async` ¶

stop()

Stop polling, drain tasks, and remove the worker registration.

Repeated calls share one shutdown result. A failed deregistration is therefore propagated to every caller without sending another DELETE.