diff --git a/README.md b/README.md index 2fcb461b12..764606bafd 100644 --- a/README.md +++ b/README.md @@ -326,6 +326,18 @@ 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` includes `skip_reasons` so an empty candidate pool is directly diagnosable. +Repository-tree HTTP failures are reported as retriable inspection failures, +not cached as empty repositories. Repeated normalized file paths count once +(using the largest reported size for that path), while different weight formats +still contribute separately to the full-repository OOM gate. + +For an empty-candidate investigation, read `daily_runs//summary.json`, +then follow `waveResults[].summary.runDir` to `skipped.jsonl` and `failed.jsonl`. +OOM skips now include size/capacity/source evidence, and `scanStages` includes +`discoveryQueries` with raw rows, accepted rows, cache hits, and stop reasons. +Reaching a page/result budget does not mean the catalog is exhausted. These +diagnostics contain selected fields only, never full configs or HTTP headers. + Community deduplication is scoped to the exact model/GPU combination. A model adapted on one GPU remains eligible for another GPU. Before each submission the worker performs an uncached community check; a lookup failure defers the task diff --git a/modelhub_submmit_api/candidate_preflight.py b/modelhub_submmit_api/candidate_preflight.py index d0e0f3e68b..d7c2e42da6 100644 --- a/modelhub_submmit_api/candidate_preflight.py +++ b/modelhub_submmit_api/candidate_preflight.py @@ -191,6 +191,8 @@ class CandidatePreflightAdvisor: "quantizationMethod": inspection.quantization_method, "estimatedLoadBytes": inspection.estimated_load_bytes(framework), "repositoryOnDiskBytes": inspection.repository_size_bytes, + "recursiveRepositoryBytes": inspection.exact_repository_size_bytes, + "publishedSizeBytes": inspection.published_size_bytes, "gpuMemoryGiB": self.gpu_memory_gib.get(target_gpu), "gpuMemoryEvidence": self.gpu_memory_evidence.get(target_gpu), } @@ -272,7 +274,9 @@ class CandidatePreflightAdvisor: required_gib = memory_sizing_bytes / (1024**3) * MODEL_LOAD_OVERHEAD metadata["estimatedRequiredGiB"] = round(required_gib, 3) metadata["memorySizingBasis"] = ( - "recursive_repository_on_disk" if repository_bytes else "selected_weights_fallback" + "published_repository_upper_bound" + if (inspection.published_size_bytes or 0) > (inspection.exact_repository_size_bytes or 0) + else ("recursive_repository_on_disk" if repository_bytes else "selected_weights_fallback") ) metadata["maximumRepositorySizeGiB"] = round(memory_gib / MODEL_LOAD_OVERHEAD, 3) if required_gib > memory_gib: diff --git a/modelhub_submmit_api/hf_discovery.py b/modelhub_submmit_api/hf_discovery.py index 5cb9bf49b9..091a509b69 100644 --- a/modelhub_submmit_api/hf_discovery.py +++ b/modelhub_submmit_api/hf_discovery.py @@ -110,6 +110,7 @@ class HuggingFaceDiscovery: self._model_card_lock = threading.Lock() self._popular_page_lock = threading.Lock() self._popular_page_cursor = 0 + self.last_listing_diagnostics: list[dict[str, Any]] = [] @staticmethod def _remember_lru(cache: OrderedDict, key: Any, value: Any, limit: int) -> None: @@ -133,6 +134,7 @@ class HuggingFaceDiscovery: return [] deduped: dict[str, HFModelSummary] = {} + self.last_listing_diagnostics = [] del read_concurrency per_tag_limit = limit if len(pipeline_tags) <= 1 else max(10, (limit + len(pipeline_tags) - 1) // len(pipeline_tags)) @@ -156,6 +158,7 @@ class HuggingFaceDiscovery: updated_after=updated_after, sort=sort, start_page=start_page, + diagnostics=self.last_listing_diagnostics, ): current = deduped.get(model.repo_id) if current is None or (model.last_modified or parse_datetime("1970-01-01")) > ( @@ -175,6 +178,7 @@ class HuggingFaceDiscovery: updated_after=None, sort: str = "last_modified", start_page: int = 1, + diagnostics: list[dict[str, Any]] | None = None, ) -> list[HFModelSummary]: # ModelScope OpenAPI rejects values above 50 with InputParameterError. page_size = min(max(1, limit), 50) @@ -183,12 +187,23 @@ class HuggingFaceDiscovery: filter_disabled = self._unsupported_task_filters.get(task_tag, 0.0) > time.monotonic() models: list[HFModelSummary] = [] page_count = (max_items + page_size - 1) // page_size + diagnostic = { + "pipelineTag": pipeline_tag, "sort": sort, + "requestedLimit": limit, "effectiveLimit": max_items, + "pageSize": page_size, "startPage": start_page, + "pagesRead": 0, "cachedPages": 0, "rawRowsScanned": 0, + "acceptedModels": 0, "stopReason": "page_budget", + "catalogExhausted": False, + } + if diagnostics is not None: + diagnostics.append(diagnostic) for page_number in range(start_page, min(61, start_page + page_count)): cache_key = (sort, "*" if filter_disabled else task_tag, page_number, page_size) cached = self._model_page_cache.get(cache_key) if cached is not None and time.monotonic() - cached[0] < self._model_page_cache_ttl: self._model_page_cache.move_to_end(cache_key) items = list(cached[1]) + diagnostic["cachedPages"] += 1 else: elapsed = time.monotonic() - self._last_model_page_request_at if self._last_model_page_request_at > 0 and elapsed < self._page_interval_seconds: @@ -211,6 +226,7 @@ class HuggingFaceDiscovery: self._last_model_page_request_at = time.monotonic() if exc.status_code == 400 and not filter_disabled: self._unsupported_task_filters[task_tag] = time.monotonic() + 86_400 + diagnostic["stopReason"] = "unsupported_filter_retry" print( f"[modelscope] task_filter_unsupported task={task_tag} " "fallback=unfiltered ttl=86400s", @@ -223,12 +239,15 @@ class HuggingFaceDiscovery: updated_after=updated_after, sort=sort, start_page=start_page, + diagnostics=diagnostics, ) print( f"[modelscope] list_models_error task={task_tag} page={page_number} " f"partial_models={len(models)} retry_next_cycle=true error={exc}", flush=True, ) + diagnostic["stopReason"] = "http_error" + diagnostic["httpStatus"] = exc.status_code break items = self._extract_models(payload) @@ -238,7 +257,11 @@ class HuggingFaceDiscovery: (time.monotonic(), list(items)), self._page_cache_limit, ) + diagnostic["pagesRead"] += 1 + diagnostic["rawRowsScanned"] += len(items) if not items: + diagnostic["stopReason"] = "empty_page" + diagnostic["catalogExhausted"] = True break for item in items: model = self._parse_model( @@ -249,9 +272,13 @@ class HuggingFaceDiscovery: ) if model is not None: models.append(model) + diagnostic["acceptedModels"] += 1 if len(models) >= max_items: + diagnostic["stopReason"] = "result_limit" return models if len(items) < page_size: + diagnostic["stopReason"] = "short_page" + diagnostic["catalogExhausted"] = True break return models @@ -412,7 +439,10 @@ class HuggingFaceDiscovery: ) except HttpJsonError as exc: print(f"[modelscope] repo_tree_error repo={repo_id} error={exc}", flush=True) - payload = None + # A failed lookup is not an empty repository. Let the caller defer + # inspection and retry next cycle instead of caching missing + # weights as a permanent framework-incompatibility decision. + raise entries = self._extract_files(payload) @@ -570,7 +600,7 @@ def inspect_repo_tree(repo_id: str, entries: list[dict[str, Any]]) -> ModelInspe break if size_present: try: - file_sizes[path] = max(0, int(size_value)) + file_sizes[path] = max(file_sizes.get(path, 0), int(size_value)) except (TypeError, ValueError): pass filename = PurePosixPath(path).name.lower() @@ -584,12 +614,12 @@ def inspect_repo_tree(repo_id: str, entries: list[dict[str, Any]]) -> ModelInspe selected_gguf = choose_best_gguf(gguf_files) return ModelInspection( repo_id=repo_id, - file_paths=sorted(file_paths), + file_paths=sorted(set(file_paths)), file_sizes=file_sizes, - gguf_files=sorted(gguf_files), + gguf_files=sorted(set(gguf_files)), selected_gguf=PurePosixPath(selected_gguf).name if selected_gguf else None, - weight_files=sorted(vllm_weight_files), - onnx_files=sorted(onnx_files), + weight_files=sorted(set(vllm_weight_files)), + onnx_files=sorted(set(onnx_files)), ) diff --git a/modelhub_submmit_api/main.py b/modelhub_submmit_api/main.py index c96f31e4d9..00dac95eea 100644 --- a/modelhub_submmit_api/main.py +++ b/modelhub_submmit_api/main.py @@ -377,6 +377,7 @@ def choose_candidate_for_gpu_detailed( require_official_config: bool = False, config_optimizer: SafeConfigOptimizer | None = None, excluded_frameworks: set[str] | None = None, + rejection_details: list[dict[str, Any]] | None = None, ) -> tuple[CandidateModel | None, str | None]: last_rejection_reason: str | None = None for task_type in task_types: @@ -473,6 +474,23 @@ def choose_candidate_for_gpu_detailed( ) if not assessment.allowed: last_rejection_reason = assessment.reason + if rejection_details is not None: + rejection_details.append({ + "taskType": task_type, + "framework": framework, + "reason": assessment.reason, + "metadata": { + key: assessment.metadata[key] + for key in ( + "modelType", "architectures", "quantizationMethod", + "estimatedLoadBytes", "repositoryOnDiskBytes", + "recursiveRepositoryBytes", "publishedSizeBytes", + "gpuMemoryGiB", "gpuMemoryEvidence", + "estimatedRequiredGiB", "memorySizingBasis", + "maximumRepositorySizeGiB", + ) if key in assessment.metadata + }, + }) continue config_params = assessment.config_params warnings.extend(assessment.warnings) @@ -651,6 +669,14 @@ def process_model_for_candidates( "repoId": model.repo_id, "targetGpu": target_gpu, "reason": "preflight_predicted_oom", + "preflightDetails": { + "stage": "published_size_precheck", + "publishedSizeBytes": model.file_size, + "estimatedRequiredGiB": model.file_size / (1024**3) * MODEL_LOAD_OVERHEAD, + "gpuMemoryGiB": memory_gib, + "gpuMemoryEvidence": preflight_advisor.gpu_memory_evidence.get(target_gpu), + "loadOverhead": MODEL_LOAD_OVERHEAD, + }, }) continue @@ -717,6 +743,7 @@ def process_model_for_candidates( for target_gpu, task_types, excluded_frameworks in pending_routes: if len(candidates) - before_count >= MAX_ROUTES_PER_MODEL: break + rejection_details: list[dict[str, Any]] = [] best, preflight_reason = choose_candidate_for_gpu_detailed( model=model, inspection=inspection, @@ -728,6 +755,7 @@ def process_model_for_candidates( require_official_config=official_registry is not None, config_optimizer=config_optimizer, excluded_frameworks=excluded_frameworks, + rejection_details=rejection_details, ) if best is None: reason = ( @@ -738,7 +766,10 @@ def process_model_for_candidates( else "no_compatible_auto_template_or_framework" ) ) - skipped.append({"repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason}) + skipped.append({ + "repoId": model.repo_id, "targetGpu": target_gpu, "reason": reason, + **({"preflightRejections": rejection_details} if rejection_details else {}), + }) continue record = candidate_to_record(best) if excluded_frameworks: @@ -1631,6 +1662,7 @@ def run_submission( "candidateModelsAfterStage": candidate_model_count(candidates), "skippedAdded": len(skipped) - skipped_before_stage, "failedAdded": len(stage_failed), + "discoveryQueries": list(getattr(hf_discovery, "last_listing_diagnostics", [])), } scan_stages.append(stage_summary) print( diff --git a/modelhub_submmit_api/models.py b/modelhub_submmit_api/models.py index a1dc2b38e4..f62bae3312 100644 --- a/modelhub_submmit_api/models.py +++ b/modelhub_submmit_api/models.py @@ -140,12 +140,17 @@ class ModelInspection: return None @property - def repository_size_bytes(self) -> int | None: + def exact_repository_size_bytes(self) -> int | None: """Return exact recursive on-disk size when every file has a size.""" - exact: int | None = None if self.file_paths and not any(path not in self.file_sizes for path in self.file_paths): total = sum(max(0, int(self.file_sizes[path])) for path in self.file_paths) - exact = total if total > 0 else None + return total if total > 0 else None + return None + + @property + def repository_size_bytes(self) -> int | None: + """Use the conservative maximum of recursive and published sizes.""" + exact = self.exact_repository_size_bytes published = int(self.published_size_bytes or 0) or None values = [value for value in (exact, published) if value is not None] return max(values) if values else None diff --git a/tests/test_candidate_preflight.py b/tests/test_candidate_preflight.py index 7a27d34b32..ab46350c72 100644 --- a/tests/test_candidate_preflight.py +++ b/tests/test_candidate_preflight.py @@ -243,6 +243,26 @@ class CandidatePreflightTests(unittest.TestCase): self.assertFalse(assessment.allowed) self.assertEqual("recursive_repository_on_disk", assessment.metadata["memorySizingBasis"]) + def test_memory_diagnostics_distinguish_published_upper_bound(self) -> None: + gib = 1024 ** 3 + inspection = ModelInspection( + repo_id="owner/published-larger", + file_paths=["config.json", "tokenizer.json", "model.safetensors"], + file_sizes={"config.json": 0, "tokenizer.json": 0, "model.safetensors": 20 * gib}, + weight_files=["model.safetensors"], + model_config={"model_type": "qwen2"}, + published_size_bytes=40 * gib, + ) + assessment = CandidatePreflightAdvisor(gpu_memory_gib={"gpu": 32}).assess( + inspection=inspection, task_type="text-generation", target_gpu="gpu", + framework="vllm", config_params="max_model_len: 4096", + ) + self.assertEqual("preflight_predicted_oom", assessment.reason) + self.assertEqual(20 * gib, assessment.metadata["recursiveRepositoryBytes"]) + self.assertEqual(40 * gib, assessment.metadata["publishedSizeBytes"]) + self.assertEqual(48.0, assessment.metadata["estimatedRequiredGiB"]) + self.assertEqual("published_repository_upper_bound", assessment.metadata["memorySizingBasis"]) + def test_p800_verified_capacity_accepts_80_gib_repository_boundary(self) -> None: gib = 1024**3 diff --git a/tests/test_modelscope_discovery.py b/tests/test_modelscope_discovery.py index b3f8e5e4c0..795951ba3a 100644 --- a/tests/test_modelscope_discovery.py +++ b/tests/test_modelscope_discovery.py @@ -11,7 +11,7 @@ if str(PACKAGE_DIR) in sys.path: sys.path.remove(str(PACKAGE_DIR)) sys.path.insert(0, str(PACKAGE_DIR)) -from hf_discovery import HuggingFaceDiscovery # noqa: E402 +from hf_discovery import HuggingFaceDiscovery, inspect_repo_tree # noqa: E402 from http_json import HttpJsonError # noqa: E402 @@ -58,6 +58,56 @@ class ModelMetadataClient: class ModelScopeDiscoveryTests(unittest.TestCase): + def test_listing_diagnostics_distinguish_filtered_prefix_from_exhaustion(self) -> None: + class FilteredPageClient: + def request_json(self, *_args, **_kwargs): + return {"data": {"models": [ + {"id": f"owner/model-{index}", "downloads": 0} + for index in range(50) + ]}} + + discovery = HuggingFaceDiscovery(http_client=FilteredPageClient()) # type: ignore[arg-type] + self.assertEqual([], discovery.list_recent_models( + pipeline_tags=["text-generation"], limit=50, min_downloads=50, + )) + diagnostic = discovery.last_listing_diagnostics[0] + self.assertEqual(50, diagnostic["rawRowsScanned"]) + self.assertEqual(0, diagnostic["acceptedModels"]) + self.assertEqual("page_budget", diagnostic["stopReason"]) + self.assertFalse(diagnostic["catalogExhausted"]) + + def test_failed_repo_tree_is_retried_instead_of_cached_as_empty(self) -> None: + class RecoveringClient: + calls = 0 + + def request_json(self, *_args, **_kwargs): + self.calls += 1 + if self.calls == 1: + raise HttpJsonError("temporary outage", status_code=503) + return {"Data": {"Files": [{"Path": "model.safetensors", "Size": 1024}]}} + + client = RecoveringClient() + discovery = HuggingFaceDiscovery(legacy_http_client=client) # type: ignore[arg-type] + with self.assertRaises(HttpJsonError): + discovery.list_repo_tree("owner/recoverable") + self.assertNotIn("owner/recoverable", discovery._repo_tree_cache) + entries = discovery.list_repo_tree("owner/recoverable") + self.assertEqual("model.safetensors", entries[0]["Path"]) + self.assertEqual(entries, discovery.list_repo_tree("owner/recoverable")) + self.assertEqual(2, client.calls) + + def test_duplicate_normalized_paths_do_not_inflate_repository_size(self) -> None: + gib = 1024 ** 3 + inspection = inspect_repo_tree("owner/model", [ + {"Path": "model.safetensors", "Size": 20 * gib}, + {"Path": "./model.safetensors", "Size": 19 * gib}, + {"Path": "/model.safetensors", "Size": 20 * gib}, + ]) + self.assertEqual(["model.safetensors"], inspection.file_paths) + self.assertEqual(["model.safetensors"], inspection.weight_files) + self.assertEqual(20 * gib, inspection.repository_size_bytes) + self.assertEqual(20 * gib, inspection.estimated_load_bytes("vllm")) + def test_model_last_modified_uses_detail_api_and_is_cached(self) -> None: metadata_client = ModelMetadataClient() discovery = HuggingFaceDiscovery(legacy_http_client=metadata_client) # type: ignore[arg-type]