Merge pull request #66 from scub-france/fix/audit-robustness
[FIX] Robustness audit — settings validation, timeout cascade, cancellation
This commit is contained in:
Pier-Jean Malandrino
GitHub
commit
18f3b91af4
5 changed files with 210 additions and 10 deletions
|
|
@ -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:
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
@ -25,6 +26,41 @@ 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.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})"
|
||||
)
|
||||
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}')"
|
||||
)
|
||||
# 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))
|
||||
|
||||
@classmethod
|
||||
def from_env(cls) -> Settings:
|
||||
"""Build a Settings instance from environment variables."""
|
||||
|
|
@ -37,6 +73,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")),
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
@ -232,7 +243,10 @@ 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))
|
||||
|
||||
|
||||
# Keep the module-level function as the default, but AnalysisService uses its own method.
|
||||
|
||||
|
||||
def _schedule_mark_failed(job_id: str, error: str) -> None:
|
||||
|
|
|
|||
|
|
@ -3,6 +3,7 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import functools
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
|
@ -20,7 +21,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 +31,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):
|
||||
|
|
@ -71,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."""
|
||||
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
@ -29,6 +30,79 @@ 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_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
|
||||
|
||||
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
|
||||
|
||||
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")
|
||||
|
|
@ -36,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")
|
||||
|
|
@ -50,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"
|
||||
|
|
|
|||
Loading…
Reference in a new issue