Great question. Short version: by default, a Prefect 3.x runner will cancel in‑flight runs when it’s sent a termination signal. If you need “finish what’s running, then exit” draining, you’ll want to add a small bit of orchestration around the runner and coordinate with your Azure Container Apps scale‑in behavior. A few key points about Prefect 3 runners (ProcessRunner): - There is no

prefect runner

CLI; you typically run runners via

prefect flow serve ...

or programmatically with

flow.serve()

/

Runner()

. - On SIGTERM, the runner stops taking new work and cancels active runs. Child processes are given a short grace before being killed. There isn’t a built‑in “drain until all runs finish” mode. - The Runner can expose a tiny HTTP server with useful endpoints (health and run_count). You can enable it with

PREFECT_RUNNER_SERVER_ENABLE=true

or

Runner(webserver=True)

. Given that, you have two practical patterns depending on your tolerance for canceling vs draining: Option A: Allow cancellation on scale-in (simplest) - Accept that in‑flight runs will be cancelled during scale‑in. - Ensure your container’s termination grace window is long enough for graceful cancellation to complete. - In Azure Container Apps, set a termination grace period that’s comfortably above 30 seconds to give the runner time to stop cleanly. - Keep your runner concurrency (

limit

/

PREFECT_RUNNER_PROCESS_LIMIT

) at a level you’re comfortable cancelling. Option B: True draining (finish current work, then exit) - Wrap the runner with a tiny entrypoint that: 1) Enables the runner’s webserver so you can see active in‑flight runs (run count). 2) On SIGTERM, stops the runner from accepting new work and waits until

run_count

drops to 0, then lets the process exit. - Sample entrypoint you can adapt (the runner webserver listens on localhost:8080 by default):

#!/usr/bin/env bash
set -euo pipefail

# 1) Enable the runner webserver so we can see in-flight runs
export PREFECT_RUNNER_SERVER_ENABLE=true
export PREFECT_RUNNER_SERVER_PORT=8080

# Start your runner — using flow serve as an example
prefect flow serve ./path/to/flows.py:my_flow -n my-deployment &
RUNNER_PID=$!

drain_and_exit() {
  echo "Scale-in signal received: initiating drain"

  # Optional: stop taking new work. If you’re starting via Python, prefer programmatic control.
  # If using the runner webserver, you can choose to pause new intake in your own code path.
  # Otherwise, just wait for run_count to drop.

  # 2) Wait for in-flight runs to finish (bounded wait)
  end=$((SECONDS+900))  # wait up to 15 minutes; adjust
  while [ $SECONDS -lt $end ]; do
    rc=$(curl -fsS <http://127.0.0.1:8080/run_count> || echo "0")
    if [ "$rc" = "0" ]; then
      break
    fi
    sleep 2
  done

  echo "Forwarding SIGTERM to runner"
  kill -TERM "$RUNNER_PID" || true
  wait "$RUNNER_PID" || true
}

trap drain_and_exit TERM INT
wait "$RUNNER_PID"

- In Azure Container Apps: - Set a termination grace period that’s long enough for your worst‑case drain time. - Prefer scale rules that don’t pick instances with active work during scale‑in. If you can’t express “in‑flight runs” as a metric, use conservative scale‑down stabilization windows so instances don’t get terminated immediately after picking up work. - If you can influence scale-in selection, try to bias toward idle instances (e.g., via custom metrics). The runner’s

run_count

endpoint can serve as a signal if you can surface it to the scaler. Notes and knobs you can use - Enable the runner webserver: - Env vars:

PREFECT_RUNNER_SERVER_ENABLE=true

,

PREFECT_RUNNER_SERVER_HOST

,

PREFECT_RUNNER_SERVER_PORT

- Programmatic:

Runner(webserver=True)

- Limit concurrency to reduce “drain time”:

limit

in

flow.serve(...)

or

Runner(limit=...)

, or

PREFECT_RUNNER_PROCESS_LIMIT

. -

prefect flow serve --help

is available if you’re invoking via CLI to confirm flags. A couple of clarifying questions to tailor this:

- How are you starting the runner today:

prefect flow serve ...

in the container entrypoint, or a custom Python script using

Runner()

/

flow.serve()

? - Is it acceptable if in‑flight runs are cancelled during scale‑in, or do they need to finish? - What scale rules are you using in Azure Container Apps (CPU/memory, queue length, custom)? If you can consume a custom/HTTP metric, we can surface the runner’s active run count to the scaler.

Thanks for clarifying — since you’re using deployments with

prefect worker start

(Process worker), you’re on Prefect 2.x. In 2.x, workers handle shutdown gracefully: - When the container gets SIGTERM, the worker stops polling for new flow runs and waits for in‑flight runs to finish. - For Process work pools, each running flow is a subprocess; the worker waits for those subprocesses to exit cleanly before the worker process exits. What you should do for autoscaling in Azure Container Apps (ACA): 1) Give enough termination time - Ensure the container has a generous termination grace period so in‑flight runs can finish. Pick something ≥ your worst‑case flow duration. - If ACA scaling config has a cooldown or scale‑down stabilization window, increase it so instances don’t get terminated immediately after they pick up work. 2) Cap concurrency to bound drain time - Use a limit so you know the maximum amount of work that may be draining on a single instance.

prefect worker start --pool <your-pool> --limit <max-concurrent-flows>

3) Bias scale‑in away from busy instances - If you scale on a queue/trigger (e.g., KEDA), tune rules so scale‑in only happens when there’s been sustained low demand. - If possible, prefer selecting idle instances for termination. If that’s not available, rely on the termination grace period + concurrency limit above. 4) Plan for the “not enough time” case - If a container is still terminated before runs finish, those flow runs will be interrupted. Use flow‑level retries where appropriate so they’re rescheduled. If you later move to Prefect 3.x (runners), behavior changes: runners cancel in‑flight runs on shutdown by default, so you’d need a custom drain wrapper. But on Prefect 2.x workers, you already get the “finish current runs, then exit” behavior — just make sure ACA gives the container enough time and you cap concurrency. Quick questions to tailor this: - About how long do your longest flows run? - What ACA scale trigger are you using (CPU, queue length, custom) and what are your current cooldown/stabilization settings?

Thanks for the clarification. Important note: in Prefect 3.x there are no workers. The

prefect worker start

command is from Prefect 2.x. If you’re truly on 3.x, you’ll want to run a Runner via

prefect flow serve ...

(or programmatically with

flow.serve()

/

Runner(...)

). Could you share the output of

prefect version

? That will confirm which behavior you get on shutdown. Assuming Prefect 3.x and Azure Container Apps: - Default behavior: when the container receives SIGTERM, the Runner stops accepting new work and cancels all active flow runs. It gives each child process a short grace before forcing termination. There is no built‑in “drain until current runs finish” mode. - To ensure “done before scale-in” you have two practical patterns: Pattern 1: Only terminate idle runners (recommended) - Enable the runner’s tiny webserver to expose

/run_count

(active flow runs). - Configure your scaler/scale-in logic to only pick instances with

/run_count == 0

. - Set a termination grace period long enough for clean shutdown. How to enable the webserver and serve your deployment:

# In your container env:
PREFECT_RUNNER_SERVER_ENABLE=true
PREFECT_RUNNER_SERVER_HOST=0.0.0.0
PREFECT_RUNNER_SERVER_PORT=8080

# Start the runner serving your flow
prefect flow serve ./path/to/file.py:my_flow -n my-deployment

- Your scaler (or a pre-termination step) can poll: - GET http//<container ip>8080/run_count → only scale in when it returns 0. Pattern 2: Add a drain wrapper (stop intake, let current runs finish, then exit) - Trap SIGTERM yourself, stop intake, wait for active runs to finish, then exit. Here’s a simple entrypoint script that waits for

/run_count

to reach 0 before forwarding SIGTERM to the runner:

#!/usr/bin/env bash
set -euo pipefail

export PREFECT_RUNNER_SERVER_ENABLE=true
export PREFECT_RUNNER_SERVER_HOST=0.0.0.0
export PREFECT_RUNNER_SERVER_PORT=8080

prefect flow serve ./path/to/file.py:my_flow -n my-deployment &
RUNNER_PID=$!

drain_and_exit() {
  echo "Scale-in signal received: draining..."
  # Wait up to 15 minutes for active runs to complete; adjust for your flows
  end=$((SECONDS+900))
  while [ $SECONDS -lt $end ]; do
    rc=$(curl -fsS <http://127.0.0.1:8080/run_count> || echo "0")
    if [ "$rc" = "0" ]; then
      break
    fi
    sleep 2
  done
  echo "Forwarding SIGTERM to runner"
  kill -TERM "$RUNNER_PID" || true
  wait "$RUNNER_PID" || true
}

trap drain_and_exit TERM INT
wait "$RUNNER_PID"

Additional recommendations for ACA - Termination grace time: give the container enough time to drain (e.g., worst‑case flow duration or your chosen drain timeout). - Concurrency: cap per-runner concurrency so your drain time is bounded (use

limit

in

flow.serve(...)

or the corresponding environment if you set it programmatically). - Retries: if a run does get cancelled due to scale-in, configure retries on the flow/run steps so it will be rescheduled. Open questions so I can tailor this precisely: - Can you share

prefect version

and how you’re starting the process today (CLI

prefect flow serve

vs a Python script)? - Do you need strict draining (finish in‑flight runs) or is cancel-and-retry acceptable? - What ACA scale-in trigger are you using, and can it check an HTTP endpoint on the instance (so we can use

/run_count

)?

Perfect — strict draining is doable with a small wrapper around the Prefect 3.x Runner. Since the Runner cancels in‑flight runs on shutdown by default, the wrapper’s job is to: - Stop intake - Wait until no runs are active - Then terminate the runner Below is a ready-to-use Bash entrypoint you can run in Azure Container Apps. It uses the Runner webserver’s /run_count endpoint to detect when it’s safe to terminate. What this script does - Starts your runner via

prefect flow serve ...

in the background - Enables the runner webserver so we can read /run_count - On SIGTERM, polls /run_count until it reaches 0 (or timeout), then forwards SIGTERM to the runner - Exits after the runner shuts down Drop-in entrypoint for ACA

#!/usr/bin/env bash
set -euo pipefail

# Configure the runner webserver so we can observe active runs
: "${PREFECT_RUNNER_SERVER_ENABLE:=true}"
: "${PREFECT_RUNNER_SERVER_HOST:=0.0.0.0}"
: "${PREFECT_RUNNER_SERVER_PORT:=8080}"

# Optional: bound concurrency so drain time is predictable
# export PREFECT_RUNNER_PROCESS_LIMIT=4

# REQUIRED: update your flow entrypoint and deployment name
ENTRYPOINT="${ENTRYPOINT:-./path/to/file.py:my_flow}"
DEPLOYMENT_NAME="${DEPLOYMENT_NAME:-my-deployment}"

# Start the runner
prefect flow serve "$ENTRYPOINT" -n "$DEPLOYMENT_NAME" &
RUNNER_PID=$!

# Function to read active run count
get_run_count() {
  curl -fsS "<http://127.0.0.1:${PREFECT_RUNNER_SERVER_PORT}/run_count>" || echo "0"
}

drain_and_exit() {
  echo "[drain] Termination signal received: beginning drain"

  # Wait up to DRAIN_TIMEOUT seconds for active runs to finish
  DRAIN_TIMEOUT="${DRAIN_TIMEOUT:-900}"  # default 15 minutes
  end=$((SECONDS + DRAIN_TIMEOUT))

  # Poll until run_count == 0 or timeout
  while [ $SECONDS -lt $end ]; do
    rc="$(get_run_count)"
    # rc should be an integer; if curl failed, we default to "0" above
    if [ "$rc" = "0" ]; then
      echo "[drain] No active runs; safe to stop"
      break
    fi
    echo "[drain] Active runs: $rc; waiting..."
    sleep 2
  done

  echo "[drain] Forwarding SIGTERM to runner (PID $RUNNER_PID)"
  kill -TERM "$RUNNER_PID" || true

  # Give the runner time to exit cleanly
  wait "$RUNNER_PID" || true
  echo "[drain] Runner exited; shutting down entrypoint"
}

trap drain_and_exit TERM INT

# Block until runner exits or a signal is trapped
wait "$RUNNER_PID"

How to use it in your container - Set these env vars (or bake into the image): - ENTRYPOINT:

./path/to/file.py:my_flow

- DEPLOYMENT_NAME:

my-deployment

- Optional:

PREFECT_RUNNER_PROCESS_LIMIT

to cap concurrency - Optional:

DRAIN_TIMEOUT

to cap how long you’ll wait before forcing termination - Ensure your ACA termination grace period is greater than your

DRAIN_TIMEOUT

so the drain completes before the platform kills the container. - Scale-in selection: if you can, configure ACA to avoid terminating instances that still have active runs. If ACA can’t read a per-instance metric, the drain script still protects in-flight runs; the platform just needs to allow enough time. Notes - This pattern stops intake implicitly by not killing the runner immediately; while draining, the runner may still pick up new work unless you prevent it. To prevent new intake: - Run with fewer or zero eligible schedules during drain windows, or - Use tags/filters to isolate which runner instances pick up which deployments, or - If you control the scale-in signal timing, briefly disable schedules (CLI/API) before signaling with SIGTERM. - If you start multiple deployments in the same runner, /run_count reflects all active runs across that runner. If you’d prefer a Python-based wrapper that flips the runner into a true “no new work” state during drain, I can share a small script that sets

runner.stopping = True

and waits on

runner._flow_run_process_map

before shutting down.

Great—can you share a bit more about what you want this wrapper to cover? For example: just trigger-and-wait for a deployment, or also listing deployments, reading states, variables, etc. In the meantime, here’s a simple, production-friendly Python wrapper for Prefect 3.x that: - Triggers a deployment by name or ID - Polls until the flow run reaches a terminal state - Reads flow runs - Lists deployments It includes both async and sync versions. ``` from future import annotations import asyncio import time from typing import Any, Dict, Optional, Union from uuid import UUID from prefect.client.orchestration import get_client # ========== ASYNC WRAPPER ========== class AsyncPrefect: def __init__(self): self._client_cm = None self.client = None async def __aenter__(self) -> "AsyncPrefect": self._client_cm = get_client() self.client = await self._client_cm.__aenter__() return self async def __aexit__(self, exc_type, exc, tb): await self._client_cm.__aexit__(exc_type, exc, tb) async def trigger_deployment( self, deployment: Union[str, UUID], parameters: Optional[Dict[str, Any]] = None, tags: Optional[list[str]] = None, name: Optional[str] = None, work_queue_name: Optional[str] = None, idempotency_key: Optional[str] = None, ): # Accept "flow_name/deployment_name" or UUID if isinstance(deployment, str): dep = await self.client.read_deployment_by_name(deployment) deployment_id = dep.id else: deployment_id = deployment return await self.client.create_flow_run_from_deployment( deployment_id=deployment_id, parameters=parameters or {}, tags=tags or [], name=name, work_queue_name=work_queue_name, idempotency_key=idempotency_key, ) async def read_flow_run(self, flow_run_id: Union[str, UUID]): return await self.client.read_flow_run(UUID(str(flow_run_id))) async def wait_for_flow_run( self, flow_run_id: Union[str, UUID], timeout: float = 3600, poll_interval: float = 5.0, ): start = time.time() flow_run_id = UUID(str(flow_run_id)) while True: fr = await self.client.read_flow_run(flow_run_id) if fr.state.is_terminal(): return fr if time.time() - start > timeout: raise TimeoutError(f"Flow run {flow_run_id} did not complete within {timeout}s") await asyncio.sleep(poll_interval) async def list_deployments(self, limit: int = 200, offset: int = 0): return await self.client.read_deployments(limit=limit, offset=offset) # ========== SYNC WRAPPER ========== class Prefect: def __init__(self): self._client_cm = None self.client = None def __enter__(self) -> "Prefect": self._client_cm = get_client(sync_client=True) self.client = self._client_cm.__enter__() return self def __exit__(self, exc_type, exc, tb): self._client_cm.__exit__(exc_type, exc, tb) def trigger_deployment( self, deployment: Union[str, UUID], parameters: Optional[Dict[str, Any]] = None, tags: Optional[list[str]] = None, name: Optional[str] = None, work_queue_name: Optional[str] = None, idempotency_key: Optional[str] = None, ): if isinstance(deployment, str): dep = self.client.read_deployment_by_name(deployment) deployment_id = dep.id else: deployment_id = deployment return self.client.create_flow_run_from_deployment( deployment_id=deployment_id, parameters=parameters or {}, tags=tags or [], name=name, work_queue_name=work_queue_name, idempotency_key=idempotency_key, )

def read_flow_run(self, flow_run_id: Union[str, UUID]): return self.client.read_flow_run(UUID(str(flow_run_id))) def wait_for_flow_run( self, flow_run_id: Union[str, UUID], timeout: float = 3600, poll_interval: float = 5.0, ): start = time.time() flow_run_id = UUID(str(flow_run_id)) while True: fr = self.client.read_flow_run(flow_run_id) if fr.state.is_terminal(): return fr if time.time() - start > timeout: raise TimeoutError(f"Flow run {flow_run_id} did not complete within {timeout}s") time.sleep(poll_interval) def list_deployments(self, limit: int = 200, offset: int = 0): return self.client.read_deployments(limit=limit, offset=offset) # ========== USAGE EXAMPLES ========== # Async async def main_async(): async with AsyncPrefect() as p: # Trigger by "flow_name/deployment_name" fr = await p.trigger_deployment( "my-flow/production", parameters={"param1": "value"}, tags=["manual"], idempotency_key="example-123", ) print("Created flow run:", fr.id) # Wait for completion final = await p.wait_for_flow_run(fr.id, timeout=1800, poll_interval=5) print("Final state:", final.state.type, "-", final.state.message) # asyncio.run(main_async()) # Sync def main_sync(): with Prefect() as p: fr = p.trigger_deployment( "my-flow/production", parameters={"param1": "value"}, tags=["manual"], ) print("Created flow run:", fr.id) final = p.wait_for_flow_run(fr.id, timeout=1800, poll_interval=5) print("Final state:", final.state.type, "-", final.state.message) # main_sync() ``` Would you like this wrapper to also include: - Deployment lookup by ID or name with validation - Variable get/set helpers - Pause/resume deployment - Listing flow runs with filters - Event-based waiting (WebSocket) when available If you share your exact needs, I can tailor the wrapper and add tests.

Short answer: that’s a perfectly fine way to run things. In Prefect 3.x, running a single

prefect worker start

that polls a work pool (and optionally multiple queues) is a common, supported pattern. It can execute any deployment that targets the same work pool/queues. If you prefer a shell script wrapper around the CLI, that’s also a great, simple approach. A few tips and examples: - Start one worker that can handle “any” deployment in a pool (optionally multiple queues)

prefect worker start -p my-pool -q default -q high-priority -n single-worker

Notes: - Point your deployments at the same work pool (and queue(s)) so this worker will pick them up. - A single worker can run multiple flow runs depending on your infra type and concurrency settings; scale by running more workers if needed. - For process workers, you usually run one per machine. For Docker/Kubernetes pools, a single worker can orchestrate many ephemeral jobs. - Trigger a deployment and watch it (nice for a shell wrapper)

prefect deployment run "my-flow/production" --param key=value --watch

You can also trigger by id with

--id <deployment-id>

, pass multiple

--param

flags, or add

--job-variable

flags to set job variables. - Minimal shell script pattern

#!/usr/bin/env bash
set -euo pipefail

POOL="my-pool"
QUEUE="default"
DEPLOYMENT="my-flow/production"

# Start the worker under a supervisor in practice (systemd, tmux, Docker, K8s, etc.)
prefect worker start -p "$POOL" -q "$QUEUE" -n "single-worker" &

# Trigger a run and watch logs/state until terminal
prefect deployment run "$DEPLOYMENT" --param run_date="$(date +%F)" --watch

When a single worker is not ideal: - You want strict workload isolation by team or environment (use separate pools/workers). - You need differing infrastructure (e.g., some jobs on Docker, others on K8s). - You need higher parallelism than one worker can comfortably handle. If you want, I can help you tailor a tiny shell wrapper for your pool/queues and common parameters. For background reading, see work pools and workers: Work pools and workers.