[core][taskEvents out of GCS][5/n] In-memory task event store implementation on task events head by karticam · Pull Request #65123 · ray-project/ray

karticam · 2026-07-30T01:57:21Z

Part of the effort to move task events out of GCS.
This PR builds on top of #65057

This is the first part of implementing GcsTaskManager inside task_events_head
Changes done:

ray_event_converter.py: the incoming requests are AddEventsRequest but the storage stores TaskEvents, and dropped_task_attempts. This layer converts the incoming requests to storage friendly format.
gc_policy.py: FinishedTaskActorTaskGcPolicy: 3 priority tiers (finished tasks, actor, others) that decide eviction order of task events when needed. Mimics the existing policy in GcsTaskManager.
task_event_storage.py: Based on TaskEventStorage in GcsTaskManager.

The entire storage is organized in three tiered lists (actually implemented using insertion ordered-dicts keyed by taskAttempt in python). For every task attempt, we hold one task event entry, which goes to one of the tiers based on whether the task attempt is finished, is actor task or none of this.
When more task events for an attempt arrive, they are merged with an existing entry and moved across tiers if required.
Task attempts for which events are dropped (either at the source level i.e. core_worker level, or due to eviction here) are maintained in dropped_task_attempts per job.
Priority-tiered eviction once over MAX_NUM_TASK_EVENTS (100k), evicting the oldest in the lowest-priority tier first.
dropped_task_attempts maintained per job which garbage when the job finishes. If per job list grows past a threshold, its pruned every 5 seconds.
Tracks the counters (stored / reported / attempts-dropped / profile-dropped / per-type). These will be emitted in a followup PR.

Followup PRs:

Changing task states based on worker death and job finished notification. Receiving those notifications is already wired up.
Metric emissions.
Serving state API from task events head.

gemini-code-assist

Code Review

This pull request introduces a new in-memory task event storage and reconciliation system on the Ray dashboard head, including the TaskEventsHead module, a priority-based garbage collection policy, and an HTTP publisher client to send events from aggregator agents. The review feedback is highly constructive, highlighting critical areas for improvement: optimizing set eviction in TaskEventStorage to avoid blocking the event loop, adding a sleep delay in subscription loops to prevent tight-spinning on persistent failures, wrapping the event processing pipeline in a try-except block for robust error handling, and addressing a potential memory leak in job summary tracking.

gemini-code-assist · 2026-07-30T01:59:13Z

+        to_evict = list(self._dropped_task_attempts)[:num_to_evict]
+        self._dropped_task_attempts.difference_update(to_evict)


Converting the entire _dropped_task_attempts set (which can contain up to 1,000,000 elements) to a list just to slice the first num_to_evict elements is highly inefficient and will block the single-threaded asyncio event loop. Instead, use an iterator to retrieve only the required number of elements in O(num_to_evict) time.

Suggested change

to_evict = list(self._dropped_task_attempts)[:num_to_evict]

self._dropped_task_attempts.difference_update(to_evict)

iterator = iter(self._dropped_task_attempts)

to_evict = [next(iterator) for _ in range(num_to_evict)]

self._dropped_task_attempts.difference_update(to_evict)

gemini-code-assist · 2026-07-30T01:59:13Z

+        while True:
+            try:
+                for _, worker_delta in await subscriber.poll(
+                    batch_size=_SUBSCRIBER_POLL_BATCH_SIZE
+                ):
+                    self._handle_worker_delta(worker_delta)
+            except Exception:
+                logger.exception("Failed handling worker-death notifications.")


If subscriber.poll or the message handling raises an exception (e.g., due to GCS connection issues), this loop will tight-spin and spam the logs at maximum speed, leading to high CPU usage and log disk exhaustion. Adding a short sleep (e.g., asyncio.sleep(1)) in the except block prevents tight-looping during persistent failures.

Suggested change

while True:

try:

for _, worker_delta in await subscriber.poll(

batch_size=_SUBSCRIBER_POLL_BATCH_SIZE

):

self._handle_worker_delta(worker_delta)

except Exception:

logger.exception("Failed handling worker-death notifications.")

while True:

try:

for _, worker_delta in await subscriber.poll(

batch_size=_SUBSCRIBER_POLL_BATCH_SIZE

):

self._handle_worker_delta(worker_delta)

except Exception:

logger.exception("Failed handling worker-death notifications.")

await asyncio.sleep(1)

gemini-code-assist · 2026-07-30T01:59:13Z

+        while True:
+            try:
+                for _, job_data in await subscriber.poll(
+                    batch_size=_SUBSCRIBER_POLL_BATCH_SIZE
+                ):
+                    self._handle_job_update(job_data)
+            except Exception:
+                logger.exception("Failed handling job-finished notifications.")


If subscriber.poll or the message handling raises an exception (e.g., due to GCS connection issues), this loop will tight-spin and spam the logs at maximum speed, leading to high CPU usage and log disk exhaustion. Adding a short sleep (e.g., asyncio.sleep(1)) in the except block prevents tight-looping during persistent failures.

Suggested change

while True:

try:

for _, job_data in await subscriber.poll(

batch_size=_SUBSCRIBER_POLL_BATCH_SIZE

):

self._handle_job_update(job_data)

except Exception:

logger.exception("Failed handling job-finished notifications.")

while True:

try:

for _, job_data in await subscriber.poll(

batch_size=_SUBSCRIBER_POLL_BATCH_SIZE

):

self._handle_job_update(job_data)

except Exception:

logger.exception("Failed handling job-finished notifications.")

await asyncio.sleep(1)

gemini-code-assist · 2026-07-30T01:59:13Z

+        body = await request.read()
+        try:
+            add_events_request = self._deserialize_request(body)
+        except Exception as e:
+            logger.warning(f"Failed to deserialize task events request: {e}")
+            return dashboard_optional_utils.rest_response(
+                status_code=dashboard_utils.HTTPStatusCode.INTERNAL_ERROR,
+                message=f"Failed to deserialize task events request: {e}",
+            )
+
+        task_events, dropped_task_attempts = convert_to_task_events(add_events_request)
+        self._store.record_data_loss_from_worker(dropped_task_attempts)
+        for task_event in task_events:
+            self._store.add_or_replace_task_event(task_event)


The parsing, conversion, and storage of task events are executed outside of the try-except block. If convert_to_task_events or any of the store methods raise an exception (e.g., ValueError from invalid IDs or AssertionError from unsupported event types), it will propagate unhandled, causing the HTTP request to fail with a 500 error without graceful logging. Wrapping the entire processing pipeline in the try-except block ensures robust error handling.

Suggested change

body = await request.read()

try:

add_events_request = self._deserialize_request(body)

except Exception as e:

logger.warning(f"Failed to deserialize task events request: {e}")

return dashboard_optional_utils.rest_response(

status_code=dashboard_utils.HTTPStatusCode.INTERNAL_ERROR,

message=f"Failed to deserialize task events request: {e}",

)

task_events, dropped_task_attempts = convert_to_task_events(add_events_request)

self._store.record_data_loss_from_worker(dropped_task_attempts)

for task_event in task_events:

self._store.add_or_replace_task_event(task_event)

body = await request.read()

try:

add_events_request = self._deserialize_request(body)

task_events, dropped_task_attempts = convert_to_task_events(add_events_request)

self._store.record_data_loss_from_worker(dropped_task_attempts)

for task_event in task_events:

self._store.add_or_replace_task_event(task_event)

except Exception as e:

logger.warning(f"Failed to process task events request: {e}")

return dashboard_optional_utils.rest_response(

status_code=dashboard_utils.HTTPStatusCode.INTERNAL_ERROR,

message=f"Failed to process task events request: {e}",

)

gemini-code-assist · 2026-07-30T01:59:13Z

+        self._task_index: Dict[bytes, Set[TaskAttempt]] = {}
+        self._job_index: Dict[bytes, Set[TaskAttempt]] = {}
+        self._worker_index: Dict[bytes, Set[TaskAttempt]] = {}
+        self._job_task_summary: Dict[bytes, JobTaskSummary] = {}


The _job_task_summary dictionary grows indefinitely as new jobs are executed, because there is currently no mechanism to clean up or evict summaries for finished jobs. Although JobTaskSummary.on_job_ends is defined and tested, it is never called in production code. Consider implementing a cleanup mechanism (e.g., when job finished notifications are processed) to prevent a memory leak over time.

this is taken care of in future PRs. This is the first PR just introducing stuff.

this is done in PR 6/n (#65141)

sampan-s-nayak · 2026-08-02T06:50:25Z

+from ray.core.generated.common_pb2 import TaskStatus, TaskType
+
+
+def is_task_finished(task_event: gcs_pb2.TaskEvents) -> bool:


can this be a private member function instead of a module level function if its not being used anywhere?

sampan-s-nayak · 2026-08-02T06:50:32Z

+    return TaskStatus.FINISHED in task_event.state_updates.state_ts_ns
+
+
+def is_actor_task(task_event: gcs_pb2.TaskEvents) -> bool:


this as well

sampan-s-nayak · 2026-08-02T06:51:28Z

+    """Buckets task events into priority tiers; a higher tier is evicted later."""
+
+    # Number of priority tiers, i.e. valid tiers are 0 .. MAX_PRIORITY - 1.
+    MAX_PRIORITY = 3


should we expose via a static getter instead of direct access?

sampan-s-nayak

we should probably test this as well by enabling flags and running all unit tests once the state api changes are ready.

karticam · 2026-08-03T08:14:08Z

we should probably test this as well by enabling flags and running all unit tests once the state api changes are ready.

True. Triggered a test from this PR: #65248
Premerge tests are passing

cursor · 2026-08-05T09:23:47Z

+            to_drop = num_profile - MAX_NUM_PROFILE_EVENTS_PER_TASK
+            del existing.profile_events.events[:to_drop]
+            self._summary(existing.job_id).record_profile_events_dropped(to_drop)
+            self._stats[STAT_TOTAL_PROFILE_DROPPED] += to_drop


Unlimited profile cap mishandled

Low Severity

MAX_NUM_PROFILE_EVENTS_PER_TASK documents -1 as unlimited, but _update_existing truncates whenever num_profile > MAX_NUM_PROFILE_EVENTS_PER_TASK. With -1, that condition is always true, so profile events are incorrectly dropped. The same module already treats max_num_task_events <= 0 as unlimited.

^{Reviewed by Cursor Bugbot for commit f8c8d41. Configure here.}

sampan-s-nayak · 2026-08-06T17:21:02Z


        c_bool enable_task_events_to_dashboard_head() const
+
+        int64_t task_events_max_num_task_in_gcs() const


we should probably deprecate this in the future, maybe we can add a todo so we dont forget

and have dashboard specific configs (just needs a rename)

same for all the other configs which we are reusing from GCS

Right. I did it right now for an easier migration for users. Once we deprecate the GcsTaskManager, we can get rid of the flags or change their name.

sampan-s-nayak · 2026-08-06T17:31:20Z

might be worth looking into the automated bot comments

cursor

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

There are 2 total unresolved issues (including 1 from previous review).

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit d311e8a. Configure here.}

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

- Copy each set TaskLogInfo field into the state update, matching the C++ converter - Field-level presence copy so a later end-offset event doesn't clobber start offsets - Add a converter round-trip test Signed-off-by: Kartica Modi <karticamodi@gmail.com>

- The GCS-specific cap is obsolete once task events move out of GCS Signed-off-by: Kartica Modi <karticamodi@gmail.com>

karticam requested review from MengjinYan and sampan-s-nayak July 30, 2026 01:57

karticam requested a review from a team as a code owner July 30, 2026 01:57

karticam added core Issues that should be addressed in Ray Core go add ONLY when ready to merge, run all tests labels Jul 30, 2026

gemini-code-assist Bot reviewed Jul 30, 2026

View reviewed changes

karticam mentioned this pull request Jul 31, 2026

[core][taskEvents out of GCS][6/n] Reconcile task states on worker death and job notifications #65141

Open

sampan-s-nayak reviewed Aug 2, 2026

View reviewed changes

karticam force-pushed the karticam/gcs-task-mgr-to-db-head-1 branch from b42066f to ac27fdd Compare August 3, 2026 03:11

cursor Bot reviewed Aug 3, 2026

View reviewed changes

Comment thread python/ray/dashboard/modules/task_events/task_event_storage.py

karticam force-pushed the karticam/gcs-task-mgr-to-db-head-1 branch from ac27fdd to 67a84b4 Compare August 3, 2026 07:48

cursor Bot reviewed Aug 3, 2026

View reviewed changes

Comment thread python/ray/dashboard/head.py

karticam changed the base branch from master to karticam/gcs-notify-task-events-head August 4, 2026 17:25

karticam force-pushed the karticam/gcs-notify-task-events-head branch from 610aacb to dd3e4b4 Compare August 5, 2026 08:52

karticam force-pushed the karticam/gcs-task-mgr-to-db-head-1 branch from 67a84b4 to f8c8d41 Compare August 5, 2026 09:18

cursor Bot reviewed Aug 5, 2026

View reviewed changes

sampan-s-nayak reviewed Aug 6, 2026

View reviewed changes

sampan-s-nayak approved these changes Aug 6, 2026

View reviewed changes

karticam mentioned this pull request Aug 6, 2026

[core][taskEvents out of GCS][3/n] Export task events from aggregator agent to dashboard head #65028

Merged

karticam force-pushed the karticam/gcs-notify-task-events-head branch from dd3e4b4 to 09fcfc7 Compare August 7, 2026 02:58

karticam force-pushed the karticam/gcs-task-mgr-to-db-head-1 branch from f8c8d41 to d311e8a Compare August 7, 2026 02:59

cursor Bot reviewed Aug 7, 2026

View reviewed changes

Comment thread python/ray/dashboard/modules/task_events/ray_event_converter.py

karticam force-pushed the karticam/gcs-task-mgr-to-db-head-1 branch from 6bd709f to 39b9368 Compare August 7, 2026 07:51

karticam added 2 commits August 7, 2026 11:00

ray_event_converter to convert ray events to task events for storage

002114e

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

Implement task storage, gc policy and jobSummar

14019c3

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

karticam added 5 commits August 7, 2026 11:00

wiring storage, gc and converter to task events head

9a02efb

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

Adding event storage model comments

a1279ec

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

Resolving comments - make methods private + getter

204518f

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

Using existing config vars for delays, cnt in python using _config

134496a

Signed-off-by: Kartica Modi <karticamodi@gmail.com>

karticam force-pushed the karticam/gcs-task-mgr-to-db-head-1 branch from 39b9368 to d8f53f0 Compare August 7, 2026 18:01

karticam requested review from a team as code owners August 7, 2026 18:01

karticam changed the base branch from karticam/gcs-notify-task-events-head to master August 7, 2026 18:02

Add TODO to deprecate task_events_max_num_task_in_gcs

2a2df09

- The GCS-specific cap is obsolete once task events move out of GCS Signed-off-by: Kartica Modi <karticamodi@gmail.com>

karticam force-pushed the karticam/gcs-task-mgr-to-db-head-1 branch from d8f53f0 to 2a2df09 Compare August 7, 2026 18:36

		to_evict = list(self._dropped_task_attempts)[:num_to_evict]
		self._dropped_task_attempts.difference_update(to_evict)

		from ray.core.generated.common_pb2 import TaskStatus, TaskType


		def is_task_finished(task_event: gcs_pb2.TaskEvents) -> bool:

		return TaskStatus.FINISHED in task_event.state_updates.state_ts_ns


		def is_actor_task(task_event: gcs_pb2.TaskEvents) -> bool:


		c_bool enable_task_events_to_dashboard_head() const

		int64_t task_events_max_num_task_in_gcs() const

Uh oh!

Conversation

karticam commented Jul 30, 2026

Uh oh!

gemini-code-assist Bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist Bot Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

gemini-code-assist Bot Jul 30, 2026

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sampan-s-nayak left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

karticam commented Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

cursor Bot Aug 5, 2026

Choose a reason for hiding this comment

Unlimited profile cap mishandled

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

karticam Aug 6, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sampan-s-nayak commented Aug 6, 2026

Uh oh!

cursor Bot left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

karticam commented Aug 3, 2026 •

edited

Loading

karticam Aug 6, 2026 •

edited

Loading