Record Ray actor and placement group ids as strings

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()

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(
         [

raydar/task_tracker/task_tracker.py

@@ -131,15 +131,15 @@ 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",
                             "node_id": "string",
                             "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,

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)]}})

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()