fix(batches): bound the batch rate limiter's input-file read

litellm/constants.py

@@ -1356,6 +1356,13 @@
 BATCH_STATUS_POLL_INTERVAL_SECONDS = int(os.getenv("BATCH_STATUS_POLL_INTERVAL_SECONDS", 3600))  # 1 hour
 BATCH_STATUS_POLL_MAX_ATTEMPTS = int(os.getenv("BATCH_STATUS_POLL_MAX_ATTEMPTS", 24))  # for 24 hours
 
+# Deadline for the batch rate limiter's input-file read. The read happens inline
+# in POST /v1/batches, so it must resolve well within a client's read timeout;
+# unbounded, the OpenAI SDK default (600s, max_retries=2) applies and a stalled
+# Files API holds the request open indefinitely. Override with
+# general_settings.batch_input_file_read_timeout.
+DEFAULT_BATCH_INPUT_FILE_READ_TIMEOUT_SECONDS = 10.0
+
 HEALTH_CHECK_TIMEOUT_SECONDS = int(os.getenv("HEALTH_CHECK_TIMEOUT_SECONDS", 60))  # 60 seconds
 _background_health_check_max_tokens_env = os.getenv("BACKGROUND_HEALTH_CHECK_MAX_TOKENS")
 try:

litellm/proxy/_types.py

@@ -2408,6 +2408,10 @@ class ConfigGeneralSettings(LiteLLMPydanticObjectBase):
         None,
         description="By default, the user calling /team/new is automatically added to the new team as a team admin. If True, proxy admins are no longer auto-added; members explicitly listed in members_with_roles are unaffected. Default is False.",
     )
+    batch_input_file_read_timeout: Optional[float] = Field(
+        None,
+        description="Seconds the batch rate limiter may spend reading a batch input file to count tokens (default 10). The read runs inline in POST /v1/batches, so this must stay well inside client read timeouts. On timeout, keys whose model allowlist must be validated against the file are rejected; keys with unrestricted model access are admitted without rate limiting.",
+    )
     maximum_spend_logs_retention_period: Optional[str] = Field(
         None,
         description="Maximum retention period for spend logs (e.g., '7d' for 7 days). Logs older than this will be deleted.",

litellm/proxy/hooks/batch_rate_limiter.py

@@ -33,10 +33,12 @@
 from fastapi import HTTPException
 from pydantic import BaseModel
 
+import asyncio
 import json
 
 import litellm
 from litellm._logging import verbose_proxy_logger
+from litellm.constants import DEFAULT_BATCH_INPUT_FILE_READ_TIMEOUT_SECONDS
 from litellm.batches.batch_utils import (
     _count_entry_tokens,
     _estimate_batch_entry_tokens,
@@ -96,6 +98,24 @@ class BatchFileUsage(BaseModel):
     request_count: int
 
 
+class BatchInputFileReadTimeout(Exception):
+    """The batch input-file read exceeded its deadline.
+
+    Distinct from a generic failure because the read serves two purposes and the
+    two have opposite safe defaults: it counts tokens for rate limiting (where
+    admitting the batch unmetered is tolerable) and it validates every
+    ``body.model`` in the JSONL against the caller's allowlist (where admitting
+    the batch unchecked is a privilege escalation). Carrying its own type lets
+    ``async_pre_call_hook`` fail closed only for keys that need the allowlist
+    check, instead of the blanket fail-open its generic handler applies.
+    """
+
+    def __init__(self, file_id: str, timeout_seconds: float) -> None:
+        self.file_id = file_id
+        self.timeout_seconds = timeout_seconds
+        super().__init__(f"Timed out after {timeout_seconds}s reading batch input file {file_id}")
+
+
 class _PROXY_BatchRateLimiter(CustomLogger):
     """
     Rate limiter for batch API requests.
@@ -292,6 +312,28 @@ def _warn_if_unsupported_model_skip_configured(self, general_settings: Dict) ->
                 "disable_batch_input_file_rate_limiting instead."
             )
 
+    @staticmethod
+    def _batch_input_file_read_timeout() -> float:
+        """Seconds the input-file read may take before it is abandoned.
+
+        Falls back to the default when the operator's value is missing or not a
+        positive number: a zero/negative deadline would make wait_for expire
+        immediately and reject every batch from a restricted key.
+        """
+        from litellm.proxy.proxy_server import general_settings
+
+        configured = general_settings.get("batch_input_file_read_timeout")
+        if isinstance(configured, bool) or not isinstance(configured, (int, float)):
+            return DEFAULT_BATCH_INPUT_FILE_READ_TIMEOUT_SECONDS
+        if configured <= 0:
+            verbose_proxy_logger.warning(
+                "Ignoring general_settings.batch_input_file_read_timeout=%s: must be > 0. Using %ss.",
+                configured,
+                DEFAULT_BATCH_INPUT_FILE_READ_TIMEOUT_SECONDS,
+            )
+            return DEFAULT_BATCH_INPUT_FILE_READ_TIMEOUT_SECONDS
+        return float(configured)
+
     @staticmethod
     def _key_requires_batch_model_access_check(
         user_api_key_dict: UserAPIKeyAuth,
@@ -334,6 +376,12 @@ def _resolve_batch_input_file_fetch_params(
         Model-embedded IDs (``file-<base64>``) are not unified managed-file IDs;
         without decoding them, ``afile_content`` is called with the encoded ID
         and the upstream provider returns 404.
+
+        The returned kwargs may carry a ``timeout`` from the deployment's
+        credentials (one of ``_extract_file_access_credentials``' keys). Callers
+        that need their own deadline must set it on the result rather than passing
+        it alongside ``**fetch_kwargs``, which would raise TypeError for a
+        duplicate kwarg.
         """
         from litellm.proxy.openai_files_endpoints.common_utils import (
             decode_model_from_file_id,
@@ -518,8 +566,11 @@ async def count_input_file_usage(
             # For managed files the unified file id encodes the proxy model
             # alias(es) the file was uploaded for; auth validates against those.
             target_model_names = get_models_from_unified_file_id(is_managed_file) if is_managed_file else []
+            # Resolved before the coroutine is built so a failure here can never
+            # leave an un-awaited coroutine behind.
+            timeout_seconds = self._batch_input_file_read_timeout()
             if is_managed_file and user_api_key_dict is not None:
-                file_content = await self._fetch_managed_file_content(
+                fetch = self._fetch_managed_file_content(
                     file_id=file_id,
                     user_api_key_dict=user_api_key_dict,
                 )
@@ -529,13 +580,35 @@ async def count_input_file_usage(
                     custom_llm_provider=custom_llm_provider,
                     data=data or {},
                 )
-                # For non-managed files, use the standard litellm.afile_content
-                file_content = await litellm.afile_content(
+                # Set on the resolved kwargs, not passed as a separate keyword:
+                # they may already carry the deployment's own `timeout`, and
+                # passing both raises TypeError for a duplicate kwarg. This
+                # read's budget wins on purpose; a deployment timeout is sized
+                # for serving traffic, not for a read that blocks POST /v1/batches.
+                fetch_kwargs["timeout"] = timeout_seconds
+                # `timeout` reaches file_content's TIMEOUT LOGIC via
+                # GenericLiteLLMParams, capping the upstream HTTP request itself
+                # rather than only the await, so an abandoned read stops
+                # occupying its executor thread too.
+                fetch = litellm.afile_content(
                     file_id=provider_file_id,
                     user_api_key_dict=user_api_key_dict,
                     **fetch_kwargs,
                 )
 
+            # Bound the read: it runs inline in POST /v1/batches, so unbounded the
+            # SDK default (600s x 3 attempts) outlives every client timeout and the
+            # request just hangs (LIT-5027).
+            #
+            # wait_for is what guarantees the handler stops waiting, and it is the
+            # only bound the managed-files path has (that hook takes no timeout
+            # argument), so an abandoned managed read may hold its executor thread
+            # until the SDK's own timeout fires.
+            try:
+                file_content = await asyncio.wait_for(fetch, timeout=timeout_seconds)
+            except asyncio.TimeoutError as exc:
+                raise BatchInputFileReadTimeout(file_id=file_id, timeout_seconds=timeout_seconds) from exc
+
             file_content_bytes = getattr(file_content, "content", None)
             if not isinstance(file_content_bytes, bytes):
                 raise ValueError(
@@ -604,6 +677,10 @@ async def count_input_file_usage(
                     f"Batch input file rejected for {file_id}: status={e.status_code} detail={e.detail}"
                 )
             raise
+        except BatchInputFileReadTimeout:
+            # The caller decides the policy (reject vs admit unmetered) and logs
+            # accordingly; a generic error line here would just duplicate it.
+            raise
         except Exception as e:
             verbose_proxy_logger.error(f"Error counting input file usage for {file_id}: {str(e)}")
             raise
@@ -854,6 +931,38 @@ async def async_pre_call_hook(
         except HTTPException:
             # Re-raise HTTP exceptions (rate limit exceeded)
             raise
+        except BatchInputFileReadTimeout as e:
+            # The read is both the token count and the JSONL model-allowlist
+            # check, so the two cases diverge. A key restricted to a subset of
+            # models cannot be admitted without validating the file: doing so
+            # would grant exactly the bypass _should_skip_batch_input_file_processing
+            # refuses to allow via operator config. An unrestricted key has only
+            # rate-limit accuracy at stake, so it is admitted unmetered, matching
+            # the generic fail-open below.
+            if self._key_requires_batch_model_access_check(user_api_key_dict):
+                verbose_proxy_logger.error(
+                    "Rejecting batch: could not read input file %s within %ss to validate "
+                    "the models it references against the key's allowlist.",
+                    e.file_id,
+                    e.timeout_seconds,
+                )
+                raise ProxyException(
+                    message=(
+                        f"Could not read the batch input file within {e.timeout_seconds}s to "
+                        "validate the models it references. Retry, or contact your proxy admin "
+                        "if the files API is degraded."
+                    ),
+                    type=ProxyErrorTypes.internal_server_error,
+                    param="input_file_id",
+                    code=504,
+                ) from e
+            verbose_proxy_logger.warning(
+                "Batch admitted without rate limiting: reading input file %s timed out after %ss. "
+                "Its tokens and requests are not counted against this key's limits.",
+                e.file_id,
+                e.timeout_seconds,
+            )
+            return data
         except Exception as e:
             verbose_proxy_logger.error(f"Error in batch rate limiting: {str(e)}", exc_info=True)
             # Don't block the request if rate limiting fails

tests/e2e/batches/test_batches_e2e.py

@@ -397,16 +397,6 @@ def unattributed_rows(rows: list[SpendLogRow]) -> list[SpendLogRow]:
     return [row for row in rows if not row.api_key]
 
 
-@pytest.mark.skip(
-    reason=(
-        "LIT-5027: the path under test hangs. The batch rate limiter reads the input file "
-        "to count tokens by awaiting litellm.afile_content with no timeout, so a slow Files "
-        "API holds POST /v1/batches open past any client deadline (63.6s observed on stage "
-        "against a 60s read timeout). The unattributed-spend-row contract below is never "
-        "reached, so the test reports a timeout rather than the behavior it guards. Unskip "
-        "once the fetch is bounded."
-    )
-)
 def test_rate_limited_batch_create_leaves_no_unattributed_spend_row(
     client: BatchClient, resources: ResourceManager, batch_deployments: None
 ) -> None: