https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> I'm getting the error below randomly ...
# ask-marvin
f
@Marvin I'm getting the error below randomly when running a flow daily (most of the time there is no exception). Can you give me some pointers to fix it?
Copy code
07:01:17 AM
Error
Encountered exception during execution: AttributeError("'TaskRun' object has no attribute 'state'")
Traceback (most recent call last):
  File "/app/.venv/lib/python3.11/site-packages/prefect/task_engine.py", line 1220, in initialize_run
    validated_state=self.task_run.state,
                    ^^^^^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/pydantic/main.py", line 994, in __getattr__
    raise AttributeError(f'{type(self).__name__!r} object has no attribute {item!r}')
AttributeError: 'TaskRun' object has no attribute 'state'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/app/.venv/lib/python3.11/site-packages/prefect/flow_engine.py", line 1332, in run_context
    yield self
  File "/app/.venv/lib/python3.11/site-packages/prefect/flow_engine.py", line 1394, in run_flow_async
    await engine.call_flow_fn()
  File "/app/.venv/lib/python3.11/site-packages/prefect/flow_engine.py", line 1346, in call_flow_fn
    result = await call_with_parameters(self.flow.fn, self.parameters)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/data_platform/internal_alerting/src/prefect/flows/internal_alerting.py", line 57, in internal_alerting
    task_end_states = resolve_futures_to_states(tasks)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/prefect/futures.py", line 643, in resolve_futures_to_states
    future.wait()
  File "/app/.venv/lib/python3.11/site-packages/prefect/futures.py", line 197, in wait
    result = self._wrapped_future.result(timeout=timeout)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 449, in result
    return self.__get_result()
           ^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/concurrent/futures/_base.py", line 401, in __get_result
    raise self._exception
  File "/usr/local/lib/python3.11/concurrent/futures/thread.py", line 58, in run
    result = self.fn(*self.args, **self.kwargs)
             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/asyncio/base_events.py", line 654, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/prefect/task_engine.py", line 1412, in run_task_async
    async with engine.start(task_run_id=task_run_id, dependencies=dependencies):
  File "/usr/local/lib/python3.11/contextlib.py", line 210, in __aenter__
    return await anext(self.gen)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/prefect/task_engine.py", line 1287, in start
    async with self.initialize_run(
  File "/usr/local/lib/python3.11/contextlib.py", line 210, in __aenter__
    return await anext(self.gen)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/prefect/task_engine.py", line 1258, in initialize_run
    self.log_finished_message()
  File "/app/.venv/lib/python3.11/site-packages/prefect/task_engine.py", line 251, in log_finished_message
    display_state = repr(self.state) if PREFECT_DEBUG_MODE else str(self.state)
                                                                    ^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/prefect/task_engine.py", line 136, in state
    if not self.task_run or not self.task_run.state:
                                ^^^^^^^^^^^^^^^^^^^
  File "/app/.venv/lib/python3.11/site-packages/pydantic/main.py", line 994, in __getattr__
    raise AttributeError(f'{type(self).__name__!r} object has no attribute {item!r}')
AttributeError: 'TaskRun' object has no attribute 'state'
And here is the flow code:
Copy code
@flow(
    result_storage=common_settings.PREFECT_RESULT_STORAGE_S3_BLOCK_NAME,
    cache_result_in_memory=False,
    on_cancellation=[
        slack_notifier.send_flow_run_finished_slack_notification,
    ],
    on_crashed=[slack_notifier.send_flow_run_finished_slack_notification],
    log_prints=True,
)
async def internal_alerting() -> State | None:

    logging = Logging(logging_level=common_settings.LOGGING_LEVEL)
    logger = logging.get_logger()

    app_container = AwsInternalAlertingAppContainer(
        config=aws_internal_alerting_settings,
        common_config=common_settings,
        logging=logging,
    )

    alert_service = app_container.services.alert_service

    configs = aws_internal_alerting_settings.CONFIGS
    tasks = [
        alert_service.process_alert.submit(cfg)
        for cfg in configs
        if cfg.is_active()
    ]

    task_end_states = resolve_futures_to_states(tasks)
    has_failures = has_failed_tasks(task_end_states)
    if has_failures:
        message = (
            "Internal alerting failed. Please check the logs for more information."
        )
    else:
        message = "Internal alerting completed successfully."
    return await handle_flow_completion(
        slack_client=slack_notifier,
        logger=logger,
        message=message,
        is_success=not has_failures,
        send_slack=True,
    )
m
✅ thought for 513.9 seconds
Thanks for the detailed traceback and the flow snippet — that helps a lot. Short answer: this looks like a known intermittent bug in Prefect 3.x where, under bursts of task submissions, the task engine tries to log a finished message while the 
TaskRun
object hasn’t finished initializing its 
state
attribute. We’ve seen it happen more often with lots of near-simultaneous 
.submit()
calls. There’s an open issue tracking this exact error with a minimal repro and stack trace very similar to yours: Submitting many tasks simultaneously raises TaskRun AttributeError. What you can do right now - Upgrade Prefect to the latest 3.x - Even though the release notes don’t explicitly call out this fix, staying current has reduced the frequency of these engine-side race conditions for many users. - Latest notes: 3.6.4 release (full changelog: 3.6.3…3.6.4) - Reduce the burstiness of task submissions - Set a task runner with bounded concurrency so you don’t submit a very large number of tasks at once:
Copy code
from prefect import flow
  from prefect.task_runners import ThreadPoolTaskRunner

  @flow(
      task_runner=ThreadPoolTaskRunner(max_workers=8),
      result_storage=common_settings.PREFECT_RESULT_STORAGE_S3_BLOCK_NAME,
      cache_result_in_memory=False,
      on_cancellation=[slack_notifier.send_flow_run_finished_slack_notification],
      on_crashed=[slack_notifier.send_flow_run_finished_slack_notification],
      log_prints=True,
  )
  async def internal_alerting():
      ...
- Batch your submissions to avoid a stampede:
Copy code
active = [cfg for cfg in configs if cfg.is_active()]
  batch_size = 10
  tasks = []
  for i in range(0, len(active), batch_size):
      batch = active[i : i + batch_size]
      batch_tasks = [alert_service.process_alert.submit(cfg) for cfg in batch]
      # Option A: wait per-batch (smooths load)
      resolve_futures_to_states(batch_tasks)
      tasks.extend(batch_tasks)

  # Or, if you’d rather wait once at the end:
  # resolve_futures_to_states(tasks)
- If you keep the async flow, a tiny yield between submissions can help smooth scheduling:
Copy code
import asyncio

  tasks = []
  for cfg in configs:
      if not cfg.is_active():
          continue
      tasks.append(alert_service.process_alert.submit(cfg))
      await asyncio.sleep(0)  # yield to event loop to avoid a submission burst
- Alternative to resolve_futures_to_states - It’s fine to keep using it, but if you want to test whether a different wait path avoids the crash, you can do:
Copy code
for f in tasks:
      f.wait()
  task_end_states = [f.state for f in tasks]
- Functionally equivalent for your “did anything fail?” check, but takes a slightly different path through the engine. A couple of questions to help us pinpoint and give you a more tailored workaround: - Which Prefect version and Python version are you on? - Roughly how many tasks are you submitting per run (number of active configs)? - What task runner/worker type and environment (e.g., Kubernetes, Docker, local) are you using? If you’re okay with it, I can add your traceback and pattern (async flow + many concurrent `.submit`s) to the GitHub issue above to help the team prioritize.
23 Views
Open in Slack
PreviousNext
Powered by