fix(docling_serve): address async job review feedback

Author: 2830500285

integrations/docling_serve/src/haystack_integrations/components/converters/docling_serve/__init__.py

@@ -6,7 +6,14 @@
     ConversionMode,
     DoclingServeConversionError,
     DoclingServeConverter,
+    DoclingServeTimeoutError,
     ExportType,
 )
 
-__all__ = ["ConversionMode", "DoclingServeConversionError", "DoclingServeConverter", "ExportType"]
+__all__ = [
+    "ConversionMode",
+    "DoclingServeConversionError",
+    "DoclingServeConverter",
+    "DoclingServeTimeoutError",
+    "ExportType",
+]

integrations/docling_serve/src/haystack_integrations/components/converters/docling_serve/converter.py

@@ -57,6 +57,10 @@ class DoclingServeConversionError(Exception):
     """Raised when DoclingServe reports an async task or conversion failure."""
 
 
+class DoclingServeTimeoutError(DoclingServeConversionError):
+    """Raised when a DoclingServe async task exceeds job_timeout."""
+
+
 def _is_url(source: str) -> bool:
     parsed = urlparse(source)
     return parsed.scheme in ("http", "https")
@@ -137,23 +141,25 @@ def __init__(
             Conversion mode. `sync` uses DoclingServe's synchronous endpoints. `async` submits
             conversion jobs to DoclingServe's async endpoints and polls until completion.
         :param poll_interval:
-            Maximum server-side long-poll wait in seconds when `mode="async"`.
+            Controls both the server-side long-poll wait (?wait= parameter) and the maximum local sleep between polls.
+            A higher value reduces round-trips; a lower value increases polling frequency.
         :param job_timeout:
             Maximum time in seconds to wait for each async conversion job.
         """
-        if poll_interval <= 0:
-            msg = "poll_interval must be greater than 0."
-            raise ValueError(msg)
-        if job_timeout <= 0:
-            msg = "job_timeout must be greater than 0."
-            raise ValueError(msg)
+        self.mode = ConversionMode(mode)
+        if self.mode == ConversionMode.ASYNC:
+            if poll_interval <= 0:
+                msg = "poll_interval must be greater than 0."
+                raise ValueError(msg)
+            if job_timeout <= 0:
+                msg = "job_timeout must be greater than 0."
+                raise ValueError(msg)
 
         self.base_url = base_url.rstrip("/")
         self.export_type = ExportType(export_type)
         self.convert_options = dict(convert_options) if convert_options else {}
         self.timeout = timeout
         self.api_key = api_key
-        self.mode = ConversionMode(mode)
         self.poll_interval = poll_interval
         self.job_timeout = job_timeout
 
@@ -262,7 +268,9 @@ def _post_file(self, client: httpx.Client, source: str | Path | ByteStream) -> d
             headers=self._headers(),
         )
         response.raise_for_status()
-        return response.json()
+        data = response.json()
+        self._raise_for_failed_conversion(data)
+        return data
 
     def _submit_file_job(self, client: httpx.Client, source: str | Path | ByteStream) -> str:
         filename = _resolve_filename(source)
@@ -293,7 +301,9 @@ def _post_url(self, client: httpx.Client, url: str) -> dict[str, Any]:
             headers=self._headers(),
         )
         response.raise_for_status()
-        return response.json()
+        data = response.json()
+        self._raise_for_failed_conversion(data)
+        return data
 
     def _submit_url_job(self, client: httpx.Client, url: str) -> str:
         response = client.post(
@@ -319,7 +329,7 @@ def _wait_for_job(self, client: httpx.Client, task_id: str) -> None:
             remaining = deadline - time.monotonic()
             if remaining <= 0:
                 msg = f"Timed out waiting for DoclingServe task {task_id} after {self.job_timeout:.2f}s."
-                raise DoclingServeConversionError(msg)
+                raise DoclingServeTimeoutError(msg)
 
             wait = min(self.poll_interval, remaining)
             poll_started = time.monotonic()
@@ -369,7 +379,9 @@ async def _post_file_async(self, client: httpx.AsyncClient, source: str | Path |
             headers=self._headers(),
         )
         response.raise_for_status()
-        return response.json()
+        data = response.json()
+        self._raise_for_failed_conversion(data)
+        return data
 
     async def _submit_file_job_async(self, client: httpx.AsyncClient, source: str | Path | ByteStream) -> str:
         filename = _resolve_filename(source)
@@ -400,7 +412,9 @@ async def _post_url_async(self, client: httpx.AsyncClient, url: str) -> dict[str
             headers=self._headers(),
         )
         response.raise_for_status()
-        return response.json()
+        data = response.json()
+        self._raise_for_failed_conversion(data)
+        return data
 
     async def _submit_url_job_async(self, client: httpx.AsyncClient, url: str) -> str:
         response = await client.post(
@@ -426,7 +440,7 @@ async def _wait_for_job_async(self, client: httpx.AsyncClient, task_id: str) ->
             remaining = deadline - time.monotonic()
             if remaining <= 0:
                 msg = f"Timed out waiting for DoclingServe task {task_id} after {self.job_timeout:.2f}s."
-                raise DoclingServeConversionError(msg)
+                raise DoclingServeTimeoutError(msg)
 
             wait = min(self.poll_interval, remaining)
             poll_started = time.monotonic()
@@ -517,6 +531,12 @@ def run(
                         source=source,
                         error=e,
                     )
+                except DoclingServeTimeoutError as e:
+                    logger.warning(
+                        "Timed out waiting for DoclingServe conversion for {source}. Skipping it. Error: {error}",
+                        source=source,
+                        error=e,
+                    )
                 except DoclingServeConversionError as e:
                     logger.warning(
                         "DoclingServe conversion failed for {source}. Skipping it. Error: {error}",
@@ -585,6 +605,12 @@ async def run_async(
                         source=source,
                         error=e,
                     )
+                except DoclingServeTimeoutError as e:
+                    logger.warning(
+                        "Timed out waiting for DoclingServe conversion for {source}. Skipping it. Error: {error}",
+                        source=source,
+                        error=e,
+                    )
                 except DoclingServeConversionError as e:
                     logger.warning(
                         "DoclingServe conversion failed for {source}. Skipping it. Error: {error}",

integrations/docling_serve/tests/test_converter.py

@@ -37,6 +37,11 @@ def _mock_status_response(
     return httpx.Response(200, json=json_data, request=httpx.Request("GET", "http://test"))
 
 
+def _mock_failed_result_response(error_message: str = "conversion failed", status: str = "failure") -> httpx.Response:
+    json_data = {"status": status, "errors": [{"error_message": error_message}]}
+    return httpx.Response(200, json=json_data, request=httpx.Request("GET", "http://test"))
+
+
 class TestDoclingServeConverterInit:
     def test_defaults(self, monkeypatch):
         monkeypatch.delenv("DOCLING_SERVE_API_KEY", raising=False)
@@ -88,10 +93,16 @@ def test_api_key_none_overrides_env(self, monkeypatch):
 
     def test_invalid_async_job_settings(self):
         with pytest.raises(ValueError, match="poll_interval"):
-            DoclingServeConverter(poll_interval=0)
+            DoclingServeConverter(mode=ConversionMode.ASYNC, poll_interval=0)
 
         with pytest.raises(ValueError, match="job_timeout"):
-            DoclingServeConverter(job_timeout=0)
+            DoclingServeConverter(mode=ConversionMode.ASYNC, job_timeout=0)
+
+    def test_sync_mode_allows_unused_async_job_settings(self):
+        converter = DoclingServeConverter(poll_interval=0, job_timeout=0)
+        assert converter.mode == ConversionMode.SYNC
+        assert converter.poll_interval == 0
+        assert converter.job_timeout == 0
 
 
 class TestDoclingServeConverterSerialization:
@@ -290,6 +301,30 @@ def test_run_skips_when_no_content(self, caplog):
         assert result["documents"] == []
         assert "No content returned" in caplog.text
 
+    def test_run_skips_failed_sync_url_response_details(self, caplog):
+        converter = DoclingServeConverter(api_key=None)
+        mock_resp = _mock_failed_result_response("url conversion failed")
+
+        with patch("httpx.Client.post", return_value=mock_resp):
+            with caplog.at_level(logging.WARNING):
+                result = converter.run(sources=["https://example.com/doc.pdf"])
+
+        assert result["documents"] == []
+        assert "url conversion failed" in caplog.text
+
+    def test_run_skips_failed_sync_file_response_details(self, tmp_path, caplog):
+        pdf = tmp_path / "test.pdf"
+        pdf.write_bytes(b"%PDF-test")
+        converter = DoclingServeConverter(api_key=None)
+        mock_resp = _mock_failed_result_response("file conversion failed")
+
+        with patch("httpx.Client.post", return_value=mock_resp):
+            with caplog.at_level(logging.WARNING):
+                result = converter.run(sources=[pdf])
+
+        assert result["documents"] == []
+        assert "file conversion failed" in caplog.text
+
     def test_run_text_export(self):
         converter = DoclingServeConverter(export_type=ExportType.TEXT, api_key=None)
         mock_resp = _mock_httpx_response("plain text content", ExportType.TEXT)
@@ -400,6 +435,50 @@ def test_run_async_mode_skips_failed_job(self, caplog):
         assert result["documents"] == []
         assert "conversion failed" in caplog.text
 
+    def test_run_async_mode_skips_failed_result(self, caplog):
+        converter = DoclingServeConverter(api_key=None, mode=ConversionMode.ASYNC)
+
+        with (
+            patch("httpx.Client.post", return_value=_mock_task_response()),
+            patch("httpx.Client.get", side_effect=[_mock_status_response(), _mock_failed_result_response("bad file")]),
+        ):
+            with caplog.at_level(logging.WARNING):
+                result = converter.run(sources=["https://example.com/doc.pdf"])
+
+        assert result["documents"] == []
+        assert "bad file" in caplog.text
+
+    def test_run_async_mode_skips_skipped_result(self, caplog):
+        converter = DoclingServeConverter(api_key=None, mode=ConversionMode.ASYNC)
+
+        with (
+            patch("httpx.Client.post", return_value=_mock_task_response()),
+            patch(
+                "httpx.Client.get",
+                side_effect=[_mock_status_response(), _mock_failed_result_response("skipped file", status="skipped")],
+            ),
+        ):
+            with caplog.at_level(logging.WARNING):
+                result = converter.run(sources=["https://example.com/doc.pdf"])
+
+        assert result["documents"] == []
+        assert "skipped file" in caplog.text
+
+    def test_run_async_mode_skips_timed_out_job(self, caplog):
+        converter = DoclingServeConverter(
+            api_key=None, mode=ConversionMode.ASYNC, poll_interval=0.001, job_timeout=0.001
+        )
+
+        with (
+            patch("httpx.Client.post", return_value=_mock_task_response()),
+            patch("httpx.Client.get", return_value=_mock_status_response(status="pending")),
+        ):
+            with caplog.at_level(logging.WARNING):
+                result = converter.run(sources=["https://example.com/doc.pdf"])
+
+        assert result["documents"] == []
+        assert "Timed out" in caplog.text
+
 
 class TestDoclingServeConverterFilename:
     def test_bytestream_filename_from_file_name(self):
@@ -463,6 +542,37 @@ async def test_run_async_url_source(self):
         assert len(result["documents"]) == 1
         assert result["documents"][0].content == "# Async content"
 
+    @pytest.mark.asyncio
+    async def test_run_async_file_source(self, tmp_path):
+        pdf = tmp_path / "test.pdf"
+        pdf.write_bytes(b"%PDF-test")
+        converter = DoclingServeConverter(api_key=None)
+        mock_resp = httpx.Response(
+            200,
+            json={"document": {"md_content": "# Async file content"}, "status": "success"},
+            request=httpx.Request("POST", "http://test"),
+        )
+
+        with patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=mock_resp) as mock_post:
+            result = await converter.run_async(sources=[pdf])
+
+        post_url = mock_post.call_args[0][0]
+        assert post_url.endswith("/v1/convert/file")
+        assert result["documents"][0].content == "# Async file content"
+
+    @pytest.mark.asyncio
+    async def test_run_async_skips_failed_sync_response_details(self, caplog):
+        converter = DoclingServeConverter(api_key=None)
+
+        with patch(
+            "httpx.AsyncClient.post", new_callable=AsyncMock, return_value=_mock_failed_result_response("async failure")
+        ):
+            with caplog.at_level(logging.WARNING):
+                result = await converter.run_async(sources=["https://example.com/doc.pdf"])
+
+        assert result["documents"] == []
+        assert "async failure" in caplog.text
+
     @pytest.mark.asyncio
     async def test_run_async_skips_on_status_error(self, caplog):
         converter = DoclingServeConverter(api_key=None)
@@ -534,6 +644,66 @@ async def test_run_async_async_mode_uses_job_endpoints(self):
         assert status_call.kwargs["params"] == {"wait": 1.5}
         assert result_call.args[0].endswith("/v1/result/task-1")
 
+    @pytest.mark.asyncio
+    async def test_run_async_async_mode_uses_job_endpoints_for_file_source(self, tmp_path):
+        pdf = tmp_path / "test.pdf"
+        pdf.write_bytes(b"%PDF-test")
+        converter = DoclingServeConverter(api_key=None, mode=ConversionMode.ASYNC)
+
+        with (
+            patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=_mock_task_response()) as mock_post,
+            patch(
+                "httpx.AsyncClient.get",
+                new_callable=AsyncMock,
+                side_effect=[_mock_status_response(), _mock_httpx_response("async file content")],
+            ),
+        ):
+            result = await converter.run_async(sources=[pdf])
+
+        post_url = mock_post.call_args[0][0]
+        data = mock_post.call_args[1]["data"]
+        assert post_url.endswith("/v1/convert/file/async")
+        assert data["target_type"] == "inbody"
+        assert result["documents"][0].content == "async file content"
+
+    @pytest.mark.asyncio
+    async def test_run_async_async_mode_skips_failed_result(self, caplog):
+        converter = DoclingServeConverter(api_key=None, mode=ConversionMode.ASYNC)
+
+        with (
+            patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=_mock_task_response()),
+            patch(
+                "httpx.AsyncClient.get",
+                new_callable=AsyncMock,
+                side_effect=[_mock_status_response(), _mock_failed_result_response("bad async result")],
+            ),
+        ):
+            with caplog.at_level(logging.WARNING):
+                result = await converter.run_async(sources=["https://example.com/doc.pdf"])
+
+        assert result["documents"] == []
+        assert "bad async result" in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_run_async_async_mode_skips_timed_out_job(self, caplog):
+        converter = DoclingServeConverter(
+            api_key=None, mode=ConversionMode.ASYNC, poll_interval=0.001, job_timeout=0.001
+        )
+
+        with (
+            patch("httpx.AsyncClient.post", new_callable=AsyncMock, return_value=_mock_task_response()),
+            patch(
+                "httpx.AsyncClient.get",
+                new_callable=AsyncMock,
+                return_value=_mock_status_response(status="pending"),
+            ),
+        ):
+            with caplog.at_level(logging.WARNING):
+                result = await converter.run_async(sources=["https://example.com/doc.pdf"])
+
+        assert result["documents"] == []
+        assert "Timed out" in caplog.text
+
 
 class TestDoclingServeConverterIntegration:
     @pytest.mark.integration