diff --git a/README.md b/README.md index 9eae70fb34..fc6342bdc2 100644 --- a/README.md +++ b/README.md @@ -2,7 +2,23 @@ This repository is packaged for the ModelHub XC agent platform. -## Capacity refresh fix 2026.10.27.1 +## Capacity pagination verification 2026.10.28.1 + +Supersedes the 2026.10.27.1 policy of using the larger reported total. +The platform can return inflated totals, incorrect page counts, and short pages. +Capacity now reads waiting/running pages through an explicit empty page, ignores +total/pages for termination, and deduplicates task IDs. It repeats the scan with +page sizes 100 and 50; only matching active ID sets confirm occupancy. Reported +totals are diagnostic only. Repeated pages, malformed records, request failures, +changing snapshots or the 100-page-per-status budget fail closed for that account. +This costs additional read requests but scans active tasks, not full history. +Two agreeing reads are not an atomic snapshot or a guarantee against a platform +omission shared by both queries; server submission limits remain authoritative. +Logs report `method=dual_page_scan`, `confirmed_active`, and safe failure reasons. +No age cleanup, routing, success-rate gates, recent-five-slot reserve, or weekly +capacity exploration policy changes. Existing tasks are not cancelled. + +## Previous 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 diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index b4eda2983e..a2f683c56f 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -325,47 +325,63 @@ class ModelHubClient: return active_count def count_active_tasks_exact(self) -> int: - """Conservative active occupancy from filtered, validated pages. + """Confirm active IDs with two independently paginated filtered reads. - Totals can disagree with records. Count unique IDs, but retain the - larger reported total rather than inventing extra free slots. + The platform's total/pages fields can describe a different population. + Only an explicit empty page terminates a scan, never those fields or a + short page. Two page sizes must agree before advertising any free slots. + This is not an atomic snapshot: concurrent queue changes fail closed. """ - total = 0 - seen: set[str] = set() - details = [] - for status in ("waiting", "running"): - 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 invalid(reason: str) -> ModelHubAPIError: + error = ModelHubAPIError(reason) + error.capacity_reason = reason + return error + + def scan(page_size: int) -> tuple[set[str], list[dict[str, Any]]]: + active: set[str] = set() + details = [] + for status in ("waiting", "running"): + ids: set[str] = set() + reported = None + for current in range(1, 101): + response = self.list_tasks_page( + current=current, page_size=page_size, only_mine=True, status=status, + ) + data = response.get("data") if isinstance(response, dict) else None + if not isinstance(data, dict) or not isinstance(data.get("records"), list): + raise invalid("invalid_active_response") + try: + value = int(data["total"]) + if value >= 0: + reported = max(reported or 0, value) + except (KeyError, ValueError, TypeError): + pass # Metadata is diagnostic, not an occupancy source. + rows = data["records"] + if not rows: + break + if len(rows) > page_size: + raise invalid("page_size_not_respected") + previous = len(ids) + for row in rows: + if not isinstance(row, dict) or not row.get("taskId") or row.get("status") != status: + raise invalid("invalid_active_record") + ids.add(str(row["taskId"])) + if len(ids) == previous: + raise invalid("active_pagination_no_progress") + else: + raise invalid("active_pagination_budget_exhausted") + active.update(ids) + details.append({"status": status, "reported": reported, "unique": len(ids), + "mismatch": reported != len(ids), "pageSize": page_size}) + return active, details + + self.active_count_details = [] + first, details = scan(100) + confirmed, confirmation_details = scan(50) + if first != confirmed: + raise invalid("active_snapshot_changed") + self.active_count_details = details + confirmation_details + return len(confirmed) def _request( self, @@ -742,7 +758,8 @@ class ModelHubClientPool: 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) + reason = getattr(exc, "capacity_reason", "request_failed") + print(f"[capacity] refresh_failed error={type(exc).__name__} code={code} reason={reason} count=unknown", flush=True) return None def _safe_list_tasks(self, client: ModelHubClient, kwargs: dict[str, Any]) -> list[dict[str, Any]]: @@ -819,7 +836,7 @@ class ModelHubClientPool: 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) + f"confirmed_active={new_remote_count} method=dual_page_scan details={details}", flush=True) if new_remote_count > self._account_caps[index]: self._account_caps[index] = new_remote_count caps_changed = True diff --git a/modelhub_submmit_api/version.py b/modelhub_submmit_api/version.py index 3e91dc89fa..ca9f027432 100644 --- a/modelhub_submmit_api/version.py +++ b/modelhub_submmit_api/version.py @@ -1 +1 @@ -AGENT_VERSION = "2026.10.27.1" +AGENT_VERSION = "2026.10.28.1" diff --git a/tests/test_capacity_refresh.py b/tests/test_capacity_refresh.py index 13b27b8f8e..297e8bae90 100644 --- a/tests/test_capacity_refresh.py +++ b/tests/test_capacity_refresh.py @@ -15,10 +15,11 @@ class CapacityRefreshTests(unittest.TestCase): def client(self, total=97, observed=94): client = ModelHubClient(token="test") - def page(*, status, page_size, **kwargs): + def page(*, status, page_size, current=1, **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)]}} + {"taskId": str(i), "status": status} for i in range( + (current - 1) * page_size, min(current * page_size, observed if status == "waiting" else 0))]}} client.list_tasks_page = page return client @@ -26,8 +27,8 @@ class CapacityRefreshTests(unittest.TestCase): 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.assertEqual([94], pool.refresh_active_counts()) + self.assertEqual(6, pool.available_submit_slots()) self.assertTrue(client.active_count_details[0]["mismatch"]) def test_error_is_unknown_and_recovery_reopens_slots(self): @@ -40,22 +41,82 @@ class CapacityRefreshTests(unittest.TestCase): 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()) + self.assertEqual(6, pool.available_submit_slots()) def test_incomplete_or_repeated_pages_do_not_open_slots(self): client = self.client(total=200, observed=100) + client.list_tasks_page = lambda **kwargs: {"data": {"records": [ + {"taskId": "same", "status": kwargs["status"]}]}} with self.assertRaises(ModelHubAPIError): client.count_active_tasks_exact() - client.list_tasks_page = lambda **kwargs: {"data": {"records": []}} + client.list_tasks_page = lambda **kwargs: {"data": {}} 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": [ + client.list_tasks_page = lambda **kwargs: {"data": {"total": 1, "records": [] if kwargs['current'] > 1 else [ {"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()) + + def test_inflated_total_does_not_hide_capacity(self): + self.assertEqual(100, self.client(total=773, observed=100).count_active_tasks_exact()) + + def test_underreported_pages_do_not_truncate_scan(self): + client = self.client(total=58, observed=175) + original = client.list_tasks_page + def page(**kwargs): + result = original(**kwargs) + result['data']['pages'] = 1 + return result + client.list_tasks_page = page + self.assertEqual(175, client.count_active_tasks_exact()) + + def test_different_confirmation_ids_fail_closed_even_with_equal_counts(self): + client = self.client(total=1, observed=1) + original = client.list_tasks_page + def page(**kwargs): + result = original(**kwargs) + if kwargs['page_size'] == 50: + for row in result['data']['records']: + row['taskId'] = 'replacement' + return result + client.list_tasks_page = page + with self.assertRaisesRegex(ModelHubAPIError, 'active_snapshot_changed'): + client.count_active_tasks_exact() + + def test_confirmation_error_does_not_open_slots(self): + client = self.client(total=773, observed=94) + original = client.list_tasks_page + def page(**kwargs): + if kwargs['page_size'] == 50: + raise ModelHubAPIError('temporary failure') + return original(**kwargs) + client.list_tasks_page = page + pool = ModelHubClientPool([client], active_task_cap=100) + pool.refresh_active_counts() + self.assertEqual(0, pool.available_submit_slots()) + self.assertFalse(pool.active_count_status()[0]['known']) + + def test_empty_queue_confirmed_despite_stale_total(self): + self.assertEqual(0, self.client(total=773, observed=0).count_active_tasks_exact()) + + def test_short_page_is_not_end_of_scan(self): + client = ModelHubClient(token='test') + def page(**kwargs): + rows = [{'taskId': str(kwargs['current']), 'status': 'waiting'}] if ( + kwargs['status'] == 'waiting' and kwargs['current'] <= 3) else [] + return {'data': {'records': rows}} + client.list_tasks_page = page + self.assertEqual(3, client.count_active_tasks_exact()) + + def test_scan_budget_is_bounded(self): + client = ModelHubClient(token='test') + client.list_tasks_page = lambda **kw: {'data': {'records': [ + {'taskId': str(kw['current']), 'status': kw['status']}]}} + with self.assertRaisesRegex(ModelHubAPIError, 'active_pagination_budget_exhausted'): + client.count_active_tasks_exact() diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index b4d369b4a6..0d7151e534 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -290,16 +290,17 @@ class ClientPoolConcurrencyTests(unittest.TestCase): client = ModelHubClient(token="test-token") calls: list[str] = [] - def page(*, status: str, current: int = 1, **_kwargs) -> dict: # noqa: ANN003 + def page(*, status: str, current: int = 1, page_size: int = 100, **_kwargs) -> dict: # noqa: ANN003 calls.append(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))]}} + for i in range((current - 1) * page_size, min(current * page_size, total))]}} client.list_tasks_page = page # type: ignore[method-assign] self.assertEqual(200, client.count_active_tasks_exact()) - self.assertEqual(["waiting", "waiting", "running"], calls) + self.assertEqual(8, calls.count("waiting")) + self.assertEqual(4, calls.count("running")) @staticmethod def monday_calibration_time(hour: int = 10) -> datetime: