From 0af3b81b8fe76d85f518d95bd2e7d74fcc31ead2 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 31 Mar 2026 11:13:51 +0200 Subject: [PATCH] Fix zombie jobs and unprotected JSON parse Bug #1: _on_task_done now receives job_id via functools.partial and calls _mark_failed when the background task raises or is cancelled, preventing jobs from being stuck in RUNNING state forever. Bug #5: _parse_response wraps json.loads in try/except JSONDecodeError so malformed json_content strings fall back gracefully instead of crashing. Co-Authored-By: Claude Opus 4.6 --- document-parser/infra/serve_converter.py | 6 +- document-parser/services/analysis_service.py | 13 ++-- .../tests/test_analysis_service.py | 71 +++++++++++++++++++ document-parser/tests/test_serve_converter.py | 15 ++++ 4 files changed, 99 insertions(+), 6 deletions(-) create mode 100644 document-parser/tests/test_analysis_service.py 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": "

Hello

", + "json_content": "NOT VALID JSON {{{", + } + } + result = _parse_response(data) + assert isinstance(result, ConversionResult) + assert result.content_markdown == "# Hello" + assert result.pages == [] + assert result.page_count == 1 + def test_tables_and_pictures(self): data = { "document": {