From 168b7cad5f41fb08c4e9d19058daa8b3bd08d7ad Mon Sep 17 00:00:00 2001 From: oeway Date: Wed, 12 Aug 2026 15:35:32 +0200 Subject: [PATCH 1/2] =?UTF-8?q?fix(startup):=20bound=20+=20parallelize=20c?= =?UTF-8?q?heck=5Fand=5Fcleanup=5Fservers=20(#0043)=20=E2=80=94=200.21.134?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Sibling of #0015. #0015 moved the orphaned-CLIENT reap off the readiness path and made it concurrent, but left the SERVER-check phase (RedisStore.check_and_cleanup_servers, awaited inline in init()) with the same O(N x timeout) shape that CrashLooped the pod: * list_servers() scans services:*|*:public/*:built-in@* — the built-in of EVERY public-workspace client, so after a crash/rollout a non-reset (prod) Redis carries a pile of DEAD built-in registrations. * The old loop pinged each SEQUENTIALLY; a dead client never answers, so get_remote_service(..., {"timeout": 2}) burned the full timeout per server, and svc.ping("ping") itself had NO explicit timeout — a half-open peer that resolves but never replies could hang boot forever. Fix (readiness path stays bounded): * Probe servers CONCURRENTLY, concurrency-capped (HYPHA_SERVER_CHECK_CONCURRENCY, default 50) → one pass ~ceil(N/cap) x timeout, not N x timeout. * Bound EACH probe: both get_remote_service resolution AND the ping round-trip get an explicit timeout (HYPHA_SERVER_CHECK_TIMEOUT, default 2). * Bound the WHOLE phase with an overall deadline (HYPHA_SERVER_CHECK_DEADLINE, default 60); on deadline, log + continue startup (the background reaper trims the rest) rather than block readiness. * Preserve the duplicate-self-id guard (a LIVE server already holding this server's id still raises RuntimeError) and the exact dead-server cleanup key patterns (server-owned services only; user services are untouched, as before). Also bound the ping in _cleanup_orphaned_client_services._probe with asyncio.wait_for (item 4): resolution was bounded but the ping was not. Tests (real, docker-free, fakeredis; reproduce-before-fix): seed genuine dead built-in registrations in a non-reset Redis (prod-like — a reset boot flushes the pile and can never reproduce it) and assert boot is prompt and the dead registrations are cleaned. Pre-fix: init took 24.3s for 12 dead servers (12 x ~2s serial); post-fix ~1s. Co-Authored-By: Claude Opus 4.8 --- docker-compose/README.md | 2 +- docker-compose/docker-compose.yml | 2 +- helm-charts/aks-hypha.md | 2 +- helm-charts/hypha-server-kit/Chart.lock | 2 +- helm-charts/hypha-server-kit/Chart.yaml | 6 +- helm-charts/hypha-server-kit/values.yaml | 2 +- helm-charts/hypha-server/Chart.yaml | 2 +- helm-charts/hypha-server/README.md | 2 +- helm-charts/hypha-server/values.yaml | 2 +- hypha/VERSION | 2 +- hypha/core/store.py | 179 +++++++++++++++++------ tests/test_server_check_bounded.py | 149 +++++++++++++++++++ 12 files changed, 295 insertions(+), 57 deletions(-) create mode 100644 tests/test_server_check_bounded.py diff --git a/docker-compose/README.md b/docker-compose/README.md index ffe87f28..966e499f 100644 --- a/docker-compose/README.md +++ b/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 diff --git a/docker-compose/docker-compose.yml b/docker-compose/docker-compose.yml index 2f57cbc3..076effe8 100644 --- a/docker-compose/docker-compose.yml +++ b/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: diff --git a/helm-charts/aks-hypha.md b/helm-charts/aks-hypha.md index 2850ca41..f2bc1c01 100644 --- a/helm-charts/aks-hypha.md +++ b/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 diff --git a/helm-charts/hypha-server-kit/Chart.lock b/helm-charts/hypha-server-kit/Chart.lock index f9cc37aa..738e19ae 100644 --- a/helm-charts/hypha-server-kit/Chart.lock +++ b/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 diff --git a/helm-charts/hypha-server-kit/Chart.yaml b/helm-charts/hypha-server-kit/Chart.yaml index 896a75ca..4b6ab370 100644 --- a/helm-charts/hypha-server-kit/Chart.yaml +++ b/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 diff --git a/helm-charts/hypha-server-kit/values.yaml b/helm-charts/hypha-server-kit/values.yaml index 27667e2b..2d21c7dd 100644 --- a/helm-charts/hypha-server-kit/values.yaml +++ b/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 diff --git a/helm-charts/hypha-server/Chart.yaml b/helm-charts/hypha-server/Chart.yaml index 46eb1041..f4dc583c 100644 --- a/helm-charts/hypha-server/Chart.yaml +++ b/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 diff --git a/helm-charts/hypha-server/README.md b/helm-charts/hypha-server/README.md index 89462b3e..e30a61a1 100644 --- a/helm-charts/hypha-server/README.md +++ b/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` | diff --git a/helm-charts/hypha-server/values.yaml b/helm-charts/hypha-server/values.yaml index 973aea34..d22f89c0 100644 --- a/helm-charts/hypha-server/values.yaml +++ b/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: "" diff --git a/hypha/VERSION b/hypha/VERSION index 880b18af..3c4dcad0 100644 --- a/hypha/VERSION +++ b/hypha/VERSION @@ -1 +1 @@ -{"version": "0.21.133"} +{"version": "0.21.134"} diff --git a/hypha/core/store.py b/hypha/core/store.py index 315d3aa4..e7fc23a2 100644 --- a/hypha/core/store.py +++ b/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) diff --git a/tests/test_server_check_bounded.py b/tests/test_server_check_bounded.py new file mode 100644 index 00000000..92879520 --- /dev/null +++ b/tests/test_server_check_bounded.py @@ -0,0 +1,149 @@ +"""Startup ``check_and_cleanup_servers`` must be BOUNDED and CONCURRENT (Task A). + +Sibling of #0015 (``test_orphan_reaper.py``). #0015 fixed the *orphaned client* +reap (moved it off the readiness path + made it concurrent). But the +*server-check* phase — ``RedisStore.check_and_cleanup_servers`` — was left on the +readiness path (``init()`` awaits it at ``store.py`` ~1026) and still had the +SAME O(N x timeout) shape that CrashLooped the pod: + + * ``list_servers()`` scans ``services:*|*:public/*:built-in@*`` — that is the + built-in service of EVERY public-workspace client, not just live hypha + instances. After a crash/rollout a pile of DEAD public built-in + registrations survives in a non-reset (prod) Redis. + * The old loop pinged each one SEQUENTIALLY; a dead client never answers, so + ``get_remote_service(..., {"timeout": 2})`` burned the full timeout for each, + and the ``svc.ping("ping")`` call itself had NO explicit timeout at all — + a half-open peer that resolves but never replies could hang boot forever. + +So N dead server registrations added O(N x 2s) — or unbounded — to the readiness +path, exactly the #0015 failure mode on a different phase. + +The fix (asserted here, docker-free, fakeredis): + 1. Probe servers CONCURRENTLY (concurrency-capped), so one pass over a pile is + ~ceil(N/cap) x timeout, not N x timeout. + 2. Bound EACH probe: both the ``get_remote_service`` resolution AND the + ``svc.ping`` are given an explicit timeout (``HYPHA_SERVER_CHECK_TIMEOUT``). + 3. Bound the WHOLE phase with an overall deadline + (``HYPHA_SERVER_CHECK_DEADLINE``); on deadline, log + continue rather than + block readiness. + +These are real tests: they seed genuine dead built-in registrations in fakeredis +and boot a real ``RedisStore`` WITHOUT resetting Redis (the prod-like condition — +a reset boot flushes the pile and can never reproduce it), then assert boot is +prompt and the dead registrations are cleaned. +""" + +import asyncio +import time + +import pytest + +from hypha.core.store import RedisStore + +pytestmark = pytest.mark.asyncio + + +async def _seed_dead_server(redis, server_id): + """Seed a dead server's public built-in registration + a couple of its + services, exactly the shape ``list_servers()`` enumerates and + ``check_and_cleanup_servers`` cleans. Values are never parsed (scan+delete by + key pattern), so an empty JSON blob is sufficient.""" + keys = [ + f"services:public|built-in:public/{server_id}:built-in@default", + f"services:public|test:public/{server_id}:svc-a@default", + f"services:public|test:public/{server_id}-worker1:svc-b@default", + ] + for k in keys: + await redis.set(k, b"{}") + return keys + + +async def _keys_for_server(store, server_id): + return await store._scan_keys(f"services:*|*:public/{server_id}:*@*") + + +async def test_dead_server_pile_does_not_block_boot_and_is_cleaned(monkeypatch): + """A pile of dead server registrations in a non-reset Redis must NOT make + boot O(N x timeout), and must be cleaned during boot. + + Old (serial, hardcoded 2s resolution timeout): N=12 dead servers add ~24s to + init. New (concurrent, honoring HYPHA_SERVER_CHECK_TIMEOUT=1): ~1s. + """ + monkeypatch.setenv("HYPHA_SERVER_CHECK_TIMEOUT", "1") + monkeypatch.setenv("HYPHA_SERVER_CHECK_CONCURRENCY", "50") + # Park the background orphan reaper so it can't interfere with timing. + monkeypatch.setenv("HYPHA_ORPHAN_REAP_INITIAL_DELAY", "60") + + store = RedisStore(None, redis_uri=None) + + n = 12 + for i in range(n): + await _seed_dead_server(store._redis, f"deadsrv{i}") + for i in range(n): + assert await _keys_for_server(store, f"deadsrv{i}"), "seed failed" + + t0 = time.perf_counter() + await store.init(reset_redis=False) + elapsed = time.perf_counter() - t0 + try: + # Boot was not serialized over the dead pile. Serial-old would be + # ~n x 2s = ~24s on top of base init; concurrent is ~1s. Generous bound + # that still fails the old sequential implementation. + assert elapsed < 12, ( + f"init() took {elapsed:.1f}s for {n} dead servers — server-check " + f"phase not concurrent/bounded (expected ~base+1s)" + ) + + # The dead servers' service keys were cleaned during boot. + for i in range(n): + assert not await _keys_for_server(store, f"deadsrv{i}"), ( + f"deadsrv{i} keys should have been cleaned by check_and_cleanup_servers" + ) + finally: + await store.teardown() + + +async def test_server_check_overall_deadline_does_not_hang_boot(monkeypatch): + """Even if per-probe timeouts are large, the whole server-check phase is + bounded by an overall deadline — boot proceeds (log + continue) instead of + blocking on a slow/half-open pile. + + Per-probe timeout 30s but deadline 2s: the phase must return in ~2s, not 30s. + """ + monkeypatch.setenv("HYPHA_SERVER_CHECK_TIMEOUT", "30") + monkeypatch.setenv("HYPHA_SERVER_CHECK_DEADLINE", "2") + monkeypatch.setenv("HYPHA_SERVER_CHECK_CONCURRENCY", "50") + monkeypatch.setenv("HYPHA_ORPHAN_REAP_INITIAL_DELAY", "60") + + store = RedisStore(None, redis_uri=None) + + n = 5 + for i in range(n): + await _seed_dead_server(store._redis, f"slowsrv{i}") + + t0 = time.perf_counter() + await store.init(reset_redis=False) + elapsed = time.perf_counter() - t0 + try: + # Deadline (2s) cuts the phase well before the 30s per-probe timeout. + assert elapsed < 12, ( + f"init() took {elapsed:.1f}s — overall server-check deadline did not " + f"bound the phase (per-probe timeout was 30s)" + ) + finally: + await store.teardown() + + +async def test_clean_boot_unaffected(monkeypatch): + """Guard: with no dead pile, a normal boot still succeeds and registers this + server (the hardening must not regress the happy path).""" + monkeypatch.setenv("HYPHA_ORPHAN_REAP_INITIAL_DELAY", "60") + store = RedisStore(None, redis_uri=None) + await store.init(reset_redis=True) + try: + servers = await store.list_servers() + assert store._server_id in servers, ( + f"this server should be registered after boot, got {servers}" + ) + finally: + await store.teardown() From 730c6075836f7e8bb0c58f926f0510010819cdd9 Mon Sep 17 00:00:00 2001 From: oeway Date: Wed, 12 Aug 2026 15:38:11 +0200 Subject: [PATCH 2/2] docs(claude): record #0043 sibling lesson to #0015 readiness-path bound Co-Authored-By: Claude Opus 4.8 --- CLAUDE.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/CLAUDE.md b/CLAUDE.md index 8be16c13..04dea2bb 100644 --- a/CLAUDE.md +++ b/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