diff --git a/tests/utilities/aws_lambda/test_task_handler.py b/tests/utilities/aws_lambda/test_task_handler.py index 418669bcd..9c538bce7 100644 --- a/tests/utilities/aws_lambda/test_task_handler.py +++ b/tests/utilities/aws_lambda/test_task_handler.py @@ -8,11 +8,12 @@ import pytest from sqlalchemy import Engine from sqlmodel import Session +from domain.tasks.subtasks import SubTaskStatus from domain.tasks.tasks import Source from orchestration.task_orchestrator import TaskOrchestrator from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository from repositories.tasks.task_postgres_repository import TaskPostgresRepository -from utilities.aws_lambda.task_handler import task_handler +from utilities.aws_lambda.task_handler import NonRetriableTaskError, task_handler @dataclass @@ -104,6 +105,35 @@ def test_task_handler_passes_orchestrator_and_task_id_when_flag_is_true( assert recv_task_id == UUID(result[0]["task_id"]) +def test_task_handler_does_not_requeue_a_record_failing_non_retriably( + harness: Harness, +) -> None: + # Arrange + @task_handler( + task_source="abri", + source=Source.HUBSPOT_DEAL, + orchestrator_cm=harness.factory, + ) + def handler(body: dict[str, Any], context: Any) -> None: + raise NonRetriableTaskError("job logged but write-back failed") + + event = { + "Records": [ + {"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'} + ] + } + + # Act + result = handler(event, context=None) + + # Assert: the failure is recorded, but the message is not requeued. + assert result["batchItemFailures"] == [] + subtask_id = result["tasks"][0]["subtask_id"] + failed = harness.subtasks.get(UUID(subtask_id)) + assert failed.status is SubTaskStatus.FAILED + assert failed.outputs == {"error": "job logged but write-back failed"} + + def test_task_handler_leaves_cloudwatch_url_unset_outside_lambda( harness: Harness, monkeypatch: pytest.MonkeyPatch ) -> None: diff --git a/utilities/aws_lambda/task_handler.py b/utilities/aws_lambda/task_handler.py index f2064f573..bbcf3f5cd 100644 --- a/utilities/aws_lambda/task_handler.py +++ b/utilities/aws_lambda/task_handler.py @@ -20,6 +20,17 @@ logger = logging.getLogger(__name__) OrchestratorCM = Callable[[], AbstractContextManager[TaskOrchestrator]] +class NonRetriableTaskError(Exception): + """A failure that must not be retried by SQS redelivery. + + The SubTask is marked failed (the failure record is the visibility + surface) but the message is not returned to the queue, so the work is + never re-run. Raise it for failures where a retry could duplicate an + irreversible side effect — e.g. a job already logged in OpenHousing + whose HubSpot write-back failed. + """ + + def task_handler( *, task_source: str,