fix: Prevent marker CPU issues from semaphore loops and orphaned processes - #193
Conversation
📝 WalkthroughWalkthroughAdds configurable timeout and batched concurrency for chunk contextualization, a Ray actor timeout utility, CPU-spin prevention in cleanup loops, per-call timeouts for PDF marker processing, expanded mock VLLM endpoints, config/env updates, and extended integration tests for indexing and tools APIs. Changes
Sequence DiagramsequenceDiagram
participant Client
participant ChunkContextualizer
participant BatchProcessor
participant RayActor
participant TimeoutUtil
Client->>ChunkContextualizer: contextualize_chunks(chunks)
ChunkContextualizer->>BatchProcessor: split into batches (MAX_CONCURRENT)
loop per batch
BatchProcessor->>RayActor: submit contextualization tasks
RayActor->>TimeoutUtil: run call with timeout (CONTEXTUALIZATION_TIMEOUT)
alt timeout
TimeoutUtil->>BatchProcessor: return warning / empty results
else success
TimeoutUtil->>BatchProcessor: return results
end
BatchProcessor->>ChunkContextualizer: aggregate batch results
end
ChunkContextualizer->>Client: return contextualized Documents
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Suggested labels
Suggested reviewers
Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧹 Recent nitpick comments
📜 Recent review detailsConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (3)
🚧 Files skipped from review as they are similar to previous changes (1)
🧰 Additional context used🧬 Code graph analysis (1)tests/api_tests/test_tools.py (2)
🪛 Ruff (0.14.11)tests/api_tests/test_indexer.py28-28: Avoid specifying long messages outside the exception class (TRY003) 36-36: Avoid specifying long messages outside the exception class (TRY003) 40-40: Avoid specifying long messages outside the exception class (TRY003) 49-49: Avoid specifying long messages outside the exception class (TRY003) ⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
🔇 Additional comments (11)
✏️ Tip: You can disable this entire section by setting 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 |
7c70ad3 to
444f775
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 0
🧹 Nitpick comments (2)
openrag/components/indexer/loaders/pdf_loaders/marker.py (1)
198-207: Good pool-level timeout implementation.The
asyncio.wait_forwrapper prevents indefinite hangs at the pool level. However, consider usingself.logger.exception()instead ofself.logger.error()on line 206 to capture the full traceback for debugging timeout issues.🔧 Suggested improvement for better debugging
except asyncio.TimeoutError: - self.logger.error(f"MarkerPool timeout for {file_path}") + self.logger.exception(f"MarkerPool timeout for {file_path} after {timeout}s") raiseopenrag/components/utils.py (1)
64-64: Makemax_restartsconfigurable via config object for consistency.The change to
max_restarts=5appropriately prevents unbounded restart loops that could exhaust resources. However, the codebase establishes a pattern of externalizing Ray parameters through the config object (e.g.,config.ray.indexer.max_task_retries,config.ray.serve.num_replicas,config.ray.semaphore.concurrency). Consider addingmax_restartsas a configurable parameter inconfig.ray.semaphore.max_restartsrather than hardcoding it to5, enabling operators to tune this value for their production workload.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
docker-compose.yamldocs/content/docs/documentation/env_vars.mdopenrag/components/indexer/chunker/chunker.pyopenrag/components/indexer/loaders/pdf_loaders/marker.pyopenrag/components/utils.py
🧰 Additional context used
🪛 Ruff (0.14.10)
openrag/components/indexer/chunker/chunker.py
72-72: Consider moving this statement to an else block
(TRY300)
openrag/components/indexer/loaders/pdf_loaders/marker.py
155-156: try-except-pass detected, consider logging the exception
(S110)
155-155: Do not catch blind exception: Exception
(BLE001)
204-204: Consider moving this statement to an else block
(TRY300)
206-206: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: index-backup-restore
🔇 Additional comments (10)
openrag/components/utils.py (2)
57-61: LGTM - Sleep prevents CPU spin in cleanup loop.The
time.sleep(0.001)addition correctly prevents the busy-wait loop from consuming 100% CPU during cleanup. However, consider whether the sleep should also occur whenlocked()returnsFalseto handle edge cases where the semaphore state changes rapidly.
75-78: LGTM - Consistent sleep pattern in distributed actor cleanup.The sleep addition mirrors the fix in
LLMSemaphore.cleanup(), maintaining consistency across both semaphore implementations.docker-compose.yaml (2)
39-39: Good addition to reduce idle CPU usage.
VLLM_SLEEP_WHEN_IDLE=1is a valid VLLM environment variable that helps reduce CPU consumption when the model is not actively processing requests.
156-158: Good fix for network naming consistency.Explicitly naming the network
openrag_defaultprevents "network still in use" errors duringdocker-compose downby ensuring consistent network identification across compose operations.docs/content/docs/documentation/env_vars.md (1)
103-104: LGTM - Clear documentation for new configuration variables.The documentation accurately describes the two new environment variables with correct types, defaults, and descriptions that match the implementation in
chunker.py.openrag/components/indexer/chunker/chunker.py (3)
19-24: LGTM - Well-structured configuration with sensible defaults.The configuration variables are clearly documented and use reasonable defaults (120s timeout, batch size of 10) that balance throughput with resource protection.
68-83: Good timeout handling with proper exception differentiation.The
asyncio.wait_forwrapper with specificTimeoutErrorhandling is well-implemented. Both timeout and general exceptions gracefully degrade by returning an empty context string, maintaining system resilience.Regarding the static analysis hint (TRY300): moving
return output.contentto anelseblock is a minor style improvement but not required here since the code is clear and correct.
96-118: Well-implemented batch processing for concurrent LLM calls.The batching logic correctly limits concurrent requests using
MAX_CONCURRENT_CONTEXTUALIZATION. The progress indicator showing batch ranges (e.g.,[1-10/50]) provides good visibility during processing.One minor observation: if
MAX_CONCURRENT_CONTEXTUALIZATIONis set very high via config, this batching provides no protection. Consider adding a sanity check or upper bound.openrag/components/indexer/loaders/pdf_loaders/marker.py (2)
69-72: Correct cleanup order: terminate before join.The reordering to
close() → terminate() → join()is the proper sequence for forceful pool shutdown. This prevents the previous issue wherejoin()could block indefinitely waiting for workers that might be stuck.
148-156: Good addition of destructor for resource cleanup.The
__del__method provides best-effort cleanup of the multiprocessing pool when the actor is destroyed, helping prevent orphaned processes.Regarding the static analysis hints: the bare
except Exception: passis acceptable in__del__since logging infrastructure may be unavailable during interpreter shutdown. The comment "Best effort cleanup" adequately documents the intent.
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@openrag/components/indexer/chunker/chunker.py`:
- Around line 71-88: The code currently only catches openai.APITimeoutError but
asyncio.wait_for raises asyncio.TimeoutError (or TimeoutError on Py3.11+), so
add explicit handlers for asyncio.TimeoutError and builtins.TimeoutError before
the generic Exception; update the try/except around the await
self.context_generator.ainvoke(messages) call to catch (asyncio.TimeoutError,
TimeoutError) and log the same timeout message using CONTEXTUALIZATION_TIMEOUT
and filename (similar to the existing openai.APITimeoutError handler), then keep
the generic Exception handler for other errors so timeouts are not lumped into
the generic error path.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (2)
.hydra_config/chunker/base.yamlopenrag/components/indexer/chunker/chunker.py
🧰 Additional context used
🪛 Ruff (0.14.11)
openrag/components/indexer/chunker/chunker.py
75-75: Consider moving this statement to an else block
(TRY300)
82-82: Do not catch blind exception: Exception
(BLE001)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (1)
- GitHub Check: index-backup-restore
🔇 Additional comments (4)
.hydra_config/chunker/base.yaml (1)
2-3: LGTM!The new configuration keys for
contextualization_timeoutandmax_concurrent_contextualizationare well-structured and follow the existing pattern. The defaults (120s and 10 concurrent) are reasonable for LLM operations.openrag/components/indexer/chunker/chunker.py (3)
1-25: LGTM!The imports and constants are well-defined. Using
config.chunker.get()with fallback defaults ensures graceful handling if config keys are missing.
36-39: LGTM!Good practice to copy the config dict before modifying it to avoid side effects on the original.
90-139: LGTM!The batch processing implementation is well-structured:
- Correctly limits concurrent LLM calls to
MAX_CONCURRENT_CONTEXTUALIZATION- Maintains proper
prev_chunkscontext calculation across batch boundaries- Progress reporting clearly shows batch ranges
✏️ Tip: You can disable this entire section by setting review_details to false in your review settings.
284975d to
ca2f571
Compare
ca2f571 to
2849eb4
Compare
|
|
||
|
|
||
| @ray.remote(max_restarts=-1, max_concurrency=config.ray.semaphore.concurrency) | ||
| @ray.remote(max_restarts=5, max_concurrency=config.ray.semaphore.concurrency) |
There was a problem hiding this comment.
I saw in this PR's description that max_restarts=-1 is potentially prone to restart storms, hence this change. I wonder if that actually happened in production.
I’ve prepared a PR #195 to improve Ray actors’ resilience and have added max_restarts=-1 in some actors. The intuition being to reboot actors whenever they are dead to ensure service continuity.
There was a problem hiding this comment.
I don't know if it's legitimate to try to reboot ray actors potentially forever? In my view, this is dangerous, and might be a reason of what we witnessed with processes loops
There was a problem hiding this comment.
Then for safety we can limit it at 5 and we will see in practice
- Add time.sleep(0.001) to semaphore cleanup loops to prevent 100% CPU usage when cleanup() is called while semaphore is locked - Change DistributedSemaphoreActor max_restarts from -1 (infinite) to 5 to prevent restart storms if the actor keeps crashing These issues could cause sudden CPU spikes in production. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Fix multiprocessing pool cleanup order: terminate() before join() to prevent blocking indefinitely on stuck workers - Add __del__ method to MarkerWorker to clean up multiprocessing pool when the Ray actor is destroyed, preventing orphaned processes - Add asyncio.wait_for() timeout at MarkerPool level to prevent indefinite hangs on remote calls These changes prevent CPU accumulation from orphaned processes and ensure proper cleanup of resources. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Prevents CPU exhaustion when indexing many files simultaneously: - Add 120s timeout to individual LLM contextualization calls - Process chunks in batches of 10 instead of all at once - Both values are configurable via chunker config This addresses the root cause of the Dec 30 incident where 131 files indexed in 2 minutes created 2600+ concurrent LLM tasks. 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
Enable vLLM sleep mode to prevent 100% CPU usage when the server is idle. This addresses the busy-wait loop in vLLM's shared memory broadcast mechanism (PR #16226). 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
… its reponse in another time, we use its inherent openai.APITimeoutError
Document the new CONTEXTUALIZATION_TIMEOUT and MAX_CONCURRENT_CONTEXTUALIZATION environment variables introduced in commit 184a34f to prevent CPU exhaustion during batch indexing. Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Add configurable SERIALIZE_TIMEOUT from config (default 3600s) - Fix bug: use ray.get() instead of awaiting ObjectRef directly - Handle asyncio.CancelledError to cancel Ray task on caller cancellation - Handle RayTaskError with proper error wrapping Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- Add ray_utils.py with call_ray_actor_with_timeout() helper - Provides consistent timeout via ray.wait() with proper ray.cancel() - Handles asyncio.CancelledError to propagate cancellation to Ray tasks - Handles TaskCancelledError and RayTaskError uniformly Updated call sites: - serialize_file: use helper instead of inline implementation - MarkerPool.process_pdf: replace asyncio.wait_for() with helper - MarkerLoader.aload_document: add timeout/cancellation handling Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
- TestMarkerPDFIndexing: PDF upload, document creation, task state transitions - TestExtractTextPDF: extractText tool with PDF files - TestPDFErrorHandling: Invalid PDF graceful handling Tests verify the full pipeline: upload -> MarkerLoader -> chunking -> storage Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
| self.logger.exception( | ||
| "Error processing PDF with MarkerWorker", error=str(e) | ||
| future = worker.process_pdf.remote(file_path) | ||
| return await call_ray_actor_with_timeout( |
There was a problem hiding this comment.
Sorry, i've a doubt here: call_ray_actor_with_timeout raises errors. Shouldn't we catch them here or not with try...catch block?
There was a problem hiding this comment.
The error will be catched by the caller, here aload_document
- Fix import path in files.py (components.ray_utils) - Add trailing slashes to BASE_URL/VLM_BASE_URL in docker-compose - Add /v1/completions text completion endpoint to mock VLLM - Add QUEUED to valid task states in PDF tests - Fix tools endpoint path (/v1/tools/execute) - Fix tool parameter format (JSON object) - Add 404 retry handling for task status polling Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
eadf1ad to
03a45db
Compare
|
I have no further suggestions. I’ll approve once the API tests pass. |
- Add full task state machine (QUEUED → SERIALIZING → CHUNKING → INSERTING → COMPLETED) - Document DocSerializer, MarkerPool, MarkerWorker Ray actors - Add tools.py router with extractText endpoint format - Document integration tests in tests/api_tests/ and act command - Document mock VLLM for CI testing - Add Ray actor timeout/cancellation utility pattern (components/ray_utils.py) - Add import conventions (absolute imports from openrag/) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 Fix all issues with AI agents
In `@tests/api_tests/test_marker_pdf.py`:
- Around line 99-101: The test extracts task_id from response.json() via
data.get("task_status_url", "").split("/")[-1] or data.get("task_id") which can
yield an empty string; update the test around that extraction to validate and
fail fast: derive task_id from data.get("task_status_url") if present else
data.get("task_id"), then assert that task_id is truthy (non-empty) before
calling wait_for_task(api_client, task_id); if the assertion fails, include a
clear message referencing task_status_url/task_id so the test fails rather than
polling an invalid endpoint.
♻️ Duplicate comments (1)
openrag/components/utils.py (1)
64-64: Acknowledged:max_restartslimited to 5.This change from
max_restarts=-1tomax_restarts=5was already discussed and agreed upon in previous comments to prevent potential restart storms.
🧹 Nitpick comments (5)
openrag/components/indexer/chunker/chunker.py (1)
76-88: Consider catchingTimeoutErroralongsideopenai.APITimeoutError.While
openai.APITimeoutErrorhandles the OpenAI client's internal timeout, if any future code paths introduceasyncio.wait_for()wrappers or if the underlying httpx client raises a standardTimeoutError, it would be caught by the genericExceptionhandler instead of getting the timeout-specific logging.For defensive coding, consider:
Suggested improvement
- except openai.APITimeoutError: + except (openai.APITimeoutError, TimeoutError): logger.warning( f"OpenAI API timeout contextualizing chunk after {CONTEXTUALIZATION_TIMEOUT}s", filename=filename, ) return ""openrag/components/indexer/loaders/pdf_loaders/marker.py (1)
148-156: Consider logging exceptions at debug level for observability.While suppressing exceptions in
__del__is reasonable for best-effort cleanup, logging at debug level would help diagnose resource cleanup issues in production without affecting behavior.♻️ Suggested improvement
def __del__(self): """Clean up multiprocessing pool on actor destruction""" if self.pool: try: self.pool.close() self.pool.terminate() self.pool.join() except Exception: - pass # Best effort cleanup + # Best effort cleanup - log for debugging + try: + self.logger.debug("Pool cleanup in __del__ encountered an error") + except Exception: + pass # Logger might not be availableopenrag/components/ray_utils.py (2)
39-41: Minor: Prefix unused variable with underscore.The
pendingvariable is unused; prefixing with_signals intent and silences linters.- ready, pending = await asyncio.to_thread( + ready, _pending = await asyncio.to_thread( ray.wait, [future], num_returns=1, timeout=timeout )
38-45: Add handling forRayActorErrorto cover actor crashes.The function calls
ray.get()on line 45, which raisesRayActorErrorwhen the actor itself dies (e.g., OOM, process crash). Currently, onlyRayTaskErroris caught, leaving actor failures unhandled. This should be caught alongsideRayTaskErrorto provide consistent error handling.♻️ Suggested improvement
-from ray.exceptions import RayTaskError, TaskCancelledError +from ray.exceptions import RayTaskError, TaskCancelledError, RayActorError # ... in the function: except RayTaskError as e: raise RuntimeError(f"{task_description} failed") from e + + except RayActorError as e: + raise RuntimeError(f"{task_description} failed: actor died") from etests/api_tests/test_marker_pdf.py (1)
213-218: Consider stricter assertion for invalid PDF handling.Accepting
COMPLETEDfor an invalid PDF file seems permissive. If the system can gracefully recover from invalid PDFs (e.g., empty document), this is fine. Otherwise, the test should expectFAILEDspecifically.- # Either state is acceptable - task should not hang - assert state in ["COMPLETED", "FAILED"], f"Task stuck in state: {state}" + # Invalid PDF should fail, but main goal is to verify no hang + assert state == "FAILED", f"Expected FAILED for invalid PDF, got: {state}"If graceful degradation is intended, add a comment explaining why
COMPLETEDis acceptable for invalid PDFs.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (15)
.github/workflows/api_tests/docker-compose.yaml.github/workflows/api_tests/mock_vllm.py.hydra_config/chunker/base.yaml.hydra_config/config.yamlCLAUDE.mddocker-compose.yamldocs/content/docs/documentation/env_vars.mdopenrag/components/indexer/chunker/chunker.pyopenrag/components/indexer/loaders/pdf_loaders/marker.pyopenrag/components/indexer/utils/files.pyopenrag/components/ray_utils.pyopenrag/components/utils.pypytest.initests/api_tests/test_indexer.pytests/api_tests/test_marker_pdf.py
🚧 Files skipped from review as they are similar to previous changes (4)
- .hydra_config/chunker/base.yaml
- docs/content/docs/documentation/env_vars.md
- tests/api_tests/test_indexer.py
- .hydra_config/config.yaml
🧰 Additional context used
🧬 Code graph analysis (4)
openrag/components/ray_utils.py (1)
openrag/utils/logger.py (1)
get_logger(10-50)
openrag/components/indexer/utils/files.py (2)
openrag/config/config.py (1)
load_config(12-29)openrag/components/ray_utils.py (1)
call_ray_actor_with_timeout(11-57)
tests/api_tests/test_marker_pdf.py (1)
tests/api_tests/conftest.py (2)
api_client(16-19)created_partition(87-96)
openrag/components/indexer/loaders/pdf_loaders/marker.py (1)
openrag/components/ray_utils.py (1)
call_ray_actor_with_timeout(11-57)
🪛 LanguageTool
CLAUDE.md
[uncategorized] ~119-~119: The official name of this software platform is spelled with a capital “H”.
Context: ...s.yml --bind ``` Mock VLLM for CI: .github/workflows/api_tests/mock_vllm.py provi...
(GITHUB)
🪛 Ruff (0.14.11)
openrag/components/ray_utils.py
39-39: Unpacked variable pending is never used
Prefix it with an underscore or any other dummy variable pattern
(RUF059)
44-44: Avoid specifying long messages outside the exception class
(TRY003)
57-57: Avoid specifying long messages outside the exception class
(TRY003)
openrag/components/indexer/chunker/chunker.py
82-82: Do not catch blind exception: Exception
(BLE001)
tests/api_tests/test_marker_pdf.py
43-43: Avoid specifying long messages outside the exception class
(TRY003)
51-51: Avoid specifying long messages outside the exception class
(TRY003)
55-55: Avoid specifying long messages outside the exception class
(TRY003)
openrag/components/indexer/loaders/pdf_loaders/marker.py
155-156: try-except-pass detected, consider logging the exception
(S110)
155-155: Do not catch blind exception: Exception
(BLE001)
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (2)
- GitHub Check: api-tests
- GitHub Check: index-backup-restore
🔇 Additional comments (21)
pytest.ini (1)
12-16: LGTM!Good addition of test markers for categorizing integration vs unit tests. This enables selective test runs (e.g.,
pytest -m unitto skip integration tests requiring external services).docker-compose.yaml (1)
39-39: LGTM!Good addition of
VLLM_SLEEP_WHEN_IDLE=1to prevent vLLM from consuming 100% CPU when idle. This directly addresses the production CPU usage concerns mentioned in the PR objectives.openrag/components/utils.py (2)
57-61: LGTM - CPU spin prevention.Adding
time.sleep(0.001)inside the cleanup loop prevents busy-waiting and addresses the CPU usage issue. The 1ms sleep is short enough to not significantly delay cleanup while effectively yielding CPU time.
75-78: LGTM - Consistent CPU spin prevention in distributed actor.Same beneficial pattern applied to
DistributedSemaphoreActor.cleanup()for consistency with the local semaphore cleanup.openrag/components/indexer/chunker/chunker.py (3)
23-28: LGTM - Configurable timeout and concurrency limits.Good addition of
CONTEXTUALIZATION_TIMEOUTandMAX_CONCURRENT_CONTEXTUALIZATIONwith sensible defaults. These constants help prevent system overload and indefinite hangs during LLM calls.
40-42: LGTM - Timeout propagated to ChatOpenAI client.Setting the timeout in the ChatOpenAI configuration ensures requests don't hang indefinitely, which aligns with the PR objective of preventing CPU issues from hung processes.
103-125: LGTM - Batched contextualization prevents system overload.The batching approach with
MAX_CONCURRENT_CONTEXTUALIZATIONeffectively limits concurrent LLM requests, preventing resource exhaustion. The progress reporting per batch is a nice UX improvement..github/workflows/api_tests/docker-compose.yaml (1)
88-99: LGTM - Consolidated mock service configuration.Good refactoring to route all LLM/embedding requests through the single mock-vllm service. Using
EMPTYas API key and descriptive model names (mock-chat-model,mock-vlm-model) clearly indicates this is a test environment.openrag/components/indexer/loaders/pdf_loaders/marker.py (3)
69-72: LGTM! Correct pool cleanup ordering.The change to
close()→terminate()→join()is the right approach for preventing stuck workers from blocking the main process. This ensures forceful cleanup happens before waiting.
200-206: Verify nested timeout behavior.The timeout is applied at both
MarkerPool.process_pdfandMarkerLoader.aload_documentlevels with the same value. The outer timeout (inaload_document) covers queue wait + health check + inner timeout, so the inner timeout will typically trigger first if the worker is slow.This pattern appears intentional based on past review discussions, but consider whether the outer timeout should be slightly larger to account for queue/health check overhead.
224-239: LGTM! Proper timeout handling with descriptive task message.The timeout wrapper correctly enforces time limits and provides useful context in the task description for debugging timeout failures.
openrag/components/indexer/utils/files.py (2)
9-14: LGTM! Consistent timeout configuration.The module-level config loading and timeout constant follow the same pattern used in
marker.py. The default of 3600 seconds (1 hour) provides a reasonable fallback.
74-86: LGTM! Clean refactoring to use the shared timeout utility.The serialization now consistently uses
call_ray_actor_with_timeoutwith proper task description for debugging. The lazy import pattern avoids potential circular import issues.openrag/components/ray_utils.py (1)
11-57: Well-designed utility with comprehensive error handling.The centralized timeout handling with proper cancellation propagation is a solid improvement. The docstring accurately documents the exception types that callers should expect.
.github/workflows/api_tests/mock_vllm.py (3)
114-128: LGTM! Deterministic fake embeddings and token counting.The MD5-based embedding generation provides reproducible test results. The approximate token counting (4 chars/token) is reasonable for mock purposes.
131-156: LGTM! Context-aware mock response generation.The function properly handles both standard chat messages and vision model requests with list-based content. The keyword-based responses provide meaningful test data.
195-252: LGTM! Well-structured mock endpoints following OpenAI API conventions.The chat and text completion endpoints properly implement the expected response formats with appropriate token usage tracking. Good addition for comprehensive CI testing.
tests/api_tests/test_marker_pdf.py (2)
28-55: LGTM! Robust task polling with 404 handling.The
wait_for_taskutility correctly handles the race condition where tasks may not be immediately registered. The polling approach with clear timeout is appropriate for integration tests.
148-180: LGTM! Good coverage of text extraction functionality.The tests verify both successful extraction and that meaningful content is returned. The assertions appropriately check response structure without being too specific about content.
CLAUDE.md (2)
60-180: Excellent documentation updates that align with the PR changes.The documentation comprehensively covers:
- TaskStateManager state flow (line 60)
- New MarkerPool/MarkerWorker components for PDF processing (lines 62-63)
- Ray actor timeout handling pattern (lines 135-153) that supports the CPU issue fixes
- Testing infrastructure with act and mock VLLM (lines 109-120)
- Clear import conventions (lines 169-180)
All code examples are syntactically correct, and the import paths are consistent with the established conventions. This will be very helpful for developers working with the timeout and marker pool changes introduced in this PR.
119-119: Minor typo: Capitalize "GitHub" properly.The official name is "GitHub" with a capital "H".
📝 Proposed fix
-**Mock VLLM for CI:** `.github/workflows/api_tests/mock_vllm.py` provides fake embeddings and completions endpoints for testing without a real LLM. +**Mock VLLM for CI:** `.GitHub/workflows/api_tests/mock_vllm.py` provides fake embeddings and completions endpoints for testing without a real LLM.Based on static analysis.
Likely an incorrect or invalid review comment.
✏️ Tip: You can disable this entire section by setting review_details to false in your review settings.
c2a5cc8 to
3504927
Compare
- Include test_file.pdf with 2 pages for PDF processing tests
d23538e to
4bada3c
Compare
2 participants
Summary
Fixes production issue where marker consumes all CPU after running for a while.
Root causes identified and fixed:
utils.py): Thecleanup()methods hadwhileloops without any sleep, causing 100% CPU spin when called while semaphore is lockedutils.py):max_restarts=-1allowed infinite restarts, potentially causing restart stormsmarker.py):join()was called beforeterminate(), which could block indefinitely on stuck workersmarker.py): Missing__del__method meant orphaned multiprocessing pool processes could accumulatemarker.py): Remote calls could hang indefinitely without timeout protectionChanges
openrag/components/utils.pytime.sleep(0.001)to semaphore cleanup loops to prevent CPU spinmax_restarts=-1tomax_restarts=5to prevent restart stormsopenrag/components/indexer/loaders/pdf_loaders/marker.pyclose()→terminate()→join()__del__method to clean up multiprocessing pool on actor destructionasyncio.wait_for()timeout wrapper at MarkerPool levelTest plan
ps aux | grep pythonshows stable process count🤖 Generated with Claude Code
Summary by CodeRabbit
New Features
Performance & Optimization
Tests
Chores
✏️ Tip: You can customize this high-level summary in your review settings.