From dc8365c2154059b2cb6ec32f196097295b294573 Mon Sep 17 00:00:00 2001 From: coolboy <131342541+wudixzy@users.noreply.github.com> Date: Fri, 2 Oct 2026 21:24:08 +0800 Subject: [PATCH] fix: prevent stale capacity poisoning and deepen recent-only refill scans Keep ordinary capacity races from lowering persistent per-account limits. Deepen recent-only discovery within the existing age and quality gates. Add offline regressions for recovery and reserved-slot refill. --- README.md | 10 +++ modelhub_submmit_api/main.py | 11 +++ modelhub_submmit_api/modelhub_client.py | 7 +- tests/test_concurrency.py | 112 +++++++++++++++++++++++- 4 files changed, 138 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 43a497fb99..258cea307a 100644 --- a/README.md +++ b/README.md @@ -301,6 +301,16 @@ model/GPU combinations, the same run progressively expands discovery to the last 7 days, the last 30 days, and finally older history (up to 3,000 models). Scanning stops as soon as enough replacement candidates have been found. +When only recent-model reserved positions remain, discovery also deepens the +same recent window up to the configured scan cap (3,000 by default). It never +admits older models into those positions or lowers the quality gates. This avoids +rechecking only an exhausted short prefix on every poll. + +An ordinary account-capacity rejection temporarily marks that account full and +tries the next account. It does not lower the persisted capacity using a cached +active count: another worker may have filled the queue since that count was +read. Downward capacity learning remains confined to weekly calibration. + Model verification results are reused for 15 minutes across poll cycles, and a locally failed model/GPU pair cools down for 24 hours instead of being excluded forever. The `[scan]` lines show every expansion stage, while `[daily] wave_done` diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index ecda225d09..c96f31e4d9 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -822,6 +822,17 @@ def build_adaptive_scan_stages( "limit": max_models, } ) + if not allow_older_than_recent_window: + # Reserved positions still need a deep search of eligible recent + # models. Widening the date window alone repeatedly scans the same + # short prefix (often already processed) on every poll cycle. + stages.append( + { + "name": "recent_window_full", + "updatedAfter": recent_cutoff, + "limit": max_models, + } + ) deduped: list[dict[str, Any]] = [] seen: set[tuple[str | None, int, str]] = set() for stage in stages: diff --git a/modelhub_submmit_api/modelhub_client.py b/modelhub_submmit_api/modelhub_client.py index 707fe0e850..a99a8fa967 100644 --- a/modelhub_submmit_api/modelhub_client.py +++ b/modelhub_submmit_api/modelhub_client.py @@ -1029,7 +1029,12 @@ class ModelHubClientPool: self._capacity_probe_cooldown_until[index] = ( self._capacity_probe_cycle + self._capacity_probe_cooldown_cycles ) - else: + elif self._weekly_target_locked() == index: + # Only the scheduled calibration may lower a learned cap. + # Ordinary rejection can race with another agent filling this + # account; our cached count is not evidence of its capacity. + # Persisting it (even zero -> one) strands otherwise usable + # positions until the next calibration. observed_capacity = max(1, self._effective_count_locked(index)) if observed_capacity < self._account_caps[index]: self._account_caps[index] = observed_capacity diff --git a/tests/test_concurrency.py b/tests/test_concurrency.py index 3f3945ab58..98e3bc4653 100644 --- a/tests/test_concurrency.py +++ b/tests/test_concurrency.py @@ -6,6 +6,7 @@ import tempfile import threading import unittest from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace from datetime import datetime, timedelta, timezone from pathlib import Path from unittest.mock import patch @@ -216,6 +217,38 @@ def make_candidate(index: int) -> dict: class ClientPoolConcurrencyTests(unittest.TestCase): + def test_stale_capacity_rejection_does_not_poison_learned_limit(self) -> None: + with tempfile.TemporaryDirectory() as temporary_dir: + state_path = Path(temporary_dir) / "capacity.json" + client = FakeClient(active_count=0, reject_capacity=True) + pool = ModelHubClientPool( + [client], # type: ignore[list-item] + active_task_cap=100, + recent_model_reserve_slots=5, + capacity_probe_interval_cycles=0, + capacity_state_path=state_path, + ) + self.assertEqual(100, pool.available_submit_slots()) + with self.assertRaises(ModelHubAPIError): + pool.add_task({"model": "rejected-during-race"}) + + self.assertEqual([100], pool.account_capacity_limits()) + client.reject_capacity = False + pool.refresh_active_counts() + self.assertEqual(95, pool.old_model_submit_slots()) + pool.add_task_for_model( + {"model": "eligible-old-model"}, + model_last_modified="2025-01-01T00:00:00Z", + submitted_at="2026-10-02T00:00:00Z", + ) + self.assertEqual(1, len(client.submitted)) + restarted = ModelHubClientPool( + [client], # type: ignore[list-item] + active_task_cap=100, + capacity_state_path=state_path, + ) + self.assertEqual([100], restarted.account_capacity_limits()) + def test_exact_active_count_uses_filtered_platform_totals(self) -> None: client = ModelHubClient(token="test-token") calls: list[str] = [] @@ -481,8 +514,37 @@ class ClientPoolConcurrencyTests(unittest.TestCase): recent_model_days=7, ) - self.assertEqual(["configured_window", "last_7_days"], [stage["name"] for stage in stages]) + self.assertEqual( + ["configured_window", "last_7_days", "recent_window_full"], + [stage["name"] for stage in stages], + ) self.assertTrue(all(stage["updatedAfter"] >= now - timedelta(days=7) for stage in stages)) + self.assertEqual(3000, stages[-1]["limit"]) + + def test_recent_only_scan_deepens_even_when_date_window_cannot_expand(self) -> None: + now = datetime(2026, 8, 11, tzinfo=timezone.utc) + for initial_cutoff in (None, now - timedelta(days=7), now - timedelta(days=30)): + with self.subTest(initial_cutoff=initial_cutoff): + stages = build_adaptive_scan_stages( + now=now, + initial_updated_after=initial_cutoff, + initial_limit=20, + max_models=800, + allow_older_than_recent_window=False, + ) + self.assertEqual([20, 800], [stage["limit"] for stage in stages]) + self.assertTrue(all(stage["updatedAfter"] == now - timedelta(days=7) for stage in stages)) + + def test_recent_only_scan_deduplicates_an_already_full_window(self) -> None: + now = datetime(2026, 8, 11, tzinfo=timezone.utc) + stages = build_adaptive_scan_stages( + now=now, + initial_updated_after=now - timedelta(days=7), + initial_limit=100, + max_models=100, + allow_older_than_recent_window=False, + ) + self.assertEqual(1, len(stages)) def test_submit_candidate_reports_old_model_policy_skip_at_dynamic_threshold(self) -> None: client = FakeClient(active_count=95) @@ -738,6 +800,54 @@ class ClientPoolConcurrencyTests(unittest.TestCase): self.assertEqual("last_modified", summary["scanStages"][-1]["sort"]) self.assertIsNone(discovery.calls[-1]) + def test_reserved_slots_find_recent_candidates_beyond_first_scan_prefix(self) -> None: + class RecentOnlyClient(ExhaustedRecentClient): + def old_model_submit_slots(self) -> int: + return 0 + + class PagedRecentDiscovery(FakeDiscovery): + def __init__(self) -> None: + super().__init__(4) + self.models[0] = replace(self.models[0], repo_id="owner/recent-exhausted") + self.limits: list[int] = [] + + def list_recent_models(self, *, limit, **_kwargs): # noqa: ANN003 + self.limits.append(limit) + return self.models[:limit] + + with tempfile.TemporaryDirectory() as temporary_dir: + root = Path(temporary_dir) + args = build_parser().parse_args([ + "--task-types", "text-generation", "--since-hours", "168", + "--limit", "1", "--max-scan-models", "4", + "--skip-outcome-sync", "--skip-history-archive", + "--disable-gpu-strategy", + ]) + for field, filename in ( + ("runs_dir", "runs"), ("ledger_path", "ledger.jsonl"), + ("outcomes_path", "outcomes.jsonl"), ("claims_path", "claims.jsonl"), + ("history_archive_path", "history.jsonl"), + ("submission_exclusions_path", "exclusions.jsonl"), + ): + setattr(args, field, str(root / filename)) + selector = TemplateSelector() + discovery = PagedRecentDiscovery() + client = RecentOnlyClient( + available=2, + processed_gpus=set(selector.supported_target_gpus("text-generation", auto_only=True)), + ) + summary = run_submission( + args, + now=datetime(2026, 1, 2, 12, tzinfo=timezone.utc), + hf_discovery=discovery, # type: ignore[arg-type] + modelhub_client=client, # type: ignore[arg-type] + template_selector=selector, + ) + self.assertEqual([1, 4], discovery.limits) + self.assertEqual(2, summary["submittedCount"]) + self.assertEqual("recent_window_full", summary["scanStages"][-1]["name"]) + self.assertEqual(4, summary["scannedModels"]) + def test_adaptive_strategy_counts_only_platform_accepted_tasks(self) -> None: with tempfile.TemporaryDirectory() as temporary_dir: root = Path(temporary_dir)