diff --git a/README.md b/README.md index 934eb03e2a..9eae70fb34 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,19 @@ This repository is packaged for the ModelHub XC agent platform. +## Capacity refresh fix 2026.10.27.1 + +All task-page requests are clamped to the platform's 100-record limit, including +community sampling. Normal and weekly capacity refreshes now use the same +filtered waiting/running pagination, validate responses and deduplicate task IDs. +When reported totals disagree with returned rows, occupancy uses the larger +value conservatively and logs the discrepancy. Missing or non-progressing pages +fail closed. Failed reads report `active_counts=unknown` and +`reason=capacity_read_unavailable`, retaining the last-success age for diagnosis; +they do not advertise cached occupancy as a freshly confirmed full queue. +Healthy accounts remain usable, and the next successful refresh reopens slots. +The five-slot recent-model reserve remains in force. + ## Single-worker takeover release 2026.10.26.1 This release is intended to replace the competing workers, using all configured diff --git a/modelhub_submmit_api/market_intelligence.py b/modelhub_submmit_api/market_intelligence.py index 58aae57ac2..77dbe623be 100644 --- a/modelhub_submmit_api/market_intelligence.py +++ b/modelhub_submmit_api/market_intelligence.py @@ -410,7 +410,7 @@ class MarketIntelligenceManager: } payload = client.list_tasks_page( current=1, - page_size=200, + page_size=100, only_mine=False, begin_time=cutoff, end_time=now, diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index 05687d43ad..b4eda2983e 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -151,7 +151,7 @@ class ModelHubClient: "/api/adapt/task/page", query={ "current": current, - "pageSize": page_size, + "pageSize": max(1, min(100, int(page_size))), "onlyMine": str(only_mine).lower(), "beginTime": query_time(begin_time), "endTime": query_time(end_time), @@ -325,14 +325,46 @@ class ModelHubClient: return active_count def count_active_tasks_exact(self) -> int: - """Use the platform's filtered totals instead of paging all history.""" + """Conservative active occupancy from filtered, validated pages. + + Totals can disagree with records. Count unique IDs, but retain the + larger reported total rather than inventing extra free slots. + """ total = 0 + seen: set[str] = set() + details = [] for status in ("waiting", "running"): - page = self.list_tasks_page(current=1, page_size=1, only_mine=True, status=status) - data = page.get("data") - if not isinstance(data, dict) or "total" not in data: - raise ModelHubAPIError(f"Active-task total is unavailable for status={status}") - total += max(0, int(data["total"])) + current, reported = 1, 0 + status_ids: set[str] = set() + while True: + data = self.list_tasks_page(current=current, page_size=100, only_mine=True, status=status).get("data") + if not isinstance(data, dict) or "total" not in data or not isinstance(data.get("records"), list): + raise ModelHubAPIError(f"Incomplete active-task response for status={status}") + count = int(data["total"]) + if count < 0: + raise ModelHubAPIError("Negative active-task total") + reported = max(reported, count) + pages = max(1, int(data.get("pages") or 0), (count + 99) // 100) + rows = data["records"] + prior = len(status_ids) + for row in rows: + if not isinstance(row, dict) or not row.get("taskId") or row.get("status") != status: + raise ModelHubAPIError(f"Invalid filtered active-task record for status={status}") + status_ids.add(str(row["taskId"])) + if current >= pages: + if current > 1 and len(status_ids) == prior: + raise ModelHubAPIError("Active-task pagination made no progress") + break + if not rows or len(status_ids) == prior or current >= 1000: + raise ModelHubAPIError("Incomplete active-task pagination") + current += 1 + # An ID can move from waiting to running between requests. + overlap = len(status_ids & seen) + total += max(reported, len(status_ids)) - overlap + seen.update(status_ids) + details.append({"status": status, "reported": reported, "unique": len(status_ids), + "mismatch": reported != len(status_ids)}) + self.active_count_details = details return total def _request( @@ -521,6 +553,7 @@ class ModelHubClientPool: # persisted capacity is not enough without a successful active-count # read for the current scheduling window. self._count_known: list[bool] = [False for _ in clients] + self._count_success_at = [0.0 for _ in clients] self._active_refresh_at: float = 0.0 self._counts_initialized = False self._state_lock = threading.Lock() @@ -703,10 +736,13 @@ class ModelHubClientPool: def _safe_count_active_tasks(self, client: ModelHubClient, max_count: int | None) -> int | None: try: - if max_count is None and hasattr(client, "count_active_tasks_exact"): + if hasattr(client, "count_active_tasks_exact"): return client.count_active_tasks_exact() - return client.count_active_tasks(max_count=max_count, page_size=200) - except Exception: + return client.count_active_tasks(max_count=max_count, page_size=100) + except Exception as exc: + # Never log request URLs/headers or credentials. + code = getattr(exc, "code", None) + print(f"[capacity] refresh_failed error={type(exc).__name__} code={code} count=unknown", flush=True) return None def _safe_list_tasks(self, client: ModelHubClient, kwargs: dict[str, Any]) -> list[dict[str, Any]]: @@ -779,6 +815,11 @@ class ModelHubClientPool: old_remote_count = self._remote_counts[index] new_remote_count = max(0, int(count)) self._count_known[index] = True + self._count_success_at[index] = refreshed_at + details = getattr(self.clients[index], "active_count_details", []) + if any(item.get("mismatch") for item in details): + print(f"[capacity] account={index + 1:02d} total_records_mismatch=true " + f"conservative_active={new_remote_count} details={details}", flush=True) if new_remote_count > self._account_caps[index]: self._account_caps[index] = new_remote_count caps_changed = True @@ -821,6 +862,7 @@ class ModelHubClientPool: 1 for index in range(len(self.clients)) if index not in self._capacity_probe_attempted + and self._count_known[index] and self._capacity_probe_cycle >= self._capacity_probe_cooldown_until[index] and self._effective_count_locked(index) >= self._account_caps[index] ) @@ -846,9 +888,20 @@ class ModelHubClientPool: normal_slots = sum( max(0, self._account_caps[index] - self._effective_count_locked(index)) for index in range(len(self.clients)) + if self._count_known[index] ) return normal_slots if normal_slots > 0 else self._probe_slot_count_locked() + def active_count_status(self) -> list[dict[str, Any]]: + self._refresh_active_counts() + with self._state_lock: + now = time.monotonic() + return [{"account": i + 1, "known": self._count_known[i], + "active": self._effective_count_locked(i) if self._count_known[i] else None, + "lastSuccessAgeSeconds": round(now - self._count_success_at[i], 1) + if self._count_success_at[i] else None} + for i in range(len(self.clients))] + def list_tasks(self, **kwargs): # noqa: ANN003, ANN001 # Default: use only the reader client to avoid fanout amplification. # The fanout_all parameter allows explicit cross-account merging when needed. @@ -935,6 +988,7 @@ class ModelHubClientPool: index: self._account_caps[index] - self._effective_count_locked(index) for index in range(len(self.clients)) if index not in excluded + and self._count_known[index] and (weekly_target is None or index == weekly_target) and ( reserve_capacity_slots is None @@ -997,6 +1051,7 @@ class ModelHubClientPool: index for index in range(len(self.clients)) if index not in excluded + and self._count_known[index] and (weekly_target is None or index == weekly_target) and (weekly_target is not None or index not in self._capacity_probe_attempted) and ( diff --git a/modelhub_submmit_api/poll_runner.py b/modelhub_submmit_api/poll_runner.py index 663a18ab4e..60982a7025 100644 --- a/modelhub_submmit_api/poll_runner.py +++ b/modelhub_submmit_api/poll_runner.py @@ -1056,8 +1056,12 @@ def run_poll_loop( if hasattr(modelhub_client, "weekly_calibration_status") else {} ) available_slots = modelhub_client.available_submit_slots() if hasattr(modelhub_client, "available_submit_slots") else None + count_status = modelhub_client.active_count_status() if hasattr(modelhub_client, "active_count_status") else [] + display_counts = [str(item["active"]) if item["known"] else "unknown" for item in count_status] + if count_status and any(not item["known"] for item in count_status): + log(f"[capacity] account_status={count_status} unknown_accounts=deferred") log( - f"[poll] cycle={cycles} active_counts={','.join(str(count) for count in active_counts) if active_counts else 'n/a'} " + f"[poll] cycle={cycles} active_counts={','.join(display_counts or [str(count) for count in active_counts]) or 'n/a'} " f"capacity_limits={','.join(str(limit) for limit in capacity_limits) if capacity_limits else 'n/a'} " f"capacity_probe={'on' if capacity_probe else 'off'} " f"weekly_calibration={'on' if calibration.get('targetAccount') else 'off'} " @@ -1084,7 +1088,8 @@ def run_poll_loop( state_sync.healthy = False state_sync.last_error = str(exc) state_sync.write_readiness(ready=False, reason="state_sync_unhealthy") - log(f"[poll] cycle={cycles} sleep={base_args.poll_interval_seconds}s reason=no_available_slots") + no_slot_reason = "capacity_read_unavailable" if any(not item["known"] for item in count_status) else "no_available_slots" + log(f"[poll] cycle={cycles} sleep={base_args.poll_interval_seconds}s reason={no_slot_reason}") time.sleep(base_args.poll_interval_seconds) continue diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 9d176a9515..3e91dc89fa 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.10.26.1" +AGENT_VERSION = "2026.10.27.1" diff --git a/tests/test_capacity_refresh.py b/tests/test_capacity_refresh.py new file mode 100644 index 0000000000..13b27b8f8e --- /dev/null +++ b/tests/test_capacity_refresh.py @@ -0,0 +1,61 @@ +import sys +import unittest +from pathlib import Path +from unittest.mock import patch +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "modelhub_submmit_api")) +from modelhub_client import ModelHubClient, ModelHubClientPool, ModelHubAPIError + + +class CapacityRefreshTests(unittest.TestCase): + def test_api_boundary_clamps_all_callers(self): + client = ModelHubClient(token="test") + with patch.object(client, "_request", return_value={}) as request: + client.list_tasks_page(page_size=200) + self.assertEqual(100, request.call_args.kwargs["query"]["pageSize"]) + + def client(self, total=97, observed=94): + client = ModelHubClient(token="test") + def page(*, status, page_size, **kwargs): + self.assertLessEqual(page_size, 100) + return {"data": {"total": total if status == "waiting" else 0, "records": [ + {"taskId": str(i), "status": status} for i in range(observed if status == "waiting" else 0)]}} + client.list_tasks_page = page + return client + + def test_full_cache_refreshes_to_free_slots_with_disagreeing_total(self): + client = self.client() + pool = ModelHubClientPool([client], active_task_cap=100) + pool._remote_counts[0] = 100 + self.assertEqual([97], pool.refresh_active_counts()) + self.assertEqual(3, pool.available_submit_slots()) + self.assertTrue(client.active_count_details[0]["mismatch"]) + + def test_error_is_unknown_and_recovery_reopens_slots(self): + client = self.client() + pool = ModelHubClientPool([client], active_task_cap=100) + pool.refresh_active_counts() + with patch.object(client, "list_tasks_page", side_effect=ModelHubAPIError("分页大小不能超过100")): + pool.refresh_active_counts() + self.assertEqual(0, pool.available_submit_slots()) + self.assertIsNone(pool.active_count_status()[0]["active"]) + self.assertIsNone(pool._reserve_account(set())) + pool.refresh_active_counts() + self.assertEqual(3, pool.available_submit_slots()) + + def test_incomplete_or_repeated_pages_do_not_open_slots(self): + client = self.client(total=200, observed=100) + with self.assertRaises(ModelHubAPIError): + client.count_active_tasks_exact() + client.list_tasks_page = lambda **kwargs: {"data": {"records": []}} + with self.assertRaises(ModelHubAPIError): + client.count_active_tasks_exact() + + def test_duplicate_ids_and_transition_are_counted_once(self): + client = ModelHubClient(token="test") + client.list_tasks_page = lambda **kwargs: {"data": {"total": 1, "records": [ + {"taskId": "same", "status": kwargs["status"]}, + {"taskId": "same", "status": kwargs["status"]}]}} + self.assertEqual(1, client.count_active_tasks_exact()) + + def test_total_lower_than_observed_uses_larger_count(self): + self.assertEqual(99, self.client(total=97, observed=99).count_active_tasks_exact()) diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index e66e7ab5b5..b4d369b4a6 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -290,13 +290,16 @@ class ClientPoolConcurrencyTests(unittest.TestCase): client = ModelHubClient(token="test-token") calls: list[str] = [] - def page(*, status: str, **_kwargs) -> dict: # noqa: ANN003 + def page(*, status: str, current: int = 1, **_kwargs) -> dict: # noqa: ANN003 calls.append(status) - return {"data": {"total": {"waiting": 175, "running": 25}[status]}} + total = {"waiting": 175, "running": 25}[status] + return {"data": {"total": total, "records": [ + {"taskId": f"{status}-{i}", "status": status} + for i in range((current - 1) * 100, min(current * 100, total))]}} client.list_tasks_page = page # type: ignore[method-assign] self.assertEqual(200, client.count_active_tasks_exact()) - self.assertEqual(["waiting", "running"], calls) + self.assertEqual(["waiting", "waiting", "running"], calls) @staticmethod def monday_calibration_time(hour: int = 10) -> datetime: