fix(startup): bound + parallelize check_and_cleanup_servers (#0043) — 0.21.134

CLAUDE.md

@@ -624,6 +624,29 @@ turns a one-off incident into a **CrashLoop that feeds itself**.
   promptly AND the orphans are still present right after** (proving the reap did
   not run inline) — a timing-independent proof.
 
+- **Sibling fix (#0043, 0.21.134):** #0015 hardened only the orphaned-**client**
+  reap. The **server**-check phase `check_and_cleanup_servers` (also awaited inline
+  in `init()`) had the *identical* `O(N × timeout)` shape — `list_servers()` scans
+  `services:*|*:public/*:built-in@*` (the built-in of EVERY public-workspace client,
+  so a non-reset prod Redis carries a pile of dead built-ins), and the old loop
+  pinged each SEQUENTIALLY with a hardcoded 2s resolution timeout **and a fully
+  unbounded `svc.ping`**. Fix mirrors #0015 but **stays on the readiness path** (it's
+  the cheap dup-self-id guard + dead-server sweep, which must run before this server
+  registers its own built-in): concurrent probe (`HYPHA_SERVER_CHECK_CONCURRENCY`,
+  50), each probe bounded on BOTH resolution and the ping
+  (`asyncio.wait_for(svc.ping(...), HYPHA_SERVER_CHECK_TIMEOUT=2)`), plus an overall
+  `HYPHA_SERVER_CHECK_DEADLINE` (60) — on deadline, log + continue (the deferred
+  reaper trims the rest). Contracts preserved: dup-self-id `RuntimeError` (via a
+  `nonlocal` flag checked after `gather`) and the exact dead-server cleanup key
+  patterns (server-owned only; user services untouched). The same unbounded-ping
+  (item 4) gap in `_cleanup_orphaned_client_services._probe` was fixed too. Reproduce
+  before fix: 12 dead servers → `init()` took **24.3s** (12 × ~2s serial), ~1s after.
+  `hypha/core/store.py::check_and_cleanup_servers`; tests
+  `tests/test_server_check_bounded.py`. **Lesson: when you fix an O(N×timeout)
+  readiness hang, grep for EVERY inline peer-ping/scan loop `init()` awaits — the fix
+  rarely covers them all in one pass, and `list_servers`/`list_clients`-style scans
+  enumerate far more than "live instances."**
+
 ### Multi-Replica: Per-Pod Caches Must Be Invalidated Across Pods (F6)
 
 Any **per-pod in-memory cache keyed by client/peer identity** becomes a

docker-compose/README.md

@@ -58,7 +58,7 @@ This directory contains a comprehensive Docker Compose setup for running Hypha S
 
 The main Hypha application server with full feature set enabled.
 
-- **Image**: `ghcr.io/amun-ai/hypha:0.21.133`
+- **Image**: `ghcr.io/amun-ai/hypha:0.21.134`
 - **Port**: 9527
 - **Features Enabled**:
   - Server Apps

docker-compose/docker-compose.yml

@@ -2,7 +2,7 @@ version: '3.8'
 
 services:
   hypha-server:
-    image: ghcr.io/amun-ai/hypha:0.21.133
+    image: ghcr.io/amun-ai/hypha:0.21.134
     ports:
       - "${HYPHA_PORT:-9527}:9527"
     environment:

helm-charts/aks-hypha.md

@@ -146,7 +146,7 @@ replicaCount: 1
 image:
   repository: ghcr.io/amun-ai/hypha
   pullPolicy: IfNotPresent
-  tag: "0.21.133"
+  tag: "0.21.134"
 
 serviceAccount:
   create: true

helm-charts/hypha-server-kit/Chart.lock

@@ -1,7 +1,7 @@
 dependencies:
 - name: hypha-server
   repository: file://../hypha-server
-  version: 0.21.133
+  version: 0.21.134
 - name: minio
   repository: https://charts.bitnami.com/bitnami
   version: 17.0.16

helm-charts/hypha-server-kit/Chart.yaml

@@ -2,12 +2,12 @@ apiVersion: v2
 name: hypha-server-kit
 description: A comprehensive Helm chart for Hypha server with integrated dependencies
 type: application
-version: 0.21.133
-appVersion: "0.21.133"
+version: 0.21.134
+appVersion: "0.21.134"
 
 dependencies:
   - name: hypha-server
-    version: "0.21.133"
+    version: "0.21.134"
     repository: "file://../hypha-server"
     condition: hypha-server.enabled
 

helm-charts/hypha-server-kit/values.yaml

@@ -29,7 +29,7 @@ hypha-server:
   image:
     repository: ghcr.io/amun-ai/hypha
     pullPolicy: IfNotPresent
-    tag: "0.21.133"
+    tag: "0.21.134"
 
   ingress:
     enabled: true

helm-charts/hypha-server/Chart.yaml

@@ -15,7 +15,7 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.21.133
+version: 0.21.134
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to

helm-charts/hypha-server/README.md

@@ -23,7 +23,7 @@ The following table lists the main configurable parameters of the Hypha Server c
 |-----------|-------------|---------|
 | `replicaCount` | Number of replicas | `1` |
 | `image.repository` | Image repository | `ghcr.io/amun-ai/hypha` |
-| `image.tag` | Image tag | `0.21.133` |
+| `image.tag` | Image tag | `0.21.134` |
 | `image.pullPolicy` | Image pull policy | `IfNotPresent` |
 | `service.type` | Kubernetes service type | `ClusterIP` |
 | `service.port` | Service port | `9520` |

helm-charts/hypha-server/values.yaml

@@ -8,7 +8,7 @@ image:
   repository: ghcr.io/amun-ai/hypha
   pullPolicy: IfNotPresent
   # Overrides the image tag whose default is the chart appVersion.
-  tag: "0.21.133"
+  tag: "0.21.134"
 
 imagePullSecrets: []
 nameOverride: ""

hypha/VERSION

@@ -1 +1 @@
-{"version": "0.21.133"}
+{"version": "0.21.134"}

hypha/core/store.py

@@ -681,54 +681,140 @@ async def upgrade(self):
         await self._redis.rpush("change_log", *map(str, database_change_log))
 
     async def check_and_cleanup_servers(self):
-        """Cleanup and check servers."""
+        """Detect dead hypha servers, clean up their leftover services, and guard
+        against booting with a server id that is already live.
+
+        BOUNDED + CONCURRENT (Task A). ``list_servers()`` scans
+        ``services:*|*:public/*:built-in@*`` — the built-in service of EVERY
+        public-workspace client — so after a crash/rollout a non-reset (prod)
+        Redis can carry a large pile of DEAD built-in registrations. This runs on
+        the readiness path (``init()``), so it must NOT be O(N x timeout):
+
+          * Each server is probed CONCURRENTLY (concurrency-capped), so one pass
+            is ~ceil(N/cap) x timeout instead of N x timeout.
+          * Each probe is time-bounded: BOTH the ``get_remote_service`` resolution
+            AND the ``svc.ping`` itself get an explicit timeout — a half-open peer
+            that resolves but never replies to ``ping`` can no longer hang boot.
+          * The WHOLE phase is bounded by an overall deadline; on deadline we log
+            and continue startup (the background reaper trims the rest) rather
+            than block readiness. This is the sibling fix to #0015, which
+            hardened the orphaned-client reap but left this server-check phase on
+            the readiness path with the same O(N x timeout) shape.
+        """
         server_ids = await self.list_servers()
         logger.info("Connected hypha servers: %s", server_ids)
-        if server_ids:
-            rpc = self.create_rpc(
-                "root", self._root_user, client_id="server-checker", silent=True
+        if not server_ids:
+            return
+
+        check_timeout = float(os.environ.get("HYPHA_SERVER_CHECK_TIMEOUT", "2"))
+        concurrency = max(
+            1, int(os.environ.get("HYPHA_SERVER_CHECK_CONCURRENCY", "50"))
+        )
+        deadline = float(os.environ.get("HYPHA_SERVER_CHECK_DEADLINE", "60"))
+        sem = asyncio.Semaphore(concurrency)
+
+        rpc = self.create_rpc(
+            "root", self._root_user, client_id="server-checker", silent=True
+        )
+
+        # Set if a LIVE server already holds THIS server's id (a fatal misconfig).
+        self_id_is_live = False
+
+        async def _probe(server_id):
+            # Returns server_id when the server is DEAD (to be cleaned), else None.
+            nonlocal self_id_is_live
+            async with sem:
+                try:
+                    svc = await rpc.get_remote_service(
+                        f"public/{server_id}:built-in", {"timeout": check_timeout}
+                    )
+                    # Bound the ping itself: get_remote_service's timeout only
+                    # covers resolution, not the ping round-trip.
+                    assert (
+                        await asyncio.wait_for(
+                            svc.ping("ping"), timeout=check_timeout
+                        )
+                        == "pong"
+                    )
+                except Exception as e:
+                    logger.warning(
+                        "Server %s is not responding (error: %s), marking for cleanup...",
+                        server_id,
+                        e,
+                    )
+                    return server_id
+                else:
+                    if server_id == self._server_id:
+                        self_id_is_live = True
+                    return None
+
+        dead_servers = []
+        try:
+            logger.info(
+                "Probing %d server registration(s) (timeout=%ss, concurrency=%d, deadline=%ss)",
+                len(server_ids),
+                check_timeout,
+                concurrency,
+                deadline,
             )
             try:
-                for server_id in server_ids:
-                    try:
-                        svc = await rpc.get_remote_service(
-                            f"public/{server_id}:built-in", {"timeout": 2}
-                        )
-                        assert await svc.ping("ping") == "pong"
-                    except Exception as e:
-                        logger.warning(
-                            f"Server {server_id} is not responding (error: {e}), cleaning up ALL its services..."
-                        )
-                        # Clean up ALL services from the dead server across all workspaces
-                        # When a server is dead, we need to clean up everything it might have left behind
-                        patterns = [
-                            f"services:*|*:*/{server_id}:*@*",  # Direct server services
-                            f"services:*|*:*/{server_id}-*:*@*",  # Services from server-generated clients
-                            f"services:*|*:*/manager-{server_id}:*@*",  # Manager services
-                        ]
-
-                        all_keys = []
-                        for pattern in patterns:
-                            keys = await self._scan_keys(pattern)
-                            all_keys.extend(keys)
-
-                        if all_keys:
-                            logger.info(f"Removing {len(all_keys)} services from dead server {server_id}")
-                            # Use pipeline for efficient bulk deletion
-                            pipeline = self._redis.pipeline()
-                            for key in all_keys:
-                                pipeline.delete(key)
-                            await pipeline.execute()
-                            logger.info(f"Successfully cleaned up dead server {server_id}")
-                    else:
-                        if server_id == self._server_id:
-                            raise RuntimeError(
-                                f"Server with the same id ({server_id}) is already running, please use a different server id by passing `--server-id=new-server-id` to the command line."
-                            )
-            except Exception as exp:
-                raise exp
-            finally:
-                await rpc.disconnect()
+                results = await asyncio.wait_for(
+                    asyncio.gather(*[_probe(sid) for sid in server_ids]),
+                    timeout=deadline,
+                )
+                dead_servers = [r for r in results if r is not None]
+            except asyncio.TimeoutError:
+                logger.warning(
+                    "Server-check phase exceeded its %ss deadline over %d server(s); "
+                    "continuing startup — remaining dead registrations will be trimmed "
+                    "by the background reaper.",
+                    deadline,
+                    len(server_ids),
+                )
+
+            # A LIVE server already running with our id is a hard misconfiguration.
+            if self_id_is_live:
+                raise RuntimeError(
+                    f"Server with the same id ({self._server_id}) is already running, "
+                    "please use a different server id by passing "
+                    "`--server-id=new-server-id` to the command line."
+                )
+
+            if dead_servers:
+                logger.info(
+                    "Cleaning up %d dead server(s): %s",
+                    len(dead_servers),
+                    dead_servers,
+                )
+                all_keys = []
+                for server_id in dead_servers:
+                    # Clean up ALL services the dead server might have left behind
+                    # across all workspaces.
+                    patterns = [
+                        f"services:*|*:*/{server_id}:*@*",  # Direct server services
+                        f"services:*|*:*/{server_id}-*:*@*",  # Server-generated clients
+                        f"services:*|*:*/manager-{server_id}:*@*",  # Manager services
+                    ]
+                    for pattern in patterns:
+                        keys = await self._scan_keys(pattern)
+                        all_keys.extend(keys)
+
+                if all_keys:
+                    logger.info(
+                        "Removing %d service(s) from %d dead server(s)",
+                        len(all_keys),
+                        len(dead_servers),
+                    )
+                    # Pipeline for efficient bulk deletion.
+                    pipeline = self._redis.pipeline()
+                    for key in all_keys:
+                        pipeline.delete(key)
+                    await pipeline.execute()
+                    logger.info("Successfully cleaned up dead servers")
+        except Exception as exp:
+            raise exp
+        finally:
+            await rpc.disconnect()
 
     async def _clear_client_services(self, workspace: str, client_id: str):
         """Clear a workspace."""
@@ -866,7 +952,10 @@ async def _probe(workspace, client_id):
                     svc = await rpc.get_remote_service(
                         f"{workspace}/{client_id}:built-in", {"timeout": ping_timeout}
                     )
-                    await svc.ping("ping")
+                    # Bound the ping itself: get_remote_service's timeout only
+                    # covers resolution, not the ping round-trip. A half-open peer
+                    # that resolves but never replies would otherwise hang the pass.
+                    await asyncio.wait_for(svc.ping("ping"), timeout=ping_timeout)
                     return None
                 except Exception:
                     return (workspace, client_id)