Worker Lifecycle and Resilience#

Prefect workers run as long-lived processes that poll a work pool for scheduled flow runs. The lifecycle has three distinct phases: CLI pre-flight validation, setup/backend sync, and the steady-state service loop. All three phases are implemented across src/prefect/cli/worker.py, src/prefect/cli/_worker_utils.py, and src/prefect/workers/base.py.

Phase 1: CLI Pre-flight Validation#

Before a worker object is ever instantiated, prefect worker start 1 performs several gate-keeping checks:

1. Pause state checks — _check_work_pool_paused and _check_work_queues_paused query the API and emit a warning if the pool or queues are paused. The worker still starts, but the user is warned it won't execute runs until unpaused 2.

2. Worker type resolution — If --type is not supplied, _retrieve_worker_type_from_pool reads the work pool from the API and discovers the type. If the pool doesn't exist, it defaults to "process". Push/managed pools are rejected here with a hard exit 3.

3. Package install — _load_worker_class attempts to resolve the worker type via the plugin registry. If that fails, _find_package_for_worker_type queries the Prefect collections metadata API to find the right integration package. Depending on --install-policy (always, if-not-present, prompt, never), the package is installed via _install_package before proceeding 4.

4. Signal handlers — setup_signal_handlers_worker is registered so SIGINT/SIGTERM produce clean shutdowns rather than tracebacks 5.

Phase 2: Setup and Backend Sync#

BaseWorker.start() calls async with self as worker, which invokes setup():

• Validates that PREFECT_API_URL is set 6.
• Creates the PrefectClient and capacity limiter, and seeds os.environ["PREFECT__WORKER_NAME"] for attribution headers 7.
• Calls sync_with_backend() immediately to register the worker and obtain its backend_id before the main loop starts.
• Optionally starts a FlowRunCancellingObserver when worker.enable_cancellation is true.

Worker Channel (sync_with_backend)#

sync_with_backend delegates to a WorkPoolWorkerChannel, introduced in PR #21885. The channel uses a WebSocket-first, REST-fallback strategy:

• Primary: a persistent WebSocket connection delivers work-pool snapshots, heartbeats, and configuration updates in real time.
• Fallback: when the WebSocket endpoint is unavailable, the channel falls back to REST heartbeat calls.

PR #21919 hardened the REST fallback against a race condition where two workers start simultaneously: if create_work_pool raises ObjectAlreadyExists, the worker now re-reads the pool created by the other worker instead of crashing.

_record_worker_id captures the backend-assigned UUID into self.backend_id and sets os.environ["PREFECT__WORKER_ID"] 8. _record_work_pool_snapshot refreshes the local copy of the work pool, warning if the remote type doesn't match the local worker type 9.

Phase 3: The Service Loop#

BaseWorker.start() spins up two concurrent critical_service_loop tasks inside an anyio task group:

Both loops use backoff=4 (up to ~4 doublings, ~1 minute max interval) and jitter_range=0.3 10.

critical_service_loop Resilience#

critical_service_loop wraps any workload with structured retry/backoff logic:

• Transient errors tolerated: httpx.TransportError (connection/timeout) and HTTP 5xx responses are caught and recorded without crashing 11.
• Consecutive-error tracking: a fixed-size deque of consecutive (default 3) booleans tracks the most recent outcomes. If all entries are False, backoff is triggered .
• Exponential backoff: on each backoff event, the sleep interval becomes interval × 2^backoff_count. Workers call these loops with backoff=4, so they survive up to 4 consecutive-failure cycles before raising RuntimeError("Service exceeded error threshold.") 13.
• Reset on success: any successful workload invocation resets backoff_count to 0 14.
• Non-transient errors: anything other than TransportError or 5xx immediately re-raises, exiting the loop 15.

Healthcheck#

When started with --with-healthcheck, a uvicorn ASGI server runs on a daemon thread. is_worker_still_polling() returns False if _last_polled_time is more than query_interval_seconds × 30 seconds old, allowing an external orchestrator (e.g., Kubernetes liveness probe) to restart a stuck worker.

Teardown#

teardown() cancels all scheduled task scopes, stops the worker channel, emits a prefect.worker.stopped event, and closes the client. Environment variables PREFECT__WORKER_NAME and PREFECT__WORKER_ID are cleaned up only if they belong to this worker instance (guarding against multi-worker-per-process scenarios).

Key Settings#

Related Files#

• src/prefect/workers/base.py — BaseWorker implementation
• src/prefect/cli/worker.py — prefect worker start command
• src/prefect/cli/_worker_utils.py — pre-flight helpers
• src/prefect/utilities/services.py — critical_service_loop
• PR #21885 — worker channel / WebSocket sync
• PR #21919 — concurrent work-pool creation fix