A non-retriable failure marks the task failed without requeueing the message 🟥

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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:

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,