Enhance audio job processing tests and handling

tests/audio/test_audio_job_consumer.py

@@ -78,6 +78,10 @@ async def test_processes_day_job_successfully(
         job_id = uuid4()
         day_id = uuid4()
         mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value}
+        mock_update_status.side_effect = [
+            {"job_id": str(job_id), "status": AudioJobStatus.PROCESSING.value},
+            {"job_id": str(job_id), "status": AudioJobStatus.COMPLETED.value},
+        ]
         mock_generate.return_value = {
             "audio_url": "https://example.com/a.wav",
             "audio_duration_ms": 1200,
@@ -120,6 +124,10 @@ async def test_marks_failed_on_generation_error(
     ):
         job_id = uuid4()
         mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value}
+        mock_update_status.side_effect = [
+            {"job_id": str(job_id), "status": AudioJobStatus.PROCESSING.value},
+            {"job_id": str(job_id), "status": AudioJobStatus.FAILED.value},
+        ]
         mock_generate.side_effect = HTTPException(status_code=404, detail={"message": "Sub task not found"})
 
         message = {
@@ -168,6 +176,81 @@ async def test_skips_completed_job(
         mock_update_status.assert_not_awaited()
         mock_delete.assert_called_once_with("abc")
 
+    @pytest.mark.asyncio
+    @patch("worker_api.audio.services.audio_job_consumer.delete_audio_job_message")
+    @patch("worker_api.audio.services.audio_job_consumer.generate_plan_audio_service", new_callable=AsyncMock)
+    @patch("worker_api.audio.services.audio_job_consumer.update_audio_job_status", new_callable=AsyncMock)
+    @patch("worker_api.audio.services.audio_job_consumer.get_audio_job_status", new_callable=AsyncMock)
+    async def test_skips_when_claim_conflicts(
+        self,
+        mock_get_job,
+        mock_update_status,
+        mock_generate,
+        mock_delete,
+    ):
+        job_id = uuid4()
+        mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value}
+        mock_update_status.side_effect = HTTPException(
+            status_code=409,
+            detail={"error": "Bad request", "message": "Audio job is already being processed"},
+        )
+
+        message = {
+            "ReceiptHandle": "abc",
+            "Body": json.dumps(
+                {
+                    "job_id": str(job_id),
+                    "day_id": str(uuid4()),
+                    "language": "bo",
+                    "type": "TEXT_READING",
+                }
+            ),
+        }
+
+        await process_audio_job_message(message)
+
+        mock_generate.assert_not_called()
+        mock_update_status.assert_awaited_once()
+        assert mock_update_status.await_args.kwargs["status"] == AudioJobStatus.PROCESSING
+        mock_delete.assert_called_once_with("abc")
+
+    @pytest.mark.asyncio
+    @patch("worker_api.audio.services.audio_job_consumer.delete_audio_job_message")
+    @patch("worker_api.audio.services.audio_job_consumer.generate_plan_audio_service", new_callable=AsyncMock)
+    @patch("worker_api.audio.services.audio_job_consumer.update_audio_job_status", new_callable=AsyncMock)
+    @patch("worker_api.audio.services.audio_job_consumer.get_audio_job_status", new_callable=AsyncMock)
+    async def test_skips_when_claim_returns_terminal(
+        self,
+        mock_get_job,
+        mock_update_status,
+        mock_generate,
+        mock_delete,
+    ):
+        job_id = uuid4()
+        mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value}
+        mock_update_status.return_value = {
+            "job_id": str(job_id),
+            "status": AudioJobStatus.COMPLETED.value,
+        }
+
+        message = {
+            "ReceiptHandle": "abc",
+            "Body": json.dumps(
+                {
+                    "job_id": str(job_id),
+                    "day_id": str(uuid4()),
+                    "language": "bo",
+                    "type": "TEXT_READING",
+                }
+            ),
+        }
+
+        await process_audio_job_message(message)
+
+        mock_generate.assert_not_called()
+        mock_update_status.assert_awaited_once()
+        mock_delete.assert_called_once_with("abc")
+
     @pytest.mark.asyncio
     @patch("worker_api.audio.services.audio_job_consumer.delete_audio_job_message")
     @patch("worker_api.audio.services.audio_job_consumer.parse_audio_job_message_body", return_value=None)
@@ -210,6 +293,10 @@ async def test_fails_when_language_missing(
     ):
         job_id = uuid4()
         mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value}
+        mock_update_status.side_effect = [
+            {"job_id": str(job_id), "status": AudioJobStatus.PROCESSING.value},
+            {"job_id": str(job_id), "status": AudioJobStatus.FAILED.value},
+        ]
         message = {
             "ReceiptHandle": "abc",
             "Body": json.dumps(
@@ -243,6 +330,10 @@ async def test_fails_when_target_ids_missing(
     ):
         job_id = uuid4()
         mock_get_job.return_value = {"job_id": str(job_id), "status": AudioJobStatus.PENDING.value}
+        mock_update_status.side_effect = [
+            {"job_id": str(job_id), "status": AudioJobStatus.PROCESSING.value},
+            {"job_id": str(job_id), "status": AudioJobStatus.FAILED.value},
+        ]
         message = {
             "ReceiptHandle": "abc",
             "Body": json.dumps({"job_id": str(job_id), "language": "en"}),

worker_api/audio/services/audio_job_consumer.py

@@ -100,7 +100,23 @@ async def process_audio_job_message(message: Dict[str, Any]) -> None:
             delete_audio_job_message(receipt_handle)
         return
 
-    await update_audio_job_status(job_id=job_id, status=AudioJobStatus.PROCESSING)
+    try:
+        claimed = await update_audio_job_status(job_id=job_id, status=AudioJobStatus.PROCESSING)
+    except HTTPException as exc:
+        # Another worker already claimed this job (duplicate SQS delivery).
+        if exc.status_code == 409:
+            logger.info("Skipping already claimed audio job %s", job_id)
+            if receipt_handle:
+                delete_audio_job_message(receipt_handle)
+            return
+        raise
+
+    claimed_status = str(claimed.get("status") or "")
+    if claimed_status != AudioJobStatus.PROCESSING.value:
+        logger.info("Skipping audio job %s after claim (%s)", job_id, claimed_status)
+        if receipt_handle:
+            delete_audio_job_message(receipt_handle)
+        return
 
     try:
         day_id = _parse_uuid(body.get("day_id"))