diff --git a/document-parser/infra/serve_converter.py b/document-parser/infra/serve_converter.py index 635e8fb..2edbb91 100644 --- a/document-parser/infra/serve_converter.py +++ b/document-parser/infra/serve_converter.py @@ -138,7 +138,11 @@ def _parse_response(data: dict) -> ConversionResult: # json_content contains the full DoclingDocument with pages, elements, bboxes json_content = document.get("json_content") if isinstance(json_content, str): - json_content = json.loads(json_content) + try: + json_content = json.loads(json_content) + except json.JSONDecodeError: + logger.warning("Failed to parse json_content as JSON, ignoring structured data") + json_content = None pages: list[PageDetail] = [] if json_content: diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 6f7f543..b9b9113 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -7,6 +7,7 @@ from the conversion implementation (local Docling lib vs remote Docling Serve). from __future__ import annotations import asyncio +import functools import json import logging from dataclasses import asdict @@ -42,7 +43,7 @@ class AnalysisService: task = asyncio.create_task( self._run_analysis(job.id, doc.storage_path, doc.filename, pipeline_options) ) - task.add_done_callback(_on_task_done) + task.add_done_callback(functools.partial(_on_task_done, job_id=job.id)) return job @@ -99,14 +100,16 @@ class AnalysisService: await _mark_failed(job_id, str(e)) -def _on_task_done(task: asyncio.Task) -> None: - """Log unhandled exceptions from background analysis tasks.""" +def _on_task_done(task: asyncio.Task, *, job_id: str) -> None: + """Log unhandled exceptions from background analysis tasks and mark job as FAILED.""" if task.cancelled(): - logger.warning("Analysis task was cancelled") + logger.warning("Analysis task was cancelled: %s", job_id) + asyncio.ensure_future(_mark_failed(job_id, "Task was cancelled")) return exc = task.exception() if exc: - logger.error("Unhandled exception in analysis task: %s", exc, exc_info=True) + logger.error("Unhandled exception in analysis task %s: %s", job_id, exc, exc_info=True) + asyncio.ensure_future(_mark_failed(job_id, str(exc))) async def _mark_failed(job_id: str, error: str) -> None: diff --git a/document-parser/tests/test_analysis_service.py b/document-parser/tests/test_analysis_service.py new file mode 100644 index 0000000..7a9cbaa --- /dev/null +++ b/document-parser/tests/test_analysis_service.py @@ -0,0 +1,71 @@ +"""Tests for AnalysisService — focus on _on_task_done callback (bug #1 fix).""" + +from __future__ import annotations + +import asyncio +from unittest.mock import AsyncMock, patch + +import pytest + +from services.analysis_service import _on_task_done + + +class TestOnTaskDone: + """Bug #1: _on_task_done must call _mark_failed when the task raises.""" + + @pytest.mark.asyncio + async def test_exception_marks_job_failed(self): + """When a background task raises, the job should be marked FAILED.""" + job_id = "job-123" + + # Create a task that raises + async def failing_task(): + raise RuntimeError("boom") + + task = asyncio.create_task(failing_task()) + await asyncio.sleep(0) # let the task fail + + with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark: + _on_task_done(task, job_id=job_id) + # ensure_future schedules it; give the event loop a tick + await asyncio.sleep(0) + + mock_mark.assert_called_once_with(job_id, "boom") + + @pytest.mark.asyncio + async def test_cancelled_task_marks_job_failed(self): + """When a background task is cancelled, the job should be marked FAILED.""" + job_id = "job-456" + + async def slow_task(): + await asyncio.sleep(999) + + task = asyncio.create_task(slow_task()) + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + + with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark: + _on_task_done(task, job_id=job_id) + await asyncio.sleep(0) + + mock_mark.assert_called_once_with(job_id, "Task was cancelled") + + @pytest.mark.asyncio + async def test_successful_task_does_not_mark_failed(self): + """When a background task succeeds, _mark_failed should not be called.""" + job_id = "job-789" + + async def ok_task(): + return "done" + + task = asyncio.create_task(ok_task()) + await task + + with patch("services.analysis_service._mark_failed", new_callable=AsyncMock) as mock_mark: + _on_task_done(task, job_id=job_id) + await asyncio.sleep(0) + + mock_mark.assert_not_called() diff --git a/document-parser/tests/test_serve_converter.py b/document-parser/tests/test_serve_converter.py index b9c4b36..09de6b3 100644 --- a/document-parser/tests/test_serve_converter.py +++ b/document-parser/tests/test_serve_converter.py @@ -147,6 +147,21 @@ class TestParseResponse: result = _parse_response(data) assert result.page_count == 1 + def test_json_content_malformed_string_falls_back(self): + """Bug #5: malformed JSON string in json_content must not crash.""" + data = { + "document": { + "md_content": "# Hello", + "html_content": "