From dcf2de0052f2c78a3bad84c067bf454206d396a3 Mon Sep 17 00:00:00 2001 From: "Abiba (pi)" Date: Sat, 27 Jun 2026 05:50:13 +0000 Subject: [PATCH] fix(zulip): detect BAD_EVENT_QUEUE_ID instead of silently swallowing it Root cause of Tanko not responding to DMs after initial connection: Zulip event queues expire after ~10min of dont_block polling. When the queue expired, _api_call() returned None for all 400-level errors, which _fetch_events() treated as 'no events' and returned []. The poll loop never knew the queue died, so it never reconnected. Fix: 1. _api_call() now returns (Optional[Dict], status_code) tuple 2. _fetch_events() explicitly checks for 400 status and raises RuntimeError('BAD_EVENT_QUEUE_ID') for the poll loop to catch 3. All callers updated to unpack the tuple Now when a queue expires, the poll loop catches the error and calls _reconnect() to register a fresh queue. --- plugins/platforms/zulip/adapter.py | 45 ++++++++++++++++++++---------- 1 file changed, 30 insertions(+), 15 deletions(-) diff --git a/plugins/platforms/zulip/adapter.py b/plugins/platforms/zulip/adapter.py index 51ebd60..13e50a6 100644 --- a/plugins/platforms/zulip/adapter.py +++ b/plugins/platforms/zulip/adapter.py @@ -242,7 +242,7 @@ class ZulipAdapter(BasePlatformAdapter): self._health_stats["started_at"] = datetime.now(timezone.utc).isoformat() # Register event queue - queue_resp = await self._api_call( + queue_resp, _ = await self._api_call( "POST", "/api/v1/register", data={ "event_types": '["message"]', @@ -381,11 +381,15 @@ class ZulipAdapter(BasePlatformAdapter): await asyncio.sleep(self._poll_interval) async def _fetch_events(self) -> List[Dict[str, Any]]: - """Fetch events from the Zulip event queue.""" + """Fetch events from the Zulip event queue. + + Raises RuntimeError with BAD_EVENT_QUEUE_ID when the queue + expires, so _poll_forever can reconnect. + """ if not self._queue_id: return [] - resp = await self._api_call( + resp, raw_status = await self._api_call( "GET", "/api/v1/events", params={ "queue_id": self._queue_id, @@ -393,6 +397,9 @@ class ZulipAdapter(BasePlatformAdapter): "dont_block": "true", }, ) + # Detect queue expiry from HTTP response + if raw_status == 400: + raise RuntimeError("BAD_EVENT_QUEUE_ID: queue expired") if not resp: return [] @@ -409,7 +416,7 @@ class ZulipAdapter(BasePlatformAdapter): """Re-register the event queue.""" self._queue_id = None try: - resp = await self._api_call( + resp, _ = await self._api_call( "POST", "/api/v1/register", data={ "event_types": '["message"]', @@ -506,7 +513,7 @@ class ZulipAdapter(BasePlatformAdapter): Falls back to the configured or default value if resolution fails. """ try: - resp = await self._api_call("GET", "/api/v1/users") + resp, _ = await self._api_call("GET", "/api/v1/users") if not resp: return members = resp.get("members", []) @@ -766,7 +773,7 @@ class ZulipAdapter(BasePlatformAdapter): "content": truncated, } try: - resp = await self._api_call("PATCH", "/api/v1/messages", data=payload) + resp, _ = await self._api_call("PATCH", "/api/v1/messages", data=payload) return resp is not None except Exception as e: logger.warning("[%s] Edit message %s error: %s", self.name, message_id, e) @@ -782,6 +789,7 @@ class ZulipAdapter(BasePlatformAdapter): "POST", "/api/v1/typing", data={"to": user_ids, "op": "start"}, ) + # typing indicator failure is non-critical except Exception: pass # Non-critical @@ -795,6 +803,7 @@ class ZulipAdapter(BasePlatformAdapter): "POST", "/api/v1/typing", data={"to": user_ids, "op": "stop"}, ) + # typing indicator failure is non-critical except Exception: pass # Non-critical @@ -1009,10 +1018,15 @@ class ZulipAdapter(BasePlatformAdapter): path: str, data: Optional[Dict] = None, params: Optional[Dict] = None, - ) -> Optional[Dict]: - """Make an API call to Zulip.""" + ) -> Tuple[Optional[Dict], int]: + """Make an API call to Zulip. + + Returns (response_json, status_code). status_code is useful for + callers that need to detect specific HTTP errors like 400 + (BAD_EVENT_QUEUE_ID). Returns (None, 0) on transport errors. + """ if not self._http_client: - return None + return None, 0 url = f"{self._site}{path}" headers = { @@ -1033,7 +1047,7 @@ class ZulipAdapter(BasePlatformAdapter): url, data=data, headers=headers, ) else: - return None + return None, 0 if response.status_code >= 400: logger.warning( @@ -1041,20 +1055,21 @@ class ZulipAdapter(BasePlatformAdapter): self.name, method, path, response.status_code, response.text[:200], ) - return None + return None, response.status_code - return response.json() + return response.json(), response.status_code except httpx.TimeoutException: logger.debug("[%s] Timeout on %s %s", self.name, method, path) - return None + return None, 0 except Exception as e: logger.warning("[%s] API error %s %s: %s", self.name, method, path, e) - return None + return None, 0 async def _send_api_call(self, payload: Dict) -> Optional[Dict]: """Send a message to Zulip.""" - return await self._api_call("POST", "/api/v1/messages", data=payload) + result, _ = await self._api_call("POST", "/api/v1/messages", data=payload) + return result # ------------------------------------------------------------------ # Utilities