From 3a6415089ed2a6a695c5d4b29778933ff8bab5af Mon Sep 17 00:00:00 2001 From: Tim Paine <3105306+timkpaine@users.noreply.github.com> Date: Fri, 21 Aug 2026 19:26:17 -0400 Subject: [PATCH] Release tracker actors between tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The new actor test failed on main with an empty dataframe, having passed on the PR branch and locally. It is not a bug in the actor id fix. Both tracker actors pin 0.1 of the head node's `node:__internal_head__` budget, which totals 1.0, so each RayTaskTracker costs 0.2 and a cluster admits exactly five. Measured: the eleventh pinned actor cannot be scheduled at all. Tests only stopped the dashboard, never the detached actors, so the budget accumulated across the module. Four trackers held 0.8; adding the actor test made it five and 1.0 — exactly the cap, with no headroom. Whichever tracker lost the race got an unschedulable callback actor, which never reported back, so the dataframe stayed empty. Sitting on the boundary is why it passed in one environment and failed in another. Build trackers through a fixture that calls exit() afterwards. Verified that exit() returns the pin: seven trackers in sequence each take 0.2 and release back to 1.0. The five-tracker ceiling itself is untouched here and is a real limit on concurrent trackers per cluster. Signed-off-by: Tim Paine <3105306+timkpaine@users.noreply.github.com> --- raydar/tests/test_task_tracker.py | 131 ++++++++++++++++-------------- 1 file changed, 72 insertions(+), 59 deletions(-) diff --git a/raydar/tests/test_task_tracker.py b/raydar/tests/test_task_tracker.py index b6962b0..cf1aca2 100644 --- a/raydar/tests/test_task_tracker.py +++ b/raydar/tests/test_task_tracker.py @@ -1,3 +1,4 @@ +import logging import time import httpx @@ -30,72 +31,84 @@ def wait_for(fetch, ready, timeout=120, interval=0.5): return value +@pytest.fixture +def trackers(): + """Build trackers and release them afterwards. + + Each RayTaskTracker pins 0.2 of the head node's 1.0 `node:__internal_head__` + budget across its two detached actors, so leaving them alive starves the + fifth tracker in a module and its metadata never arrives. + """ + created = [] + + def make(**kwargs): + tracker = RayTaskTracker(**kwargs) + created.append(tracker) + return tracker + + yield make + + for tracker in created: + try: + tracker.exit() + except Exception: + logging.getLogger(__name__).exception("Failed to release tracker %s", tracker.name) + + @pytest.mark.usefixtures("unittest_ray_cluster") class TestRayTaskTracker: - def test_construction_and_dataframe(self): - task_tracker = RayTaskTracker(dashboard="local") - try: - assert len(task_tracker.namespace.split("-")) == 2 - refs = [do_some_work.remote() for _ in range(10)] - task_tracker.process(refs) - - # Metadata arrives via GCS polling, so wait on the result rather than - # on a fixed sleep, which was slow on a fast box and flaky on a slow one. - df = wait_for(task_tracker.get_df, lambda d: not d.is_empty()) - assert not df.is_empty(), "tracker recorded no finished tasks" - assert df[["name", "state"]].row(0) == ("do_some_work", "FINISHED") - finally: - task_tracker.dashboard.stop() - - def test_dashboard_is_off_by_default(self): - task_tracker = RayTaskTracker() + def test_construction_and_dataframe(self, trackers): + task_tracker = trackers(dashboard="local") + assert len(task_tracker.namespace.split("-")) == 2 + refs = [do_some_work.remote() for _ in range(10)] + task_tracker.process(refs) + + # Metadata arrives via GCS polling, so wait on the result rather than + # on a fixed sleep, which was slow on a fast box and flaky on a slow one. + df = wait_for(task_tracker.get_df, lambda d: not d.is_empty()) + assert not df.is_empty(), "tracker recorded no finished tasks" + assert df[["name", "state"]].row(0) == ("do_some_work", "FINISHED") + + 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): + 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 = RayTaskTracker(dashboard="local") - try: - actor = SomeActor.remote() - refs = [actor.do_some_work.remote() for _ in range(3)] - task_tracker.process(refs) - ray.get(refs) + 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" + 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() + 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): + def test_dashboard_options_reach_the_dashboard(self, trackers): layout = {"sizes": [1], "viewers": {}} - task_tracker = RayTaskTracker(dashboard="local", dashboard_options={"title": "custom", "layout": layout}) - try: - dashboard = task_tracker.dashboard.dashboard - dashboard.apply({"schemas": {"t": {"a": "integer"}}}) - assert dashboard.state.layout == layout - assert "custom" in httpx.get(task_tracker.dashboard_url).text - finally: - task_tracker.dashboard.stop() - - def test_local_dashboard_serves_tables_pulled_from_the_actor(self): - task_tracker = RayTaskTracker(dashboard="local") - try: - assert task_tracker.dashboard_url.startswith("http://127.0.0.1:") - task_tracker.create_table("custom", {"a": "string", "b": "integer"}) - task_tracker.update_table("custom", [{"a": "foo", "b": 1}]) - - 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() + task_tracker = trackers(dashboard="local", dashboard_options={"title": "custom", "layout": layout}) + dashboard = task_tracker.dashboard.dashboard + dashboard.apply({"schemas": {"t": {"a": "integer"}}}) + assert dashboard.state.layout == layout + assert "custom" in httpx.get(task_tracker.dashboard_url).text + + def test_local_dashboard_serves_tables_pulled_from_the_actor(self, trackers): + task_tracker = trackers(dashboard="local") + assert task_tracker.dashboard_url.startswith("http://127.0.0.1:") + task_tracker.create_table("custom", {"a": "string", "b": "integer"}) + task_tracker.update_table("custom", [{"a": "foo", "b": 1}]) + + 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