< Marvin> What is this error ```Task run failed with excepti Prefect Community #marvin-ai

<@ULVA73B9P> What is this error: ```Task run faile...

Federico Ferlito

12/19/2025, 8:08 AM

@Marvin What is this error:

Copy code

Task run failed with exception: AttributeError("'PrefectFlowRunFuture' object has no attribute '_task_run_id'")
Traceback (most recent call last):
  File "/app/.venv/lib/python3.10/site-packages/prefect/task_engine.py", line 1596, in run_context
    yield self
  File "/app/.venv/lib/python3.10/site-packages/prefect/task_engine.py", line 1675, in run_task_async
    await engine.call_task_fn(txn)
  File "/app/.venv/lib/python3.10/site-packages/prefect/task_engine.py", line 1613, in call_task_fn
    result = await call_with_parameters(self.task.fn, parameters)
  File "/app/barlowtwins-cyc-431-preprocessing/src/bt_process_data.py", line 218, in trigger_parallel_s2_processing
    s3_data_path = data_processing_s2(
  File "/app/.venv/lib/python3.10/site-packages/prefect/flows.py", line 2151, in __call__
    return run_coro_as_sync(
  File "/app/.venv/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 207, in run_coro_as_sync
    return call.result()
  File "/app/.venv/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 365, in result
    return self.future.result(timeout=timeout)
  File "/app/.venv/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 192, in result
    return self.__get_result()
  File "/usr/local/lib/python3.10/concurrent/futures/_base.py", line 403, in __get_result
    raise self._exception
  File "/app/.venv/lib/python3.10/site-packages/prefect/_internal/concurrency/calls.py", line 441, in _run_async
    result = await coro
  File "/app/.venv/lib/python3.10/site-packages/prefect/utilities/asyncutils.py", line 188, in coroutine_wrapper
    return await task
  File "/app/.venv/lib/python3.10/site-packages/prefect/flows.py", line 2127, in modified_call
    async with self.worker_cls(work_pool_name=self.work_pool) as worker:
  File "/app/.venv/lib/python3.10/site-packages/prefect_aws/workers/ecs_worker.py", line 1586, in __aexit__
    return await super().__aexit__(*exc_info)
  File "/app/.venv/lib/python3.10/site-packages/prefect/workers/base.py", line 1629, in __aexit__
    raise exceptions[0] from None
  File "/app/.venv/lib/python3.10/site-packages/prefect/flows.py", line 2137, in modified_call
    return await future.aresult()
  File "/app/.venv/lib/python3.10/site-packages/prefect/futures.py", line 428, in aresult
    f"Task run {self.task_run_id} did not complete within {timeout} seconds"
  File "/app/.venv/lib/python3.10/site-packages/prefect/futures.py", line 61, in task_run_id
    return self._task_run_id
AttributeError: 'PrefectFlowRunFuture' object has no attribute '_task_run_id'. Did you mean: 'task_run_id'?

I executed parallel tasks in parallel:

Copy code

asyncio.run(
                process_tiles(
                   ...
                )
            )

....

@flow(name="process_tiles", task_runner=ThreadPoolTaskRunner(max_workers=16))
async def process_tiles(
    ....
) -> None:
    """Process tiles for a given sensor."""
    logger = get_run_logger()
    <http://logger.info|logger.info>(f"Processing tiles for sensor {sensor}")
    processing_func = SENSOR_DATA_PROCESSING[sensor]
    flow_runs = []

    for tile in tiles_list:
        data_dir = f"{disk}/{tile}/{sensor}"
        flow_runs.append(
            processing_func.submit(
                ....
            )
        )
        time.sleep(5)
    wait(flow_runs)
    s3_data_paths = [fr.result() for fr in flow_runs]
    <http://logger.info|logger.info>(f"Results: {s3_data_paths}")
....

@task(
    persist_result=True,
    result_storage=s3_bucket_block,
)
async def processing_func(
    ...
) -> str:
    """Trigger parallel data processing for a tile."""
    logging = get_run_logger()
    <http://logging.info|logging.info>(f"Submitting tile {tile}")
    s3_data_path = data_processing_s2(
        ....
    )
    return s3_data_path

....

@ecs(
    work_pool="ec2-pool",
    image="....",
    region_name="...",
)
@flow(
    name="data_processing",
    persist_result=True,
    result_storage=s3_bucket_block,
)
def data_processing_s2(
    .....
) -> str:
do stuff here

This works fine usually. I got the error above only once, even tho the actual task finished sucessfully. How can I avoid this? Is it about the multithreding, pydantic, or something like that? I need to trigger the processing_func tasks in parallel so that the ecs flows jobs data_processing_s2 can be triggered together , as they are long jobs. what would be a fix that implies the least changes, and the least dependance from asking you more things?

Marvin

12/19/2025, 8:08 AM

✅ thought for 368.3 seconds

Marvin

12/19/2025, 8:15 AM

Short answer: this is a Prefect bug, not your code. What happened - You’re calling an infrastructure-bound subflow (

data_processing_s2

decorated with

@ecs

) from within a task. - Under certain timing conditions (typically when the worker context is exiting and waiting on the subflow), Prefect raises a timeout-related error message for a flow-run future. - In Prefect 3.x, that error message incorrectly references a task-run attribute (

task_run_id

) on a flow-run future. Since

PrefectFlowRunFuture

doesn’t have

_task_run_id

, you see: 'PrefectFlowRunFuture' object has no attribute '_task_run_id' - Your subflow actually finished; the error is raised while formatting the timeout message, so it’s intermittent and only shows up when that timeout/error path is hit. How to avoid it with minimal changes 1) Upgrade to the latest Prefect and prefect-aws - This bug has been addressed in recent 3.x work. Upgrading is the least invasive fix. - Commands:

Copy code

pip install -U prefect prefect-aws

- Latest release notes: Prefect 3.6.7 changelog 2) Two very small code tweaks that improve stability - Don’t wrap a Prefect flow call in

asyncio.run(...)

. Prefect’s flow call already manages async/sync for you. - Instead of:

Copy code

asyncio.run(process_tiles(...))

do:

Copy code

process_tiles(...)

- In your async flow, replace

time.sleep(5)

with

await asyncio.sleep(5)

so you don’t block the event loop:

Copy code

await asyncio.sleep(5)

Optional: simplify the pattern (if you’re open to it) - Instead of calling the ECS subflow from inside a task, submit the ECS subflow directly from the parent flow. This avoids mixing a task future with a flow-run future and removes the path that triggers the buggy message.

Copy code

@flow(name="process_tiles", task_runner=ThreadPoolTaskRunner(max_workers=16))
  async def process_tiles(...):
      logger = get_run_logger()
      <http://logger.info|logger.info>(f"Processing tiles for sensor {sensor}")

      frs = []
      for tile in tiles_list:
          frs.append(
              data_processing_s2.submit(
                  ...
              )
          )
          await asyncio.sleep(5)

      # Wait for all ECS subflows and collect their results
      wait(frs)
      s3_data_paths = [fr.result() for fr in frs]
      <http://logger.info|logger.info>(f"Results: {s3_data_paths}")

- This keeps the parallelism and behavior you want with minimal changes, and avoids the error path that referenced

task_run_id

. Not related to Pydantic or your multithreading logic - This isn’t a Pydantic issue. - ThreadPoolTaskRunner is fine for your use case; the intermittent error stems from that incorrect attribute reference in Prefect’s flow future error message. If you share your Prefect and prefect-aws versions, I can confirm whether they include the fix.

3 Views

Open in Slack

Previous Next