feat(serverless): coordinate early health checks once per container

ARCHITECTURE.md

@@ -599,12 +599,12 @@ log.error(message, job_id=None)
 
 ### Fitness Checks: `modules/rp_fitness.py`
 
-**Location**: `runpod/serverless/modules/rp_fitness.py`
+**Location**: `runpod/_health/fitness.py` (legacy `serverless.modules.rp_fitness` imports remain aliases)
 
 **Responsibilities**:
 - Validate worker health at startup before handler initialization
 - Support both synchronous and asynchronous check functions
-- Exit immediately with sys.exit(1) on any check failure
+- Exit immediately with os._exit(1) on any check failure
 - Enable fail-fast deployment validation
 
 **Key Functions**:
@@ -613,11 +613,11 @@ log.error(message, job_id=None)
 - `clear_fitness_checks()`: Clear registry (testing only)
 
 **Execution Flow**:
-1. Called from `worker.py:40` before heartbeat starts: `asyncio.run(run_fitness_checks())`
+1. The first top-level import with both `RUNPOD_ENDPOINT_ID` and `RUNPOD_WEBHOOK_GET_JOB` runs shared hardware checks, excluding test invocations and `RUNPOD_TEST`. A Linux file lock and container-start-scoped result prevent concurrent/repeated execution across processes; saved failures propagate to later workers. Identity includes host boot, PID namespace, and PID 1 start time. Network, Python CUDA initialization, compute, and custom checks remain at worker start (realtime uses serving lifespan). Unsupported/unwritable coordination defers to worker-start checks. Import lock waiting is bounded at 35 seconds; worker-start waiting covers the configured GPU timeout, GPU fallback, both CUDA-version probes, and a five-second overhead allowance (minimum 35 seconds). A worker-start timeout fails closed. `RUNPOD_DEFER_FITNESS_CHECKS=true` postpones early checks. No custom launcher or PID environment variable is required.
 2. Runs only in production mode (skipped for local testing)
 3. Auto-detects sync vs async using `inspect.iscoroutinefunction()`
 4. Executes checks in registration order (list preserves order)
-5. On failure: log detailed error, call `sys.exit(1)`
+5. On health failure: log, best-effort unhealthy report, force-kill via `os._exit(1)`. Registration is atomic; early setup errors defer, unresolved worker-start setup errors report `fitness_check_setup` and force-exit.
 6. On success: log completion, proceed with worker startup
 
 **Performance**: ~0.5ms framework overhead per check, total depends on check logic
@@ -765,7 +765,7 @@ sequenceDiagram
             CHECK->>CHECK: Log success
         else Check fails
             CHECK->>SYS: Log error + traceback
-            CHECK->>SYS: sys.exit(1)
+            CHECK->>SYS: os._exit(1)
         end
     end
 

README.md

@@ -143,7 +143,9 @@ runpod.serverless.start({"handler": handler})
 
 **Key Features:**
 - Supports both synchronous and asynchronous check functions
-- Checks run only once at worker startup (production mode)
+- Shared hardware checks run once per container at the first Serverless import; network and process-specific checks run at worker start
+- Local tests/helper imports remain exempt; network readiness and custom checks run at worker start
+- Successful early checks are reused unless their configuration changes
 - Runs before handler initialization and job processing begins
 - Any check failure exits with code 1 (worker marked unhealthy)
 

docs/serverless/worker_fitness_checks.md

@@ -41,6 +41,22 @@ if __name__ == "__main__":
     runpod.serverless.start({"handler": handler})
 ```
 
+## When Checks Run
+
+On Serverless, the first `import runpod` runs RAM, disk, CUDA-version, and native GPU health checks. Eligibility requires both `RUNPOD_ENDPOINT_ID` and `RUNPOD_WEBHOOK_GET_JOB`. Platform tests (`RUNPOD_TEST`), `--test_input`, and local `--rp_serve_api` invocations skip early checks.
+
+A shared Linux file lock serializes these checks across Python processes. Successful results are reused by helpers and the actual worker; failures are saved before reporting unhealthy and exiting, so another process cannot silently ignore the failure. Results are scoped to the host boot, PID namespace, and container init process start time, rather than just the Pod ID. Checks with changed settings or SDK version are rerun. The first import may occur after model loading; no earlier timing is guaranteed in that case.
+
+Network connectivity, Python CUDA initialization, GPU compute, and customer-registered checks run in the worker process at `.start()`, before accepting jobs. Network checks retry against the worker API with a bounded budget. Keeping Python CUDA initialization out of imports protects subsequent customer forks. Production realtime mode runs its final checks in serving lifespan.
+
+Coordination uses a fixed `/tmp` path shared by container processes. If procfs or shared state is unavailable, early checks defer to worker start. Imports wait at most 35 seconds for another checking process; a timeout defers to worker start. At worker start the wait budget covers the configured GPU timeout, its fallback, both CUDA-version probes, and five seconds of overhead (minimum 35 seconds). If the lock remains busy after that budget, the worker reports failure and exits rather than accepting jobs without validation. An owner crash releases its OS lock, allowing a later process to retry unfinished checks. Processes with separate filesystems or incompatible file permissions cannot share results and use the fallback.
+
+`RUNPOD_DEFER_FITNESS_CHECKS=true` restores worker-start timing. `RUNPOD_SKIP_FITNESS_CHECKS=true` disables all checks. Set early thresholds before importing the SDK; late changes are applied at worker start, but cannot undo an earlier failure. No launcher or Docker entrypoint changes are needed.
+
+### Rollout
+
+Validate in a small set of workers before broader rollout. The deferral variable provides a rollback of early timing without handler edits. This SDK change does not itself alter deployed platform configuration.
+
 ## Async Fitness Checks
 
 Fitness checks support both synchronous and asynchronous functions:
@@ -284,19 +300,17 @@ Disk space check passed: 50.00GB free (50.0% available)
 
 ### Network Connectivity
 
-Tests basic internet connectivity for API calls and job processing.
+Tests TCP reachability of the worker API host at worker start.
 
-- **Default**: 5 second timeout to 8.8.8.8:53
-- **Configure**: `RUNPOD_NETWORK_CHECK_TIMEOUT=10`
-
-What it checks:
-- Connection to Google DNS (8.8.8.8 port 53)
-- Response latency
-- Overall internet accessibility
+- **Default**: Up to three attempts within a 5-second total connection/cleanup budget.
+- **Configure**: `RUNPOD_NETWORK_CHECK_TIMEOUT=10` (positive seconds).
+- **Target**: Host and port from `RUNPOD_WEBHOOK_GET_JOB`; defaults to `api.runpod.ai:443` if absent. URL paths and credentials are not sent or logged by this probe.
+- Tests connection reachability, not API authentication or full application readiness.
+- Retries temporary connection failures; persistent failure exits through the worker failure path.
 
 Example log output:
 ```
-Network connectivity passed: Connected to 8.8.8.8 (45ms)
+Network connectivity passed: Connected to api.runpod.ai:443
 ```
 
 ### CUDA Version (GPU workers only)
@@ -343,15 +357,15 @@ ERROR  | Fitness check failed: _cuda_init_check | RuntimeError: Failed to initia
 
 Quick matrix multiplication to verify GPU compute functionality and responsiveness. Skips silently on CPU-only workers.
 
-- **Default**: 100ms maximum execution time
-- **Configure**: `RUNPOD_GPU_BENCHMARK_TIMEOUT=2`
+- **Default**: 2 seconds maximum execution time
+- **Configure**: `RUNPOD_GPU_BENCHMARK_TIMEOUT=2` (seconds)
 
 What it tests:
 - GPU compute capability (matrix multiplication)
 - GPU response time
 - Memory bandwidth to GPU
 
-If the operation takes longer than 100ms, the worker exits as the GPU is too slow for reliable job processing.
+If the operation takes longer than the timeout, the worker exits as the GPU is too slow for reliable job processing.
 
 Example log output:
 ```
@@ -371,13 +385,15 @@ ENV RUNPOD_NETWORK_CHECK_TIMEOUT=10
 ENV RUNPOD_GPU_BENCHMARK_TIMEOUT=2
 ```
 
-Or in Python:
+For deferred launches, settings can also be configured in Python before worker start:
 
 ```python
 import os
 
 os.environ["RUNPOD_MIN_MEMORY_GB"] = "8.0"
 os.environ["RUNPOD_MIN_DISK_PERCENT"] = "15.0"
+
+import runpod
 ```
 
 ### Disabling Built-in Checks
@@ -388,6 +404,8 @@ For testing or specialized deployments, built-in checks can be disabled via envi
 |---|---|
 | `RUNPOD_SKIP_AUTO_SYSTEM_CHECKS=true` | Skips auto-registration of memory, disk, network, CUDA version, CUDA init, and GPU benchmark checks |
 | `RUNPOD_SKIP_GPU_CHECK=true` | Skips auto-registration of the native GPU memory allocation test (`gpu_test` binary) |
+| `RUNPOD_SKIP_FITNESS_CHECKS=true` | Skips every fitness check, built-in **and** user-registered |
+| `RUNPOD_DEFER_FITNESS_CHECKS=true` | Keeps the checks but runs them only at `runpod.serverless.start()`, not at import |
 
 ```python
 import os
@@ -397,15 +415,19 @@ os.environ["RUNPOD_SKIP_AUTO_SYSTEM_CHECKS"] = "true"
 
 # Disable the automatic GPU memory allocation test
 os.environ["RUNPOD_SKIP_GPU_CHECK"] = "true"
+
+import runpod
 ```
 
-User-registered checks via `@register_fitness_check` still run regardless of these flags.
+For early checks, set these before launching the handler. For deferred launches, set them before worker start.
+
+User-registered checks via `@register_fitness_check` still run regardless of `RUNPOD_SKIP_AUTO_SYSTEM_CHECKS` and `RUNPOD_SKIP_GPU_CHECK`. Only `RUNPOD_SKIP_FITNESS_CHECKS` disables those too.
 
 ## Behavior
 
 ### Execution Timing
 
-- Fitness checks run **only once at worker startup**
+- Early checks run in eligible Serverless containers; the final pass runs before job processing. Successful checks are reused unless their configuration changes.
 - They run **before the first job is processed**
 - They run **only on the actual Runpod serverless platform**
 - Local development and testing modes skip fitness checks
@@ -555,7 +577,7 @@ async def check_api_with_retry():
 
 ## Testing
 
-When developing locally, fitness checks don't run. To test them, you can manually invoke the runner:
+When developing locally, fitness checks don't run. To test them, you can manually invoke the runner. Note that each check runs once per process: a second `run_fitness_checks()` call skips checks that already passed, so call `clear_fitness_checks()` (as below) between runs:
 
 ```python
 import asyncio

runpod/__init__.py

@@ -3,6 +3,10 @@
 import logging
 import os
 
+from ._startup import run_import_checks
+
+run_import_checks()
+
 from . import serverless
 from .api.ctl_commands import (
     create_container_registry_auth,

runpod/_health/__init__.py

@@ -0,0 +1,22 @@
+"""Lightweight Serverless environment detection; no SDK imports."""
+
+import os
+import sys
+
+
+def is_serverless_environment() -> bool:
+    """Recognize production worker configuration, excluding platform tests."""
+    return (
+        bool(os.environ.get("RUNPOD_ENDPOINT_ID", "").strip())
+        and bool(os.environ.get("RUNPOD_WEBHOOK_GET_JOB", "").strip())
+        and os.environ.get("RUNPOD_TEST", "").strip().lower()
+        not in ("1", "true", "yes", "on")
+    )
+
+
+def is_early_check_eligible() -> bool:
+    """Eligibility for shared early checks, not an assertion of process identity."""
+    return is_serverless_environment() and not any(
+        arg.split("=", 1)[0] in ("--test_input", "--rp_serve_api")
+        for arg in sys.argv[1:]
+    )

runpod/_health/coordination.py

@@ -0,0 +1,108 @@
+"""Linux container-start-scoped coordination for shared health checks."""
+
+import asyncio
+import hashlib
+import json
+import os
+import time
+from pathlib import Path
+
+
+class CoordinationUnavailable(Exception):
+    """Shared state cannot be used; worker-start checks remain available."""
+
+
+class CoordinationBusy(Exception):
+    """Another process did not finish within the bounded wait."""
+
+
+def container_start_id() -> str:
+    """PID namespace + init start ticks + host boot distinguish container restarts.
+
+    Use fixed /tmp rather than TMPDIR (which can differ between processes).
+    No pod-id-only marker: container files can survive a restart.
+    """
+    boot = Path("/proc/sys/kernel/random/boot_id").read_text().strip()
+    stat = Path("/proc/1/stat").read_text()
+    start_ticks = stat.rsplit(")", 1)[1].split()[19]
+    namespace = os.readlink("/proc/1/ns/pid")
+    return hashlib.sha256(f"{boot}:{namespace}:{start_ticks}".encode()).hexdigest()
+
+
+class ContainerChecks:
+    """Hold one flock while reading, executing, and recording early checks."""
+
+    def __init__(self, timeout: float = 35):
+        self.timeout = timeout
+        self.fd = None
+        self.state = {"passed": [], "failure": None}
+
+    async def __aenter__(self):
+        try:
+            import fcntl
+
+            identity = container_start_id()
+            path = f"/tmp/runpod-fitness-{identity}.json"
+            self.fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+            os.set_inheritable(self.fd, False)
+        except (OSError, ValueError, IndexError, ImportError) as exc:
+            self.close()
+            raise CoordinationUnavailable(str(exc)) from exc
+        try:
+            await self._acquire_lock(fcntl)
+            self._load_state()
+            return self
+        except (OSError, ValueError) as exc:
+            self.close()
+            raise CoordinationUnavailable(str(exc)) from exc
+        except BaseException:
+            self.close()
+            raise
+
+    async def _acquire_lock(self, fcntl) -> None:
+        deadline = time.monotonic() + self.timeout
+        while True:
+            try:
+                fcntl.flock(self.fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+                return
+            except BlockingIOError:
+                if time.monotonic() >= deadline:
+                    raise CoordinationBusy("Timed out waiting for early health checks")
+                await asyncio.sleep(0.05)
+
+    def _load_state(self) -> None:
+        raw = os.read(self.fd, 65536)
+        if not raw:
+            return
+        state = json.loads(raw)
+        if not isinstance(state, dict):
+            raise ValueError("Health-check state must be an object")
+        passed = state.get("passed")
+        if not isinstance(passed, list) or not all(
+            isinstance(key, str) for key in passed
+        ):
+            raise ValueError("Passed health checks must be a list of cache keys")
+        if not isinstance(state.get("failure"), (str, type(None))):
+            raise ValueError("Health-check failure must be a string or null")
+        self.state = state
+
+    def save(self) -> None:
+        """Persist before releasing the lock or terminating on failure."""
+        data = json.dumps(self.state).encode()
+        os.lseek(self.fd, 0, os.SEEK_SET)
+        remaining = memoryview(data)
+        while remaining:
+            written = os.write(self.fd, remaining)
+            if written <= 0:
+                raise OSError("Unable to persist health-check state")
+            remaining = remaining[written:]
+        os.ftruncate(self.fd, len(data))
+        os.fsync(self.fd)
+
+    def close(self) -> None:
+        if self.fd is not None:
+            os.close(self.fd)
+            self.fd = None
+
+    async def __aexit__(self, *args):
+        self.close()

runpod/_health/cuda.py

@@ -0,0 +1,22 @@
+"""
+Provides some of the torch.cuda functionality without requiring torch.
+"""
+
+import subprocess
+
+
+def is_available():
+    """
+    Returns True if CUDA is available, False otherwise.
+    """
+    try:
+        # Bounded: this runs at `import runpod` on real workers, where a wedged
+        # nvidia-smi must not hang the boot forever.
+        output = subprocess.check_output(
+            ["nvidia-smi"], stderr=subprocess.DEVNULL, timeout=5
+        )
+        if "NVIDIA-SMI" in output.decode():
+            return True
+    except Exception:  # pylint: disable=broad-except
+        pass
+    return False