fix: added concurrency limits and retry for calls in docling component and service#2083
fix: added concurrency limits and retry for calls in docling component and service#2083lucaseduoli wants to merge 2 commits into
Conversation
github-actions
Bot
added
backend
WalkthroughDocling integration now retries transient HTTP failures, limits concurrent operations with shared semaphores, retains slots through optional polling lifecycles, guarantees slot release, and updates status/result error handling. Remote component flows and unit tests were updated accordingly. ChangesDocling reliability controls
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
github-actions
Bot
added
bug
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
flows/components/docling_remote.py (1)
278-294: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winKeep polling after transient status-check retries are exhausted.
Three failed GET retries here abort an in-progress Docling task even though the server may just not be ready yet; catch that boundary and continue polling untilmax_poll_timeoutin bothflows/components/docling_remote.pyandflows/ingestion_flow.json.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@flows/components/docling_remote.py` around lines 278 - 294, Update the polling loops surrounding _get_with_retry in flows/components/docling_remote.py (lines 278-294) and flows/ingestion_flow.json (line 710) to catch exhausted transient status-check retries and continue polling until max_poll_timeout, while preserving normal success/failure handling and timeout behavior.src/services/docling_service.py (1)
233-274: 🩺 Stability & Availability | 🔴 Critical | ⚡ Quick winSemaphore slot leaks if preparation fails before the
tryblock.The slot is acquired at Line 237, but everything from
_build_docling_optionsthrough_get_client(Lines 239-257) runs outside thetrywhoseexceptat Line 296 performs the release. Ifget_openrag_config(),json.dumps(...), or client creation raises, the acquired slot is never released. Under repeated failures the semaphore drains to zero and all subsequent uploads block indefinitely.Acquire immediately before the guarded
try(option/header/data building don't need the slot):🔒 Proposed fix
- logger.debug("Acquiring Docling concurrency slot", filename=filename) - await DoclingService._semaphore.acquire() - options = self._build_docling_options( ocr_override=ocr, picture_descriptions_override=picture_descriptions, ) headers = self._get_auth_headers(user_id, auth_header) data = { k: str(v).lower() if isinstance(v, bool) else v for k, v in options.items() if not isinstance(v, dict) } if "picture_description_local" in options: data["picture_description_local"] = json.dumps(options["picture_description_local"]) files = {"files": (filename, file_content)} client = self._get_client() should_close = client != self.httpx_client `@retry`( stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=1, max=10), retry=retry_if_exception(is_transient_error), reraise=True, ) async def _post_with_retry(): response = await client.post( f"{self.docling_url}/v1/convert/file/async", files=files, data=data, headers=headers, ) response.raise_for_status() return response + logger.debug("Acquiring Docling concurrency slot", filename=filename) + await DoclingService._semaphore.acquire() try:🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/services/docling_service.py` around lines 233 - 274, Move semaphore acquisition in the Docling request flow to immediately before the existing guarded try block, after _build_docling_options, _get_auth_headers, data preparation, file setup, and _get_client complete. Keep the existing release logic paired with that acquisition so preparation or client-creation failures cannot consume a slot permanently.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@flows/components/docling_remote.py`:
- Around line 47-52: Validate max_concurrency is at least 1 before semaphore
initialization or acquisition in _acquire_semaphore, preserving the existing
default when it is None. Apply the same non-positive concurrency guard in
flows/components/docling_remote.py lines 47-52 and flows/ingestion_flow.json
line 710 so process_files and embedded component execution reject invalid values
consistently.
- Around line 70-79: Remove automatic retry behavior from _post_with_retry for
/convert/source/async submissions, or otherwise ensure post-submission failures
cannot resubmit the task without a server-supported deduplication key. Update
the call site in flows/components/docling_remote.py at line 386 and the
corresponding flow configuration in flows/ingestion_flow.json at line 710 as
needed to use the non-retrying submission path; all three sites must preserve
single-submission behavior.
---
Outside diff comments:
In `@flows/components/docling_remote.py`:
- Around line 278-294: Update the polling loops surrounding _get_with_retry in
flows/components/docling_remote.py (lines 278-294) and flows/ingestion_flow.json
(line 710) to catch exhausted transient status-check retries and continue
polling until max_poll_timeout, while preserving normal success/failure handling
and timeout behavior.
In `@src/services/docling_service.py`:
- Around line 233-274: Move semaphore acquisition in the Docling request flow to
immediately before the existing guarded try block, after _build_docling_options,
_get_auth_headers, data preparation, file setup, and _get_client complete. Keep
the existing release logic paired with that acquisition so preparation or
client-creation failures cannot consume a slot permanently.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: e53e1426-e4b1-43bc-9afe-61f785a64938
📒 Files selected for processing (7)
flows/components/docling_remote.pyflows/ingestion_flow.jsonsrc/config/settings.pysrc/services/docling_polling_service.pysrc/services/docling_service.pytests/unit/test_docling_service.pytests/unit/test_docling_service_status_check.py
| def _acquire_semaphore(cls, max_concurrency: int | None): | ||
| val = max_concurrency if max_concurrency is not None else 2 | ||
| with cls._semaphore_lock: | ||
| if cls._global_semaphore is None: | ||
| cls._global_semaphore = threading.Semaphore(val) | ||
| cls._global_semaphore.acquire() |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== docling_remote.py relevant slice ==\n'
wc -l flows/components/docling_remote.py
sed -n '1,140p' flows/components/docling_remote.py
printf '\n== ingestion_flow.json relevant slice ==\n'
# show around the embedded component code only if file exists
wc -l flows/ingestion_flow.json
sed -n '680,760p' flows/ingestion_flow.json
printf '\n== search for max_concurrency uses ==\n'
rg -n "max_concurrency|_acquire_semaphore|Semaphore\(" flows/components/docling_remote.py flows/ingestion_flow.jsonRepository: langflow-ai/openrag
Length of output: 40381
🏁 Script executed:
#!/bin/bash
set -euo pipefail
python3 - <<'PY'
import threading
for val in [2, 1, 0, -1, -5]:
print(f"val={val}")
try:
s = threading.Semaphore(val)
print(" created:", s)
# try nonblocking acquire to understand immediate behavior
got = s.acquire(blocking=False)
print(" acquire(blocking=False):", got)
except Exception as e:
print(" exception:", type(e).__name__, e)
PYRepository: langflow-ai/openrag
Length of output: 588
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== max_concurrency config block in ingestion_flow.json ==\n'
sed -n '832,860p' flows/ingestion_flow.json
printf '\n== search for IntInput definition / minimum validation ==\n'
rg -n "class IntInput|def .*IntInput|min_value|min=|minimum|greater than 0|max_concurrency" flows lfx . -g '!flows/ingestion_flow.json'Repository: langflow-ai/openrag
Length of output: 50376
Reject non-positive concurrency
max_concurrency == 0can hang_process_task_id()onSemaphore.acquire()and also makesThreadPoolExecutor(max_workers=0)invalid inprocess_files().max_concurrency < 0raises fromthreading.Semaphore(...).
Validate max_concurrency >= 1 before acquiring the semaphore, and mirror the same guard in the embedded component code.
📍 Affects 2 files
flows/components/docling_remote.py#L47-L52(this comment)flows/ingestion_flow.json#L710-L710
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@flows/components/docling_remote.py` around lines 47 - 52, Validate
max_concurrency is at least 1 before semaphore initialization or acquisition in
_acquire_semaphore, preserving the existing default when it is None. Apply the
same non-positive concurrency guard in flows/components/docling_remote.py lines
47-52 and flows/ingestion_flow.json line 710 so process_files and embedded
component execution reject invalid values consistently.
| @retry( | ||
| stop=stop_after_attempt(3), | ||
| wait=wait_exponential(multiplier=1, min=1, max=10), | ||
| retry=retry_if_exception(is_sync_transient_error), | ||
| reraise=True, | ||
| ) | ||
| def _post_with_retry(self, client: httpx.Client, url: str, **kwargs) -> httpx.Response: | ||
| response = client.post(url, **kwargs) | ||
| response.raise_for_status() | ||
| return response |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== file list =="
git ls-files | rg '(^|/)(flows/components/docling_remote\.py|flows/ingestion_flow\.json|README\.md|.*docling.*|.*ingestion.*)$' || true
echo "== docling_remote.py outline =="
ast-grep outline flows/components/docling_remote.py --view expanded || true
echo "== search idempotency / retry / docling =="
rg -n "idempot|retry_if_exception|is_sync_transient_error|/convert/source/async|Docling Serve|convert/source/async|task_id" flows/components/docling_remote.py flows/ingestion_flow.json README.md . || trueRepository: langflow-ai/openrag
Length of output: 50376
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== local docling references =="
rg -n -C 3 "convert/source/async|idempot|idempotency|task_id|Docling Serve|docling-serve|/v1/convert/source/async" \
README.md docs flows src tests scripts kubernetes plugins sdks 2>/dev/null || true
echo "== docling service/client code =="
rg -n -C 3 "/convert/source/async|task_id|idempot|retry|post.*async|convert/source" src flows tests 2>/dev/null || trueRepository: langflow-ai/openrag
Length of output: 50376
🌐 Web query:
Docling Serve /convert/source/async idempotency key retry timeout 5xx
💡 Result:
The Docling Serve /v1/convert/source/async endpoint is designed to facilitate asynchronous document conversion, returning a task descriptor upon submission [1][2]. Key technical details regarding your query include: Idempotency Key There is no explicit support for a client-provided idempotency key in the current Docling Serve API protocol [1][2]. The async endpoints return a unique task_id upon successful submission, which acts as the reference for polling status and fetching results [1][2]. Retry Policy for 5xx Errors The official Docling client SDK implements automatic retries with exponential backoff and jitter for transient errors, including 5xx status codes (such as 500, 502, and 503) and transport-level connection errors [3][4][5]. The default retry behavior is configured via the client's http_retries parameter, which typically defaults to 3 attempts [6][5]. Timeouts There are multiple layers of timeouts in the Docling Serve ecosystem: 1. Client-Side Timeouts: The Python client SDK allows users to configure http_connect_timeout (default 10.0s), http_read_timeout (default 60.0s), and a global job_timeout (default 300.0s) for waiting on async tasks [6][5]. Integrations (like Haystack) may expose their own default timeout settings, such as a 600s job timeout [7]. 2. Server-Side Timeouts: The Docling Serve backend manages various internal timeouts for its orchestration engine, including eng_ray_document_timeout (default 300s) and eng_ray_task_timeout (default 3600s), which govern the maximum allowed time for background extraction and task processing [8][9]. Synchronous requests also have a maximum wait time controlled by the DOCLING_SERVE_MAX_SYNC_WAIT configuration [10]. If you encounter 5xx errors or timeouts, ensure your client-side job_timeout is sufficiently long to accommodate the server-side processing time for your specific documents [6][7].
Citations:
- 1: https://docling-project.github.io/docling/usage/api_server/rest_api/
- 2: https://github.com/docling-project/docling-serve/blob/0276ef1f/docs/usage.md
- 3: docling-project/docling@bcd5509
- 4: https://pypi.org/project/docling-serve-client/1.14.0/
- 5: https://github.com/docling-project/docling/blob/f847b6cc/docling/service_client/_async_client.py
- 6: https://github.com/docling-project/docling/blob/aba7f155/docling/service_client/client.py
- 7: https://docs.haystack.deepset.ai/reference/integrations-docling_serve
- 8: Docling-Serve request timeout docling-project/docling#2226
- 9: https://github.com/docling-project/docling-serve/blob/main/docling_serve/settings.py
- 10: https://github.com/docling-project/docling-serve/blob/0ec67a37/docling_serve/app.py
Do not retry /convert/source/async automatically. The endpoint returns a fresh task_id on submission and doesn’t expose an idempotency key, so a read timeout or 5xx after acceptance can enqueue duplicate conversion tasks. Restrict this POST from retrying on post-submission failures, or add a server-side dedupe key if the API grows one.
flows/components/docling_remote.py#L70-L79flows/components/docling_remote.py#L386-L386flows/ingestion_flow.json#L710-L710
📍 Affects 2 files
flows/components/docling_remote.py#L70-L79(this comment)flows/components/docling_remote.py#L386-L386flows/ingestion_flow.json#L710-L710
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@flows/components/docling_remote.py` around lines 70 - 79, Remove automatic
retry behavior from _post_with_retry for /convert/source/async submissions, or
otherwise ensure post-submission failures cannot resubmit the task without a
server-supported deduplication key. Update the call site in
flows/components/docling_remote.py at line 386 and the corresponding flow
configuration in flows/ingestion_flow.json at line 710 as needed to use the
non-retrying submission path; all three sites must preserve single-submission
behavior.
1 participant
This pull request improves the reliability and concurrency control of remote Docling API calls in
DoclingRemoteComponent. The main changes include adding retry logic for transient HTTP errors, centralizing concurrency management with a global semaphore, and updating API calls to use the new retry and concurrency mechanisms.Reliability improvements:
tenacitylibrary, with exponential backoff, in new helper methods_get_with_retryand_post_with_retry. These are now used for all key HTTP requests instead of manual retry logic. [1] [2] [3] [4] [5]Concurrency control:
max_concurrency. Semaphore acquisition and release is handled in all major processing methods to prevent exceeding allowed parallelism. [1] [2] [3] [4]Code simplification:
These changes make remote processing more robust and prevent overloading the remote service with too many simultaneous requests.
Summary by CodeRabbit
Reliability
Performance
DOCLING_MAX_CONCURRENCY, defaulting to 2.Bug Fixes
Tests