From e91ccf71fb7dd87165ac63c6d73bdb62b83044a3 Mon Sep 17 00:00:00 2001 From: Tim Paine <3105306+timkpaine@users.noreply.github.com> Date: Fri, 21 Aug 2026 19:37:24 -0400 Subject: [PATCH 1/2] Test the id schema without a Ray cluster MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The actor test I added was unreliable on CI. It failed on main twice with an empty dataframe while every other test passed, including the plain-function equivalent, and there were no scheduling errors or actor tracebacks in the logs. My earlier explanation, that the head-node resource budget was exhausted, was wrong: it did not fix this. The defect was a type declaration, so assert that directly. Building a polars frame from the shipped schema with a real Ray hex id reproduces it exactly — Float32 raises TypeError, Utf8 round-trips — with no Ray, in 0.09s rather than a 120s poll. Reverting the schema turns 4 of the new tests red. Extend the same check to every id column, since task_id, job_id, node_id, worker_id and parent_task_id are all hex strings too. Why actor tasks record no metadata on Linux CI while they do on macOS is still unexplained and is not covered here. Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com> --- raydar/tests/test_schema.py | 28 ++++++++++++++++++++++++++++ raydar/tests/test_task_tracker.py | 23 ----------------------- 2 files changed, 28 insertions(+), 23 deletions(-) create mode 100644 raydar/tests/test_schema.py diff --git a/raydar/tests/test_schema.py b/raydar/tests/test_schema.py new file mode 100644 index 0000000..88e2a26 --- /dev/null +++ b/raydar/tests/test_schema.py @@ -0,0 +1,28 @@ +"""The schema must match what Ray actually reports. + +These run without a Ray cluster: the defect they guard is a type declaration, +and asserting it directly is both deterministic and immediate. +""" + +import polars as pl +import pytest + +from raydar.task_tracker.schema import schema + +# Ray reports ids as hex strings, e.g. TaskState.actor_id. +RAY_ID = "22c21d18d3081db73e11271a01000000" + +ID_COLUMNS = ("task_id", "actor_id", "job_id", "node_id", "worker_id", "parent_task_id", "placement_group_id") + + +@pytest.mark.parametrize("column", ID_COLUMNS) +def test_id_columns_hold_ray_hex_ids(column): + # Declaring these numeric made get_df raise on any actor task and rendered + # every id as 0.0 in the dashboard. + frame = pl.DataFrame({column: [RAY_ID]}, schema_overrides={column: schema[column]}) + assert frame[column][0] == RAY_ID + + +@pytest.mark.parametrize("column", ID_COLUMNS) +def test_id_columns_are_declared_as_strings(column): + assert schema[column] == pl.Utf8 diff --git a/raydar/tests/test_task_tracker.py b/raydar/tests/test_task_tracker.py index cf1aca2..0a4df5b 100644 --- a/raydar/tests/test_task_tracker.py +++ b/raydar/tests/test_task_tracker.py @@ -14,13 +14,6 @@ def do_some_work(): return True -@ray.remote -class SomeActor: - def do_some_work(self): - time.sleep(0.1) - return True - - def wait_for(fetch, ready, timeout=120, interval=0.5): """Poll `fetch` until `ready` accepts the value, then return it.""" deadline = time.time() + timeout @@ -73,22 +66,6 @@ def test_dashboard_is_off_by_default(self, trackers): task_tracker = trackers() assert task_tracker.dashboard_url is None - def test_actor_task_ids_survive_as_strings(self, trackers): - # Ray reports actor_id as a hex string. Declaring it numeric made get_df - # raise and rendered every id as 0.0 in the dashboard. - task_tracker = trackers(dashboard="local") - actor = SomeActor.remote() - refs = [actor.do_some_work.remote() for _ in range(3)] - task_tracker.process(refs) - ray.get(refs) - - df = wait_for(task_tracker.get_df, lambda d: not d.is_empty()) - assert not df.is_empty(), "tracker recorded no finished actor tasks" - - actor_ids = [a for a in df["actor_id"].to_list() if a] - assert actor_ids, "actor_id was not recorded" - assert all(isinstance(a, str) and int(a, 16) for a in actor_ids) - def test_dashboard_options_reach_the_dashboard(self, trackers): layout = {"sizes": [1], "viewers": {}} task_tracker = trackers(dashboard="local", dashboard_options={"title": "custom", "layout": layout}) From a2b4c44ef257d49c38d9eeac89a07e93bd6c5b84 Mon Sep 17 00:00:00 2001 From: Tim Paine <3105306+timkpaine@users.noreply.github.com> Date: Fri, 21 Aug 2026 19:59:30 -0400 Subject: [PATCH 2/2] Record tasks the GCS has not published yet The tracker drops task metadata, and that is what has been failing CI. When ray.wait reports an object ready the GCS has not yet published the task's state. Measured on a quiet box, this is not an edge case: all 60 of 60 tasks read back as None at that instant. callback files those tasks in pending_tasks, and only a later callback revisits that list. The processor sends callbacks while it still has unfinished tasks, so whatever lands in the final batch is never looked at again and never reaches the dataframe. Ten tasks that all finish in one wait round produce exactly one callback, so every task can be stranded at once and get_df stays empty forever. That is the empty dataframe CI reported, and it explains why only the two tests that wait on metadata ever failed while the other three passed. Have the processor keep asking until the tracker reports nothing pending. Calling back with no tasks re-resolves the list on its own. This reproduces only under CI timing. On macOS the hop between the two actors is enough for the GCS to catch up, so the suite passes here with and without this change. Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com> --- raydar/task_tracker/task_tracker.py | 20 +++++++++++++++++++- raydar/tests/test_task_tracker.py | 11 +++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/raydar/task_tracker/task_tracker.py b/raydar/task_tracker/task_tracker.py index 1e80d1e..74be3ae 100644 --- a/raydar/task_tracker/task_tracker.py +++ b/raydar/task_tracker/task_tracker.py @@ -1,6 +1,7 @@ import asyncio import itertools import logging +import time from collections.abc import Iterable from typing import Literal @@ -34,7 +35,7 @@ class AsyncMetadataTrackerCallback: def __init__(self, name: str, namespace: str): self.actor = ray.get_actor(name, namespace) - def process(self, obj_refs: Iterable[ray.ObjectRef]) -> None: + def process(self, obj_refs: Iterable[ray.ObjectRef], settle_timeout: float = 60.0, settle_interval: float = 0.5) -> None: """Processes an interable collection of ray.ObjectRefs. Iterates through the collection, finds completed references, and returns those references to the @@ -42,6 +43,9 @@ def process(self, obj_refs: Iterable[ray.ObjectRef]) -> None: Args: obj_refs: An iterable collection of (possibly) in-progress ray object references + settle_timeout: How long to keep asking the tracker to re-resolve tasks whose + metadata the GCS has not published yet. + settle_interval: How long to wait between those attempts. """ active_tasks = set(obj_refs) while len(active_tasks) > 0: @@ -55,6 +59,16 @@ def process(self, obj_refs: Iterable[ray.ObjectRef]) -> None: if len(finished_tasks) > 0: self.actor.callback.remote(finished_tasks) + # ray.wait reports an object ready before the GCS has published the task's final + # state, so tasks the tracker could not resolve are held in its pending list. Only + # a later callback revisits that list, and the loop above has just run out of + # tasks to send, so the last batch would otherwise never be recorded. Calling back + # with nothing re-resolves the pending list on its own. + deadline = time.monotonic() + settle_timeout + while time.monotonic() < deadline and ray.get(self.actor.has_pending_tasks.remote()): + time.sleep(settle_interval) + ray.get(self.actor.callback.remote([])) + def exit(self) -> None: """Terminate this actor""" ray.actor.exit_actor() @@ -165,6 +179,10 @@ def drain(self) -> dict | None: def get_dashboard_mode(self) -> str | None: return self.dashboard_mode + def has_pending_tasks(self) -> bool: + """Whether any task is still waiting on the GCS to publish its final state.""" + return bool(self.pending_tasks) + def callback(self, tasks: Iterable[ray.ObjectRef]) -> None: """A remote function used by this actor's processor actor attribute. Will be called by a separate actor with a collection of ray object references once those ObjectReferences are not in the "RUNNING" or diff --git a/raydar/tests/test_task_tracker.py b/raydar/tests/test_task_tracker.py index 0a4df5b..03a1819 100644 --- a/raydar/tests/test_task_tracker.py +++ b/raydar/tests/test_task_tracker.py @@ -66,6 +66,17 @@ def test_dashboard_is_off_by_default(self, trackers): task_tracker = trackers() assert task_tracker.dashboard_url is None + def test_a_single_task_is_still_recorded(self, trackers): + # One task completes in one wait round, so the tracker gets exactly one + # callback. The GCS has not published the task's state that early, so + # nothing was recorded until the processor learned to wait for it. + task_tracker = trackers(dashboard="local") + task_tracker.process([do_some_work.remote()]) + + df = wait_for(task_tracker.get_df, lambda d: not d.is_empty(), timeout=90) + assert not df.is_empty(), "the only task the tracker was given went unrecorded" + assert df[["name", "state"]].row(0) == ("do_some_work", "FINISHED") + def test_dashboard_options_reach_the_dashboard(self, trackers): layout = {"sizes": [1], "viewers": {}} task_tracker = trackers(dashboard="local", dashboard_options={"title": "custom", "layout": layout})