diff --git a/docker-compose/README.md b/docker-compose/README.md index 35e165bc..f4b777ac 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.135` +- **Image**: `ghcr.io/amun-ai/hypha:0.21.136` - **Port**: 9527 - **Features Enabled**: - Server Apps diff --git a/docker-compose/docker-compose.yml b/docker-compose/docker-compose.yml index 6e9f6cfa..32bf9cbf 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.135 + image: ghcr.io/amun-ai/hypha:0.21.136 ports: - "${HYPHA_PORT:-9527}:9527" environment: diff --git a/helm-charts/aks-hypha.md b/helm-charts/aks-hypha.md index f824e8a4..08acae3a 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.135" + tag: "0.21.136" serviceAccount: create: true diff --git a/helm-charts/hypha-server-kit/Chart.lock b/helm-charts/hypha-server-kit/Chart.lock index 85e5f77c..91b96c0e 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.135 + version: 0.21.136 - 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 1232805a..6b234a9c 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.135 -appVersion: "0.21.135" +version: 0.21.136 +appVersion: "0.21.136" dependencies: - name: hypha-server - version: "0.21.135" + version: "0.21.136" 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 dfc40cd0..fb676354 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.135" + tag: "0.21.136" ingress: enabled: true diff --git a/helm-charts/hypha-server/Chart.yaml b/helm-charts/hypha-server/Chart.yaml index 85cf7af5..333563c8 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.135 +version: 0.21.136 # 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 d0747a6a..c92d9e28 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.135` | +| `image.tag` | Image tag | `0.21.136` | | `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 d136d63b..4c47fef2 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.135" + tag: "0.21.136" imagePullSecrets: [] nameOverride: "" diff --git a/hypha/VERSION b/hypha/VERSION index 171cf295..4ad2fef0 100644 --- a/hypha/VERSION +++ b/hypha/VERSION @@ -1 +1 @@ -{"version": "0.21.135"} +{"version": "0.21.136"} diff --git a/hypha/core/__init__.py b/hypha/core/__init__.py index fa5681d5..43d223c4 100644 --- a/hypha/core/__init__.py +++ b/hypha/core/__init__.py @@ -1245,7 +1245,21 @@ def __init__(self, redis) -> None: self._redis_event_bus = EventBus(logger) # Track local clients for optimized routing self._local_clients = set() # Set of "workspace/client_id" strings - self._subscribed_patterns = set() # Track which patterns we've subscribed to + # `_subscribed_patterns` = the DESIRED set (what we want wired to Redis). + # `_confirmed_patterns` = the subset actually confirmed on the live + # pubsub. Separating them fixes the #1053 convergence race: a psubscribe + # that times out/errors at register time used to leave the pattern in the + # desired set but unwired, and the desired-set membership guard then made + # every later subscribe a no-op — a permanent cross-pod black hole. Now + # the guard is keyed on `_confirmed_patterns`, so a desired-but-unconfirmed + # pattern is retried (on the next subscribe call, on reconnect, and by the + # background reconciler) until it is actually wired. + self._subscribed_patterns = set() # DESIRED patterns + self._confirmed_patterns = set() # patterns confirmed on the live pubsub + self._reconcile_task = None + self._reconcile_interval = float( + os.environ.get("HYPHA_SUBSCRIPTION_RECONCILE_INTERVAL", "5.0") + ) self._pubsub = None # Store the pubsub object for dynamic subscriptions # Track recently-disconnected clients for fast dead-peer detection self._recently_disconnected = {} # {client_key: disconnect_timestamp} @@ -1301,34 +1315,95 @@ async def unregister_local_client(self, workspace: str, client_id: str): except Exception: pass + async def _ensure_pattern_subscribed(self, pattern: str) -> bool: + """Wire a DESIRED pattern to the live pubsub, idempotently. + + Returns True if the pattern is confirmed on the pubsub after this call. + A no-op (returns True) if already confirmed. Leaves the pattern + unconfirmed (returns False) if the pubsub is unavailable, the circuit + breaker is open, or the psubscribe times out/errors — so a later call + (subscribe / reconnect / reconciler) retries it rather than treating a + recorded-but-unwired pattern as done (#1053). + """ + if pattern in self._confirmed_patterns: + return True + if not self._pubsub or self._circuit_breaker_open: + return False + try: + # Add timeout to prevent hanging on Redis connection issues + await asyncio.wait_for(self._pubsub.psubscribe(pattern), timeout=5.0) + self._confirmed_patterns.add(pattern) + logger.debug("Subscribed to client events: %s", pattern) + RedisEventBus._patterns_subscribed_total.inc() + RedisEventBus._patterns_subscribed_total_int += 1 + # Mark successful operation + self._consecutive_failures = 0 + return True + except asyncio.TimeoutError: + logger.warning("Timeout subscribing to client events %s", pattern) + self._consecutive_failures += 1 + if self._consecutive_failures >= self._max_failures: + self._circuit_breaker_open = True + return False + except Exception as e: + logger.warning("Failed to subscribe to client events %s: %s", pattern, e) + self._consecutive_failures += 1 + if self._consecutive_failures >= self._max_failures: + self._circuit_breaker_open = True + return False + + async def _rewire_desired_patterns(self): + """Re-wire ALL desired targeted patterns onto the current pubsub. + + Called after (re)connecting the pubsub. The fresh pubsub has nothing + wired, so `_confirmed_patterns` is reset first; then every desired + pattern is re-attempted WITHOUT discarding on failure — a transient + re-subscribe failure must not permanently lose a client's cross-pod + events (the reconciler retries any that stay unconfirmed) — #1053. + """ + self._confirmed_patterns = set() + for pattern in list(self._subscribed_patterns): + if await self._ensure_pattern_subscribed(pattern): + logger.debug("Re-subscribed to pattern: %s", pattern) + + async def _reconcile_subscriptions(self): + """Retry wiring every DESIRED pattern that is not yet confirmed. + + Called on an interval by the background reconcile loop. In steady state + the desired and confirmed sets are equal, so this is a cheap no-op; it + only does work after a transient psubscribe failure left a pattern + recorded-but-unwired (#1053). Bounded by the number of unconfirmed + patterns. + """ + pending = self._subscribed_patterns - self._confirmed_patterns + for pattern in pending: + await self._ensure_pattern_subscribed(pattern) + + async def _reconcile_loop(self): + """Periodically reconcile desired vs confirmed subscriptions.""" + while not self._stop: + try: + await asyncio.sleep(self._reconcile_interval) + if self._stop: + break + await self._reconcile_subscriptions() + except asyncio.CancelledError: + break + except Exception as e: + logger.warning("Subscription reconcile loop error: %s", e) + async def subscribe_to_client_events(self, workspace: str, client_id: str): """Subscribe to events for a specific client using targeted prefix.""" client_key = f"{workspace}/{client_id}" # Subscribe to targeted messages for this client pattern = f"targeted:{client_key}:*" - if pattern not in self._subscribed_patterns: - # Record desired subscription regardless of pubsub availability - self._subscribed_patterns.add(pattern) - RedisEventBus._active_patterns_gauge.set(len(self._subscribed_patterns)) - if self._pubsub and not self._circuit_breaker_open: - try: - # Add timeout to prevent hanging on Redis connection issues - await asyncio.wait_for(self._pubsub.psubscribe(pattern), timeout=5.0) - logger.debug("Subscribed to client events: %s", pattern) - RedisEventBus._patterns_subscribed_total.inc() - RedisEventBus._patterns_subscribed_total_int += 1 - # Mark successful operation - self._consecutive_failures = 0 - except asyncio.TimeoutError: - logger.warning("Timeout subscribing to client events %s", pattern) - self._consecutive_failures += 1 - if self._consecutive_failures >= self._max_failures: - self._circuit_breaker_open = True - except Exception as e: - logger.warning("Failed to subscribe to client events %s: %s", pattern, e) - self._consecutive_failures += 1 - if self._consecutive_failures >= self._max_failures: - self._circuit_breaker_open = True + # Always record the DESIRED subscription regardless of pubsub + # availability. The guard is on `_confirmed_patterns` (inside + # `_ensure_pattern_subscribed`), NOT on the desired set, so a + # desired-but-unconfirmed pattern is retried instead of no-op'd (#1053). + self._subscribed_patterns.add(pattern) + RedisEventBus._active_patterns_gauge.set(len(self._subscribed_patterns)) + await self._ensure_pattern_subscribed(pattern) async def unsubscribe_from_client_events(self, workspace: str, client_id: str): """Unsubscribe from events for a specific client.""" @@ -1354,7 +1429,8 @@ async def unsubscribe_from_client_events(self, workspace: str, client_id: str): except Exception as e: logger.warning("Failed to unsubscribe from client events %s: %s", pattern, e) self._subscribed_patterns.discard(pattern) - + self._confirmed_patterns.discard(pattern) + RedisEventBus._active_patterns_gauge.set(len(self._subscribed_patterns)) def is_local_client(self, workspace: str, client_id: str) -> bool: @@ -1444,6 +1520,7 @@ async def cleanup_orphaned_patterns(self): try: await asyncio.wait_for(self._pubsub.punsubscribe(pattern), timeout=5.0) self._subscribed_patterns.discard(pattern) + self._confirmed_patterns.discard(pattern) logger.debug(f"Cleaned up orphaned pattern: {pattern}") except asyncio.TimeoutError: logger.warning(f"Timeout cleaning up orphaned pattern: {pattern}") @@ -1460,10 +1537,11 @@ async def _ensure_subscription(self, workspace: str, client_id: str): client_key = f"{workspace}/{client_id}" if not self.is_local_client(workspace, client_id): pattern = f"event:*:{client_key}:*" - if pattern not in self._subscribed_patterns and self._pubsub: + if pattern not in self._confirmed_patterns and self._pubsub: + self._subscribed_patterns.add(pattern) try: await self._pubsub.psubscribe(pattern) - self._subscribed_patterns.add(pattern) + self._confirmed_patterns.add(pattern) logger.debug(f"Subscribed to pattern: {pattern}") except Exception as e: logger.warning(f"Failed to subscribe to pattern {pattern}: {e}") @@ -1476,6 +1554,10 @@ async def init(self): # Start the Redis subscription task self._subscribe_task = loop.create_task(self._subscribe_redis()) + # Start the subscription reconciler (retries desired-but-unconfirmed + # patterns so a transient psubscribe failure never permanently + # black-holes a client's cross-pod events — #1053). + self._reconcile_task = loop.create_task(self._reconcile_loop()) # Wait for readiness signal await self._ready @@ -1637,11 +1719,13 @@ async def stop(self): # Cancel tasks first if self._subscribe_task: self._subscribe_task.cancel() + if self._reconcile_task: + self._reconcile_task.cancel() # Wait for tasks to complete try: await asyncio.gather( - self._subscribe_task, return_exceptions=True + self._subscribe_task, self._reconcile_task, return_exceptions=True ) except asyncio.CancelledError: pass @@ -1668,24 +1752,22 @@ async def _subscribe_redis(self): # Subscribe to all broadcast messages (server-wide events) await pubsub.psubscribe("broadcast:*") - - # Re-subscribe to any existing targeted client patterns - for pattern in list(self._subscribed_patterns): - try: - await pubsub.psubscribe(pattern) - logger.debug(f"Re-subscribed to pattern: {pattern}") - except Exception as e: - logger.warning(f"Failed to re-subscribe to pattern {pattern}: {e}") - self._subscribed_patterns.discard(pattern) - - if not self._ready.done(): - self._ready.set_result(True) - self._counter.labels(event="subscription", status="success").inc() - # Mark healthy on successful subscription + + # The fresh pubsub just accepted a psubscribe, so it is healthy: + # clear the breaker BEFORE re-wiring targeted patterns, otherwise + # a stale-open breaker would make `_ensure_pattern_subscribed` + # refuse to wire them on this good connection. self._last_successful_connection = time.time() self._consecutive_failures = 0 self._circuit_breaker_open = False + # Re-wire all DESIRED targeted patterns onto the fresh pubsub. + await self._rewire_desired_patterns() + + if not self._ready.done(): + self._ready.set_result(True) + self._counter.labels(event="subscription", status="success").inc() + while not self._stop: try: msg = await pubsub.get_message( diff --git a/hypha/core/store.py b/hypha/core/store.py index e7fc23a2..ae0e5c63 100644 --- a/hypha/core/store.py +++ b/hypha/core/store.py @@ -378,6 +378,12 @@ def set_sqlite_pragma(dbapi_conn, connection_record): self._leader_lease = None self._malloc_trim_task = None self._orphan_reaper_task = None + # Per-client CONSECUTIVE orphan-probe failure counts, tracked across + # reaper passes. A client's services are only reaped after it fails + # `HYPHA_ORPHAN_REAP_MIN_FAILURES` consecutive probes (#1052), so a + # single dropped cross-pod ping (e.g. a brief subscription-convergence + # window on reconnect) can never delete a live client's registration. + self._orphan_probe_failures = {} # {"workspace/client_id": int} # self._house_keeping_task = None self._shared_anonymous_user = None @@ -937,6 +943,14 @@ async def _cleanup_orphaned_client_services(self): # Bounding concurrency keeps one pass at ~ceil(N/concurrency) x timeout. concurrency = max(1, int(os.environ.get("HYPHA_ORPHAN_REAP_CONCURRENCY", "50"))) ping_timeout = float(os.environ.get("HYPHA_ORPHAN_REAP_PING_TIMEOUT", "3")) + # A client is only reaped after this many CONSECUTIVE failed probes + # (#1052). The cross-pod ping is best-effort — a single dropped message + # during a reconnect/subscription-convergence window must NOT delete a + # live client's registration. With the default 300s interval, 3 failures + # span ~10 min — far longer than any transient unreachability window + # (and longer than the subscription reconcile interval), so only a + # genuinely dead client accumulates enough failures to be reaped. + min_failures = max(1, int(os.environ.get("HYPHA_ORPHAN_REAP_MIN_FAILURES", "3"))) sem = asyncio.Semaphore(concurrency) # Create a temporary RPC to ping clients @@ -960,34 +974,69 @@ async def _probe(workspace, client_id): except Exception: return (workspace, client_id) - orphaned_clients = [] try: results = await asyncio.gather( *[_probe(ws, cid) for ws, cid in candidates] ) - orphaned_clients = [r for r in results if r is not None] + unreachable = {f"{r[0]}/{r[1]}" for r in results if r is not None} + + # Update the CONSECUTIVE-failure counters (#1052): increment for + # every unreachable client, reset any client that answered, and drop + # counters for clients no longer present (reconnected away / already + # gone). Only clients that have failed `min_failures` consecutive + # passes are actually reaped, so one dropped cross-pod ping cannot + # delete a live client's registration. + candidate_keys = {f"{ws}/{cid}" for ws, cid in candidates} + self._orphan_probe_failures = { + k: v for k, v in self._orphan_probe_failures.items() + if k in candidate_keys + } + confirmed_orphans = [] + for ws, cid in candidates: + client_key = f"{ws}/{cid}" + if client_key in unreachable: + count = self._orphan_probe_failures.get(client_key, 0) + 1 + self._orphan_probe_failures[client_key] = count + if count >= min_failures: + confirmed_orphans.append((ws, cid)) + else: + self._orphan_probe_failures.pop(client_key, None) - if orphaned_clients: + if unreachable: + logger.info( + "Orphan reaper: %d client(s) unreachable this pass, " + "%d confirmed dead (>=%d consecutive failures)", + len(unreachable), + len(confirmed_orphans), + min_failures, + ) + + if confirmed_orphans: logger.warning( "Found %d orphaned clients, cleaning up their services...", - len(orphaned_clients), + len(confirmed_orphans), ) pipeline = self._redis.pipeline() total_keys = 0 - for workspace, client_id in orphaned_clients: + for workspace, client_id in confirmed_orphans: svc_pattern = f"services:*|*:{workspace}/{client_id}:*@*" svc_keys = await self._scan_keys(svc_pattern) for k in svc_keys: pipeline.delete(k) total_keys += 1 + # Clear the counter once reaped so a client that later + # reconnects under the same id starts fresh. + self._orphan_probe_failures.pop( + f"{workspace}/{client_id}", None + ) if total_keys: await pipeline.execute() logger.info( "Removed %d service keys from %d orphaned clients", total_keys, - len(orphaned_clients), + len(confirmed_orphans), ) - else: + elif not unreachable: logger.info("No orphaned client services found") except Exception as e: logger.error("Error during orphaned client cleanup: %s", e) diff --git a/tests/test_cross_pod_subscription_convergence.py b/tests/test_cross_pod_subscription_convergence.py new file mode 100644 index 00000000..13153cbd --- /dev/null +++ b/tests/test_cross_pod_subscription_convergence.py @@ -0,0 +1,192 @@ +"""Issue #1053: cross-pod RPC subscription-convergence race. + +When a provider pod subscribes to a client's targeted events, it psubscribes the +Redis pattern ``targeted:/:*``. The old code recorded the pattern +in ``_subscribed_patterns`` *optimistically* — BEFORE the psubscribe was +confirmed — and on a psubscribe timeout/error left the pattern recorded but NOT +wired to Redis. The ``if pattern not in self._subscribed_patterns`` guard then +turned every later subscribe call into a no-op, so the pattern was a **permanent +black hole**: cross-pod targeted RPC to that client was silently dropped forever +(masked for same-pod callers by the local short-circuit in ``emit``). + +This reproduces the failure with two ``RedisEventBus`` instances sharing one +fakeredis (= two pods on one Redis), mirroring +``tests/test_cross_pod_reconnect.py`` — no Docker, deterministic. +""" +import asyncio + +import pytest +from fakeredis import aioredis as fakeredis + +from hypha.core import RedisEventBus + +pytestmark = pytest.mark.asyncio + + +async def _make_bus(redis): + bus = RedisEventBus(redis) + await bus.init() + return bus + + +async def _deliver(sender_bus, receiver_bus, ws, client_id, timeout=1.5): + """Emit a targeted message from ``sender_bus`` and return the payload if it + is delivered to a handler on ``receiver_bus`` within ``timeout`` (else None). + + ``sender_bus`` must NOT have the target client registered locally, otherwise + ``emit`` short-circuits to a local delivery and never hits Redis pub/sub. + """ + event = f"{ws}/{client_id}:msg" + payload = {"hello": client_id} + loop = asyncio.get_running_loop() + fut = loop.create_future() + + def handler(data): + if not fut.done(): + fut.set_result(data) + + receiver_bus.on(event, handler) + try: + deadline = loop.time() + timeout + while loop.time() < deadline and not fut.done(): + res = sender_bus.emit(event, payload) + if asyncio.iscoroutine(res): + await res + try: + return await asyncio.wait_for(asyncio.shield(fut), timeout=0.15) + except asyncio.TimeoutError: + continue + return fut.result() if fut.done() else None + finally: + receiver_bus.off(event, handler) + + +def _break_psubscribe_for(bus, pattern): + """Make ``bus``'s pubsub raise TimeoutError for exactly ``pattern`` (as the + real ``asyncio.wait_for(psubscribe(...), 5.0)`` timeout would), reproducing + the recorded-but-unwired state. Returns the original psubscribe to restore. + """ + orig = bus._pubsub.psubscribe + + async def flaky(*channels, **kwargs): + if channels and channels[0] == pattern: + raise asyncio.TimeoutError() + return await orig(*channels, **kwargs) + + bus._pubsub.psubscribe = flaky + return orig + + +async def test_targeted_subscribe_timeout_is_reconciled_via_retry(): + """A psubscribe timeout at register time must NOT permanently black-hole the + client: once Redis recovers, a subsequent subscribe call must re-wire the + pattern and cross-pod delivery must succeed. + + Reproduce-before-fix: on old code the second subscribe is a no-op (guard on + the desired set) so delivery stays broken -> this test fails. + """ + redis = fakeredis.FakeRedis.from_url("redis://localhost:9997/11") + pod_a = await _make_bus(redis) # provider pod + pod_b = await _make_bus(redis) # caller pod + ws, client_id = "wsA", "clientA" + pattern = f"targeted:{ws}/{client_id}:*" + try: + # Register clientA as local on pod_a so the receive path is realistic, + # but keep pod_b unaware of it (forces pod_b -> Redis pub/sub). + pod_a.register_local_client_sync(ws, client_id) + + # 1) psubscribe times out at register time -> recorded but not wired. + restore = _break_psubscribe_for(pod_a, pattern) + await pod_a.subscribe_to_client_events(ws, client_id) + assert pattern in pod_a._subscribed_patterns # desired recorded + assert pattern not in pod_a._confirmed_patterns # but not wired + + # Cross-pod delivery must fail while the pattern is unwired. + assert await _deliver(pod_b, pod_a, ws, client_id, timeout=0.8) is None + + # 2) Redis recovers; a subsequent subscribe call must re-wire it. + pod_a._pubsub.psubscribe = restore + await pod_a.subscribe_to_client_events(ws, client_id) + assert pattern in pod_a._confirmed_patterns + + # Now cross-pod delivery works. + got = await _deliver(pod_b, pod_a, ws, client_id, timeout=2.0) + assert got == {"hello": client_id}, "cross-pod targeted delivery must converge" + finally: + await pod_a.stop() + await pod_b.stop() + + +async def test_reconcile_loop_rewires_without_a_second_subscribe_call(): + """subscribe_to_client_events is called ONCE per client (at register), so a + register-time timeout needs an active reconciler to converge — not another + subscribe call. The background reconcile must re-wire an unconfirmed desired + pattern once Redis recovers.""" + redis = fakeredis.FakeRedis.from_url("redis://localhost:9997/12") + pod_a = await _make_bus(redis) + pod_b = await _make_bus(redis) + ws, client_id = "wsA", "clientB" + pattern = f"targeted:{ws}/{client_id}:*" + try: + pod_a.register_local_client_sync(ws, client_id) + restore = _break_psubscribe_for(pod_a, pattern) + await pod_a.subscribe_to_client_events(ws, client_id) + assert pattern not in pod_a._confirmed_patterns + + # Redis recovers; the ONLY convergence mechanism now is the reconciler. + pod_a._pubsub.psubscribe = restore + await pod_a._reconcile_subscriptions() + assert pattern in pod_a._confirmed_patterns + + got = await _deliver(pod_b, pod_a, ws, client_id, timeout=2.0) + assert got == {"hello": client_id} + finally: + await pod_a.stop() + await pod_b.stop() + + +async def test_rewire_on_reconnect_preserves_desired_and_never_discards(): + """On a pubsub reconnect, a transiently-failing psubscribe for one desired + pattern must NOT discard it (the old code did ``discard`` on failure -> + permanent cross-pod loss). Every desired pattern must survive; the healthy + one is confirmed and the flaky one is left unconfirmed for the reconciler.""" + redis = fakeredis.FakeRedis.from_url("redis://localhost:9997/13") + pod_a = await _make_bus(redis) + pod_b = await _make_bus(redis) + ws = "wsA" + good = "clientGood" + flaky_client = "clientFlaky" + good_pat = f"targeted:{ws}/{good}:*" + flaky_pat = f"targeted:{ws}/{flaky_client}:*" + try: + pod_a.register_local_client_sync(ws, good) + pod_a.register_local_client_sync(ws, flaky_client) + await pod_a.subscribe_to_client_events(ws, good) + await pod_a.subscribe_to_client_events(ws, flaky_client) + assert {good_pat, flaky_pat} <= pod_a._confirmed_patterns + + # Simulate a reconnect: swap in a fresh pubsub where re-subscribing the + # flaky client's pattern fails, then run the re-wire step directly. + fresh = redis.pubsub() + pod_a._pubsub = fresh + restore = _break_psubscribe_for(pod_a, flaky_pat) + await pod_a._rewire_desired_patterns() + + # Neither desired pattern is discarded; the good one is re-wired now, + # the flaky one is left for the reconciler. + assert {good_pat, flaky_pat} <= pod_a._subscribed_patterns, ( + "desired patterns must survive a re-subscribe failure" + ) + assert good_pat in pod_a._confirmed_patterns + assert flaky_pat not in pod_a._confirmed_patterns + + # Redis recovers -> the reconciler converges the flaky one too. + pod_a._pubsub.psubscribe = restore + await pod_a._reconcile_subscriptions() + assert flaky_pat in pod_a._confirmed_patterns + + got = await _deliver(pod_b, pod_a, ws, good, timeout=2.0) + assert got == {"hello": good} + finally: + await pod_a.stop() + await pod_b.stop() diff --git a/tests/test_orphan_reaper.py b/tests/test_orphan_reaper.py index c3ef9a9d..7ed6099c 100644 --- a/tests/test_orphan_reaper.py +++ b/tests/test_orphan_reaper.py @@ -52,6 +52,10 @@ async def test_reap_removes_dead_clients_concurrently(monkeypatch): pings CONCURRENTLY so a pile of orphans is bounded, not O(N x timeout).""" monkeypatch.setenv("HYPHA_ORPHAN_REAP_PING_TIMEOUT", "1") monkeypatch.setenv("HYPHA_ORPHAN_REAP_CONCURRENCY", "50") + # This test asserts the concurrency/timing of a SINGLE reap pass; require + # only one failed probe so one pass reaps (the consecutive-failure guard is + # covered separately in test_orphan_reaper_confirmation.py — #1052). + monkeypatch.setenv("HYPHA_ORPHAN_REAP_MIN_FAILURES", "1") store = RedisStore(None, redis_uri=None) await store.init(reset_redis=True) try: diff --git a/tests/test_orphan_reaper_confirmation.py b/tests/test_orphan_reaper_confirmation.py new file mode 100644 index 00000000..f381d800 --- /dev/null +++ b/tests/test_orphan_reaper_confirmation.py @@ -0,0 +1,165 @@ +"""Issue #1052: the orphan reaper must CONFIRM death before deleting. + +`RedisStore._cleanup_orphaned_client_services` (the #0015 continuous reaper) +used to delete ALL of a client's `services:*` keys after a SINGLE failed +cross-pod ping. The cross-pod ping is best-effort (Redis pub/sub has no +buffering), so a single dropped message during a reconnect / subscription- +convergence window made the reaper delete a **live** client's registration — +a silent, permanent outage of that client's services until manual restart. + +The fix requires `HYPHA_ORPHAN_REAP_MIN_FAILURES` (default 3) CONSECUTIVE failed +probes, tracked across passes, before an irreversible delete. A client that +answers any probe has its counter reset; a client that disappears from the +candidate set has its counter pruned. This is defense-in-depth on top of the +#1053 subscription-convergence root-cause fix. + +Docker-free / fakeredis, reproduce-before-fix: on the old single-ping code +`test_single_dropped_ping_does_not_reap_live_client`'s first assertion (keys +survive one failed pass) fails — the client is reaped immediately. +""" +import asyncio + +import pytest + +from hypha.core.store import RedisStore + +pytestmark = pytest.mark.asyncio + + +async def _seed_dead_client(redis, workspace, client_id): + """Seed a client's built-in + a real service key (values are never parsed — + the reaper scans + deletes by key pattern only). Makes (ws, client_id) a + reap candidate that no live client answers for.""" + keys = [ + f"services:public|built-in:{workspace}/{client_id}:built-in@default", + f"services:public|test:{workspace}/{client_id}:user-svc@default", + ] + for k in keys: + await redis.set(k, b"{}") + return keys + + +async def _svc_keys(store, workspace, client_id): + return await store._scan_keys( + f"services:*|*:{workspace}/{client_id}:*@*" + ) + + +async def test_single_dropped_ping_does_not_reap_live_client(monkeypatch): + """One failed probe must NOT delete the client's services (reproduce-before- + fix: old code reaped on the first failure). Only after MIN_FAILURES + consecutive failures is the client reaped.""" + monkeypatch.setenv("HYPHA_ORPHAN_REAP_PING_TIMEOUT", "1") + monkeypatch.setenv("HYPHA_ORPHAN_REAP_MIN_FAILURES", "3") + monkeypatch.setenv("HYPHA_ORPHAN_REAP_INITIAL_DELAY", "60") # park bg reaper + store = RedisStore(None, redis_uri=None) + await store.init(reset_redis=True) + ws, cid = "ws-live", "live-client" + key = f"{ws}/{cid}" + try: + await _seed_dead_client(store._redis, ws, cid) + assert await _svc_keys(store, ws, cid) + + # Pass 1: one failed probe — must NOT reap (this is the #1052 bug). + await store._cleanup_orphaned_client_services() + assert await _svc_keys(store, ws, cid), ( + "a live client was reaped after a SINGLE failed cross-pod ping (#1052)" + ) + assert store._orphan_probe_failures.get(key) == 1 + + # Pass 2: still below threshold — survives. + await store._cleanup_orphaned_client_services() + assert await _svc_keys(store, ws, cid) + assert store._orphan_probe_failures.get(key) == 2 + + # Pass 3: reaches MIN_FAILURES=3 — now (and only now) reaped. + await store._cleanup_orphaned_client_services() + assert not await _svc_keys(store, ws, cid), ( + "a genuinely dead client must be reaped after MIN_FAILURES passes" + ) + # Counter is cleared once reaped. + assert key not in store._orphan_probe_failures + finally: + await store.teardown() + + +async def test_recovered_client_resets_failure_counter(monkeypatch): + """A client that answers a probe after previous failures has its counter + reset, so accumulated transient failures never add up to a reap for a + client that is intermittently reachable.""" + monkeypatch.setenv("HYPHA_ORPHAN_REAP_PING_TIMEOUT", "1") + monkeypatch.setenv("HYPHA_ORPHAN_REAP_MIN_FAILURES", "3") + monkeypatch.setenv("HYPHA_ORPHAN_REAP_INITIAL_DELAY", "60") + store = RedisStore(None, redis_uri=None) + await store.init(reset_redis=True) + ws, cid = "ws-recover", "recover-client" + key = f"{ws}/{cid}" + await store.register_workspace( + { + "id": ws, + "name": ws, + "description": "recover test", + "persistent": True, + "owners": ["root"], + "read_only": False, + }, + overwrite=False, + ) + try: + # Phase A: dead — two failed probes accumulate. + await _seed_dead_client(store._redis, ws, cid) + await store._cleanup_orphaned_client_services() + await store._cleanup_orphaned_client_services() + assert store._orphan_probe_failures.get(key) == 2 + + # Phase B: the client comes ALIVE (a real RPC client with the same id, + # whose built-in service answers ping). One probe now succeeds -> + # the counter must reset, and the client is not reaped. + async with store.get_workspace_interface( + store._root_user, ws, client_id=cid, silent=False + ): + await store._cleanup_orphaned_client_services() + assert key not in store._orphan_probe_failures, ( + "a successful probe must reset the consecutive-failure counter" + ) + assert await _svc_keys(store, ws, cid), ( + "a live, reachable client must never be reaped" + ) + finally: + await store.teardown() + + +async def test_disappeared_client_counter_is_pruned(monkeypatch): + """A client that drops out of the candidate set (its keys already gone / + reconnected away) must have its stale failure counter pruned, so counts + never leak across unrelated client lifecycles.""" + monkeypatch.setenv("HYPHA_ORPHAN_REAP_PING_TIMEOUT", "1") + # High threshold so nothing is reaped during this test — we assert on the + # counter bookkeeping, not on deletion. + monkeypatch.setenv("HYPHA_ORPHAN_REAP_MIN_FAILURES", "10") + monkeypatch.setenv("HYPHA_ORPHAN_REAP_INITIAL_DELAY", "60") + store = RedisStore(None, redis_uri=None) + await store.init(reset_redis=True) + ws = "ws-prune" + a, b = "client-a", "client-b" + try: + await _seed_dead_client(store._redis, ws, a) + b_keys = await _seed_dead_client(store._redis, ws, b) + + # Pass 1: both fail -> both counted. + await store._cleanup_orphaned_client_services() + assert store._orphan_probe_failures.get(f"{ws}/{a}") == 1 + assert store._orphan_probe_failures.get(f"{ws}/{b}") == 1 + + # b disappears entirely (keys removed by some other path / reconnect). + for k in b_keys: + await store._redis.delete(k) + + # Pass 2: only a is a candidate -> b's stale counter is pruned. + await store._cleanup_orphaned_client_services() + assert store._orphan_probe_failures.get(f"{ws}/{a}") == 2 + assert f"{ws}/{b}" not in store._orphan_probe_failures, ( + "a client no longer present must have its failure counter pruned" + ) + finally: + await store.teardown() diff --git a/tests/test_server_disconnection.py b/tests/test_server_disconnection.py index bee1870f..5d1783d6 100644 --- a/tests/test_server_disconnection.py +++ b/tests/test_server_disconnection.py @@ -833,7 +833,7 @@ async def test_normal_return_cleanup(): @pytest.mark.asyncio -async def test_orphaned_client_cleanup_on_startup(): +async def test_orphaned_client_cleanup_on_startup(monkeypatch): """Test that orphaned client services are cleaned up when a new server starts. Bug 2: When a server crashes without graceful shutdown, user-connected @@ -847,6 +847,12 @@ async def test_orphaned_client_cleanup_on_startup(): from fastapi import FastAPI from fakeredis import aioredis + # This test drives a single reap pass over a genuinely-dead client; require + # only one failed probe so that one pass reaps it. The consecutive-failure + # guard (which protects live clients racing a reconnect) is covered in + # test_orphan_reaper_confirmation.py (#1052). + monkeypatch.setenv("HYPHA_ORPHAN_REAP_MIN_FAILURES", "1") + app = FastAPI() # Use shared fakeredis