feat(zulip): Gen 4 — connection pool recovery, heartbeat logging, log sanitization
CI / validate (pull_request) Failing after 6s
CI / validate (pull_request) Failing after 6s
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
This commit is contained in:
@@ -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
|
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.
|
are routed to the agent via the Hermes Gateway's handle_message() interface.
|
||||||
Replies use a placeholder->edit streaming pattern for UX feedback.
|
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):
|
Gen 3 Improvements (2026-06-26):
|
||||||
1. Malformed message resilience — try/except wraps each event, no crash on bad JSON
|
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
|
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
|
ALL_BOTS_REFRESH_INTERVAL = 3600 # Re-resolve @all-bots user ID every hour
|
||||||
SELFTEST_TIMEOUT = 30 # seconds to wait for self-test response
|
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
|
# Regex to strip Zulip @mention markup
|
||||||
MENTION_CLEANER = re.compile(r"@\*\*[^*]+\*\*")
|
MENTION_CLEANER = re.compile(r"@\*\*[^*]+\*\*")
|
||||||
|
|
||||||
@@ -158,7 +174,18 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
or os.getenv("ZULIP_POLL_INTERVAL", str(DEFAULT_POLL_INTERVAL))
|
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 ---
|
# --- 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._auth_header: str = _build_auth_header(self._email, self._api_key)
|
||||||
self._queue_id: Optional[str] = None
|
self._queue_id: Optional[str] = None
|
||||||
self._last_event_id: int = -1
|
self._last_event_id: int = -1
|
||||||
@@ -201,7 +228,7 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
|
|
||||||
# --- Gen 3: Health stats callback ---
|
# --- Gen 3: Health stats callback ---
|
||||||
self._health_callback = None
|
self._health_callback = None
|
||||||
self._health_callback_interval: int = 600 # 10 min
|
self._health_callback_interval: int = 300 # 5 min
|
||||||
|
|
||||||
@staticmethod
|
@staticmethod
|
||||||
def _resolve_int(val: Any, default: int) -> int:
|
def _resolve_int(val: Any, default: int) -> int:
|
||||||
@@ -238,9 +265,14 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
try:
|
try:
|
||||||
self._http_client = httpx.AsyncClient(timeout=30.0)
|
await self._create_http_client()
|
||||||
self._health_stats["started_at"] = datetime.now(timezone.utc).isoformat()
|
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
|
# Register event queue
|
||||||
queue_resp, _ = await self._api_call(
|
queue_resp, _ = await self._api_call(
|
||||||
"POST", "/api/v1/register",
|
"POST", "/api/v1/register",
|
||||||
@@ -342,9 +374,15 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
try:
|
try:
|
||||||
events = await self._fetch_events()
|
events = await self._fetch_events()
|
||||||
self._health_stats["poll_count"] += 1
|
self._health_stats["poll_count"] += 1
|
||||||
|
self._total_poll_count += 1
|
||||||
self._health_stats["last_poll_at"] = datetime.now(
|
self._health_stats["last_poll_at"] = datetime.now(
|
||||||
timezone.utc
|
timezone.utc
|
||||||
).isoformat()
|
).isoformat()
|
||||||
|
|
||||||
|
if events:
|
||||||
|
now = time.time()
|
||||||
|
self._consecutive_empty_polls = 0
|
||||||
|
self._last_event_received_time = now
|
||||||
for event in events:
|
for event in events:
|
||||||
try:
|
try:
|
||||||
await self._process_zulip_event(event)
|
await self._process_zulip_event(event)
|
||||||
@@ -360,6 +398,70 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
self._health_stats["malformed_events"] = (
|
self._health_stats["malformed_events"] = (
|
||||||
self._health_stats.get("malformed_events", 0) + 1
|
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] 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,
|
||||||
|
)
|
||||||
|
|
||||||
|
# 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
|
backoff_idx = 0
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
break
|
break
|
||||||
@@ -375,7 +477,7 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
err_str = str(e)
|
err_str = str(e)
|
||||||
if "BAD_EVENT_QUEUE_ID" in err_str or "queue_id" in err_str.lower():
|
if "BAD_EVENT_QUEUE_ID" in err_str or "queue_id" in err_str.lower():
|
||||||
logger.info("[%s] Queue expired, re-registering...", self.name)
|
logger.info("[%s] Queue expired, re-registering...", self.name)
|
||||||
await self._reconnect()
|
await self._reconnect_with_fresh_client()
|
||||||
backoff_idx = 0
|
backoff_idx = 0
|
||||||
continue
|
continue
|
||||||
logger.warning("[%s] Poll error: %s", self.name, e)
|
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"]
|
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:
|
async def _reconnect(self) -> None:
|
||||||
"""Re-register the event queue."""
|
"""Re-register the event queue."""
|
||||||
self._queue_id = None
|
self._queue_id = None
|
||||||
@@ -434,9 +558,21 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
self._last_event_id = resp.get("last_event_id", -1)
|
self._last_event_id = resp.get("last_event_id", -1)
|
||||||
self._health_stats["reconnects"] += 1
|
self._health_stats["reconnects"] += 1
|
||||||
logger.info("[%s] Reconnected, queue=%s", self.name, self._queue_id)
|
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:
|
except Exception as e:
|
||||||
logger.error("[%s] Reconnect failed: %s", self.name, 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
|
# Gen 2: Background dedup maintenance
|
||||||
# ------------------------------------------------------------------
|
# ------------------------------------------------------------------
|
||||||
@@ -1023,6 +1159,10 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
stats["all_bots_user_id"] = self._all_bots_user_id
|
stats["all_bots_user_id"] = self._all_bots_user_id
|
||||||
stats["connected"] = self._connected
|
stats["connected"] = self._connected
|
||||||
stats["checked_at"] = now.isoformat()
|
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
|
# Compute error rate
|
||||||
total_polls = stats.get("poll_count", 0) or 1
|
total_polls = stats.get("poll_count", 0) or 1
|
||||||
@@ -1074,10 +1214,12 @@ class ZulipAdapter(BasePlatformAdapter):
|
|||||||
return None, 0
|
return None, 0
|
||||||
|
|
||||||
if response.status_code >= 400:
|
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(
|
logger.warning(
|
||||||
"[%s] API %s %s: %d %s",
|
"[%s] API %s %s: %d %s",
|
||||||
self.name, method, path,
|
self.name, method, path,
|
||||||
response.status_code, response.text[:200],
|
response.status_code, body,
|
||||||
)
|
)
|
||||||
return None, response.status_code
|
return None, response.status_code
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user