Improve ray actors resilience - #195
Conversation
📝 WalkthroughWalkthroughAdds a new Indexer "serialize" concurrency group and a serialize_file Ray method, enables Ray actor restarts (max_restarts=5) for marker and serializer actors, adds related environment variables to docs and .env.example, and updates serializer error logging. Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes 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
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 |
|
@CodeRabbit review |
✅ Actions performedReview triggered.
|
![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
🤖 Fix all issues with AI agents
In @.hydra_config/config.yaml:
- Line 156: The env var name used for the index concurrency is inconsistent:
change the occurrence of INDEXER_index_CONCURRENCY to use the uppercase
convention (INDEXER_INDEX_CONCURRENCY) so it matches the other keys (e.g.,
INDEXER_UPDATE_CONCURRENCY); update the line that reads index:
${oc.decode:${oc.env:INDEXER_index_CONCURRENCY, 50}} to reference
${oc.env:INDEXER_INDEX_CONCURRENCY, 50} instead.
In @docs/content/docs/documentation/env_vars.md:
- Line 315: Add a table row documenting the missing environment variable
INDEXER_index_CONCURRENCY in the "Indexer Concurrency Groups" section: include
the name `INDEXER_index_CONCURRENCY`, type `int`, default `50`, and a short
description such as "Maximum concurrent indexing operations" so the docs match
the definition in .hydra_config/config.yaml.
🧹 Nitpick comments (2)
openrag/components/indexer/loaders/serializer.py (1)
23-26: Verify unlimited restarts strategy aligns with operational requirements.Setting
max_restarts=-1enables unlimited actor restarts, which increases fault tolerance but could mask recurring failures or lead to resource exhaustion if an actor repeatedly crashes. This is a significant operational decision.Ensure:
- Monitoring/alerting is in place to detect restart loops
- The root causes of actor failures are investigated rather than indefinitely restarted
- This aligns with your operational requirements for the Ray-based indexing system
The
max_task_retries=2limit is appropriate and prevents indefinite task retry loops.openrag/components/indexer/loaders/pdf_loaders/marker.py (1)
27-27: Addmax_task_retriestoMarkerWorkerandMarkerPoolfor consistency and task failure handling.While
max_restarts=-1improves actor resilience, bothMarkerWorker(line 27) andMarkerPool(line 149) lack task retry limits. TheDocSerializerinserializer.pyfollows a better pattern with bothmax_restarts=-1andmax_task_retries=2, limiting individual task retries while allowing actor restarts.Without
max_task_retries, tasks that consistently fail will retry indefinitely on these actors, potentially wasting resources. Consider addingmax_task_retries=2(or an appropriate value from config) to both decorators to align with the existingDocSerializerpattern and prevent unbounded resource consumption on failed tasks.Also applies to: 149-149
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
.env.example.hydra_config/config.yamldocs/content/docs/documentation/env_vars.mdopenrag/components/indexer/indexer.pyopenrag/components/indexer/loaders/pdf_loaders/marker.pyopenrag/components/indexer/loaders/serializer.pyopenrag/utils/dependencies.py
🔇 Additional comments (6)
.env.example (1)
44-45: LGTM! Clear documentation for optional configuration.The commented-out example with an explanatory comment is helpful for users who need to disable memory-based task killing. This aligns well with the documentation added in
env_vars.md.openrag/components/indexer/loaders/serializer.py (1)
90-92: LGTM! Enhanced exception logging improves debuggability.The improved exception handling now captures and logs the error string before re-raising, which will help with troubleshooting serialization failures. This aligns well with the PR's goal of improving error handling.
openrag/utils/dependencies.py (1)
11-15: Verify that broad exception handling doesn't mask unexpected errors.The change from
except ValueError:toexcept Exception:makes actor creation more resilient by catching all exceptions and falling back to creating a new actor. However, this broader scope could mask unexpected errors beyond the "actor not found" case thatValueErrorspecifically handles.Consider:
- Logging unexpected exceptions (non-ValueError) before creating a new actor
- This helps distinguish between normal actor-not-found scenarios vs. actual errors (network issues, permission problems, etc.)
🔍 Suggested enhancement for better observability
def get_or_create_actor(name, cls, namespace="openrag", **options): + from utils.logger import get_logger + logger = get_logger() try: return ray.get_actor(name, namespace=namespace) - except Exception: + except ValueError: + # Actor not found - create it return cls.options(name=name, namespace=namespace, **options).remote() + except Exception as e: + # Unexpected error - log and attempt recovery + logger.warning(f"Unexpected error getting actor {name}: {e}, creating new actor") + return cls.options(name=name, namespace=namespace, **options).remote()This approach provides better observability while maintaining the resilience benefits.
openrag/components/indexer/indexer.py (3)
24-35: LGTM! Concurrency group properly configured.The new "index" concurrency group is correctly added following the existing pattern for other groups. The configuration is properly wired to the config file with a sensible default of 50 concurrent operations.
59-59: LGTM! Appropriate concurrency group separation.Moving
add_fileto the "index" concurrency group is a good design choice. This method orchestrates the entire indexing flow (serialization, chunking, insertion) and should have separate concurrency control from the chunking operations it invokes.
94-101: LGTM! Excellent defensive error handling.The conditional check gracefully handles serialization failures by skipping chunking when no document is produced. The warning log provides visibility into the skip, which aids debugging. This prevents potential downstream errors and aligns well with the PR's resilience goals.
6884b73 to
b4d1f70
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In @openrag/components/indexer/loaders/pdf_loaders/marker.py:
- Line 27: The actor is configured with unlimited restarts via @ray.remote(...,
max_restarts=-1) which can cause tight restart loops on persistent failures;
change the MarkerWorker actor’s Ray options to use a bounded restart count
(e.g., max_restarts=3 or another reasonable finite number) instead of -1 and/or
add simple backoff logic inside the MarkerWorker constructor/startup path (use
try/except around initialization and sleep with exponential backoff before
re-raising to trigger a delayed restart) while keeping MARKER_NUM_GPUS
unchanged; update any tests or docs that assume unlimited restarts accordingly.
🧹 Nitpick comments (2)
openrag/utils/dependencies.py (1)
11-15: Consider logging the exception for better observability.Broadening the exception handling from
ValueErrortoExceptionimproves resilience by catching more failure modes when looking up Ray actors. However, silently catching all exceptions might hide unexpected errors during debugging.♻️ Optional: Add logging for better observability
+from utils.logger import get_logger + +logger = get_logger() + def get_or_create_actor(name, cls, namespace="openrag", **options): try: return ray.get_actor(name, namespace=namespace) - except Exception: + except Exception as e: + logger.debug(f"Actor {name} not found or unavailable, creating new: {e}") return cls.options(name=name, namespace=namespace, **options).remote()openrag/components/indexer/loaders/pdf_loaders/marker.py (1)
149-149: Consider the cascade effect of unlimited pool restarts.Setting
max_restarts=-1forMarkerPoolmeans each restart will reinitialize allMarkerWorkeractors (line 160), potentially causing expensive resource allocation and disruption. If the pool encounters a persistent issue, unlimited restarts could lead to repeated expensive reinitialization cycles.Consider whether a bounded restart limit with monitoring and alerting would be more appropriate for the pool coordinator.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
.hydra_config/config.yamldocs/content/docs/documentation/env_vars.mdopenrag/components/indexer/indexer.pyopenrag/components/indexer/loaders/pdf_loaders/marker.pyopenrag/components/indexer/loaders/serializer.pyopenrag/utils/dependencies.py
🚧 Files skipped from review as they are similar to previous changes (3)
- .hydra_config/config.yaml
- docs/content/docs/documentation/env_vars.md
- openrag/components/indexer/loaders/serializer.py
⏰ 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 (5)
openrag/utils/dependencies.py (2)
41-42: LGTM!The new
get_indexer()function follows the established pattern for actor creation and properly exposes the Indexer actor with a detached lifetime.
53-53: LGTM!Module-level initialization of the
indexeractor follows the existing pattern and provides convenient access to the Indexer actor throughout the codebase.openrag/components/indexer/indexer.py (3)
33-33: LGTM!Adding the "index" concurrency group enables dedicated concurrency control for indexing operations, consistent with the existing pattern for other operation types.
59-59: LGTM!Assigning
add_fileto the "index" concurrency group enables independent concurrency control for indexing operations, improving resource management and aligning with the PR's goal of strengthening concurrency control.
94-101: Good defensive programming for serialization failures.The conditional check properly handles cases where serialization returns no document, preventing downstream errors during chunking. The warning message clearly indicates the skip condition.
b4d1f70 to
a2309b6
Compare
a2309b6 to
e604ad6
Compare
| | `RAY_ENABLE_RECORD_ACTOR_TASK_LOGGING` | `number` | `1` | Enables logs at task level in the Ray dashboard for better debugging and monitoring. | | ||
| | `RAY_task_retry_delay_ms` | `number` | `3000` | Delay (in milliseconds) before retrying a failed task. Controls the wait time between retry attempts. | | ||
| | `RAY_ENABLE_UV_RUN_RUNTIME_ENV` | `number` | `0` | Controls UV runtime environment integration. **Critical**: Must be set to `0` when using the newest version of UV to avoid compatibility issues. | | ||
| |`RAY_memory_monitor_refresh_ms`| `number` | 250 ms | To control the frequency of memory usage checks and task or actor termination if needed. If you set this value to 0, task killing is disabled. | |
There was a problem hiding this comment.
Can you make it in all capitalized letter? So be consistant with the other config variables (except for RAY_task_retry_delay_ms , that should be eventually changed as well)
There was a problem hiding this comment.
Also, I don't see how this env var is used?
There was a problem hiding this comment.
Can you make it in all capitalized letter? So be consistant with the other config variables (except for RAY_task_retry_delay_ms , that should be eventually changed as well)
It's an innate RAY variable so we don't have control over the naming. I didn't used anywhere in my code; i documented it here cause according to the RAY documentation it can help with memory issues for example disable actor killing.
There was a problem hiding this comment.
I find it a bit weird to add vars from external lib just to document it exists, but I don't have a strong opposition
| |----------|------|---------|-------------| | ||
| | `INDEXER_DEFAULT_CONCURRENCY` | int | 1000 | Default concurrency limit for general operations | | ||
| | `INDEXER_UPDATE_CONCURRENCY` | int | 100 | Maximum concurrent document update operations | | ||
| | `INDEXER_INDEX_CONCURRENCY` | int | 50 | Maximum concurrent indexing operations | |
There was a problem hiding this comment.
I find the naming a bit confusing, since insert/update/delete are also indexing operations. In this case, this is more add or insert, but I see that we already have INDEXER_INSERT_CONCURRENCY (surprisingly set to a small default of 10). But from what I see, insert_documents is only used by add_file, so in the index concurrency group. Maybe we can:
- rename INDEXER_INDEX_CONCURRENCY into INDEXER_ADD_CONCURRENCY
- remove INDEXER_INDEX_INSERT_CONCURRENCY and the associated concurrency group?
There was a problem hiding this comment.
- INDEXER_INSERT_CONCURRENCY controls vdb insertion concurrency. The value is indeed small; i set it up to 100
There was a problem hiding this comment.
Here’s what’s happening: whenever a file is uploaded, add_file is triggered, which then calls serialize_file. This function relies on the DocSerializer Ray actor.
openrag/openrag/components/indexer/indexer.py
Lines 58 to 89 in 739a44c
Under heavy load, the DocSerializer tends to crash. My intuition is that this is due to a large number of calls to serializer.serialize_document, combined with the fact that these calls are not executed concurrently.
openrag/openrag/components/indexer/utils/files.py
Lines 74 to 80 in 739a44c
Introducing INDEXER_INDEX_CONCURRENCY (which has since been refactored and renamed to INDEXER_SERIALIZE_CONCURRENCY) helped mitigate these errors.
There was a problem hiding this comment.
Ok, makes more sense now, thank you
…ith tasks which Ray error prone.
e604ad6 to
897e487
Compare
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@openrag/components/indexer/indexer.py`:
- Around line 59-65: The serialize_file signature uses a mutable default for
metadata and an incorrect type hint for task_id; change the signature in
serialize_file to accept metadata: Optional[Dict] = None and task_id:
Optional[str] = None, and inside the function normalize metadata to a fresh dict
(e.g., metadata = dict(metadata or {}) or metadata = {} if None) before any
.update() or assignment so callers don't share state; update any references in
serialize_file and callers (e.g., eml_loader.py) to rely on the normalized
metadata variable.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (5)
.hydra_config/config.yamldocs/content/docs/documentation/env_vars.mdopenrag/components/indexer/indexer.pyopenrag/components/indexer/loaders/pdf_loaders/marker.pyopenrag/components/indexer/loaders/serializer.py
🚧 Files skipped from review as they are similar to previous changes (2)
- docs/content/docs/documentation/env_vars.md
- .hydra_config/config.yaml
🧰 Additional context used
🪛 Ruff (0.14.11)
openrag/components/indexer/indexer.py
63-63: Do not use mutable data structures for argument defaults
Replace with None; initialize within function
(B006)
64-64: PEP 484 prohibits implicit Optional
Convert to T | None
(RUF013)
⏰ 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 (6)
openrag/components/indexer/loaders/serializer.py (2)
23-23: Confirm Ray actor restart policy is intentional.
Bounded restarts are a good resilience improvement; please verify thatmax_restarts=5matches the desired failure-recovery behavior for this actor in Ray 2.47.1.
87-89: Improved exception logging context.
Including the exception detail in structured logs is helpful for debugging while preserving failure propagation.openrag/components/indexer/loaders/pdf_loaders/marker.py (2)
27-27: Bounded restart policy looks reasonable—please verify count.
Confirm thatmax_restarts=5is the intended resilience level forMarkerWorkerunder your failure scenarios.
159-159: MarkerPool restart policy aligned—verify desired behavior.
Please double-check thatmax_restarts=5is sufficient for pool availability targets in Ray 2.47.1.openrag/components/indexer/indexer.py (2)
100-113: Guarding chunking on serialization failure is a solid resilience win.
Skipping chunking when no document is produced prevents cascading errors.
33-33: No action required. Theserializeconcurrency group is already defined in.hydra_config/config.yaml:162with a default value of 50 (INDEXER_SERIALIZE_CONCURRENCY), so no KeyError will occur at runtime.
✏️ Tip: You can disable this entire section by setting review_details to false in your review settings.
2 participants
This PR improves the reliability and configurability of the Ray-based indexing system by strengthening concurrency control, worker resilience, and error handling.
Key changes:
indexconcurrency group for the Ray indexer, enabling finer-grained control over indexing workloads. Updated theIndexer.add_fileflow to use the new concurrency group and safely skip indexing when document serialization fails, with clearer logging.RAY_memory_monitor_refresh_msto allow control over Ray’s memory monitoring behavior.Overall, these updates make indexing more robust and better suited for concurrent and distributed workloads.
Summary by CodeRabbit
New Features
Documentation
Improvements
Bug Fixes
✏️ Tip: You can customize this high-level summary in your review settings.