ewilliams28/turnstone
1
0
Fork 0
mirror of https://github.com/turnstonelabs/turnstone.git synced 2026-08-24 21:04:48 -06:00
Code Issues Packages Projects Releases Wiki Activity

fix: harden Discord bot against gateway disconnects and SSE failures (#269)

Browse Source 
* fix: harden Discord bot against gateway disconnects and SSE failures

- Isolate Discord API failures from SSE stream — _on_ws_event exceptions
  no longer kill the SSE connection and cause missed events
- Fix broken exponential backoff on 4xx/5xx (delay was reset on every
  attempt); skip aiter_sse() on error responses
- Add read timeout (90s) to SSE httpx client so half-open TCP
  connections are detected and recovered
- Re-resolve node URL on each SSE reconnect attempt
- Add on_resumed handler to recover SSE tasks that died during brief
  gateway disconnects (on_ready is not called on session resume)
- Sync slash commands only on first on_ready to avoid Discord rate limits

* fix: SSE backoff on 4xx/5xx and retrieve dead task exceptions

- Replace `continue` with raise+catch so 4xx/5xx errors hit the
  exponential backoff path instead of tight-looping
- Retrieve task exceptions in _purge_dead_sse_tasks to suppress
  "Task exception was never retrieved" warnings and log the cause
This commit is contained in:
Patrick Buckley
2026-03-31 17:28:59 -07:00
committed by GitHub
parent fd47c23177
commit e17cbe35a5
1 changed files with 83 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files
turnstone/channels/discord/bot.py
+83 -17
View File
@@ -180,6 +180,7 @@ class TurnstoneBot:
server_token_factory=server_token_factory,
)
self._commands_synced: bool = False
self._subscribed_ws: set[str] = set()
self._sse_tasks: dict[str, asyncio.Task[None]] = {}
self._streaming: dict[str, StreamingMessage] = {}
@@ -204,12 +205,17 @@ class TurnstoneBot:
# response message can be re-tracked for multi-turn DM conversations.
self._notify_reply_channels: dict[str, tuple[discord.abc.Messageable, str]] = {}
# Shared HTTP client for SSE connections (long-lived, no timeout).
# Shared HTTP client for SSE connections.
# Read timeout detects half-open connections (server sends ping=5s
# keepalives, so 90s is very conservative).
# Token factory provides auto-rotating JWTs; static token is fallback.
headers: dict[str, str] = {}
if api_token and not server_token_factory:
headers["Authorization"] = f"Bearer {api_token}"
self._http_client = httpx.AsyncClient(headers=headers, timeout=None)
self._http_client = httpx.AsyncClient(
headers=headers,
timeout=httpx.Timeout(connect=10.0, read=90.0, write=10.0, pool=10.0),
)
intents = discord.Intents.default()
intents.message_content = True
@@ -230,6 +236,10 @@ class TurnstoneBot:
async def on_ready() -> None:
await self._on_ready()
@self._bot.event
async def on_resumed() -> None:
await self._on_resumed()
# -- lifecycle -----------------------------------------------------------
async def _setup_hook(self) -> None:
@@ -247,23 +257,57 @@ class TurnstoneBot:
log.info("discord.setup_hook_complete")
async def _on_ready(self) -> None:
"""Sync slash commands and recover existing routes."""
"""Sync slash commands (once) and recover existing routes."""
import discord
bot = self._bot
log.info("discord.ready", user=str(bot.user), guild_count=len(bot.guilds))
if self.config.guild_id:
guild = discord.Object(id=self.config.guild_id)
bot.tree.copy_global_to(guild=guild)
await bot.tree.sync(guild=guild)
log.info("discord.commands_synced", guild_id=self.config.guild_id)
else:
await bot.tree.sync()
log.info("discord.commands_synced_global")
if not self._commands_synced:
if self.config.guild_id:
guild = discord.Object(id=self.config.guild_id)
bot.tree.copy_global_to(guild=guild)
await bot.tree.sync(guild=guild)
log.info("discord.commands_synced", guild_id=self.config.guild_id)
else:
await bot.tree.sync()
log.info("discord.commands_synced_global")
self._commands_synced = True
self._purge_dead_sse_tasks("ready")
await self._recover_routes()
async def _on_resumed(self) -> None:
"""Recover dead SSE tasks after a gateway session resume.
Unlike ``on_ready``, ``on_resumed`` fires when discord.py resumes
an existing session after a brief disconnect — ``on_ready`` is NOT
called in that case. Any SSE listener tasks that died during the
blip need to be cleaned up and re-subscribed.
"""
self._purge_dead_sse_tasks("resumed")
await self._recover_routes()
def _purge_dead_sse_tasks(self, trigger: str) -> None:
"""Remove completed/failed SSE tasks so they can be re-subscribed."""
dead = [ws_id for ws_id, task in self._sse_tasks.items() if task.done()]
for ws_id in dead:
task = self._sse_tasks.pop(ws_id)
self._subscribed_ws.discard(ws_id)
# Retrieve exception to suppress "Task exception was never
# retrieved" warnings and log the underlying failure.
if not task.cancelled():
exc = task.exception()
if exc is not None:
log.warning(
"discord.sse_task_failed",
trigger=trigger,
ws_id=ws_id,
error=str(exc),
)
if dead:
log.info("discord.purged_dead_tasks", trigger=trigger, count=len(dead), ws_ids=dead)
async def _recover_routes(self) -> None:
"""Re-subscribe to event channels for existing discord routes.
@@ -361,14 +405,15 @@ class TurnstoneBot:
"""
import httpx_sse
# When routing through the console, connect SSE directly to the
# assigned server node (node_url from the create response).
node_base = await self.router.get_node_url(ws_id)
url = f"{node_base}/v1/api/events"
delay = _SSE_RECONNECT_DELAY
url = "" # set before loop so exception handlers can reference it
while True:
try:
# Re-resolve node URL on each attempt so reconnects pick up
# changes after bot restarts or router cache expiry.
node_base = await self.router.get_node_url(ws_id)
url = f"{node_base}/v1/api/events"
# Refresh auth header per-connection (token may have rotated)
sse_headers: dict[str, str] | None = None
if self._token_factory is not None:
@@ -392,7 +437,13 @@ class TurnstoneBot:
ws_id=ws_id,
status=status,
)
# Fall through to backoff/retry for transient errors.
# Don't try to parse a non-SSE error body —
# fall through to backoff/retry below.
raise httpx.HTTPStatusError(
f"SSE upstream {status}",
request=event_source.response.request,
response=event_source.response,
)
delay = _SSE_RECONNECT_DELAY # reset on successful connect
async for sse in event_source.aiter_sse():
if sse.event == "message" or not sse.event:
@@ -406,12 +457,27 @@ class TurnstoneBot:
)
continue
event = ServerEvent.from_dict(data)
await self._on_ws_event(ws_id, thread, event)
try:
await self._on_ws_event(ws_id, thread, event)
except Exception:
# Discord API failures (rate limits, outages)
# must not kill the SSE connection.
log.warning(
"discord.event_dispatch_failed",
ws_id=ws_id,
exc_info=True,
)
except httpx.HTTPStatusError:
pass # already logged above; fall through to backoff
except httpx.RemoteProtocolError:
# Server closed connection (normal on stream_end or shutdown).
log.debug("discord.sse_remote_closed", ws_id=ws_id)
except asyncio.CancelledError:
return # unsubscribe or shutdown
except httpx.ReadTimeout:
# No data received within read timeout — likely a half-open
# connection. Reconnect to recover.
log.info("discord.sse_read_timeout", ws_id=ws_id)
except (httpx.ConnectError, httpx.ConnectTimeout) as exc:
log.warning(
"discord.sse_connect_failed",