From 48bc66b42f0c4f8bd234548e559741698432a083 Mon Sep 17 00:00:00 2001 From: "Abiba (pi)" Date: Sat, 27 Jun 2026 14:12:52 +0000 Subject: [PATCH] =?UTF-8?q?feat(zulip):=20Gen=204=20=E2=80=94=20connection?= =?UTF-8?q?=20pool=20recovery,=20heartbeat=20logging,=20log=20sanitization?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fixes 4 critical reliability issues discovered during Zulip server outage recovery: 1. CLOSE-WAIT socket leak: _reconnect() now creates a fresh httpx.AsyncClient() when the queue expires, preventing stuck connection pool. Old client is explicitly aclose()'d. 2. Silent poll loop death: After sustained 502 errors, the poll loop was returning [] without logging — transport errors are swallowed by _api_call returning (None, 0). Added heartbeat logging every 5 minutes and silence detection warnings after 60s of no events. 3. HTTP client stuck pool: Added proactive client recreation after 20 consecutive empty polls (connection pool exhaustion detection) and periodic refresh every 500 polls (connection reuse limit). 4. Log spam: 502 error responses from Netbird contain full HTML pages — now truncated to 80 chars with newlines stripped. Also: - Support ZULIP_URL env var (in addition to ZULIP_SITE) - expose silence_seconds, consecutive_empty_polls, client_pool_resets in health stats - Reduced health callback interval from 600s to 300s --- plugins/platforms/zulip/adapter.py | 176 ++++++++++++++++++++++++++--- 1 file changed, 159 insertions(+), 17 deletions(-) diff --git a/plugins/platforms/zulip/adapter.py b/plugins/platforms/zulip/adapter.py index 9cc7b50..99918bd 100644 --- a/plugins/platforms/zulip/adapter.py +++ b/plugins/platforms/zulip/adapter.py @@ -1,10 +1,19 @@ """ -Zulip platform adapter (Hermes plugin) — Gen 3. +Zulip platform adapter (Hermes plugin) — Gen 4. Connects a Hermes agent to Zulip via event queue polling. DMs and @mentions are routed to the agent via the Hermes Gateway's handle_message() interface. Replies use a placeholder->edit streaming pattern for UX feedback. +Gen 4 Improvements (2026-06-27): + 1. HTTP client recreation on reconnect — new httpx.AsyncClient() after queue expiry + or sustained failures to prevent CLOSE-WAIT socket leaks + 2. Heartbeat logging — periodic "still alive" message to prove poll loop is running + 3. Log sanitization — truncates 502/error HTML to first 80 chars (no Netbird HTML spam) + 4. Connection recovery — creates fresh HTTP client after N empty poll cycles + 5. Poll silence detection — logs warnings when no events received for extended period + 6. Connection reuse limit — periodic client recreation to prevent connection pool exhaustion + Gen 3 Improvements (2026-06-26): 1. Malformed message resilience — try/except wraps each event, no crash on bad JSON 2. Periodic @all-bots refresh — background task re-resolves user ID every hour @@ -66,6 +75,13 @@ DEDUP_CLEANUP_INTERVAL = 120 # Clean old entries every 2 minutes ALL_BOTS_REFRESH_INTERVAL = 3600 # Re-resolve @all-bots user ID every hour SELFTEST_TIMEOUT = 30 # seconds to wait for self-test response +# Gen 4: Connection pool health +HEARTBEAT_INTERVAL = 300 # Log heartbeat every 5 minutes +MAX_CONSECUTIVE_EMPTY_POLLS = 20 # Reset client after this many empty polls +CLIENT_REUSE_LIMIT = 500 # Create fresh client every N polls to prevent connection leaks +POLL_SILENCE_WARN_INTERVAL = 60 # Warn if no events received for this many seconds +MAX_ERROR_LOG_LEN = 80 # Truncate error response bodies to avoid HTML spam + # Regex to strip Zulip @mention markup MENTION_CLEANER = re.compile(r"@\*\*[^*]+\*\*") @@ -158,7 +174,18 @@ class ZulipAdapter(BasePlatformAdapter): or os.getenv("ZULIP_POLL_INTERVAL", str(DEFAULT_POLL_INTERVAL)) ) + # --- Gen 4: Connection health --- + self._consecutive_empty_polls: int = 0 + self._total_poll_count: int = 0 + self._last_event_received_time: float = time.time() + self._last_heartbeat_time: float = 0.0 + self._client_pool_reset_count: int = 0 + # --- State --- + # Support both ZULIP_SITE and ZULIP_URL env vars + if not self._site: + self._site = (os.getenv("ZULIP_URL", "") or "").rstrip("/") + self._auth_header: str = _build_auth_header(self._email, self._api_key) self._queue_id: Optional[str] = None self._last_event_id: int = -1 @@ -201,7 +228,7 @@ class ZulipAdapter(BasePlatformAdapter): # --- Gen 3: Health stats callback --- self._health_callback = None - self._health_callback_interval: int = 600 # 10 min + self._health_callback_interval: int = 300 # 5 min @staticmethod def _resolve_int(val: Any, default: int) -> int: @@ -238,9 +265,14 @@ class ZulipAdapter(BasePlatformAdapter): return False try: - self._http_client = httpx.AsyncClient(timeout=30.0) + await self._create_http_client() self._health_stats["started_at"] = datetime.now(timezone.utc).isoformat() + # Reset connection health counters + self._consecutive_empty_polls = 0 + self._total_poll_count = 0 + self._last_event_received_time = time.time() + # Register event queue queue_resp, _ = await self._api_call( "POST", "/api/v1/register", @@ -342,24 +374,94 @@ class ZulipAdapter(BasePlatformAdapter): try: events = await self._fetch_events() self._health_stats["poll_count"] += 1 + self._total_poll_count += 1 self._health_stats["last_poll_at"] = datetime.now( timezone.utc ).isoformat() - for event in events: - try: - await self._process_zulip_event(event) - except Exception as e: - # Gen 3: Malformed message resilience — catch per-event - # failures so one bad message never kills the poll loop + + if events: + now = time.time() + self._consecutive_empty_polls = 0 + self._last_event_received_time = now + for event in events: + try: + await self._process_zulip_event(event) + except Exception as e: + # Gen 3: Malformed message resilience — catch per-event + # failures so one bad message never kills the poll loop + logger.warning( + "[%s] Malformed event skipped: %s. " + "Event id=%s", + self.name, e, + event.get("id", "unknown"), + ) + self._health_stats["malformed_events"] = ( + self._health_stats.get("malformed_events", 0) + 1 + ) + else: + # Gen 4: Track consecutive empty polls to detect silent failures + self._consecutive_empty_polls += 1 + + # Gen 4: Detect prolonged silence (no events for too long) + silence_duration = time.time() - self._last_event_received_time + if silence_duration > POLL_SILENCE_WARN_INTERVAL: logger.warning( - "[%s] Malformed event skipped: %s. " - "Event id=%s", - self.name, e, - event.get("id", "unknown"), + "[%s] No events received for %.0fs " + "(consecutive_empty=%d, total_polls=%d, " + "queue=%s)", + self.name, silence_duration, + self._consecutive_empty_polls, + self._total_poll_count, + self._queue_id, ) - self._health_stats["malformed_events"] = ( - self._health_stats.get("malformed_events", 0) + 1 + + # Gen 4: Reset HTTP client if too many empty polls in a row + # (connection pool likely stuck in CLOSE-WAIT) + if ( + self._consecutive_empty_polls + >= MAX_CONSECUTIVE_EMPTY_POLLS + ): + logger.warning( + "[%s] %d consecutive empty polls — " + "recreating HTTP client and reconnecting", + self.name, self._consecutive_empty_polls, ) + await self._reconnect_with_fresh_client() + backoff_idx = 0 + self._consecutive_empty_polls = 0 + continue + + # Gen 4: Periodic client refresh to prevent connection leaks + if self._total_poll_count % CLIENT_REUSE_LIMIT == 0: + logger.info( + "[%s] Client reuse limit reached (%d polls) — " + "creating fresh HTTP client", + self.name, self._total_poll_count, + ) + old_client = self._http_client + await self._create_http_client() + if old_client: + try: + await old_client.aclose() + except Exception: + pass + + # Gen 4: Heartbeat logging — prove poll loop is alive + now = time.time() + if now - self._last_heartbeat_time > HEARTBEAT_INTERVAL: + self._last_heartbeat_time = now + silence = now - self._last_event_received_time + logger.info( + "[%s] Heartbeat — polls=%d empty=%d " + "silence=%.0fs errors=%d reconnects=%d " + "queue=%s", + self.name, self._total_poll_count, + self._consecutive_empty_polls, silence, + self._health_stats["poll_errors"], + self._health_stats["reconnects"], + self._queue_id, + ) + backoff_idx = 0 except asyncio.CancelledError: break @@ -375,7 +477,7 @@ class ZulipAdapter(BasePlatformAdapter): err_str = str(e) if "BAD_EVENT_QUEUE_ID" in err_str or "queue_id" in err_str.lower(): logger.info("[%s] Queue expired, re-registering...", self.name) - await self._reconnect() + await self._reconnect_with_fresh_client() backoff_idx = 0 continue logger.warning("[%s] Poll error: %s", self.name, e) @@ -417,6 +519,28 @@ class ZulipAdapter(BasePlatformAdapter): return [e for e in events if e.get("type") == "message"] + async def _create_http_client(self) -> None: + """Create a fresh httpx client, closing the old one if it exists. + + Gen 4: This ensures a clean connection pool after sustained failures + or queue expiry, preventing CLOSE-WAIT socket leaks. + """ + old_client = self._http_client + self._http_client = httpx.AsyncClient( + timeout=httpx.Timeout(30.0, connect=15.0), + limits=httpx.Limits( + max_keepalive_connections=5, + max_connections=10, + keepalive_expiry=60.0, + ), + ) + self._client_pool_reset_count += 1 + if old_client: + try: + await old_client.aclose() + except Exception: + pass + async def _reconnect(self) -> None: """Re-register the event queue.""" self._queue_id = None @@ -434,9 +558,21 @@ class ZulipAdapter(BasePlatformAdapter): self._last_event_id = resp.get("last_event_id", -1) self._health_stats["reconnects"] += 1 logger.info("[%s] Reconnected, queue=%s", self.name, self._queue_id) + self._consecutive_empty_polls = 0 + self._last_event_received_time = time.time() except Exception as e: logger.error("[%s] Reconnect failed: %s", self.name, e) + async def _reconnect_with_fresh_client(self) -> None: + """Re-register the event queue with a fresh HTTP client. + + Gen 4: Creates a new httpx client to break out of stuck connection + pools (CLOSE-WAIT sockets from previous failures). + """ + logger.info("[%s] Reconnecting with fresh HTTP client", self.name) + await self._create_http_client() + await self._reconnect() + # ------------------------------------------------------------------ # Gen 2: Background dedup maintenance # ------------------------------------------------------------------ @@ -1023,6 +1159,10 @@ class ZulipAdapter(BasePlatformAdapter): stats["all_bots_user_id"] = self._all_bots_user_id stats["connected"] = self._connected stats["checked_at"] = now.isoformat() + stats["total_polls"] = self._total_poll_count + stats["consecutive_empty_polls"] = self._consecutive_empty_polls + stats["client_pool_resets"] = self._client_pool_reset_count + stats["silence_seconds"] = round(time.time() - self._last_event_received_time) # Compute error rate total_polls = stats.get("poll_count", 0) or 1 @@ -1074,10 +1214,12 @@ class ZulipAdapter(BasePlatformAdapter): return None, 0 if response.status_code >= 400: + # Gen 4: Truncate error bodies to prevent HTML spam + body = response.text[:MAX_ERROR_LOG_LEN].replace("\n", " ") logger.warning( "[%s] API %s %s: %d %s", self.name, method, path, - response.status_code, response.text[:200], + response.status_code, body, ) return None, response.status_code