Anoop Sypereddi
09/09/2025, 6:14 PMcancelling state with one of the downstream sub_flows is still showing as running. The subflow in question sends and waits for Apify requests to complete and I see more Apify requests being spun up by the subflow I tried to cancel. The only way I'm able to actually cancel is to go to the ECS task and kill it there. Is this supposed to be how cancelling a parent flow works or is this a bug?Marvin
09/09/2025, 6:14 PMMarvin
09/09/2025, 6:17 PM@flow from your flow runs it in the same Python process. When you click Cancel, Prefect signals cancellation and will mark the parent run as Cancelling, then tries to cooperatively stop running tasks/subflows. If your subflow is in a blocking loop or doesn’t yield/check for cancel, it can keep running and even launch new external requests.
- Separate deployment subflows: If you trigger a child via run_deployment, it runs on its own infrastructure task. Cancelling the parent does not forcibly kill independent deployments; you need to cancel the child run as well to stop its ECS task.
- ECS and “stuck in Cancelling”
- There have been reports where runs on ECS remain in Cancelling until the task is manually stopped, especially when the process is not responding to the cancel signal. See: Prefect ECS runs stuck in Cancelling
- Prefect runs a cancellation cleanup service that will eventually reconcile stuck states, but it won’t stop external work for you: Cancellation cleanup service
What I recommend
1) Make the subflow cancellable
- Add timeouts to network calls to Apify.
- Add periodic “yield points” (e.g., async sleeps) or checkpoints between batches so the engine can propagate cancellation.
- Between iterations of your polling/submit loop, check for a cancellation request and exit early instead of launching new requests.
- If you’re using synchronous/blocking I/O, consider switching to async or inserting short sleeps between iterations to give the event loop a chance to process the cancel.
2) Consider isolating the long-running subflow
- If the Apify work is long-lived, deploy it separately and start it with run_deployment. Then you can cancel the child run directly so its ECS task is stopped by the worker.
3) Update to latest versions
- Make sure you’re on the latest Prefect 3.x and latest prefect-aws. Several cancellation and ECS stop_task fixes have landed recently.
4) If cancellation continues to stick
- Collect worker logs at the time you click Cancel and confirm whether the worker attempted to stop the ECS task.
- Share the parent/child flow run IDs, your Prefect and prefect-aws versions, and whether the subflow is called in-process or via run_deployment.
Is this a bug? It’s not the intended UX. Cancellation should propagate and stop further work, but it is cooperative and depends on your code yielding/checking for cancel. The need to manually kill the ECS task suggests either:
- the subflow code isn’t observing cancellation, or
- an ECS stop signal didn’t get delivered/handled (potential bug or version mismatch; see the issue linked above).
If you can share:
- Prefect version and prefect-aws version
- Your work pool type (ECS) and worker logs around the cancellation
- How you launch the subflow (in-process @flow call vs run_deployment)
- A small sketch of the Apify polling loop
…I can help pinpoint whether this is a code-level cancellability issue vs an ECS/worker issue and propose a concrete fix.Anoop Sypereddi
09/09/2025, 7:32 PM@flow(name="run-dataset-config-scrapes")
def run_dataset_config_scrapes(dataset_config_id: int) -> None:
"""Orchestrate dataset config scraping across categories and account lists."""
dataset_config = get_dataset_config_by_id(dataset_config_id)
Log.info(f"Running dataset config scrapes for {dataset_config.name}")
# Scrape all account product lists
for account_product_list_id in dataset_config.account_product_list_ids:
Log.info(f"Scraping account product list: {account_product_list_id}")
scrape_account_product_list(account_product_list_id)
# Scrape all category configurations
for category_config_id in dataset_config.category_scrape_config_ids:
Log.info(f"Scraping category config: {category_config_id}")
run_category_scrape_config(category_config_id)
scrape_account_product_list is a subflow within this subflow which calls the scraper jobs. Even though run_dataset_config_scrapes is in cancelling scrape it started a new scrape_account_product_list flow.
run_dataset_config_scrapes logged a flow run cancelling at 12:53:31 PM and began a new subflow of scrape_account_product_list at 2:34:31 PM well after the cancelling state
what other information do you need? I believe this is a bugMarvin
09/09/2025, 7:47 PMMarvin
09/09/2025, 7:48 PMprefect version)
- prefect-aws version
- Infra details
- Work pool type (ECS) and worker version image/tag
- How the parent deployment is launched (schedule, UI, API)
- Subflow details
- Is scrape_account_product_list strictly called in-process (direct @flow call), or ever triggered via run_deployment?
- Does scrape_account_product_list have its own deployment or schedule? (If yes, that could have launched independently.)
- Run metadata and logs
- Parent flow run ID and the child flow run ID that started at 2:34 PM
- Worker logs around 12:53 PM when you clicked Cancel and around 2:34 PM when the new subflow began
- Any lines mentioning cancellation, stopping infrastructure, or ECS task stop attempts
In parallel, here are mitigations and instrumentation that can prevent new work from starting after a cancel and help diagnose root cause:
- Add a cancellation yield/check between iterations
- Best: make the coordinating flow async and add a tiny await point before launching each subflow (gives the engine a chance to propagate cancel), plus a state check to bail early.
- If sticking with sync, introduce task boundaries so Prefect can intercept cancellation before starting the next iteration.
Example (async pattern with a state check):
from prefect import flow
from prefect.client.orchestration import get_client
from prefect.runtime import flow_run
import anyio
async def _is_stopping():
async with get_client() as client:
fr = await client.read_flow_run(flow_run.id)
return fr.state_type.value in {"CANCELLING", "CANCELLED"}
@flow(name="run-dataset-config-scrapes")
async def run_dataset_config_scrapes(dataset_config_id: int) -> None:
dataset_config = get_dataset_config_by_id(dataset_config_id)
for account_product_list_id in dataset_config.account_product_list_ids:
# Yield to the engine and observe cancellation quickly
await anyio.sleep(0)
if await _is_stopping():
return
await scrape_account_product_list(account_product_list_id)
for category_config_id in dataset_config.category_scrape_config_ids:
await anyio.sleep(0)
if await _is_stopping():
return
await run_category_scrape_config(category_config_id)
- If you prefer sync, add task boundaries:
from prefect import flow, task
@task
def _run_one_account_list(account_product_list_id: int):
scrape_account_product_list(account_product_list_id)
@flow(name="run-dataset-config-scrapes")
def run_dataset_config_scrapes(dataset_config_id: int) -> None:
dataset_config = get_dataset_config_by_id(dataset_config_id)
for account_product_list_id in dataset_config.account_product_list_ids:
_run_one_account_list(account_product_list_id) # cancellation is checked between tasks
for category_config_id in dataset_config.category_scrape_config_ids:
_run_one_category(category_config_id)
- Add timeouts and cooperative checks in the subflow that hits Apify
- Ensure network calls have timeouts.
- Between API calls / batches, add an await point or quick state check so cancellation can stop launching new Apify jobs.
- If possible, implement an “abort” path to cancel/stop in-flight Apify runs when Prefect cancellation is detected.
References
- Cancellation overview and behavior: Cancel workflows
- Runtime context (for IDs used in checks): Runtime contextMarvin
09/09/2025, 7:48 PMscrape_account_product_list start 1h+ after the parent was Cancelling points to either:
- the loop never yielded back to the engine (so cancel wasn’t observed until much later), or
- the child flow was actually launched independently (e.g., via its own deployment/schedule), or
- a bug where the worker/engine accepted new work after the cancel state was set.
If you can share the requested versions, run IDs, and logs, I can help confirm which it is. If it’s reproducible with in-process subflows and no independent deployment, I’ll help escalate with a minimal repro so we can get a fix into Prefect.Anoop Sypereddi
09/09/2025, 7:57 PMVersion: 3.4.7
API version: 0.8.4
Python version: 3.12.11
Git commit: cd81d15a
Built: Thu, Jun 26, 2025 09:16 PM
OS/Arch: linux/aarch64
Profile: local
Server type: ephemeral
Pydantic version: 2.10.6
Server:
Database: sqlite
SQLite version: 3.40.1
task_definition:
{
"containerDefinitions": [
{
"command": [
"/bin/sh",
"-c",
"pip install prefect-aws && prefect worker start --pool prod-prefect --type ecs"
],
"cpu": 512,
"essential": true,
"image": "prefecthq/prefect:3-latest",
"memory": 1024,
"name": "prefect-worker",
"secrets": [
...
]
}
],
"cpu": "512",
"executionRoleArn": ...,
"family": "prefect-prod",
"memory": "1024",
"networkMode": "awsvpc",
"requiresCompatibilities": [
"FARGATE"
],
"taskRoleArn": ...
}
Deployment was launched via the UIMarvin
09/09/2025, 7:58 PMMarvin
09/09/2025, 7:59 PMrun_dataset_config_scrapes run ID that entered Cancelling at 125331
- The scrape_account_product_list subflow run ID that started at 23431
- How scrape_account_product_list is invoked:
- Confirm it’s a direct in-process @flow call (not run_deployment, no separate deployment or schedule).
- Worker logs:
- CloudWatch logs from the ECS worker task around 1253–1255 (when you clicked Cancel) and 234–236 (when the new subflow began). Look for any lines about “cancel”, “stopping task”, or errors.
- Whether the subflow has any retries/backoff configured.
- Whether the subflow or its inner Apify calls are synchronous and long-running without timeouts.
Immediate mitigations to stop new work after cancel
- Add a quick cancellation check before launching each subflow. For your current sync flow, a lightweight helper using the sync client works well:
from prefect import flow
from prefect.runtime import flow_run
from prefect.client.orchestration import get_client
def is_cancelling():
with get_client(sync_client=True) as client:
fr = client.read_flow_run(flow_run.id)
return fr.state_type.value in {"CANCELLING", "CANCELLED"}
@flow(name="run-dataset-config-scrapes")
def run_dataset_config_scrapes(dataset_config_id: int) -> None:
dataset_config = get_dataset_config_by_id(dataset_config_id)
for account_product_list_id in dataset_config.account_product_list_ids:
if is_cancelling():
return
scrape_account_product_list(account_product_list_id)
for category_config_id in dataset_config.category_scrape_config_ids:
if is_cancelling():
return
run_category_scrape_config(category_config_id)
- Insert task boundaries so Prefect can intercept cancellation between iterations:
from prefect import flow, task
@task
def _run_one_account_list(account_product_list_id: int):
scrape_account_product_list(account_product_list_id)
@task
def _run_one_category(category_config_id: int):
run_category_scrape_config(category_config_id)
@flow(name="run-dataset-config-scrapes")
def run_dataset_config_scrapes(dataset_config_id: int) -> None:
dataset_config = get_dataset_config_by_id(dataset_config_id)
for account_product_list_id in dataset_config.account_product_list_ids:
_run_one_account_list(account_product_list_id)
for category_config_id in dataset_config.category_scrape_config_ids:
_run_one_category(category_config_id)
- Add timeouts and short pauses between Apify operations so the engine can observe cancellation and you can bail before launching more jobs.
Versioning and ECS worker setup
- Avoid installing prefect-aws at container start. Pin versions in your image instead to ensure compatibility with the Prefect core that’s baked into the image. For example, build a small image based on prefecthq/prefect:3.4.7 and pip install "prefect-aws==<pinned>" at build time.
- Using “3-latest” plus an unpinned “pip install prefect-aws” can drift, causing subtle runtime differences in cancellation behavior on ECS.Marvin
09/09/2025, 7:59 PMBring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.
Powered by