From 801a3777e04d64e61bc21b89378467a33abd449f Mon Sep 17 00:00:00 2001 From: Tim Paine <3105306+timkpaine@users.noreply.github.com> Date: Fri, 21 Aug 2026 19:11:27 -0400 Subject: [PATCH] Record Ray actor and placement group ids as strings MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ray reports actor_id and placement_group_id as hex strings, but both the polars schema and the perspective schema have declared them numeric since the initial commit. Tracking any actor task therefore raised TypeError from get_df(), and the dashboard rendered every id as 0.0 — silently wrong rather than visibly broken. No test had ever tracked an actor task, only plain remote functions, which is how this survived. Add one that asserts the ids round-trip as hex strings. Also from the same review pass: Sync whatever landed before a batch raised. `apply` mutated tables and then refreshed state at the end, so a bad row left the earlier tables applied but unsynced, and the next batch's replayed schemas reported no change — leaving them invisible until an unrelated update arrived. Check the dashboard mode mismatch for every mode, not just "local". An existing actor attached with the wrong mode was only reported in one of the three cases. Wait on the custom table's row count rather than its name, which passed before the update was applied and so would not have caught a dropped update. Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com> --- raydar/dashboard/dashboard.py | 26 +++++++++++++----------- raydar/task_tracker/schema.py | 4 ++-- raydar/task_tracker/task_tracker.py | 22 ++++++++++---------- raydar/tests/test_dashboard.py | 13 ++++++++++++ raydar/tests/test_task_tracker.py | 31 ++++++++++++++++++++++++++++- 5 files changed, 70 insertions(+), 26 deletions(-) diff --git a/raydar/dashboard/dashboard.py b/raydar/dashboard/dashboard.py index 3500208..f6c2017 100644 --- a/raydar/dashboard/dashboard.py +++ b/raydar/dashboard/dashboard.py @@ -151,19 +151,21 @@ async def _perspective_socket(self, websocket: WebSocket) -> None: def apply(self, batch: dict) -> None: """Apply a drained :class:`~raydar.ops.OpBuffer` batch to the tables.""" changed = False - for tablename, schema in (batch.get("schemas") or {}).items(): - changed |= self.tables.new_table(tablename, schema) - for tablename in batch.get("cleared") or (): - self.tables.clear(tablename) - changed = True - for tablename, rows in (batch.get("updates") or {}).items(): - if rows: - self.tables.update(tablename, rows) + try: + for tablename, schema in (batch.get("schemas") or {}).items(): + changed |= self.tables.new_table(tablename, schema) + for tablename in batch.get("cleared") or (): + self.tables.clear(tablename) changed = True - # Schemas are replayed on every drain, so most batches are empty; only - # touch the synced model when something actually moved. - if changed: - self._refresh_state() + for tablename, rows in (batch.get("updates") or {}).items(): + if rows: + self.tables.update(tablename, rows) + changed = True + finally: + # Sync whatever landed before a bad row raised, otherwise those tables + # stay invisible: replayed schemas report no change on the next batch. + if changed: + self._refresh_state() def _refresh_state(self) -> None: names = self.tables.names() diff --git a/raydar/task_tracker/schema.py b/raydar/task_tracker/schema.py index a245657..ae63cc8 100644 --- a/raydar/task_tracker/schema.py +++ b/raydar/task_tracker/schema.py @@ -6,7 +6,7 @@ "name": pl.Utf8, "state": pl.Utf8, "job_id": pl.Utf8, - "actor_id": pl.Float32, + "actor_id": pl.Utf8, "type": pl.Utf8, "func_or_class_name": pl.Utf8, "parent_task_id": pl.Utf8, @@ -29,7 +29,7 @@ ), ] ), - "placement_group_id": pl.Float32, + "placement_group_id": pl.Utf8, "events": pl.List(pl.Struct([pl.Field("state", pl.Utf8), pl.Field("created_ms", pl.Float64)])), "profiling_data": pl.Struct( [ diff --git a/raydar/task_tracker/task_tracker.py b/raydar/task_tracker/task_tracker.py index bcd9eb4..1e80d1e 100644 --- a/raydar/task_tracker/task_tracker.py +++ b/raydar/task_tracker/task_tracker.py @@ -131,7 +131,7 @@ def __init__( "name": "string", "state": "string", "job_id": "string", - "actor_id": "float", + "actor_id": "string", "type": "string", "func_or_class_name": "string", "parent_task_id": "string", @@ -139,7 +139,7 @@ def __init__( "worker_id": "string", "error_type": "string", "language": "string", - "placement_group_id": "float", + "placement_group_id": "string", "creation_time_ms": "datetime", "start_time_ms": "datetime", "end_time_ms": "datetime", @@ -357,18 +357,18 @@ def __init__( **kwargs, ) + # `get_if_exists` returns a pre-existing actor and drops these constructor + # args, so a mode mismatch would otherwise show as an empty dashboard. + active = ray.get(self.tracker.get_dashboard_mode.remote()) + if active != dashboard: + logger.warning( + f'Actor "{name}" in namespace "{namespace}" already exists with dashboard={active!r}, ' + f"not the requested {dashboard!r}. Use a new name or namespace." + ) + if dashboard == "local": from raydar.dashboard import LocalDashboard - # `get_if_exists` returns a pre-existing actor and drops these constructor - # args, so a mode mismatch would otherwise show as an empty dashboard. - active = ray.get(self.tracker.get_dashboard_mode.remote()) - if active != dashboard: - logger.warning( - f'Actor "{name}" in namespace "{namespace}" already exists with dashboard={active!r}, ' - f"so it will not feed a {dashboard!r} dashboard. Use a new name or namespace." - ) - self.dashboard = LocalDashboard( drain=lambda: ray.get(self.tracker.drain.remote()), host=dashboard_host, diff --git a/raydar/tests/test_dashboard.py b/raydar/tests/test_dashboard.py index d115e7f..42dd4d1 100644 --- a/raydar/tests/test_dashboard.py +++ b/raydar/tests/test_dashboard.py @@ -68,6 +68,19 @@ def test_update_of_an_unknown_table_raises(self, dashboard): with pytest.raises(KeyError): dashboard.apply({"updates": {"nope": [{"a": 1}]}}) + def test_a_failed_batch_still_syncs_what_landed(self, dashboard): + with pytest.raises(KeyError): + dashboard.apply( + { + "schemas": {"t": SCHEMA}, + "updates": {"t": [{"a": 1, "b": "x"}], "nope": [{"a": 1}]}, + } + ) + + # The good table landed, so it must be visible even though the batch raised. + assert dashboard.state.tables == ["t"] + assert dashboard.state.rows == "1" + def test_limit_caps_retained_rows(self): dashboard = Dashboard(limit=2) dashboard.apply({"schemas": {"t": SCHEMA}, "updates": {"t": [{"a": i, "b": "x"} for i in range(5)]}}) diff --git a/raydar/tests/test_task_tracker.py b/raydar/tests/test_task_tracker.py index 08265cc..b6962b0 100644 --- a/raydar/tests/test_task_tracker.py +++ b/raydar/tests/test_task_tracker.py @@ -13,6 +13,13 @@ 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 @@ -44,6 +51,25 @@ def test_dashboard_is_off_by_default(self): task_tracker = RayTaskTracker() assert task_tracker.dashboard_url is None + def test_actor_task_ids_survive_as_strings(self): + # 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 = RayTaskTracker(dashboard="local") + try: + 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) + finally: + task_tracker.dashboard.stop() + def test_dashboard_options_reach_the_dashboard(self): layout = {"sizes": [1], "viewers": {}} task_tracker = RayTaskTracker(dashboard="local", dashboard_options={"title": "custom", "layout": layout}) @@ -65,8 +91,11 @@ def test_local_dashboard_serves_tables_pulled_from_the_actor(self): tables = task_tracker.dashboard.dashboard.tables expected = ["custom", "task_tracker_data"] names = wait_for(lambda: sorted(tables.names()), lambda n: n == expected, timeout=60) - assert names == expected + + # Wait on the row too: names alone pass before the update is applied. + rows = wait_for(lambda: tables._tables["custom"].size(), lambda n: n > 0, timeout=60) + assert rows == 1 assert httpx.get(task_tracker.dashboard_url).status_code == 200 finally: task_tracker.dashboard.stop()