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.
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user