From 2c254382c82e8f6d52e7d5c242da1f5ba67d87c1 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 16:59:37 +0200 Subject: [PATCH 1/5] fix: add __post_init__ validation to Settings dataclass (#65) Reject invalid configuration at startup: negative timeouts, zero concurrency, bad table mode. Reports all errors at once. --- document-parser/infra/settings.py | 21 +++++++++++ document-parser/tests/test_settings.py | 51 ++++++++++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index ebb596c..60cc3c5 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -25,6 +25,27 @@ class Settings: default_factory=lambda: ["http://localhost:3000", "http://localhost:5173"] ) + def __post_init__(self) -> None: + errors: list[str] = [] + if self.document_timeout <= 0: + errors.append(f"document_timeout must be > 0 (got {self.document_timeout})") + if self.conversion_timeout <= 0: + errors.append(f"conversion_timeout must be > 0 (got {self.conversion_timeout})") + if self.max_concurrent_analyses < 1: + errors.append( + f"max_concurrent_analyses must be >= 1 (got {self.max_concurrent_analyses})" + ) + if self.max_page_count < 0: + errors.append(f"max_page_count must be >= 0 (got {self.max_page_count})") + if self.max_file_size < 0: + errors.append(f"max_file_size must be >= 0 (got {self.max_file_size})") + if self.default_table_mode not in ("accurate", "fast"): + errors.append( + f"default_table_mode must be 'accurate' or 'fast' (got '{self.default_table_mode}')" + ) + if errors: + raise ValueError("Invalid settings:\n " + "\n ".join(errors)) + @classmethod def from_env(cls) -> Settings: """Build a Settings instance from environment variables.""" diff --git a/document-parser/tests/test_settings.py b/document-parser/tests/test_settings.py index 1b8837c..aa27233 100644 --- a/document-parser/tests/test_settings.py +++ b/document-parser/tests/test_settings.py @@ -29,6 +29,57 @@ class TestSettingsDefaults: s.upload_dir = "/other" # type: ignore[misc] +class TestSettingsValidation: + def test_negative_document_timeout_rejected(self): + import pytest + + with pytest.raises(ValueError, match="document_timeout must be > 0"): + Settings(document_timeout=-1.0) + + def test_zero_document_timeout_rejected(self): + import pytest + + with pytest.raises(ValueError, match="document_timeout must be > 0"): + Settings(document_timeout=0) + + def test_negative_conversion_timeout_rejected(self): + import pytest + + with pytest.raises(ValueError, match="conversion_timeout must be > 0"): + Settings(conversion_timeout=-1) + + def test_zero_max_concurrent_rejected(self): + import pytest + + with pytest.raises(ValueError, match="max_concurrent_analyses must be >= 1"): + Settings(max_concurrent_analyses=0) + + def test_negative_max_page_count_rejected(self): + import pytest + + with pytest.raises(ValueError, match="max_page_count must be >= 0"): + Settings(max_page_count=-1) + + def test_negative_max_file_size_rejected(self): + import pytest + + with pytest.raises(ValueError, match="max_file_size must be >= 0"): + Settings(max_file_size=-1) + + def test_invalid_table_mode_rejected(self): + import pytest + + with pytest.raises(ValueError, match="default_table_mode must be"): + Settings(default_table_mode="turbo") + + def test_multiple_errors_reported(self): + import pytest + + with pytest.raises(ValueError, match="document_timeout") as exc_info: + Settings(document_timeout=-1, conversion_timeout=-1) + assert "conversion_timeout" in str(exc_info.value) + + class TestSettingsFromEnv: def test_reads_env_vars(self, monkeypatch): monkeypatch.setenv("APP_VERSION", "1.2.3") From 1657bce1f05f3ca927f913e9093db33301fe92f6 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 17:01:10 +0200 Subject: [PATCH 2/5] fix: make converter lock timeout configurable via LOCK_TIMEOUT env var (#61) Replace hardcoded _LOCK_TIMEOUT=300 with settings.lock_timeout, readable from LOCK_TIMEOUT environment variable. --- document-parser/infra/local_converter.py | 5 ++--- document-parser/infra/settings.py | 4 ++++ document-parser/tests/test_settings.py | 7 +++++++ 3 files changed, 13 insertions(+), 3 deletions(-) diff --git a/document-parser/infra/local_converter.py b/document-parser/infra/local_converter.py index a126e7b..1a3b701 100644 --- a/document-parser/infra/local_converter.py +++ b/document-parser/infra/local_converter.py @@ -49,7 +49,6 @@ logger = logging.getLogger(__name__) # Thread lock — DoclingConverter is not thread-safe. # Uses a timeout to prevent a frozen conversion from blocking all others. _converter_lock = threading.Lock() -_LOCK_TIMEOUT = 300 # seconds — fail fast rather than wait forever # US Letter page dimensions (points) — fallback when page size is unknown _DEFAULT_PAGE_WIDTH = 612.0 @@ -218,10 +217,10 @@ def _process_content_item( def _convert_sync(file_path: str, options: ConversionOptions) -> ConversionResult: - acquired = _converter_lock.acquire(timeout=_LOCK_TIMEOUT) + acquired = _converter_lock.acquire(timeout=settings.lock_timeout) if not acquired: raise TimeoutError( - f"Could not acquire converter lock within {_LOCK_TIMEOUT}s — " + f"Could not acquire converter lock within {settings.lock_timeout}s — " "a previous conversion may be frozen" ) try: diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index 60cc3c5..496e032 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -15,6 +15,7 @@ class Settings: docling_serve_api_key: str | None = None conversion_timeout: int = 900 document_timeout: float = 120.0 # Docling-level per-document timeout (seconds) + lock_timeout: int = 300 # converter lock acquisition timeout (seconds) max_concurrent_analyses: int = 3 default_table_mode: str = "accurate" # "accurate" or "fast" max_page_count: int = 0 # 0 = unlimited (upload validation) @@ -31,6 +32,8 @@ class Settings: errors.append(f"document_timeout must be > 0 (got {self.document_timeout})") if self.conversion_timeout <= 0: errors.append(f"conversion_timeout must be > 0 (got {self.conversion_timeout})") + if self.lock_timeout <= 0: + errors.append(f"lock_timeout must be > 0 (got {self.lock_timeout})") if self.max_concurrent_analyses < 1: errors.append( f"max_concurrent_analyses must be >= 1 (got {self.max_concurrent_analyses})" @@ -58,6 +61,7 @@ class Settings: docling_serve_api_key=os.environ.get("DOCLING_SERVE_API_KEY"), conversion_timeout=int(os.environ.get("CONVERSION_TIMEOUT", "900")), document_timeout=float(os.environ.get("DOCUMENT_TIMEOUT", "120.0")), + lock_timeout=int(os.environ.get("LOCK_TIMEOUT", "300")), max_concurrent_analyses=int(os.environ.get("MAX_CONCURRENT_ANALYSES", "3")), default_table_mode=os.environ.get("DEFAULT_TABLE_MODE", "accurate"), max_page_count=int(os.environ.get("MAX_PAGE_COUNT", "0")), diff --git a/document-parser/tests/test_settings.py b/document-parser/tests/test_settings.py index aa27233..9a683fd 100644 --- a/document-parser/tests/test_settings.py +++ b/document-parser/tests/test_settings.py @@ -15,6 +15,7 @@ class TestSettingsDefaults: assert s.docling_serve_api_key is None assert s.conversion_timeout == 900 assert s.document_timeout == 120.0 + assert s.lock_timeout == 300 assert s.max_page_count == 0 assert s.upload_dir == "./uploads" assert s.db_path == "./data/docling_studio.db" @@ -66,6 +67,12 @@ class TestSettingsValidation: with pytest.raises(ValueError, match="max_file_size must be >= 0"): Settings(max_file_size=-1) + def test_zero_lock_timeout_rejected(self): + import pytest + + with pytest.raises(ValueError, match="lock_timeout must be > 0"): + Settings(lock_timeout=0) + def test_invalid_table_mode_rejected(self): import pytest From 8ea0cebc1675614b1a73466abb241361619a44b0 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 17:04:09 +0200 Subject: [PATCH 3/5] fix: validate timeout cascade ordering in Settings (#62) Enforce document_timeout < lock_timeout < conversion_timeout at startup. Prevents inverted timeout configurations that cause unpredictable behavior. --- document-parser/infra/settings.py | 12 ++++++++++++ document-parser/tests/test_settings.py | 22 ++++++++++++++++++++-- 2 files changed, 32 insertions(+), 2 deletions(-) diff --git a/document-parser/infra/settings.py b/document-parser/infra/settings.py index 496e032..d8f1e48 100644 --- a/document-parser/infra/settings.py +++ b/document-parser/infra/settings.py @@ -46,6 +46,18 @@ class Settings: errors.append( f"default_table_mode must be 'accurate' or 'fast' (got '{self.default_table_mode}')" ) + # Timeout cascade: document_timeout < lock_timeout < conversion_timeout + if self.document_timeout > 0 and self.lock_timeout > 0 and self.conversion_timeout > 0: + if self.document_timeout >= self.lock_timeout: + errors.append( + f"document_timeout ({self.document_timeout}s) must be " + f"< lock_timeout ({self.lock_timeout}s)" + ) + if self.lock_timeout >= self.conversion_timeout: + errors.append( + f"lock_timeout ({self.lock_timeout}s) must be " + f"< conversion_timeout ({self.conversion_timeout}s)" + ) if errors: raise ValueError("Invalid settings:\n " + "\n ".join(errors)) diff --git a/document-parser/tests/test_settings.py b/document-parser/tests/test_settings.py index 9a683fd..604abac 100644 --- a/document-parser/tests/test_settings.py +++ b/document-parser/tests/test_settings.py @@ -79,6 +79,22 @@ class TestSettingsValidation: with pytest.raises(ValueError, match="default_table_mode must be"): Settings(default_table_mode="turbo") + def test_cascade_document_ge_lock_rejected(self): + import pytest + + with pytest.raises(ValueError, match=r"document_timeout.*< lock_timeout"): + Settings(document_timeout=400.0, lock_timeout=300, conversion_timeout=900) + + def test_cascade_lock_ge_conversion_rejected(self): + import pytest + + with pytest.raises(ValueError, match=r"lock_timeout.*< conversion_timeout"): + Settings(document_timeout=100.0, lock_timeout=900, conversion_timeout=900) + + def test_cascade_valid_ordering_accepted(self): + s = Settings(document_timeout=60.0, lock_timeout=300, conversion_timeout=900) + assert s.document_timeout < s.lock_timeout < s.conversion_timeout + def test_multiple_errors_reported(self): import pytest @@ -94,8 +110,9 @@ class TestSettingsFromEnv: monkeypatch.setenv("DEPLOYMENT_MODE", "huggingface") monkeypatch.setenv("DOCLING_SERVE_URL", "http://serve:9000") monkeypatch.setenv("DOCLING_SERVE_API_KEY", "secret-key") - monkeypatch.setenv("CONVERSION_TIMEOUT", "120") + monkeypatch.setenv("CONVERSION_TIMEOUT", "1200") monkeypatch.setenv("DOCUMENT_TIMEOUT", "60.0") + monkeypatch.setenv("LOCK_TIMEOUT", "600") monkeypatch.setenv("MAX_PAGE_COUNT", "20") monkeypatch.setenv("UPLOAD_DIR", "/data/uploads") monkeypatch.setenv("DB_PATH", "/data/test.db") @@ -108,7 +125,8 @@ class TestSettingsFromEnv: assert s.deployment_mode == "huggingface" assert s.docling_serve_url == "http://serve:9000" assert s.docling_serve_api_key == "secret-key" - assert s.conversion_timeout == 120 + assert s.conversion_timeout == 1200 + assert s.lock_timeout == 600 assert s.document_timeout == 60.0 assert s.max_page_count == 20 assert s.upload_dir == "/data/uploads" From 302b24cc4c75921346c68a66c4dd5dc29e1f1d10 Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 17:06:16 +0200 Subject: [PATCH 4/5] fix: route _on_task_done errors through _classify_error (#63) Unhandled task exceptions now produce user-friendly messages instead of raw Python tracebacks in the error_message field. --- document-parser/services/analysis_service.py | 2 +- .../tests/test_analysis_service.py | 23 +++++++++++++++++-- 2 files changed, 22 insertions(+), 3 deletions(-) diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index f630e95..36c2f5a 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -232,7 +232,7 @@ def _on_task_done(task: asyncio.Task, *, job_id: str) -> None: exc = task.exception() if exc: logger.error("Unhandled exception in analysis task %s: %s", job_id, exc, exc_info=True) - _schedule_mark_failed(job_id, str(exc)) + _schedule_mark_failed(job_id, _classify_error(exc)) def _schedule_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 index 83f6be8..66cbaac 100644 --- a/document-parser/tests/test_analysis_service.py +++ b/document-parser/tests/test_analysis_service.py @@ -20,7 +20,7 @@ class TestOnTaskDone: # Create a task that raises async def failing_task(): - raise RuntimeError("boom") + raise RuntimeError("unexpected failure") task = asyncio.create_task(failing_task()) await asyncio.sleep(0) # let the task fail @@ -30,7 +30,26 @@ class TestOnTaskDone: # ensure_future schedules it; give the event loop a tick await asyncio.sleep(0) - mock_mark.assert_called_once_with(job_id, "boom") + mock_mark.assert_called_once_with(job_id, "unexpected failure") + + @pytest.mark.asyncio + async def test_exception_uses_classify_error(self): + """_on_task_done should route exceptions through _classify_error.""" + job_id = "job-classify" + + async def timeout_task(): + raise TimeoutError("timeout exceeded while processing") + + task = asyncio.create_task(timeout_task()) + await asyncio.sleep(0) + + 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, "Processing took too long — try with fewer pages or simpler options" + ) @pytest.mark.asyncio async def test_cancelled_task_marks_job_failed(self): From af10c200f0bfe615bde759539137ecaf14ef57ca Mon Sep 17 00:00:00 2001 From: Pier-Jean Malandrino Date: Tue, 7 Apr 2026 17:07:55 +0200 Subject: [PATCH 5/5] fix: cancel running conversion when analysis job is deleted (#64) Track background tasks per job ID. On delete, cancel the task to release the converter lock instead of letting phantom jobs run to completion. --- document-parser/services/analysis_service.py | 18 +++++- .../tests/test_analysis_service.py | 55 +++++++++++++++++++ 2 files changed, 71 insertions(+), 2 deletions(-) diff --git a/document-parser/services/analysis_service.py b/document-parser/services/analysis_service.py index 36c2f5a..f003285 100644 --- a/document-parser/services/analysis_service.py +++ b/document-parser/services/analysis_service.py @@ -53,6 +53,7 @@ class AnalysisService: self._chunker = chunker self._conversion_timeout = conversion_timeout self._semaphore = asyncio.Semaphore(max_concurrent) + self._running_tasks: dict[str, asyncio.Task] = {} async def create( self, @@ -79,7 +80,8 @@ class AnalysisService: chunking_options, ) ) - task.add_done_callback(functools.partial(_on_task_done, job_id=job.id)) + self._running_tasks[job.id] = task + task.add_done_callback(functools.partial(self._on_task_done, job_id=job.id)) return job @@ -92,7 +94,11 @@ class AnalysisService: return await analysis_repo.find_by_id(job_id) async def delete(self, job_id: str) -> bool: - """Delete an analysis job. Returns True if it existed.""" + """Delete an analysis job, cancelling any running task. Returns True if it existed.""" + task = self._running_tasks.pop(job_id, None) + if task and not task.done(): + task.cancel() + logger.info("Cancelled running task for job %s", job_id) return await analysis_repo.delete(job_id) async def rechunk(self, job_id: str, chunking_options: dict) -> list[ChunkResult]: @@ -115,6 +121,11 @@ class AnalysisService: return chunks + def _on_task_done(self, task: asyncio.Task, *, job_id: str) -> None: + """Cleanup running tasks and delegate to module-level handler.""" + self._running_tasks.pop(job_id, None) + _on_task_done(task, job_id=job_id) + async def _run_analysis( self, job_id: str, @@ -235,6 +246,9 @@ def _on_task_done(task: asyncio.Task, *, job_id: str) -> None: _schedule_mark_failed(job_id, _classify_error(exc)) +# Keep the module-level function as the default, but AnalysisService uses its own method. + + def _schedule_mark_failed(job_id: str, error: str) -> None: """Schedule _mark_failed as a tracked background task.""" t = asyncio.ensure_future(_mark_failed(job_id, error)) diff --git a/document-parser/tests/test_analysis_service.py b/document-parser/tests/test_analysis_service.py index 66cbaac..a211dee 100644 --- a/document-parser/tests/test_analysis_service.py +++ b/document-parser/tests/test_analysis_service.py @@ -3,6 +3,7 @@ from __future__ import annotations import asyncio +import functools from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -90,6 +91,60 @@ class TestOnTaskDone: mock_mark.assert_not_called() +class TestAnalysisServiceCancellation: + """Verify delete cancels running tasks.""" + + @pytest.mark.asyncio + async def test_delete_cancels_running_task(self): + """Deleting a job while running should cancel its task.""" + converter = MagicMock() + service = AnalysisService(converter=converter) + + blocker = asyncio.Event() + + async def slow_analysis(): + await blocker.wait() + + task = asyncio.create_task(slow_analysis()) + service._running_tasks["j1"] = task + + with patch("services.analysis_service.analysis_repo") as mock_repo: + mock_repo.delete = AsyncMock(return_value=True) + result = await service.delete("j1") + + assert result is True + assert task.cancelling() or task.cancelled() + assert "j1" not in service._running_tasks + + @pytest.mark.asyncio + async def test_delete_completed_job_no_error(self): + """Deleting a completed job should not raise even if no task tracked.""" + converter = MagicMock() + service = AnalysisService(converter=converter) + + with patch("services.analysis_service.analysis_repo") as mock_repo: + mock_repo.delete = AsyncMock(return_value=True) + result = await service.delete("j-gone") + + assert result is True + + @pytest.mark.asyncio + async def test_task_cleaned_from_running_on_completion(self): + """After a task completes, it should be removed from _running_tasks.""" + converter = MagicMock() + service = AnalysisService(converter=converter) + + async def instant(): + pass + + task = asyncio.create_task(instant()) + service._running_tasks["j1"] = task + task.add_done_callback(functools.partial(service._on_task_done, job_id="j1")) + await task + + assert "j1" not in service._running_tasks + + class TestAnalysisServiceConcurrency: """Verify that the semaphore limits concurrent analysis jobs."""