[core][DO NOT MERGE] End-to-end testing changes for removing task events out of GCS - 3

[karticam]

#65248 + ray task event recorder optimizations.

test pr. do not merge

[gemini-code-assist]

Code Review

This pull request migrates task events out of GCS to the dashboard head by introducing a new TaskEventsHead module, an in-memory TaskEventStorage, and a background TaskEventManager. It also updates the state API and timeline queries to read from the dashboard head when enabled. The review feedback highlights several critical issues, including a missing aiohttp import that would cause a runtime error, potential infinite CPU-spinning loops in background subscription handlers on persistent exceptions, a potential KeyError when accessing task log info, overly restrictive filtering of events lacking task metadata, a potential integer overflow when narrowing 64-bit log offsets to 32-bit integers, and leftover AI-generation tags in comments.

[gemini-code-assist]

The aiohttp module is not imported in this file, which will cause a NameError when attempting to catch aiohttp.ClientConnectionError on line 398. Please add import aiohttp to the imports at the top of the file.

Suggested change

	import asyncio
	import json
	import logging
	from abc import ABC, abstractmethod
	import asyncio
	import aiohttp
	import json
	import logging
	from abc import ABC, abstractmethod

[gemini-code-assist]

The while True loop in _subscribe_for_worker_deaths does not have any sleep or backoff in the except block. If subscriber.poll persistently raises an exception (e.g., due to a lost connection to GCS), this loop will spin infinitely at 100% CPU and flood the logs. Adding a short sleep (e.g., await asyncio.sleep(1)) in the except block prevents this.

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]

The while True loop in _subscribe_for_finished_jobs does not have any sleep or backoff in the except block. If subscriber.poll persistently raises an exception, this loop will spin infinitely at 100% CPU and flood the logs. Adding a short sleep (e.g., await asyncio.sleep(1)) in the except block prevents this.

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]

Accessing task_state["task_log_info"] directly can raise a KeyError if "task_log_info" is not present in the dictionary (which happens if the field is unset in the protobuf message and MessageToDict omits it). Using .get("task_log_info") is much safer.

Suggested change

	task_log_info = task_state["task_log_info"]
	task_log_info = task_state.get("task_log_info")

[gemini-code-assist]

If task_event does not have task_info (which can happen under partial data loss or if the definition event hasn't arrived/was dropped), _passes_filters returns False immediately. This means these events are completely filtered out and invisible even for unfiltered queries or queries that only filter by task_id. We should allow them to pass if the filters don't require task_info.

Suggested change

	if not task_event.HasField("task_info"):
	return False
	if not task_event.HasField("task_info"):
	# If task_info is missing (e.g. due to partial data loss), we can still match task_filters.
	# However, we must reject if there are filters that require task_info.
	if (
	filters.exclude_driver
	or len(filters.job_filters) > 0
	or len(filters.actor_filters) > 0
	or len(filters.task_name_filters) > 0
	):
	return False
	for task_filter in filters.task_filters:
	if not _apply_predicate(
	task_filter.predicate, task_event.task_id, task_filter.task_id
	):
	return False
	return True

[gemini-code-assist]

Casting src.stdout_start() (which is int64_t) directly to int32_t via static_cast can cause integer overflow and wrap-around to negative values if the log file exceeds 2 GB. Since the export schema requires int32, we should cap the values at INT32_MAX (2147483647LL) to prevent negative offsets.

  // The export schema is public and its offsets are int32, so the wider offsets are
  // narrowed here rather than changing a field type that consumers already parse.
  // We cap the offsets at INT32_MAX to prevent wrap-around to negative values.
  dest->set_stdout_start(static_cast<int32_t>(std::min(src.stdout_start(), 2147483647LL)));
  dest->set_stdout_end(static_cast<int32_t>(std::min(src.stdout_end(), 2147483647LL)));
  dest->set_stderr_start(static_cast<int32_t>(std::min(src.stderr_start(), 2147483647LL)));
  dest->set_stderr_end(static_cast<int32_t>(std::min(src.stderr_end(), 2147483647LL)));

[gemini-code-assist]

There are several comments across the newly added/modified files in this PR that contain the (claude) prefix (likely left over from AI-assisted code generation). Please clean these up by removing the (claude) tag to keep the codebase professional and tidy.

Suggested change

	// (claude) options_.session_name shared with every task event this worker records,
	// options_.session_name shared with every task event this worker records,

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 2 potential issues.

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

[cursor]

Stale task-events client after disconnect

Medium Severity

GlobalState.disconnect clears _global_state_accessor but leaves the new cached _task_events_head_client in place. After ray.shutdown() / reconnect, ray.timeline() can keep using the old accessor and dashboard endpoint from the previous session.

Additional Locations (2)

• python/ray/_private/state.py#L118-L119
• python/ray/_private/state.py#L370-L377

 

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

[cursor]

Unclear PR title or description

Low Severity

⚠️ This PR needs a clearer title and/or description.

To help reviewers, please ensure your PR includes:

• Title: A concise summary of the change
• Description:
  • What problem does this solve?
  • How does this PR solve it?
  • Any relevant context for reviewers such as:
    • Why is the problem important to solve?
    • Why was this approach chosen over others?

See this list of PRs as examples for PRs that have gone above and beyond:

• [Core] Introduce local port service discovery  #59613
• [Core] Improve Large-Scale Resource View Synchronization Through Sync Message Batching #57641
• Remove node observability information from hot path of core components #56474
• [core][rdt] Support out-of-order actors by extracting metadata when creating #59610
• [core] fix open leak for plasma store memory (shm/fallback) by workers #52622

This violates the Clear PR Descriptions and Titles rule: the description is effectively “test pr. do not merge” plus a link, and does not explain the problem or how this change fixes it.

 

^{Triggered by project rule: Bugbot Rules}

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