Adds tenacity to retry failed task run creation

prefect_aws/workers/ecs_worker.py

@@ -70,6 +70,7 @@
 )
 from pydantic import Field, root_validator
 from slugify import slugify
+from tenacity import retry, stop_after_attempt, wait_fixed, wait_random
 from typing_extensions import Literal
 
 from prefect_aws import AwsCredentials
@@ -1421,6 +1422,7 @@ def _prepare_task_run_request(
 
         return task_run_request
 
+    @retry(stop=stop_after_attempt(3), wait=wait_fixed(1) + wait_random(0, 3))
     def _create_task_run(self, ecs_client: _ECSClient, task_run_request: dict) -> str:
         """
         Create a run of a task definition.

requirements.txt

@@ -1,5 +1,6 @@
 boto3>=1.24.53
 botocore>=1.27.53
-prefect>=2.10.11
 mypy_boto3_s3>=1.24.94
-mypy_boto3_secretsmanager>=1.26.49
+mypy_boto3_secretsmanager>=1.26.49
+prefect>=2.10.11
+tenacity>=8.0.0

tests/workers/test_ecs_worker.py

@@ -12,6 +12,7 @@
 from moto.ec2.utils import generate_instance_identity_document
 from prefect.server.schemas.core import FlowRun
 from prefect.utilities.asyncutils import run_sync_in_worker_thread
+from tenacity import RetryError
 
 from prefect_aws.workers.ecs_worker import (
     _TASK_DEFINITION_CACHE,
@@ -1900,3 +1901,26 @@ async def test_kill_infrastructure_with_grace_period(aws_credentials, caplog, fl
 
     # Logs warning
     assert "grace period of 60s requested, but AWS does not support" in caplog.text
+
+
+async def test_retry_on_failed_task_start(
+    aws_credentials: AwsCredentials, flow_run, ecs_mocks
+):
+    run_task_mock = MagicMock(return_value=[])
+
+    configuration = await construct_configuration(
+        aws_credentials=aws_credentials, command="echo test"
+    )
+
+    inject_moto_patches(
+        ecs_mocks,
+        {
+            "run_task": [run_task_mock],
+        },
+    )
+
+    with pytest.raises(RetryError):
+        async with ECSWorker(work_pool_name="test") as worker:
+            await run_then_stop_task(worker, configuration, flow_run)
+
+    assert run_task_mock.call_count == 3