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

@@ -105,6 +105,33 @@ 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_reports_an_ordinarily_failing_record_for_redelivery(
+    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 RuntimeError("transient failure")
+
+    event = {
+        "Records": [
+            {"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'}
+        ]
+    }
+
+    # Act
+    result = handler(event, context=None)
+
+    # Assert
+    assert result["batchItemFailures"] == [{"itemIdentifier": "msg-1"}]
+    subtask_id = result["tasks"][0]["subtask_id"]
+    assert harness.subtasks.get(UUID(subtask_id)).status is SubTaskStatus.FAILED
+
+
 def test_task_handler_does_not_requeue_a_record_failing_non_retriably(
     harness: Harness,
 ) -> None:

utilities/aws_lambda/task_handler.py

@@ -89,6 +89,17 @@ def task_handler(
                                 work=lambda: func(body, context),
                                 cloud_logs_url=cloud_logs_url,
                             )
+                    except NonRetriableTaskError:
+                        # Failed, but never redelivered: the SubTask's failure
+                        # record is the only follow-up surface.
+                        logger.exception(
+                            "subtask failed non-retriably "
+                            "(task_source=%s source_id=%s)",
+                            task_source,
+                            source_id,
+                        )
+                        if "Records" not in event:
+                            raise
                     except Exception:
                         logger.exception(
                             "subtask failed (task_source=%s source_id=%s)",